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 04a859f7df..04594a4a1a 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 @@ -539,6 +539,22 @@ private void attempt(Call call, ByteBuf payload, long requiredSessionGeneration) } void acquireForCall(Call call, ByteBuf payload, long requiredSessionGeneration) { + if (!isLoginCode(call.commandCode()) || !eventLoop.inEventLoop()) { + acquireChannel(call, payload, requiredSessionGeneration); + return; + } + // An off-loop discardAuthentication queues the close of the channels it abandons. A Register that a loop + // callback starts would take such a channel before that close runs, and the close would fail it. A new loop + // task runs after the close, like the acquire that the pool queues for a Register from another thread. + try { + eventLoop.execute(() -> acquireChannel(call, payload, requiredSessionGeneration)); + } catch (RejectedExecutionException loopGone) { + payload.release(); + call.fail(loopGone); + } + } + + private void acquireChannel(Call call, ByteBuf payload, long requiredSessionGeneration) { TransientFailoverState failoverState = call.state(); failoverState.linkFailed = false; failoverState.detachedSession = false; diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java index 5c8778ac92..2706665b14 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java @@ -661,6 +661,54 @@ void shouldSerializeRegisterAndLogoutThroughTheirResponses() throws Exception { } } + @Test + void shouldNotSendARegisterOnAChannelThatADiscardIsClosing() throws Exception { + InetAddress loopback = InetAddress.getLoopbackAddress(); + try (ServerSocket serverSocket = new ServerSocket(0, 1, loopback)) { + CompletableFuture server = CompletableFuture.runAsync(() -> { + try (Socket abandoned = serverSocket.accept()) { + abandoned.setSoTimeout((int) TimeUnit.SECONDS.toMillis(2)); + Request ping = readRequest(abandoned.getInputStream()); + writeResponse(abandoned.getOutputStream(), ping, new byte[0]); + try (Socket replacement = serverSocket.accept()) { + replacement.setSoTimeout((int) TimeUnit.SECONDS.toMillis(2)); + Request register = readRequest(replacement.getInputStream()); + assertThat(register.operation()).isEqualTo(OPERATION_REGISTER); + writeResponse(replacement.getOutputStream(), register, registerBody()); + } + } catch (IOException error) { + throw new IllegalStateException("Mock VSR server failed", error); + } + }); + + AsyncTcpConnection connection = newConnection(serverSocket, Duration.ofHours(1), 500); + CompletableFuture replyCallback = new CompletableFuture<>(); + CompletableFuture discarded = new CompletableFuture<>(); + try { + connection.connect().get(5, TimeUnit.SECONDS); + // The reply callback holds the event loop, so the close that the discard queues waits behind it, + // as it waits behind the reply that ends one sign-in and starts the queued one. + CompletableFuture register = connection + .send(PING_CODE, Unpooled.EMPTY_BUFFER) + .thenCompose(pong -> { + pong.release(); + replyCallback.complete(null); + discarded.join(); + return connection.send(LOGIN_CODE, loginPayload()); + }); + replyCallback.get(5, TimeUnit.SECONDS); + connection.discardAuthentication(); + discarded.complete(null); + + register.get(5, TimeUnit.SECONDS).release(); + server.get(5, TimeUnit.SECONDS); + } finally { + discarded.complete(null); + connection.close().get(5, TimeUnit.SECONDS); + } + } + } + @Test void shouldKeepATicketlessReadOnTheIdentityItStartedEncoding() throws Exception { try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) {