From 47c6c8d2fb7b354e95f67ce03af37c561b61c23f Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Tue, 1 Dec 2020 22:03:06 +0100 Subject: [PATCH 1/3] Re #8: Working on Netty http pipeline. --- .../tkowalcz/tjahzi/http/BootstrapUtil.java | 38 ++++++ .../tjahzi/http/HttpClientFactory.java | 2 +- .../tjahzi/http/HttpClientInitializer.java | 9 +- .../tkowalcz/tjahzi/http/HttpConnection.java | 5 + .../tkowalcz/tjahzi/http/NettyHttpClient.java | 111 ++++++++++-------- .../tkowalcz/tjahzi/http/ResponseHandler.java | 29 +++++ .../pl/tkowalcz/tjahzi/ReconnectTest.java | 95 +++++++++++++++ 7 files changed, 239 insertions(+), 50 deletions(-) create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java new file mode 100644 index 0000000..98c83c8 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java @@ -0,0 +1,38 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.bootstrap.Bootstrap; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.socket.nio.NioSocketChannel; + +public class BootstrapUtil { + + public static ChannelFuture initConnection( + EventLoopGroup group, + ClientConfiguration clientConfiguration + ) { + Bootstrap bootstrap = new Bootstrap(); + ResponseHandler responseHandler = new ResponseHandler(); + + ChannelFuture channelFuture = bootstrap.group(group) + .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) + .option(ChannelOption.SO_KEEPALIVE, true) + .option(ChannelOption.TCP_NODELAY, false) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfiguration.getConnectionTimeoutMillis()) + .channel(NioSocketChannel.class) + .handler( + new HttpClientInitializer( + responseHandler, + clientConfiguration.getRequestTimeoutMillis() + ) + ) + .remoteAddress( + clientConfiguration.getHost(), + clientConfiguration.getPort() + ).connect(); + + return channelFuture; + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java index bb4b57a..062639f 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientFactory.java @@ -6,7 +6,7 @@ public class HttpClientFactory { public NettyHttpClient getHttpClient( ClientConfiguration clientConfiguration, - String[] additionalHeaders) { + String... additionalHeaders) { return new NettyHttpClient( clientConfiguration, additionalHeaders diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java index 4a2d2ee..c8d447a 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpClientInitializer.java @@ -10,9 +10,13 @@ import io.netty.handler.timeout.IdleStateHandler; class HttpClientInitializer extends ChannelInitializer { + private static final int BYTES_IN_MEGABYTE = 1024 * 1024; + + private final ResponseHandler responseHandler; private final int requestTimeoutMillis; - HttpClientInitializer(int requestTimeoutMillis) { + HttpClientInitializer(ResponseHandler responseHandler, int requestTimeoutMillis) { + this.responseHandler = responseHandler; this.requestTimeoutMillis = requestTimeoutMillis; } @@ -21,7 +25,8 @@ class HttpClientInitializer extends ChannelInitializer { ChannelPipeline p = ch.pipeline(); p.addLast(new IdleStateHandler(requestTimeoutMillis, 0, 60)); p.addLast(new HttpClientCodec()); - p.addLast(new HttpObjectAggregator(1048576)); + p.addLast(new HttpObjectAggregator(BYTES_IN_MEGABYTE)); p.addLast(new HttpContentDecompressor()); + p.addLast(responseHandler); } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java new file mode 100644 index 0000000..d3ad5d2 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java @@ -0,0 +1,5 @@ +package pl.tkowalcz.tjahzi.http; + +public class HttpConnection { + +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java index 39c4ee5..da63c89 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java @@ -1,82 +1,99 @@ package pl.tkowalcz.tjahzi.http; -import io.netty.bootstrap.Bootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; -import io.netty.channel.Channel; -import io.netty.channel.ChannelOption; +import io.netty.channel.ChannelFuture; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.compression.Snappy; import io.netty.handler.codec.http.*; import java.io.Closeable; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; public class NettyHttpClient implements Closeable { public static final String PROTOBUF_MIME_TYPE = "application/x-protobuf"; - private final EventLoopGroup group; - private final String logEndpoint; - - private final Channel channel; - private final String host; + private final ClientConfiguration clientConfiguration; private final HttpHeaders headers; private final Snappy snappy = new Snappy(); + private final EventLoopGroup group; + private volatile ChannelFuture lokiConnection; +// private HttpConnection lokiConnection; + public NettyHttpClient( ClientConfiguration clientConfiguration, String[] additionalHeaders) { - host = clientConfiguration.getHost(); - logEndpoint = clientConfiguration.getLogEndpoint(); - - ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client"); - ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker"); - group = new NioEventLoopGroup(1, threadFactory); - - Bootstrap bootstrap = new Bootstrap(); - channel = bootstrap.group(group) - .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) - .option(ChannelOption.SO_KEEPALIVE, true) - .option(ChannelOption.TCP_NODELAY, false) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfiguration.getConnectionTimeoutMillis()) - .channel(NioSocketChannel.class) - .handler(new HttpClientInitializer(clientConfiguration.getRequestTimeoutMillis())) - .remoteAddress( - clientConfiguration.getHost(), - clientConfiguration.getPort() - ) - .connect() - .syncUninterruptibly() - .channel(); - - headers = new ReadOnlyHttpHeaders( + this.clientConfiguration = clientConfiguration; + this.headers = new ReadOnlyHttpHeaders( true, additionalHeaders ); + + ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client"); + ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker"); + NioEventLoopGroup group = new NioEventLoopGroup(1, threadFactory); + this.group = group; + + foobar(clientConfiguration); + } + + private void foobar(ClientConfiguration clientConfiguration) { + lokiConnection = BootstrapUtil.initConnection( + group, + clientConfiguration + ); + + lokiConnection.addListener(future -> { + System.out.println("NettyHttpClient.foobar"); + if (!future.isSuccess()) { + group.schedule(() -> { + foobar(clientConfiguration); + }, + 1000, TimeUnit.MILLISECONDS); + // TODO: increase error counters + + } + }); } public void log(ByteBuf dataBuffer) { - ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer(); - snappy.encode(dataBuffer, output, dataBuffer.readableBytes()); + log(dataBuffer, clientConfiguration.getMaxRetries()); + } - FullHttpRequest request = new DefaultFullHttpRequest( - HttpVersion.HTTP_1_1, - HttpMethod.POST, - logEndpoint, - output - ); + public void log(ByteBuf dataBuffer, int retries) { + lokiConnection.awaitUninterruptibly(); + if (lokiConnection.isSuccess()) { + ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer(); + snappy.encode(dataBuffer, output, dataBuffer.readableBytes()); - request.headers() - .add(headers) - .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE) - .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes()) - .set(HttpHeaderNames.HOST, host); + FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, + HttpMethod.POST, + clientConfiguration.getLogEndpoint(), + output + ); - channel.writeAndFlush(request); + request.headers() + .add(headers) + .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE) + .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes()) + .set(HttpHeaderNames.HOST, clientConfiguration.getHost()); + + lokiConnection.channel().writeAndFlush(request); + } else { + // TODO: increase error counters + + if (retries > 0) { + log(dataBuffer, retries - 1); + } + + // Dropping message on the floor + } } @Override diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java new file mode 100644 index 0000000..5a2e355 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ResponseHandler.java @@ -0,0 +1,29 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http.FullHttpResponse; + +@ChannelHandler.Sharable +class ResponseHandler extends SimpleChannelInboundHandler { + + ResponseHandler() { + super(false); + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) { + System.out.println("msg = " + msg); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + System.out.println("cause = " + cause); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + System.out.println("ResponseHandler.channelInactive"); + } +} diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java new file mode 100644 index 0000000..b470934 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java @@ -0,0 +1,95 @@ +package pl.tkowalcz.tjahzi; + +import com.github.tomakehurst.wiremock.WireMockServer; +import org.awaitility.Durations; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import pl.tkowalcz.tjahzi.http.ClientConfiguration; +import pl.tkowalcz.tjahzi.http.HttpClientFactory; +import pl.tkowalcz.tjahzi.http.NettyHttpClient; + +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import static com.github.tomakehurst.wiremock.client.WireMock.*; +import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig; +import static org.awaitility.Awaitility.await; + +public class ReconnectTest { + + private WireMockServer wireMockServer; + + private TjahziInitializer initializer; + private LoggingSystem loggingSystem; + + @BeforeEach + void setUp() { + wireMockServer = new WireMockServer( + wireMockConfig() + .dynamicPort() + .dynamicHttpsPort() + ); + +// wireMockServer.start(); + + initializer = new TjahziInitializer(); + } + + @AfterEach + void tearDown() { + wireMockServer.stop(); + + if (loggingSystem != null) { + loggingSystem.close( + (int) TimeUnit.SECONDS.toMillis(10), + System.out::println + ); + + loggingSystem = null; + } + } + + @Test + void shouldIncludeAdditionalHeaders() { + // Given +// wireMockServer.stubFor( +// get(urlEqualTo("/loki/api/v1/push")) +// .willReturn( +// aResponse().withStatus(200) +// )); + + ClientConfiguration clientConfiguration = ClientConfiguration.builder() + .withConnectionTimeoutMillis(10_000) + .withHost("localhost") + .withPort(12322) + .withMaxRetries(1) + .build(); + + NettyHttpClient httpClient = HttpClientFactory.defaultFactory() + .getHttpClient(clientConfiguration); + + loggingSystem = initializer.createLoggingSystem( + httpClient, + 1024 * 1024, + false + ); + + // When + TjahziLogger logger = loggingSystem.createLogger(); + logger.log( + System.currentTimeMillis(), + Map.of(), + "Test" + ); + + // Then + await() + .atMost(Durations.TEN_MINUTES) + .untilAsserted(() -> + wireMockServer.verify( + postRequestedFor(urlMatching("/loki/api/v1/push")) + .withHeader("X-Scope-OrgID", equalTo("Circus")) + )); + } +} From 59a1b1628c26f21a1dcc84befa0f5f51a8f51b21 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Fri, 4 Dec 2020 23:36:29 +0100 Subject: [PATCH 2/3] Re #8: Refactored connection code to separate class. Added retries. --- .../tkowalcz/tjahzi/http/BlockingRetry.java | 31 +++++++ .../tkowalcz/tjahzi/http/BootstrapUtil.java | 4 +- .../tjahzi/http/EventLoopGroupRetry.java | 32 ++++++++ .../http/ExponentialBackoffStrategy.java | 38 +++++++++ .../tkowalcz/tjahzi/http/HttpConnection.java | 68 +++++++++++++++- .../tkowalcz/tjahzi/http/NettyHttpClient.java | 81 +++++-------------- .../java/pl/tkowalcz/tjahzi/http/Retry.java | 6 ++ .../java/pl/tkowalcz/tjahzi/HeadersTest.java | 17 ++-- .../pl/tkowalcz/tjahzi/ReconnectTest.java | 23 +++--- .../tjahzi/ResourcesCleanupOnCloseTest.java | 6 +- .../http/ExponentialBackoffStrategyTest.java | 74 +++++++++++++++++ 11 files changed, 286 insertions(+), 94 deletions(-) create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategy.java create mode 100644 tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/Retry.java create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategyTest.java diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java new file mode 100644 index 0000000..46bc47c --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java @@ -0,0 +1,31 @@ +package pl.tkowalcz.tjahzi.http; + +import java.util.function.Consumer; + +class BlockingRetry implements Retry { + + private final ExponentialBackoffStrategy strategy = ExponentialBackoffStrategy.withDefault(); + private final Consumer operation; + + private int retries; + + BlockingRetry(Consumer operation, int retries) { + this.operation = operation; + this.retries = retries; + } + + @Override + public void retry() { + if (retries > 0) { + retries--; + + try { + Thread.sleep(strategy.getAsLong(), 0); + } catch (InterruptedException ignore) { + return; + } + + operation.accept(this); + } + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java index 98c83c8..c750967 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java @@ -16,7 +16,7 @@ public class BootstrapUtil { Bootstrap bootstrap = new Bootstrap(); ResponseHandler responseHandler = new ResponseHandler(); - ChannelFuture channelFuture = bootstrap.group(group) + return bootstrap.group(group) .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) .option(ChannelOption.SO_KEEPALIVE, true) .option(ChannelOption.TCP_NODELAY, false) @@ -32,7 +32,5 @@ public class BootstrapUtil { clientConfiguration.getHost(), clientConfiguration.getPort() ).connect(); - - return channelFuture; } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java new file mode 100644 index 0000000..5533955 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java @@ -0,0 +1,32 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.channel.nio.NioEventLoopGroup; + +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; + +class EventLoopGroupRetry implements Retry, Runnable { + + private final NioEventLoopGroup group; + private final Consumer operation; + private final ExponentialBackoffStrategy strategy = ExponentialBackoffStrategy.withDefault(); + + public EventLoopGroupRetry(NioEventLoopGroup group, Consumer operation) { + this.group = group; + this.operation = operation; + } + + @Override + public void run() { + operation.accept(this); + } + + @Override + public void retry() { + group.schedule( + this, + strategy.getAsLong(), + TimeUnit.MILLISECONDS + ); + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategy.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategy.java new file mode 100644 index 0000000..4a46091 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategy.java @@ -0,0 +1,38 @@ +package pl.tkowalcz.tjahzi.http; + +import java.util.function.LongSupplier; + +public class ExponentialBackoffStrategy implements LongSupplier { + + private final long maximumBackoffMillis; + private final double multiplier; + + private long nextBackoff; + + public ExponentialBackoffStrategy( + long initialBackoffMillis, + long maximumBackoffMillis, + double multiplier + ) { + this.maximumBackoffMillis = maximumBackoffMillis; + this.multiplier = multiplier; + + this.nextBackoff = initialBackoffMillis; + } + + @Override + public long getAsLong() { + long result = nextBackoff; + + nextBackoff = (long) Math.min(nextBackoff * multiplier, maximumBackoffMillis); + return result; + } + + public static ExponentialBackoffStrategy withDefault() { + return new ExponentialBackoffStrategy( + 250, + 30_000, + 3 + ); + } +} diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java index d3ad5d2..7b86930 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java @@ -1,5 +1,71 @@ package pl.tkowalcz.tjahzi.http; -public class HttpConnection { +import io.netty.channel.ChannelFuture; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.handler.codec.http.FullHttpRequest; +import java.io.Closeable; +import java.util.concurrent.ThreadFactory; + +public class HttpConnection implements Closeable { + + private final ClientConfiguration clientConfiguration; + private final NioEventLoopGroup group; + + private volatile ChannelFuture lokiConnection; + + public HttpConnection(ClientConfiguration clientConfiguration) { + this.clientConfiguration = clientConfiguration; + + ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client"); + ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker"); + + this.group = new NioEventLoopGroup(1, threadFactory); + + EventLoopGroupRetry retry = new EventLoopGroupRetry(group, this::recreateConnection); + recreateConnection(retry); + } + + private void recreateConnection(Retry retry) { + lokiConnection = BootstrapUtil.initConnection( + group, + clientConfiguration + ); + + lokiConnection.addListener((ChannelFuture future) -> { + if (future.isSuccess()) { + future + .channel() + .closeFuture() + .addListener(__ -> retry.retry()); + } else { + retry.retry(); + } + }); + } + + public void execute(FullHttpRequest request) { + BlockingRetry blockingRetry = new BlockingRetry( + retry -> execute(request, retry), + clientConfiguration.getMaxRetries() + ); + + execute(request, blockingRetry); + } + + private void execute(FullHttpRequest request, Retry retry) { + lokiConnection.awaitUninterruptibly(); + if (lokiConnection.isSuccess()) { + lokiConnection.channel().writeAndFlush(request); + } else { + // TODO: increase error counters + retry.retry(); + // Dropping message on the floor + } + } + + @Override + public void close() { + group.shutdownGracefully(); + } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java index da63c89..a808438 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java @@ -2,15 +2,10 @@ package pl.tkowalcz.tjahzi.http; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; -import io.netty.channel.ChannelFuture; -import io.netty.channel.EventLoopGroup; -import io.netty.channel.nio.NioEventLoopGroup; import io.netty.handler.codec.compression.Snappy; import io.netty.handler.codec.http.*; import java.io.Closeable; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; public class NettyHttpClient implements Closeable { @@ -21,9 +16,8 @@ public class NettyHttpClient implements Closeable { private final Snappy snappy = new Snappy(); - private final EventLoopGroup group; - private volatile ChannelFuture lokiConnection; -// private HttpConnection lokiConnection; + // private volatile ChannelFuture lokiConnection; + private HttpConnection lokiConnection; public NettyHttpClient( ClientConfiguration clientConfiguration, @@ -34,70 +28,31 @@ public class NettyHttpClient implements Closeable { additionalHeaders ); - ThreadGroup threadGroup = new ThreadGroup("Tjahzi Loki client"); - ThreadFactory threadFactory = r -> new Thread(threadGroup, r, "tjahzi-worker"); - NioEventLoopGroup group = new NioEventLoopGroup(1, threadFactory); - this.group = group; - - foobar(clientConfiguration); - } - - private void foobar(ClientConfiguration clientConfiguration) { - lokiConnection = BootstrapUtil.initConnection( - group, - clientConfiguration - ); - - lokiConnection.addListener(future -> { - System.out.println("NettyHttpClient.foobar"); - if (!future.isSuccess()) { - group.schedule(() -> { - foobar(clientConfiguration); - }, - 1000, TimeUnit.MILLISECONDS); - // TODO: increase error counters - - } - }); + lokiConnection = new HttpConnection(clientConfiguration); } public void log(ByteBuf dataBuffer) { - log(dataBuffer, clientConfiguration.getMaxRetries()); - } + ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer(); + snappy.encode(dataBuffer, output, dataBuffer.readableBytes()); - public void log(ByteBuf dataBuffer, int retries) { - lokiConnection.awaitUninterruptibly(); - if (lokiConnection.isSuccess()) { - ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer(); - snappy.encode(dataBuffer, output, dataBuffer.readableBytes()); + FullHttpRequest request = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, + HttpMethod.POST, + clientConfiguration.getLogEndpoint(), + output + ); - FullHttpRequest request = new DefaultFullHttpRequest( - HttpVersion.HTTP_1_1, - HttpMethod.POST, - clientConfiguration.getLogEndpoint(), - output - ); + request.headers() + .add(headers) + .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE) + .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes()) + .set(HttpHeaderNames.HOST, clientConfiguration.getHost()); - request.headers() - .add(headers) - .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE) - .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes()) - .set(HttpHeaderNames.HOST, clientConfiguration.getHost()); - - lokiConnection.channel().writeAndFlush(request); - } else { - // TODO: increase error counters - - if (retries > 0) { - log(dataBuffer, retries - 1); - } - - // Dropping message on the floor - } + lokiConnection.execute(request); } @Override public void close() { - group.shutdownGracefully(); + lokiConnection.close(); } } diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/Retry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/Retry.java new file mode 100644 index 0000000..5b53020 --- /dev/null +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/Retry.java @@ -0,0 +1,6 @@ +package pl.tkowalcz.tjahzi.http; + +public interface Retry { + + void retry(); +} diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java index 1c7ed40..800cd1d 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java @@ -69,10 +69,8 @@ class HeadersTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( clientConfiguration, - new String[]{ - "X-Scope-OrgID", "Circus", - "C", "Control" - } + "X-Scope-OrgID", "Circus", + "C", "Control" ); loggingSystem = initializer.createLoggingSystem( @@ -117,8 +115,7 @@ class HeadersTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( - clientConfiguration, - new String[0] + clientConfiguration ); loggingSystem = initializer.createLoggingSystem( @@ -163,11 +160,9 @@ class HeadersTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( clientConfiguration, - new String[]{ - "content-type", "text/plain", - "content-length", "5232423423", - "host", "remote" - } + "content-type", "text/plain", + "content-length", "5232423423", + "host", "remote" ); loggingSystem = initializer.createLoggingSystem( diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java index b470934..00ef198 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java @@ -27,11 +27,14 @@ public class ReconnectTest { void setUp() { wireMockServer = new WireMockServer( wireMockConfig() - .dynamicPort() - .dynamicHttpsPort() + .port(12322) ); -// wireMockServer.start(); + wireMockServer.stubFor( + get(urlEqualTo("/loki/api/v1/push")) + .willReturn( + aResponse().withStatus(200) + )); initializer = new TjahziInitializer(); } @@ -53,17 +56,11 @@ public class ReconnectTest { @Test void shouldIncludeAdditionalHeaders() { // Given -// wireMockServer.stubFor( -// get(urlEqualTo("/loki/api/v1/push")) -// .willReturn( -// aResponse().withStatus(200) -// )); - ClientConfiguration clientConfiguration = ClientConfiguration.builder() .withConnectionTimeoutMillis(10_000) .withHost("localhost") .withPort(12322) - .withMaxRetries(1) + .withMaxRetries(10) .build(); NettyHttpClient httpClient = HttpClientFactory.defaultFactory() @@ -83,13 +80,15 @@ public class ReconnectTest { "Test" ); + wireMockServer.start(); + // Then await() .atMost(Durations.TEN_MINUTES) .untilAsserted(() -> wireMockServer.verify( postRequestedFor(urlMatching("/loki/api/v1/push")) - .withHeader("X-Scope-OrgID", equalTo("Circus")) - )); + ) + ); } } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java index 53a8932..aadf88b 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ResourcesCleanupOnCloseTest.java @@ -54,10 +54,8 @@ class ResourcesCleanupOnCloseTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( clientConfiguration, - new String[]{ - "X-Scope-OrgID", "Circus", - "C", "Control" - } + "X-Scope-OrgID", "Circus", + "C", "Control" ); LoggingSystem loggingSystem = initializer.createLoggingSystem( diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategyTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategyTest.java new file mode 100644 index 0000000..1245681 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/ExponentialBackoffStrategyTest.java @@ -0,0 +1,74 @@ +package pl.tkowalcz.tjahzi.http; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; + +import static org.assertj.core.api.Assertions.assertThat; + +class ExponentialBackoffStrategyTest { + + @Test + void shouldReturnInitialBackoffOnFirstTry() { + // Given + int expected = 1000; + + ExponentialBackoffStrategy backoff = new ExponentialBackoffStrategy( + expected, + 10_000, + 2 + ); + + // When + long actual = backoff.getAsLong(); + + // Then + assertThat(actual).isEqualTo(expected); + } + + @Test + void shouldNotGoOverMaximumBackoff() { + // Given + int expected = 10_000; + + ExponentialBackoffStrategy backoff = new ExponentialBackoffStrategy( + 4000, + expected, + 2 + ); + + // When + for (int i = 0; i < 1000; i++) { + backoff.getAsLong(); + } + + long actual = backoff.getAsLong(); + + // Then + assertThat(actual).isEqualTo(expected); + } + + @Test + void shouldIncreaseBackoffExponentially() { + // Given + ExponentialBackoffStrategy backoff = new ExponentialBackoffStrategy( + 250, + 10_000, + 3 + ); + + ArrayList actual = new ArrayList<>(); + + // When + actual.add(backoff.getAsLong()); + actual.add(backoff.getAsLong()); + actual.add(backoff.getAsLong()); + actual.add(backoff.getAsLong()); + actual.add(backoff.getAsLong()); + + // Then + assertThat(actual).containsExactly( + 250L, 750L, 2250L, 6750L, 10_000L + ); + } +} From 961ace842f2a697f39f59dc2556182f5cfafcf09 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Tue, 8 Dec 2020 08:45:06 +0100 Subject: [PATCH 3/3] Re #8: Added tests and check if channel().isActive() before using it. --- tjahzi-core/pom.xml | 8 ++ .../tkowalcz/tjahzi/http/BlockingRetry.java | 17 ++++- .../tjahzi/http/EventLoopGroupRetry.java | 22 ++++-- .../tkowalcz/tjahzi/http/HttpConnection.java | 8 +- .../tkowalcz/tjahzi/http/NettyHttpClient.java | 4 +- .../java/pl/tkowalcz/tjahzi/HeadersTest.java | 4 +- .../pl/tkowalcz/tjahzi/LoggingSystemTest.java | 3 +- .../pl/tkowalcz/tjahzi/ReconnectTest.java | 64 +++++++++++++++- .../tjahzi/http/BlockingRetryTest.java | 51 +++++++++++++ .../tjahzi/http/EventLoopGroupRetryTest.java | 73 +++++++++++++++++++ 10 files changed, 235 insertions(+), 19 deletions(-) create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java create mode 100644 tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java diff --git a/tjahzi-core/pom.xml b/tjahzi-core/pom.xml index 01d885c..0050a1a 100644 --- a/tjahzi-core/pom.xml +++ b/tjahzi-core/pom.xml @@ -52,6 +52,7 @@ org.assertj assertj-core 3.15.0 + test @@ -61,6 +62,13 @@ test + + org.apache.logging.log4j + log4j-slf4j-impl + 2.13.3 + test + + org.junit.jupiter diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java index 46bc47c..72f2ef8 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java @@ -4,16 +4,29 @@ import java.util.function.Consumer; class BlockingRetry implements Retry { - private final ExponentialBackoffStrategy strategy = ExponentialBackoffStrategy.withDefault(); + private final ExponentialBackoffStrategy strategy; private final Consumer operation; private int retries; - BlockingRetry(Consumer operation, int retries) { + // VisibleForTesting + BlockingRetry( + Consumer operation, + ExponentialBackoffStrategy strategy, + int retries) { this.operation = operation; + this.strategy = strategy; this.retries = retries; } + BlockingRetry(Consumer operation, int retries) { + this( + operation, + ExponentialBackoffStrategy.withDefault(), + retries + ); + } + @Override public void retry() { if (retries > 0) { diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java index 5533955..37707d9 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java @@ -1,19 +1,31 @@ package pl.tkowalcz.tjahzi.http; -import io.netty.channel.nio.NioEventLoopGroup; - +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; class EventLoopGroupRetry implements Retry, Runnable { - private final NioEventLoopGroup group; + private final ScheduledExecutorService group; private final Consumer operation; - private final ExponentialBackoffStrategy strategy = ExponentialBackoffStrategy.withDefault(); + private final ExponentialBackoffStrategy strategy; - public EventLoopGroupRetry(NioEventLoopGroup group, Consumer operation) { + // VisibleForTesting + public EventLoopGroupRetry( + ScheduledExecutorService group, + Consumer operation, + ExponentialBackoffStrategy strategy) { this.group = group; this.operation = operation; + this.strategy = strategy; + } + + public EventLoopGroupRetry(ScheduledExecutorService group, Consumer operation) { + this( + group, + operation, + ExponentialBackoffStrategy.withDefault() + ); } @Override diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java index 7b86930..e6d3b2a 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java @@ -54,9 +54,11 @@ public class HttpConnection implements Closeable { } private void execute(FullHttpRequest request, Retry retry) { - lokiConnection.awaitUninterruptibly(); - if (lokiConnection.isSuccess()) { - lokiConnection.channel().writeAndFlush(request); + ChannelFuture stableReference = this.lokiConnection; + + stableReference.awaitUninterruptibly(); + if (stableReference.isSuccess() && stableReference.channel().isActive()) { + stableReference.channel().writeAndFlush(request); } else { // TODO: increase error counters retry.retry(); diff --git a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java index a808438..27282cc 100644 --- a/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java +++ b/tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java @@ -15,9 +15,7 @@ public class NettyHttpClient implements Closeable { private final HttpHeaders headers; private final Snappy snappy = new Snappy(); - - // private volatile ChannelFuture lokiConnection; - private HttpConnection lokiConnection; + private final HttpConnection lokiConnection; public NettyHttpClient( ClientConfiguration clientConfiguration, diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java index 800cd1d..b165b48 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/HeadersTest.java @@ -54,7 +54,7 @@ class HeadersTest { void shouldIncludeAdditionalHeaders() { // Given wireMockServer.stubFor( - get(urlEqualTo("/loki/api/v1/push")) + post(urlEqualTo("/loki/api/v1/push")) .willReturn( aResponse().withStatus(200) )); @@ -101,7 +101,7 @@ class HeadersTest { void shouldHandleCaseWithNoAdditionalHeaders() { // Given wireMockServer.stubFor( - get(urlEqualTo("/loki/api/v1/push")) + post(urlEqualTo("/loki/api/v1/push")) .willReturn( aResponse().withStatus(200) )); diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java index b0f53c7..17e813c 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/LoggingSystemTest.java @@ -55,8 +55,7 @@ class LoggingSystemTest { NettyHttpClient httpClient = HttpClientFactory.defaultFactory() .getHttpClient( - clientConfiguration, - new String[0] + clientConfiguration ); TjahziInitializer initializer = new TjahziInitializer(); diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java index 00ef198..73f51ab 100644 --- a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java @@ -31,7 +31,7 @@ public class ReconnectTest { ); wireMockServer.stubFor( - get(urlEqualTo("/loki/api/v1/push")) + post(urlEqualTo("/loki/api/v1/push")) .willReturn( aResponse().withStatus(200) )); @@ -54,7 +54,7 @@ public class ReconnectTest { } @Test - void shouldIncludeAdditionalHeaders() { + void shouldEventuallyReconnectIfLokiWasDownWhenStarting() { // Given ClientConfiguration clientConfiguration = ClientConfiguration.builder() .withConnectionTimeoutMillis(10_000) @@ -91,4 +91,64 @@ public class ReconnectTest { ) ); } + + @Test + void shouldReconnectIfLokiFailed() { + // Given + ClientConfiguration clientConfiguration = ClientConfiguration.builder() + .withConnectionTimeoutMillis(10_000) + .withHost("localhost") + .withPort(12322) + .withMaxRetries(10) + .build(); + + NettyHttpClient httpClient = HttpClientFactory.defaultFactory() + .getHttpClient(clientConfiguration); + + wireMockServer.start(); + + loggingSystem = initializer.createLoggingSystem( + httpClient, + 1024 * 1024, + false + ); + + TjahziLogger logger = loggingSystem.createLogger(); + logger.log( + System.currentTimeMillis(), + Map.of(), + "Test" + ); + + await() + .atMost(Durations.FIVE_SECONDS) + .untilAsserted(() -> + wireMockServer.verify( + 1, + postRequestedFor(urlMatching("/loki/api/v1/push")) + ) + ); + + wireMockServer.stop(); + await().until(() -> !wireMockServer.isRunning()); + + // When + logger.log( + System.currentTimeMillis(), + Map.of(), + "Test" + ); + + wireMockServer.start(); + + // Then + await() + .atMost(Durations.TEN_MINUTES) + .untilAsserted(() -> + wireMockServer.verify( + 1, + postRequestedFor(urlMatching("/loki/api/v1/push")) + ) + ); + } } diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java new file mode 100644 index 0000000..7d497e9 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/BlockingRetryTest.java @@ -0,0 +1,51 @@ +package pl.tkowalcz.tjahzi.http; + +import org.junit.jupiter.api.Test; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; + +class BlockingRetryTest { + + @Test + void shouldRetryAndInvokeOperation() { + // Given + AtomicInteger retryCounter = new AtomicInteger(); + + BlockingRetry retry = new BlockingRetry( + __ -> { + retryCounter.incrementAndGet(); + __.retry(); + }, + new ExponentialBackoffStrategy(1, 1, 1), + 5 + ); + + // When + retry.retry(); + + // Then + assertThat(retryCounter).hasValue(5); + } + + @Test + void shouldNotRetryOnItsOwn() { + // Given + AtomicInteger retryCounter = new AtomicInteger(); + + BlockingRetry retry = new BlockingRetry( + __ -> { + retryCounter.incrementAndGet(); + }, + new ExponentialBackoffStrategy(1, 1, 1), + 5 + ); + + // When + retry.retry(); + + // Then + assertThat(retryCounter).hasValue(1); + } +} diff --git a/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java new file mode 100644 index 0000000..1f72760 --- /dev/null +++ b/tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetryTest.java @@ -0,0 +1,73 @@ +package pl.tkowalcz.tjahzi.http; + +import io.netty.channel.nio.NioEventLoopGroup; +import org.awaitility.Durations; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +class EventLoopGroupRetryTest { + + private NioEventLoopGroup eventLoopGroup; + + @BeforeEach + void setUp() { + eventLoopGroup = new NioEventLoopGroup(Executors.defaultThreadFactory()); + } + + @AfterEach + void tearDown() { + eventLoopGroup.shutdownGracefully(); + } + + @Test + void shouldRetryAndInvokeOperation() { + // Given + AtomicInteger retryCounter = new AtomicInteger(); + + EventLoopGroupRetry retry = new EventLoopGroupRetry( + eventLoopGroup, + __ -> { + retryCounter.incrementAndGet(); + __.retry(); + }, + new ExponentialBackoffStrategy(250, 1000, 2) + ); + + // When + retry.retry(); + + // Then + await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> { + assertThat(retryCounter).hasValueGreaterThan(3); + }); + } + + @Test + void shouldNotRetryOnItsOwn() { + // Given + AtomicInteger retryCounter = new AtomicInteger(); + + EventLoopGroupRetry retry = new EventLoopGroupRetry( + eventLoopGroup, + __ -> { + retryCounter.incrementAndGet(); + }, + new ExponentialBackoffStrategy(1, 1, 1) + ); + + // When + retry.retry(); + + // Then + await().atMost(Durations.TEN_SECONDS).untilAsserted(() -> { + assertThat(retryCounter).hasValue(1); + }); + } +}