From a1dd533635a78c035f828fc1d3a22b9e965048c1 Mon Sep 17 00:00:00 2001
From: Hubert Gruszecki <h.gruszecki@gmail.com>
Date: Mon, 26 Oct 2026 15:10:15 +0100
Subject: [PATCH] fix(java): keep a queued sign-in from failing as replaced

A sign-in that started while another one was in flight could fail
with "A new sign-in replaced this one", although nothing replaced
it, which made shouldNotSignInAsThePreviousUserAfterARejectedSignIn
flaky in CI. Off the event loop, discardAuthentication only queues
the close of the abandoned channel. Netty reads ready sockets
before queued tasks, so the old Register's reply could end that
sign-in and send the queued Register on that channel before the
close ran.

A Register started on the loop now acquires its channel in a new
loop task, which runs after the queued close, as an acquire from
another thread already does.
---
 .../client/async/tcp/AsyncTcpConnection.java  | 16 +++++++
 .../AsyncTcpConnectionConcurrencyTest.java    | 48 +++++++++++++++++++
 2 files changed, 64 insertions(+)

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 bfdfb2452c..99d2f0e357 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<Void> 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<Void> replyCallback = new CompletableFuture<>();
+            CompletableFuture<Void> 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<ByteBuf> 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())) {