fix(binding-kafka): reconnect KIND_API_REQUEST pool after idle teardown

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java

@@ -809,7 +809,7 @@ private void onAppWindow(
             this.replyMax = maximum;
             this.replyPad = padding;
 
-            state = KafkaState.openedReply(state);
+            state = KafkaState.openReply(state);
 
             assert replyAck <= replySeq;
 
@@ -1712,15 +1712,10 @@ private void onNetEnd(
             state = KafkaState.closedReply(state);
             doNetEnd(traceId, authorization);
 
-            // a disconnect is not evidence the broker's supported versions changed;
-            // apiVersionRangeByApiKey survives reconnect and is only invalidated
-            // reactively, by an UNSUPPORTED_VERSION response to a real request
-            saslResolved = false;
-
-            cleanupDecodeSlot();
-            cleanupAppActive(traceId, EMPTY_OCTETS);
-
-            doNetSignalReconnect(traceId);
+            if (KafkaState.closed(state))
+            {
+                onNetClosed(traceId);
+            }
         }
 
         private void onNetAbort(
@@ -1729,8 +1724,12 @@ private void onNetAbort(
             final long traceId = abort.traceId();
 
             state = KafkaState.closedReply(state);
+            doNetAbort(traceId);
 
-            cleanupNet(traceId);
+            if (KafkaState.closed(state))
+            {
+                onNetClosed(traceId);
+            }
         }
 
         private void onNetReset(
@@ -1739,8 +1738,12 @@ private void onNetReset(
             final long traceId = reset.traceId();
 
             state = KafkaState.closedInitial(state);
+            doNetReset(traceId);
 
-            cleanupNet(traceId);
+            if (KafkaState.closed(state))
+            {
+                onNetClosed(traceId);
+            }
         }
 
         private void onNetWindow(
@@ -1803,31 +1806,6 @@ private void doNetBegin(
             long authorization,
             long affinity)
         {
-            if (KafkaState.closed(state))
-            {
-                state = 0;
-
-                initialSeq = 0;
-                initialAck = 0;
-                initialMax = 0;
-                initialPad = 0;
-                initialBudgetId = NO_BUDGET_ID;
-
-                replySeq = 0;
-                replyAck = 0;
-                replyMax = 0;
-
-                nextCorrelationId = 0;
-                responseBytesRemaining = 0;
-                apiVersionKeysRemaining = 0;
-                saslMechanismsRemaining = 0;
-
-                requestInFlight = false;
-                apiVersionsRequestExplicit = false;
-
-                decoder = decodeReject;
-            }
-
             if (!KafkaState.initialOpening(state))
             {
                 assert state == 0;
@@ -1895,6 +1873,11 @@ private void doNetEnd(
             cleanupBudget();
 
             deauthorizeGuardSession();
+
+            if (KafkaState.closed(state))
+            {
+                onNetClosed(traceId);
+            }
         }
 
         private void doNetAbort(
@@ -1912,6 +1895,11 @@ private void doNetAbort(
             cleanupBudget();
 
             deauthorizeGuardSession();
+
+            if (KafkaState.closed(state))
+            {
+                onNetClosed(traceId);
+            }
         }
 
         private void doNetReset(
@@ -1928,6 +1916,11 @@ private void doNetReset(
             cleanupDecodeSlot();
 
             deauthorizeGuardSession();
+
+            if (KafkaState.closed(state))
+            {
+                onNetClosed(traceId);
+            }
         }
 
         private void deauthorizeGuardSession()
@@ -2318,17 +2311,50 @@ private void onNetSignalReconnect(
             }
         }
 
-        private void cleanupNetPending(
-            long traceId,
-            int error)
+        private void onNetClosed(
+            long traceId)
         {
-            doNetReset(traceId);
-            doNetAbort(traceId);
+            assert KafkaState.closed(state);
 
-            apiVersionRangeByApiKey.clear();
+            cleanupAppActive(traceId, EMPTY_OCTETS);
+            cleanupDecodeSlot();
+            cleanupEncodeSlot();
+            cleanupBudget();
+
+            state = 0;
+
+            initialSeq = 0;
+            initialAck = 0;
+            initialMax = 0;
+            initialPad = 0;
+            initialBudgetId = NO_BUDGET_ID;
+
+            replySeq = 0;
+            replyAck = 0;
+            replyMax = 0;
+
+            nextCorrelationId = 0;
+            responseBytesRemaining = 0;
+            apiVersionKeysRemaining = 0;
+            saslMechanismsRemaining = 0;
+
+            requestInFlight = false;
             apiVersionsRequestExplicit = false;
+
+            // a disconnect is not evidence the broker's supported versions changed;
+            // apiVersionRangeByApiKey survives reconnect and is only invalidated
+            // reactively, by an UNSUPPORTED_VERSION response to a real request
             saslResolved = false;
 
+            decoder = decodeReject;
+
+            doNetSignalReconnect(traceId);
+        }
+
+        private void cleanupNetPending(
+            long traceId,
+            int error)
+        {
             final KafkaResetExFW kafkaResetEx = kafkaResetExRW.wrap(extBuffer, 0, extBuffer.capacity())
                 .typeId(kafkaTypeId)
                 .error(error)
@@ -2339,20 +2365,18 @@ private void cleanupNetPending(
             {
                 stream.cleanupApp(traceId, kafkaResetEx);
             }
+
+            doNetReset(traceId);
+            doNetAbort(traceId);
+
+            apiVersionRangeByApiKey.clear();
         }
 
         private void cleanupNet(
             long traceId)
         {
             doNetReset(traceId);
             doNetAbort(traceId);
-
-            // see onNetEnd: apiVersionRangeByApiKey survives an abortive
-            // disconnect the same way it survives an orderly one
-            saslResolved = false;
-
-            cleanupAppActive(traceId, EMPTY_OCTETS);
-            doNetSignalReconnect(traceId);
         }
 
         private void cleanupDecodeSlot()

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaState.java

@@ -81,6 +81,12 @@ static int openingReply(
         return state | REPLY_OPENING;
     }
 
+    static int openReply(
+        int state)
+    {
+        return state | REPLY_OPENED;
+    }
+
     static int openedReply(
         int state)
     {

runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/ClientOffsetCommitIT.java

@@ -15,6 +15,7 @@
  */
 package io.aklivity.zilla.runtime.binding.kafka.internal.stream;
 
+import static io.aklivity.zilla.runtime.binding.kafka.internal.KafkaConfigurationTest.KAFKA_CLIENT_API_VERSIONS_NAME;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.rules.RuleChain.outerRule;
 
@@ -28,6 +29,7 @@
 import io.aklivity.k3po.runtime.junit.rules.K3poRule;
 import io.aklivity.zilla.runtime.engine.test.EngineRule;
 import io.aklivity.zilla.runtime.engine.test.annotation.Configuration;
+import io.aklivity.zilla.runtime.engine.test.annotation.Configure;
 
 public class ClientOffsetCommitIT
 {
@@ -130,6 +132,22 @@ public void shouldFindCoordinatorOnly() throws Exception
         k3po.finish();
     }
 
+    // Same chaining as shouldChainFindCoordinatorThenDescribeGroups, but with api.versions
+    // negotiation enabled (the production default). Every other test in this class disables
+    // api.versions, so none of them exercise doEncodeRequest's apiVersionRangeByApiKey-driven
+    // branch for the second pooled request once apiVersionRangeByApiKey is already populated
+    // from the leading ApiVersions exchange - reproduces https://github.com/aklivity/zilla/issues/2532.
+    @Test
+    @Configuration("client.yaml")
+    @Configure(name = KAFKA_CLIENT_API_VERSIONS_NAME, value = "true")
+    @Specification({
+        "${app}/find.then.describe/client",
+        "${net}/find.then.describe.negotiated/server"})
+    public void shouldChainFindCoordinatorThenDescribeGroupsWithApiVersionsNegotiated() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("client.yaml")
     @Specification({

runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCreateTopicsIT.java

@@ -132,4 +132,16 @@ public void shouldReconnectWithoutReprobe() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Configuration("client.yaml")
+    @Configure(name = KAFKA_CLIENT_API_VERSIONS_NAME, value = "true")
+    @Configure(name = KAFKA_CLIENT_RECONNECT_DELAY_NAME, value = "0")
+    @Specification({
+        "${app}/create.topics.v7.idle.reset.reconnect/client",
+        "${net}/create.topics.v7.idle.reset.reconnect/server"})
+    public void shouldReconnectAfterIdleConnectionReset() throws Exception
+    {
+        k3po.finish();
+    }
 }