fix(java): keep a queued sign-in from failing as replaced

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;

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())) {