This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 430d976cc feat(java): let clients close without Netty's 2 s quiet
period (#4704)
430d976cc is described below
commit 430d976ccfc638a0c2ee33541c2b71bf81e99c82
Author: Maciej Modzelewski <[email protected]>
AuthorDate: Mon Nov 2 11:31:12 2026 +0100
feat(java): let clients close without Netty's 2 s quiet period (#4704)
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;
+ }
+}
