Merge pull request #11 from tkowalcz/8-complete-netty-pipeline

8 complete netty pipeline

tjahzi-core/pom.xml

@@ -52,6 +52,7 @@
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>
             <version>3.15.0</version>
+            <scope>test</scope>
         </dependency>
 
         <dependency>
@@ -61,6 +62,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-slf4j-impl</artifactId>
+            <version>2.13.3</version>
+            <scope>test</scope>
+        </dependency>
+
         <!-- Testcontainers -->
         <dependency>
             <groupId>org.junit.jupiter</groupId>

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BlockingRetry.java

@@ -0,0 +1,44 @@
+package pl.tkowalcz.tjahzi.http;
+
+import java.util.function.Consumer;
+
+class BlockingRetry implements Retry {
+
+    private final ExponentialBackoffStrategy strategy;
+    private final Consumer<BlockingRetry> operation;
+
+    private int retries;
+
+    // VisibleForTesting
+    BlockingRetry(
+            Consumer<BlockingRetry> operation,
+            ExponentialBackoffStrategy strategy,
+            int retries) {
+        this.operation = operation;
+        this.strategy = strategy;
+        this.retries = retries;
+    }
+
+    BlockingRetry(Consumer<BlockingRetry> operation, int retries) {
+        this(
+                operation,
+                ExponentialBackoffStrategy.withDefault(),
+                retries
+        );
+    }
+
+    @Override
+    public void retry() {
+        if (retries > 0) {
+            retries--;
+
+            try {
+                Thread.sleep(strategy.getAsLong(), 0);
+            } catch (InterruptedException ignore) {
+                return;
+            }
+
+            operation.accept(this);
+        }
+    }
+}

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/BootstrapUtil.java

@@ -0,0 +1,36 @@
+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();
+
+        return 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();
+    }
+}

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/EventLoopGroupRetry.java

@@ -0,0 +1,44 @@
+package pl.tkowalcz.tjahzi.http;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+class EventLoopGroupRetry implements Retry, Runnable {
+
+    private final ScheduledExecutorService group;
+    private final Consumer<EventLoopGroupRetry> operation;
+    private final ExponentialBackoffStrategy strategy;
+
+    // VisibleForTesting
+    public EventLoopGroupRetry(
+            ScheduledExecutorService group,
+            Consumer<EventLoopGroupRetry> operation,
+            ExponentialBackoffStrategy strategy) {
+        this.group = group;
+        this.operation = operation;
+        this.strategy = strategy;
+    }
+
+    public EventLoopGroupRetry(ScheduledExecutorService group, Consumer<EventLoopGroupRetry> operation) {
+        this(
+                group,
+                operation,
+                ExponentialBackoffStrategy.withDefault()
+        );
+    }
+
+    @Override
+    public void run() {
+        operation.accept(this);
+    }
+
+    @Override
+    public void retry() {
+        group.schedule(
+                this,
+                strategy.getAsLong(),
+                TimeUnit.MILLISECONDS
+        );
+    }
+}

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
+        );
+    }
+}

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

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<SocketChannel> {
 
+    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<SocketChannel> {
         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);
     }
 }

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/HttpConnection.java

@@ -0,0 +1,73 @@
+package pl.tkowalcz.tjahzi.http;
+
+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) {
+        ChannelFuture stableReference = this.lokiConnection;
+
+        stableReference.awaitUninterruptibly();
+        if (stableReference.isSuccess() && stableReference.channel().isActive()) {
+            stableReference.channel().writeAndFlush(request);
+        } else {
+            // TODO: increase error counters
+            retry.retry();
+            // Dropping message on the floor
+        }
+    }
+
+    @Override
+    public void close() {
+        group.shutdownGracefully();
+    }
+}

tjahzi-core/src/main/java/pl/tkowalcz/tjahzi/http/NettyHttpClient.java

@@ -1,62 +1,32 @@
 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.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;
 
 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 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
         );
+
+        lokiConnection = new HttpConnection(clientConfiguration);
     }
 
     public void log(ByteBuf dataBuffer) {
@@ -66,7 +36,7 @@ public class NettyHttpClient implements Closeable {
         FullHttpRequest request = new DefaultFullHttpRequest(
                 HttpVersion.HTTP_1_1,
                 HttpMethod.POST,
-                logEndpoint,
+                clientConfiguration.getLogEndpoint(),
                 output
         );
 
@@ -74,13 +44,13 @@ public class NettyHttpClient implements Closeable {
                 .add(headers)
                 .set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE)
                 .set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes())
-                .set(HttpHeaderNames.HOST, host);
+                .set(HttpHeaderNames.HOST, clientConfiguration.getHost());
 
-        channel.writeAndFlush(request);
+        lokiConnection.execute(request);
     }
 
     @Override
     public void close() {
-        group.shutdownGracefully();
+        lokiConnection.close();
     }
 }

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<FullHttpResponse> {
+
+    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");
+    }
+}

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();
+}

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)
                         ));
@@ -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(
@@ -103,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)
                         ));
@@ -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(

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();

tjahzi-core/src/test/java/pl/tkowalcz/tjahzi/ReconnectTest.java

@@ -0,0 +1,154 @@
+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()
+                        .port(12322)
+        );
+
+        wireMockServer.stubFor(
+                post(urlEqualTo("/loki/api/v1/push"))
+                        .willReturn(
+                                aResponse().withStatus(200)
+                        ));
+
+        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 shouldEventuallyReconnectIfLokiWasDownWhenStarting() {
+        // Given
+        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
+                .withConnectionTimeoutMillis(10_000)
+                .withHost("localhost")
+                .withPort(12322)
+                .withMaxRetries(10)
+                .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"
+        );
+
+        wireMockServer.start();
+
+        // Then
+        await()
+                .atMost(Durations.TEN_MINUTES)
+                .untilAsserted(() ->
+                        wireMockServer.verify(
+                                postRequestedFor(urlMatching("/loki/api/v1/push"))
+                        )
+                );
+    }
+
+    @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"))
+                        )
+                );
+    }
+}

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(

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);
+    }
+}

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);
+        });
+    }
+}

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<Long> 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
+        );
+    }
+}