This is an automated email from the ASF dual-hosted git repository. mmodzelewski pushed a commit to branch java-test-ci in repository https://gitbox.apache.org/repos/asf/iggy.git
commit dd18f3b4a264456235e4bae17766c58a17e93817 Author: Maciej Modzelewski <[email protected]> AuthorDate: Mon Nov 2 10:44:26 2026 +0100 feat(java): let clients close without Netty's 2 s quiet period A client that creates its own event loop group stopped it with Netty's shutdownGracefully(), which waits for 2 seconds with no new tasks. Every close took at least that long, so test suites that open and close many short-lived clients spent most of their time waiting. The TCP builders now take shutdownQuietPeriod and shutdownTimeout. A value that the builder leaves unset comes from the system property iggy.tcp.shutdownQuietPeriodMillis or iggy.tcp.shutdownTimeoutMillis. Without the property, the Netty defaults of 2 and 15 seconds apply. A property changes every client in the JVM at once, also clients that a library or a fixture builds, so the Gradle test task sets the quiet period to 0. build() rejects a negative value, a timeout below the quiet period, and a property that is not a whole number of milliseconds. The values do not apply to a group that the caller supplies, because the caller shuts that group down. --- foreign/java/README.md | 21 ++++ .../kotlin/iggy.java-common-conventions.gradle.kts | 3 + .../iggy/client/async/tcp/AsyncIggyTcpClient.java | 13 ++- .../async/tcp/AsyncIggyTcpClientBuilder.java | 43 ++++++- .../iggy/client/async/tcp/AsyncTcpConnection.java | 17 ++- .../iggy/client/async/tcp/EventLoopShutdown.java | 109 ++++++++++++++++++ .../client/blocking/tcp/IggyTcpClientBuilder.java | 24 ++++ .../tcp/AsyncIggyTcpClientEventLoopGroupTest.java | 38 +++++- .../tcp/AsyncIggyTcpClientFailoverSignInTest.java | 3 +- .../tcp/AsyncIggyTcpClientLoginRoutingTest.java | 3 +- .../tcp/AsyncIggyTcpClientSessionRulesTest.java | 3 +- .../client/async/tcp/EventLoopShutdownTest.java | 127 +++++++++++++++++++++ 12 files changed, 385 insertions(+), 19 deletions(-) diff --git a/foreign/java/README.md b/foreign/java/README.md index 7f52cd01b..396315686 100644 --- a/foreign/java/README.md +++ b/foreign/java/README.md @@ -296,6 +296,27 @@ group.shutdownGracefully(); Do not block in a completion callback. Callbacks run on the group's loops, so a blocked callback stalls every client that shares the group. +### Shutdown Quiet Period + +When a client closes, it shuts down the event loop group that it created for itself. Netty +waits for a quiet period of 2 seconds with no new tasks before the group stops, so each close +takes at least that long. If you open and close many short-lived clients, for example in tests, +set a shorter quiet period: + +```java +var client = Iggy.tcpClientBuilder() + .blocking() + .shutdownQuietPeriod(Duration.ZERO) + .shutdownTimeout(Duration.ofSeconds(5)) + .credentials("iggy", "iggy") + .buildAndLogin(); +``` + +To change the default for every client in the JVM, set the system properties +`iggy.tcp.shutdownQuietPeriodMillis` and `iggy.tcp.shutdownTimeoutMillis`, in milliseconds. +A value set on the builder overrides the property. These values do not apply to a group +that you supply with `eventLoopGroup(...)`. + ### Version Information ```java diff --git a/foreign/java/buildSrc/src/main/kotlin/iggy.java-common-conventions.gradle.kts b/foreign/java/buildSrc/src/main/kotlin/iggy.java-common-conventions.gradle.kts index 2a01cd7a7..683bbc840 100644 --- a/foreign/java/buildSrc/src/main/kotlin/iggy.java-common-conventions.gradle.kts +++ b/foreign/java/buildSrc/src/main/kotlin/iggy.java-common-conventions.gradle.kts @@ -61,6 +61,9 @@ jacoco { tasks.withType<Test> { useJUnitPlatform() + // Most tests close the clients they create, and Netty's default 2 second + // quiet period would hold every such close for that long. + systemProperty("iggy.tcp.shutdownQuietPeriodMillis", "0") finalizedBy(tasks.jacocoTestReport) } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java index cabb93ec9..43680b4dd 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java @@ -185,6 +185,7 @@ public class AsyncIggyTcpClient { private final Optional<File> tlsCertificate; private final Optional<IoEventLoopGroup> sharedEventLoopGroup; private IoEventLoopGroup eventLoopGroup; + private final EventLoopShutdown eventLoopShutdown; private final org.apache.iggy.client.async.tcp.vsr.ConsensusSession session = new org.apache.iggy.client.async.tcp.vsr.ConsensusSession(); private final int ioThreads; @@ -285,7 +286,8 @@ public class AsyncIggyTcpClient { false, Optional.empty(), Optional.empty(), - AsyncTcpConnection.DEFAULT_IO_THREADS); + AsyncTcpConnection.DEFAULT_IO_THREADS, + EventLoopShutdown.resolve(null, null)); } @SuppressWarnings("checkstyle:ParameterNumber") @@ -303,7 +305,8 @@ public class AsyncIggyTcpClient { boolean enableTls, Optional<File> tlsCertificate, Optional<IoEventLoopGroup> sharedEventLoopGroup, - int ioThreads) { + int ioThreads, + EventLoopShutdown eventLoopShutdown) { this.connectionInfo = new ConnectionInfo(host, port); this.seedConnectionInfo = this.connectionInfo; this.username = Optional.ofNullable(username); @@ -322,6 +325,7 @@ public class AsyncIggyTcpClient { this.tlsCertificate = tlsCertificate; this.sharedEventLoopGroup = sharedEventLoopGroup; this.ioThreads = ioThreads; + this.eventLoopShutdown = eventLoopShutdown; var poolConfigBuilder = TcpConnectionPoolConfig.builder(); this.acquireTimeout.ifPresent(timeout -> poolConfigBuilder.setAcquireTimeoutMillis(timeout.toMillis())); @@ -735,7 +739,8 @@ public class AsyncIggyTcpClient { /** * Closes the TCP connection and releases the Netty resources that this client owns. * - * <p>If the client created its own event loop group, it shuts that group down gracefully. + * <p>If the client created its own event loop group, it shuts that group down gracefully, + * see {@link AsyncIggyTcpClientBuilder#shutdownQuietPeriod(Duration)}. * The client does not shut down a group supplied through * {@link AsyncIggyTcpClientBuilder#eventLoopGroup(IoEventLoopGroup)}. The caller shuts * that group down. After you call this method, you cannot reuse the client. If you need @@ -773,7 +778,7 @@ public class AsyncIggyTcpClient { return CompletableFuture.completedFuture(null); } CompletableFuture<Void> stopped = new CompletableFuture<>(); - owned.shutdownGracefully().addListener(done -> { + eventLoopShutdown.shutdown(owned).addListener(done -> { if (done.isSuccess()) { stopped.complete(null); } else { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilder.java index 9b6dfd152..2ace16193 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilder.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilder.java @@ -103,6 +103,8 @@ public final class AsyncIggyTcpClientBuilder { private long maxVsrFrameSize = VsrFrameDecoder.DEFAULT_MAX_FRAME_SIZE; private int ioThreads = AsyncTcpConnection.DEFAULT_IO_THREADS; private IoEventLoopGroup eventLoopGroup; + private Duration shutdownQuietPeriod; + private Duration shutdownTimeout; public AsyncIggyTcpClientBuilder() {} @@ -326,6 +328,40 @@ public final class AsyncIggyTcpClientBuilder { return this; } + /** + * Sets how long a close waits for new tasks before it stops the event loop group that the client created for + * itself. Netty runs every task that arrives in this period, and each such task starts the period again, so the + * close of an idle client takes at least this long. + * + * <p>If this value is not set, the client reads the {@code iggy.tcp.shutdownQuietPeriodMillis} system property, + * in milliseconds. Without the property, the Netty default of 2 seconds applies. The value has no effect on a + * group supplied through {@link #eventLoopGroup(IoEventLoopGroup)}. + * + * @param shutdownQuietPeriod the quiet period, zero or more + * @return this builder + */ + public AsyncIggyTcpClientBuilder shutdownQuietPeriod(Duration shutdownQuietPeriod) { + this.shutdownQuietPeriod = shutdownQuietPeriod; + return this; + } + + /** + * Sets the longest time that a close waits for the event loop group that the client created for itself to + * stop, even when tasks keep arriving in the quiet period. It cannot be shorter than the quiet period. + * + * <p>If this value is not set, the client reads the {@code iggy.tcp.shutdownTimeoutMillis} system property, in + * milliseconds. Without the property, the Netty default of 15 seconds applies. The value has no effect on a group + * supplied through {@link #eventLoopGroup(IoEventLoopGroup)}. + * + * @param shutdownTimeout the shutdown timeout, zero or more + * @return this builder + * @see #shutdownQuietPeriod(Duration) + */ + public AsyncIggyTcpClientBuilder shutdownTimeout(Duration shutdownTimeout) { + this.shutdownTimeout = shutdownTimeout; + return this; + } + /** * Bounds active and retained mutations; defaults are {@value ConsensusSession#DEFAULT_MAX_PENDING_OPERATIONS} * operations and {@value ConsensusSession#DEFAULT_MAX_PENDING_OPERATION_BYTES} bytes (64 MiB). The count also @@ -355,7 +391,8 @@ public final class AsyncIggyTcpClientBuilder { * Note: You still need to call {@link AsyncIggyTcpClient#connect()} on the returned client. * * @return a new AsyncIggyTcpClient instance - * @throws IggyInvalidArgumentException if the host is null or empty, or if the port is not positive + * @throws IggyInvalidArgumentException if the host is null or empty, if the port is not positive, or if a + * setting or its system property is invalid */ public AsyncIggyTcpClient build() { validateHost(); @@ -367,6 +404,7 @@ public final class AsyncIggyTcpClientBuilder { validateMaxVsrFrameSize(); validateIoThreads(); validateEventLoopGroup(); + EventLoopShutdown eventLoopShutdown = EventLoopShutdown.resolve(shutdownQuietPeriod, shutdownTimeout); AsyncIggyTcpClient client = new AsyncIggyTcpClient( host, @@ -382,7 +420,8 @@ public final class AsyncIggyTcpClientBuilder { enableTls, Optional.ofNullable(tlsCertificate), Optional.ofNullable(eventLoopGroup), - ioThreads); + ioThreads, + eventLoopShutdown); client.configurePendingOperations(maxPendingOperations, maxPendingOperationBytes); return client; } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java index 9729ece95..66c8362bd 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java @@ -130,7 +130,7 @@ public class AsyncTcpConnection { private static final String EVENT_LOOP_THREAD_PREFIX = "iggy-tcp-io"; private final IoEventLoopGroup eventLoopGroup; - private final boolean ownsEventLoopGroup; + private final Optional<EventLoopShutdown> ownedGroupShutdown; private final EventLoop eventLoop; private final FixedChannelPool channelPool; private final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE, true); @@ -212,7 +212,8 @@ public class AsyncTcpConnection { this.consensusSession = new ConsensusSession(); this.vsrEncoder = new VsrRequestEncoder(consensusSession); - this.ownsEventLoopGroup = sharedEventLoopGroup.isEmpty(); + this.ownedGroupShutdown = + sharedEventLoopGroup.isEmpty() ? Optional.of(EventLoopShutdown.resolve(null, null)) : Optional.empty(); this.eventLoopGroup = sharedEventLoopGroup.orElseGet(() -> newEventLoopGroup(ioThreads)); this.eventLoop = eventLoopGroup.next(); this.heartbeat = new ConnectionHeartbeat( @@ -1710,13 +1711,11 @@ public class AsyncTcpConnection { private void closePool(CompletableFuture<Void> shutdownFuture) { try { channelPool.closeAsync().addListener(poolClosed -> { - if (!ownsEventLoopGroup) { - completeShutdown(shutdownFuture, poolClosed); - return; - } - eventLoopGroup - .shutdownGracefully() - .addListener(groupClosed -> completeShutdown(shutdownFuture, groupClosed)); + // The caller shuts a supplied group down, so for one the close ends with the pool. + Future<?> closed = ownedGroupShutdown + .<Future<?>>map(shutdown -> shutdown.shutdown(eventLoopGroup)) + .orElse(poolClosed); + closed.addListener(step -> completeShutdown(shutdownFuture, step)); }); } catch (RejectedExecutionException loopGone) { log.warn("Event loop rejected the pool close, channel already gone: {}", loopGone.getMessage()); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/EventLoopShutdown.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/EventLoopShutdown.java new file mode 100644 index 000000000..b6600e552 --- /dev/null +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/EventLoopShutdown.java @@ -0,0 +1,109 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.client.async.tcp; + +import io.netty.util.concurrent.EventExecutorGroup; +import io.netty.util.concurrent.Future; +import org.apache.iggy.exception.IggyInvalidArgumentException; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import java.util.function.UnaryOperator; + +/** + * How a close shuts down an event loop group that the client created for itself. Netty runs every task that + * arrives during the quiet period, and each such task starts the period again. The timeout ends the wait even + * when tasks keep arriving. + * + * <p>A value that the builder leaves unset comes from a system property in milliseconds, which changes it for every + * client in the JVM. Without the property, the Netty default applies. + */ +record EventLoopShutdown(Duration quietPeriod, Duration timeout) { + + static final String QUIET_PERIOD_PROPERTY = "iggy.tcp.shutdownQuietPeriodMillis"; + static final String TIMEOUT_PROPERTY = "iggy.tcp.shutdownTimeoutMillis"; + // What Netty's shutdownGracefully() uses without arguments. + static final Duration DEFAULT_QUIET_PERIOD = Duration.ofSeconds(2); + static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(15); + + /** + * Resolves the builder values against the JVM system properties. + * + * @param quietPeriod the builder quiet period, or {@code null} if unset + * @param timeout the builder timeout, or {@code null} if unset + * @throws IggyInvalidArgumentException if a value is negative, a property is not a whole number of + * milliseconds, or the timeout is shorter than the quiet period + */ + static EventLoopShutdown resolve(Duration quietPeriod, Duration timeout) { + return resolve(quietPeriod, timeout, System::getProperty); + } + + static EventLoopShutdown resolve(Duration quietPeriod, Duration timeout, UnaryOperator<String> properties) { + Duration resolvedQuietPeriod = quietPeriod != null + ? quietPeriod + : fromProperty(properties, QUIET_PERIOD_PROPERTY, DEFAULT_QUIET_PERIOD); + Duration resolvedTimeout = + timeout != null ? timeout : fromProperty(properties, TIMEOUT_PROPERTY, DEFAULT_TIMEOUT); + if (resolvedQuietPeriod.isNegative()) { + throw new IggyInvalidArgumentException("ShutdownQuietPeriod cannot be negative"); + } + if (resolvedTimeout.isNegative()) { + throw new IggyInvalidArgumentException("ShutdownTimeout cannot be negative"); + } + if (resolvedTimeout.compareTo(resolvedQuietPeriod) < 0) { + throw new IggyInvalidArgumentException("ShutdownTimeout cannot be shorter than ShutdownQuietPeriod"); + } + return new EventLoopShutdown(resolvedQuietPeriod, resolvedTimeout); + } + + Future<?> shutdown(EventExecutorGroup group) { + return group.shutdownGracefully(saturatedNanos(quietPeriod), saturatedNanos(timeout), TimeUnit.NANOSECONDS); + } + + private static Duration fromProperty(UnaryOperator<String> properties, String name, Duration fallback) { + String value = properties.apply(name); + if (value == null || value.isBlank()) { + return fallback; + } + long millis; + try { + millis = Long.parseLong(value.strip()); + } catch (NumberFormatException notANumber) { + throw invalidProperty(name, value); + } + if (millis < 0) { + throw invalidProperty(name, value); + } + return Duration.ofMillis(millis); + } + + private static IggyInvalidArgumentException invalidProperty(String name, String value) { + return new IggyInvalidArgumentException( + "System property " + name + " must be a whole number of milliseconds, zero or more, but is: " + value); + } + + private static long saturatedNanos(Duration value) { + try { + return value.toNanos(); + } catch (ArithmeticException tooLong) { + return Long.MAX_VALUE; + } + } +} diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilder.java index 69808ca02..fd808cce0 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilder.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilder.java @@ -169,6 +169,30 @@ public final class IggyTcpClientBuilder { return this; } + /** + * Sets how long a close waits for new tasks before it stops the event loop group that the client created for + * itself, see {@link AsyncIggyTcpClientBuilder#shutdownQuietPeriod(Duration)}. + * + * @param shutdownQuietPeriod the quiet period, zero or more + * @return this builder + */ + public IggyTcpClientBuilder shutdownQuietPeriod(Duration shutdownQuietPeriod) { + asyncBuilder.shutdownQuietPeriod(shutdownQuietPeriod); + return this; + } + + /** + * Sets the longest time that a close waits for the event loop group that the client created for itself to + * stop, see {@link AsyncIggyTcpClientBuilder#shutdownTimeout(Duration)}. + * + * @param shutdownTimeout the shutdown timeout, zero or more + * @return this builder + */ + public IggyTcpClientBuilder shutdownTimeout(Duration shutdownTimeout) { + asyncBuilder.shutdownTimeout(shutdownTimeout); + return this; + } + /** * Enables or disables TLS for the TCP connection. * diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEventLoopGroupTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEventLoopGroupTest.java index 895146398..b362bdd02 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEventLoopGroupTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEventLoopGroupTest.java @@ -26,6 +26,7 @@ import io.netty.channel.IoEventLoopGroup; import io.netty.channel.MultiThreadIoEventLoopGroup; import io.netty.channel.nio.NioIoHandler; import org.apache.iggy.client.ConnectionInfo; +import org.apache.iggy.client.blocking.tcp.IggyTcpClient; import org.junit.jupiter.api.Test; import java.io.EOFException; @@ -63,7 +64,7 @@ class AsyncIggyTcpClientEventLoopGroupTest { private static final String OWNED_THREAD_PREFIX = "iggy-tcp-io-"; private static final Duration HEARTBEAT_INTERVAL = Duration.ofMillis(25); - // Graceful shutdown waits a two-second quiet period before the loop thread exits. + // Bounds a close and the exit of its loop thread, whatever quiet period the run configures. private static final Duration SHUTDOWN_TIMEOUT = Duration.ofSeconds(15); private static final int HEADER_SIZE = 256; @@ -99,6 +100,41 @@ class AsyncIggyTcpClientEventLoopGroupTest { } } + // The test task sets the quiet period property to zero, so in these two tests only the builder value can hold + // the close. + @Test + void shouldHoldACloseForTheQuietPeriodSetOnTheBuilder() throws Exception { + Duration quietPeriod = Duration.ofMillis(500); + try (MockVsrServer server = MockVsrServer.start()) { + AsyncIggyTcpClient client = + builder(server).shutdownQuietPeriod(quietPeriod).build(); + client.connect().get(5, TimeUnit.SECONDS); + + long closeStarted = System.nanoTime(); + client.close().get(SHUTDOWN_TIMEOUT.toSeconds(), TimeUnit.SECONDS); + + assertThat(Duration.ofNanos(System.nanoTime() - closeStarted)).isGreaterThanOrEqualTo(quietPeriod); + } + } + + @Test + void shouldHoldABlockingCloseForTheQuietPeriodSetOnTheBuilder() throws Exception { + Duration quietPeriod = Duration.ofMillis(500); + try (MockVsrServer server = MockVsrServer.start()) { + IggyTcpClient client = IggyTcpClient.builder() + .host(loopback()) + .port(server.port()) + .shutdownQuietPeriod(quietPeriod) + .build(); + client.connect(); + + long closeStarted = System.nanoTime(); + client.close(); + + assertThat(Duration.ofNanos(System.nanoTime() - closeStarted)).isGreaterThanOrEqualTo(quietPeriod); + } + } + @Test void shouldLeaveASharedGroupRunningWhenClientsClose() throws Exception { IoEventLoopGroup group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientFailoverSignInTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientFailoverSignInTest.java index f1077ded5..97f1701fd 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientFailoverSignInTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientFailoverSignInTest.java @@ -361,7 +361,8 @@ class AsyncIggyTcpClientFailoverSignInTest { false, Optional.empty(), Optional.empty(), - AsyncTcpConnection.DEFAULT_IO_THREADS); + AsyncTcpConnection.DEFAULT_IO_THREADS, + EventLoopShutdown.resolve(null, null)); } @Override diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientLoginRoutingTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientLoginRoutingTest.java index 41b300338..81652e7b2 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientLoginRoutingTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientLoginRoutingTest.java @@ -212,7 +212,8 @@ class AsyncIggyTcpClientLoginRoutingTest { false, Optional.empty(), Optional.empty(), - AsyncTcpConnection.DEFAULT_IO_THREADS); + AsyncTcpConnection.DEFAULT_IO_THREADS, + EventLoopShutdown.resolve(null, null)); // Successful login publication retains the carrier's credentials even when this fixture fakes Register. var carrier = new AsyncTcpConnection( "localhost", diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientSessionRulesTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientSessionRulesTest.java index 1b3ec6c8c..161a01f1e 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientSessionRulesTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientSessionRulesTest.java @@ -1168,7 +1168,8 @@ class AsyncIggyTcpClientSessionRulesTest { false, Optional.empty(), Optional.empty(), - AsyncTcpConnection.DEFAULT_IO_THREADS); + AsyncTcpConnection.DEFAULT_IO_THREADS, + EventLoopShutdown.resolve(null, null)); } @Override diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/EventLoopShutdownTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/EventLoopShutdownTest.java new file mode 100644 index 000000000..dba777e7b --- /dev/null +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/EventLoopShutdownTest.java @@ -0,0 +1,127 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iggy.client.async.tcp; + +import io.netty.channel.IoEventLoopGroup; +import org.apache.iggy.exception.IggyInvalidArgumentException; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.time.Duration; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.function.UnaryOperator; + +import static org.apache.iggy.client.async.tcp.EventLoopShutdown.QUIET_PERIOD_PROPERTY; +import static org.apache.iggy.client.async.tcp.EventLoopShutdown.TIMEOUT_PROPERTY; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class EventLoopShutdownTest { + + private static final UnaryOperator<String> NO_PROPERTIES = name -> null; + + @Test + void shouldUseTheNettyDefaultsWithoutBuilderValuesOrProperties() { + var shutdown = EventLoopShutdown.resolve(null, null, NO_PROPERTIES); + + assertThat(shutdown.quietPeriod()).isEqualTo(Duration.ofSeconds(2)); + assertThat(shutdown.timeout()).isEqualTo(Duration.ofSeconds(15)); + } + + @Test + void shouldTakeUnsetValuesFromTheSystemProperties() { + var shutdown = EventLoopShutdown.resolve( + null, null, properties(Map.of(QUIET_PERIOD_PROPERTY, "0", TIMEOUT_PROPERTY, " 250 "))); + + assertThat(shutdown.quietPeriod()).isZero(); + assertThat(shutdown.timeout()).isEqualTo(Duration.ofMillis(250)); + } + + @Test + void shouldPreferBuilderValuesOverTheSystemProperties() { + var shutdown = EventLoopShutdown.resolve( + Duration.ofSeconds(1), + Duration.ofSeconds(3), + properties(Map.of(QUIET_PERIOD_PROPERTY, "0", TIMEOUT_PROPERTY, "0"))); + + assertThat(shutdown.quietPeriod()).isEqualTo(Duration.ofSeconds(1)); + assertThat(shutdown.timeout()).isEqualTo(Duration.ofSeconds(3)); + } + + @Test + void shouldResolveEachValueOnItsOwn() { + var shutdown = + EventLoopShutdown.resolve(null, Duration.ofSeconds(1), properties(Map.of(QUIET_PERIOD_PROPERTY, "0"))); + + assertThat(shutdown.quietPeriod()).isZero(); + assertThat(shutdown.timeout()).isEqualTo(Duration.ofSeconds(1)); + } + + @ParameterizedTest + @ValueSource(strings = {"-1", "2s", "PT2S", "1.5"}) + void shouldRejectAPropertyThatIsNotAWholeNumberOfMilliseconds(String value) { + assertThatThrownBy( + () -> EventLoopShutdown.resolve(null, null, properties(Map.of(QUIET_PERIOD_PROPERTY, value)))) + .isInstanceOf(IggyInvalidArgumentException.class) + .hasMessageContaining(QUIET_PERIOD_PROPERTY); + } + + @Test + void shouldRejectNegativeBuilderValues() { + assertThatThrownBy(() -> EventLoopShutdown.resolve(Duration.ofMillis(-1), null, NO_PROPERTIES)) + .isInstanceOf(IggyInvalidArgumentException.class) + .hasMessage("ShutdownQuietPeriod cannot be negative"); + assertThatThrownBy(() -> EventLoopShutdown.resolve(Duration.ZERO, Duration.ofMillis(-1), NO_PROPERTIES)) + .isInstanceOf(IggyInvalidArgumentException.class) + .hasMessage("ShutdownTimeout cannot be negative"); + } + + @Test + void shouldRejectATimeoutShorterThanTheQuietPeriod() { + assertThatThrownBy(() -> EventLoopShutdown.resolve(Duration.ofSeconds(20), null, NO_PROPERTIES)) + .isInstanceOf(IggyInvalidArgumentException.class) + .hasMessage("ShutdownTimeout cannot be shorter than ShutdownQuietPeriod"); + } + + @Test + void shouldRejectAnInvalidSettingWhenTheBuilderBuilds() { + var builder = AsyncIggyTcpClient.builder().shutdownQuietPeriod(Duration.ofMillis(-1)); + + assertThatThrownBy(builder::build) + .isInstanceOf(IggyInvalidArgumentException.class) + .hasMessage("ShutdownQuietPeriod cannot be negative"); + } + + @Test + void shouldSaturateDurationsTooLongForNetty() throws Exception { + IoEventLoopGroup group = AsyncTcpConnection.newEventLoopGroup(1); + + var shutdown = new EventLoopShutdown(Duration.ZERO, Duration.ofSeconds(Long.MAX_VALUE)); + + shutdown.shutdown(group).get(5, TimeUnit.SECONDS); + assertThat(group.isTerminated()).isTrue(); + } + + private static UnaryOperator<String> properties(Map<String, String> values) { + return values::get; + } +}
