From 961ace842f2a697f39f59dc2556182f5cfafcf09 Mon Sep 17 00:00:00 2001 From: "tomasz.kowalczewski" Date: Tue, 8 Dec 2020 08:45:06 +0100 Subject: [PATCH] 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); + }); + } +}