From 942e931cc342a76618df86b51220b01946451e57 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 15:43:45 +0000 Subject: [PATCH 1/4] fix(binding-kafka): reconnect KIND_API_REQUEST pool after idle teardown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit KafkaApiClient pools a single network connection per affinity across many independent Kafka API requests. When that connection was reset or aborted while idle (no request in flight, none queued), cleanup only OR'd the CLOSED bits into state without clearing the sticky OPENING/OPENED bits, so the next unrelated request's enqueue() saw a connection that looked both open and closed, took the "still open" branch, and wrote its request onto the dead stream. No BEGIN was ever resent and no response ever arrived, hanging the caller indefinitely. Each of onNetEnd/onNetAbort/onNetReset, and the internal cleanupNet/ cleanupNetPending teardown paths, now converge on a single onNetClosed() once both directions are confirmed closed: it fails the app-side backlog, releases decode/encode slots and budget, resets the connection's bookkeeping to a clean slate, and schedules the reconnect — restoring the invariant that a closed connection's state never lingers as "open" for a later caller to trip over. Fixes #2532 Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XJZqfbhycCM4bTy2u2Z7qc --- .../stream/KafkaClientApiFactory.java | 115 ++++++++------- .../internal/stream/KafkaCreateTopicsIT.java | 12 ++ .../client.rpt | 133 ++++++++++++++++++ .../client.rpt | 111 +++++++++++++++ .../server.rpt | 107 ++++++++++++++ .../streams/network/KafkaCreateTopicsIT.java | 9 ++ 6 files changed, 439 insertions(+), 48 deletions(-) create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/api/create.topics.v7.idle.reset.reconnect/client.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/api/create.topics.v7.idle.reset.reconnect/client.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/api/create.topics.v7.idle.reset.reconnect/server.rpt diff --git a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java b/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java index 0f9afb02ba..62f560accf 100644 --- a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java +++ b/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java @@ -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; @@ -2318,17 +2296,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,6 +2350,16 @@ private void cleanupNetPending( { stream.cleanupApp(traceId, kafkaResetEx); } + + doNetReset(traceId); + doNetAbort(traceId); + + apiVersionRangeByApiKey.clear(); + + if (KafkaState.closed(state)) + { + onNetClosed(traceId); + } } private void cleanupNet( @@ -2347,12 +2368,10 @@ private void cleanupNet( 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); + if (KafkaState.closed(state)) + { + onNetClosed(traceId); + } } private void cleanupDecodeSlot() diff --git a/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCreateTopicsIT.java b/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCreateTopicsIT.java index 94f4b634da..62c3472080 100644 --- a/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCreateTopicsIT.java +++ b/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(); + } } diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/api/create.topics.v7.idle.reset.reconnect/client.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/api/create.topics.v7.idle.reset.reconnect/client.rpt new file mode 100644 index 0000000000..925a0e944d --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/api/create.topics.v7.idle.reset.reconnect/client.rpt @@ -0,0 +1,133 @@ +# +# Copyright 2021-2026 Aklivity Inc. +# +# Aklivity 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. +# + +connect "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + option zilla:affinity 0 + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(0) + .apiName("api_versions") + .version(0) + .clientId("zilla") + .build() + .build()} + +connected + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(12) + .version(0) + .build() + .build()} + +read 0s # error code + 1 # api key count + 19s # create topics + 0s # min version + 7s # max version + +write close +read closed + +connect await CONNECTION_RESET + "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + option zilla:affinity 0 + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(110) + .apiName("create_topics") + .version(7) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # tagged fields + [0x03] # topic count + [0x07] "events" # name + 1 # number of partitions + 1s # replication factor + [0x02] # assignments + 0 # partition index + [0x02] # broker ids + 0 # broker id + [0x00] # tagged fields + [0x02] # configs + [0x0f] "cleanup.policy" # name + [0x07] "delete" # value + [0x00] # tagged fields + [0x00] # tagged fields + [0x0a] "snapshots" # name + 1 # number of partitions + 1s # replication factor + [0x02] # assignments + 0 # partition index + [0x02] # broker ids + 0 # broker id + [0x00] # tagged fields + [0x02] # configs + [0x0f] "cleanup.policy" # name + [0x08] "compact" # value + [0x00] # tagged fields + [0x00] # tagged fields + 0 # timeout + [0x00] # validate only + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(78) + .version(7) + .build() + .build()} + +read [0x00] # tagged fields + 0 # throttle time ms + [0x03] # topics + [0x07] "events" # name + [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] # topic id + 0s # error code + [0x00] # error message + 1 # number of partitions + 1s # replication factor + [0x01] # configs + [0x00] # tagged fields + [0x0a] "snapshots" # name + [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] # topic id + 0s # error code + [0x00] # error message + 1 # number of partitions + 1s # replication factor + [0x01] # configs + [0x00] # tagged fields + [0x00] # tagged fields + +write close +read closed diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/api/create.topics.v7.idle.reset.reconnect/client.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/api/create.topics.v7.idle.reset.reconnect/client.rpt new file mode 100644 index 0000000000..354143dbb3 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/api/create.topics.v7.idle.reset.reconnect/client.rpt @@ -0,0 +1,111 @@ +# +# Copyright 2021-2026 Aklivity Inc. +# +# Aklivity 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. +# + +property networkConnectWindow 8192 + +property apiVersionsRequestId ${kafka:newRequestId()} + +connect "zilla://streams/net0" + option zilla:window ${networkConnectWindow} + option zilla:transmission "duplex" + option zilla:byteorder "network" + +connected + +write 15 # size + 18s # api_versions + 0s # v0 + ${apiVersionsRequestId} + 5s "zilla" # client id + +read 16 # size + (int:apiVersionsRequestId) + 0s # error code + 1 # api key count + 19s # create topics + 0s # min version + 7s # max version + +read abort + +property createTopicsRequestId ${kafka:newRequestId()} + +connect "zilla://streams/net0" + option zilla:window ${networkConnectWindow} + option zilla:transmission "duplex" + option zilla:byteorder "network" + +connected + +write 125 # size + 19s # create topics + 7s # v7 + ${createTopicsRequestId} + 5s "zilla" # client id + [0x00] # tagged fields + [0x03] # topic count + [0x07] "events" # name + 1 # number of partitions + 1s # replication factor + [0x02] # assignments + 0 # partition index + [0x02] # broker ids + 0 # broker id + [0x00] # tagged fields + [0x02] # configs + [0x0f] "cleanup.policy" # name + [0x07] "delete" # value + [0x00] # tagged fields + [0x00] # tagged fields + [0x0a] "snapshots" # name + 1 # number of partitions + 1s # replication factor + [0x02] # assignments + 0 # partition index + [0x02] # broker ids + 0 # broker id + [0x00] # tagged fields + [0x02] # configs + [0x0f] "cleanup.policy" # name + [0x08] "compact" # value + [0x00] # tagged fields + [0x00] # tagged fields + 0 # timeout + [0x00] # validate only + [0x00] # tagged fields + +read 82 # size + (int:createTopicsRequestId) + [0x00] # tagged fields + 0 # throttle time ms + [0x03] # topics + [0x07] "events" # name + [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] # topic id + 0s # error code + [0x00] # error message + 1 # number of partitions + 1s # replication factor + [0x01] # configs + [0x00] # tagged fields + [0x0a] "snapshots" # name + [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] # topic id + 0s # error code + [0x00] # error message + 1 # number of partitions + 1s # replication factor + [0x01] # configs + [0x00] # tagged fields + [0x00] # tagged fields diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/api/create.topics.v7.idle.reset.reconnect/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/api/create.topics.v7.idle.reset.reconnect/server.rpt new file mode 100644 index 0000000000..37ba05efbc --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/api/create.topics.v7.idle.reset.reconnect/server.rpt @@ -0,0 +1,107 @@ +# +# Copyright 2021-2026 Aklivity Inc. +# +# Aklivity 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. +# + +property networkAcceptWindow 8192 + +accept "zilla://streams/net0" + option zilla:window ${networkAcceptWindow} + option zilla:transmission "duplex" + option zilla:byteorder "network" + +accepted + +connected + +read 15 # size + 18s # api_versions + 0s # v0 + (int:apiVersionsRequestId) + 5s "zilla" # client id + +write 16 # size + ${apiVersionsRequestId} + 0s # error code + 1 # api key count + 19s # create topics + 0s # min version + 7s # max version + +write abort +write notify CONNECTION_RESET + +accepted + +connected + +read 125 # size + 19s # create topics + 7s # v7 + (int:createTopicsRequestId) + 5s "zilla" # client id + [0x00] # tagged fields + [0x03] # topic count + [0x07] "events" # name + 1 # number of partitions + 1s # replication factor + [0x02] # assignments + 0 # partition index + [0x02] # broker ids + 0 # broker id + [0x00] # tagged fields + [0x02] # configs + [0x0f] "cleanup.policy" # name + [0x07] "delete" # value + [0x00] # tagged fields + [0x00] # tagged fields + [0x0a] "snapshots" # name + 1 # number of partitions + 1s # replication factor + [0x02] # assignments + 0 # partition index + [0x02] # broker ids + 0 # broker id + [0x00] # tagged fields + [0x02] # configs + [0x0f] "cleanup.policy" # name + [0x08] "compact" # value + [0x00] # tagged fields + [0x00] # tagged fields + 0 # timeout + [0x00] # validate only + [0x00] # tagged fields + +write 82 # size + ${createTopicsRequestId} + [0x00] # tagged fields + 0 # throttle time ms + [0x03] # topics + [0x07] "events" # name + [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] # topic id + 0s # error code + [0x00] # error message + 1 # number of partitions + 1s # replication factor + [0x01] # configs + [0x00] # tagged fields + [0x0a] "snapshots" # name + [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] # topic id + 0s # error code + [0x00] # error message + 1 # number of partitions + 1s # replication factor + [0x01] # configs + [0x00] # tagged fields + [0x00] # tagged fields diff --git a/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/network/KafkaCreateTopicsIT.java b/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/network/KafkaCreateTopicsIT.java index 53246725b1..4fcc4aaf32 100644 --- a/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/network/KafkaCreateTopicsIT.java +++ b/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/network/KafkaCreateTopicsIT.java @@ -81,4 +81,13 @@ public void shouldReconnectWithoutReprobe() throws Exception { k3po.finish(); } + + @Test + @Specification({ + "${net}/create.topics.v7.idle.reset.reconnect/client", + "${net}/create.topics.v7.idle.reset.reconnect/server"}) + public void shouldReconnectAfterIdleConnectionReset() throws Exception + { + k3po.finish(); + } } From 4c5a4988eab784cd26b0602f69910347f8518db7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 17:57:26 +0000 Subject: [PATCH 2/4] fix(binding-kafka): also guard onNetClosed inside doNetEnd/Abort/Reset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit doNetEnd/doNetAbort/doNetReset are always the last mutation for whichever direction they close, and any preceding direct state assignment in a caller already happened before they run — so checking KafkaState.closed(state) at the end of each of them is safe for every caller, not just the three onNetXxx handlers that already check explicitly. This covers cleanupNet() and cleanupNetPending() (the internal decode-error/SASL-failure teardown paths), which previously needed their own trailing check to get the same guarantee and no longer do. The onNetXxx handlers keep their own explicit check too, so onNetClosed() is evaluated twice on that path (once inside doNetEnd/Abort/Reset, once in the caller) — harmless, since the first evaluation resets state to 0, making the second trivially false. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XJZqfbhycCM4bTy2u2Z7qc --- .../stream/KafkaClientApiFactory.java | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java b/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java index 62f560accf..d5838fd72e 100644 --- a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java +++ b/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java @@ -1873,6 +1873,11 @@ private void doNetEnd( cleanupBudget(); deauthorizeGuardSession(); + + if (KafkaState.closed(state)) + { + onNetClosed(traceId); + } } private void doNetAbort( @@ -1890,6 +1895,11 @@ private void doNetAbort( cleanupBudget(); deauthorizeGuardSession(); + + if (KafkaState.closed(state)) + { + onNetClosed(traceId); + } } private void doNetReset( @@ -1906,6 +1916,11 @@ private void doNetReset( cleanupDecodeSlot(); deauthorizeGuardSession(); + + if (KafkaState.closed(state)) + { + onNetClosed(traceId); + } } private void deauthorizeGuardSession() @@ -2355,11 +2370,6 @@ private void cleanupNetPending( doNetAbort(traceId); apiVersionRangeByApiKey.clear(); - - if (KafkaState.closed(state)) - { - onNetClosed(traceId); - } } private void cleanupNet( @@ -2367,11 +2377,6 @@ private void cleanupNet( { doNetReset(traceId); doNetAbort(traceId); - - if (KafkaState.closed(state)) - { - onNetClosed(traceId); - } } private void cleanupDecodeSlot() From c95edc093fbd35e9e1bd8b863bc80ed1b89c0c49 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 18:27:46 +0000 Subject: [PATCH 3/4] test(binding-kafka.spec): add missing app-layer peer script for idle-reset-reconnect The new create.topics.v7.idle.reset.reconnect scenario had a client.rpt under streams/application/api/ but no matching server.rpt or peer-to-peer IT method, leaving the pairing/self-consistency check specs/AGENTS.md requires for every scenario unmet (an existing sibling scenario, create.topics.v7.reconnect.no.probe, carries the same pre-existing gap, left alone here as out of scope). The app-layer server.rpt also emits the CONNECTION_RESET notify the shared client.rpt awaits before its second connect, since notify/await barriers are session-wide: in the runtime IT that signal comes from the network script after the simulated idle-connection reset, but the pure app-layer peer test has no network side, so the server script provides it instead to unblock the second exchange. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XJZqfbhycCM4bTy2u2Z7qc --- .../server.rpt | 137 ++++++++++++++++++ .../application/KafkaCreateTopicsIT.java | 9 ++ 2 files changed, 146 insertions(+) create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/api/create.topics.v7.idle.reset.reconnect/server.rpt diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/api/create.topics.v7.idle.reset.reconnect/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/api/create.topics.v7.idle.reset.reconnect/server.rpt new file mode 100644 index 0000000000..a722b57461 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/api/create.topics.v7.idle.reset.reconnect/server.rpt @@ -0,0 +1,137 @@ +# +# Copyright 2021-2026 Aklivity Inc. +# +# Aklivity 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. +# + +property serverAddress "zilla://streams/app0" + +accept ${serverAddress} + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +accepted + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(0) + .api(18) + .version(0) + .clientId("zilla") + .build() + .build()} + +connected + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(12) + .version(0) + .build() + .build()} + +write 0s # error code + 1 # api key count + 19s # create topics + 0s # min version + 7s # max version + +write flush + +read closed +write close + +write notify CONNECTION_RESET + +accepted + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(110) + .api(19) + .version(7) + .clientId("zilla") + .build() + .build()} + +connected + +read [0x00] # tagged fields + [0x03] # topic count + [0x07] "events" # name + 1 # number of partitions + 1s # replication factor + [0x02] # assignments + 0 # partition index + [0x02] # broker ids + 0 # broker id + [0x00] # tagged fields + [0x02] # configs + [0x0f] "cleanup.policy" # name + [0x07] "delete" # value + [0x00] # tagged fields + [0x00] # tagged fields + [0x0a] "snapshots" # name + 1 # number of partitions + 1s # replication factor + [0x02] # assignments + 0 # partition index + [0x02] # broker ids + 0 # broker id + [0x00] # tagged fields + [0x02] # configs + [0x0f] "cleanup.policy" # name + [0x08] "compact" # value + [0x00] # tagged fields + [0x00] # tagged fields + 0 # timeout + [0x00] # validate only + [0x00] # tagged fields + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(78) + .version(7) + .build() + .build()} + +write [0x00] # tagged fields + 0 # throttle time ms + [0x03] # topics + [0x07] "events" # name + [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] # topic id + 0s # error code + [0x00] # error message + 1 # number of partitions + 1s # replication factor + [0x01] # configs + [0x00] # tagged fields + [0x0a] "snapshots" # name + [0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00 0x00] # topic id + 0s # error code + [0x00] # error message + 1 # number of partitions + 1s # replication factor + [0x01] # configs + [0x00] # tagged fields + [0x00] # tagged fields + +write flush + +read closed +write close diff --git a/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/KafkaCreateTopicsIT.java b/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/KafkaCreateTopicsIT.java index 1f505a2ad8..9432e7c2c4 100644 --- a/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/KafkaCreateTopicsIT.java +++ b/specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/KafkaCreateTopicsIT.java @@ -54,4 +54,13 @@ public void shouldRejectCreateTopicsV3WhenUnsupported() throws Exception { k3po.finish(); } + + @Test + @Specification({ + "${app}/create.topics.v7.idle.reset.reconnect/client", + "${app}/create.topics.v7.idle.reset.reconnect/server"}) + public void shouldReconnectAfterIdleConnectionReset() throws Exception + { + k3po.finish(); + } } From a9d682e56c981c12254e3a2ae0df6b08170aad53 Mon Sep 17 00:00:00 2001 From: John Fallows Date: Fri, 4 Sep 2026 14:30:20 -0700 Subject: [PATCH 4/4] fix(binding-kafka): stop premature reply-window credit from suppressing apiRequest BEGIN KafkaApiStream.onAppWindow marked the reply as both opening and opened via KafkaState.openedReply(state) whenever it received a reply-window grant, on the assumption a window could only arrive after this stream's own doAppBegin had already sent BEGIN. A proactive reply-window grant (originating from the http server binding's early-window optimization and propagating down through mcp/mcp-kafka/kafka) can now arrive before the real response does, so REPLY_OPENING was getting set before doAppBegin ever ran. doAppBegin's `if (!KafkaState.replyOpening(state))` guard then saw the bit already set and silently skipped sending BEGIN, while the response DATA/END still went out - leaving reset_offsets (and any other multi-stage apiRequest chain, e.g. FindCoordinator -> DescribeGroups) hung forever waiting for a BEGIN that would never come. Adds KafkaState.openReply(state), setting REPLY_OPENED alone, matching the independent opening/opened bit pattern HttpState already uses (see HttpClientFactory/HttpServerFactory). onAppWindow now uses openReply instead of the combined openedReply, so a window grant only ever records flow-control progress and never implies BEGIN was sent - doAppBegin remains the sole setter of REPLY_OPENING. Also adds a network-level regression test exercising the FindCoordinator -> DescribeGroups chain with api.versions negotiation enabled (the production default) rather than disabled, since every existing test in this class ran with api.versions off and so never exercised this pooled-connection request-dispatch path. Fixes #2532 Co-Authored-By: Claude Sonnet 5 --- .../stream/KafkaClientApiFactory.java | 2 +- .../kafka/internal/stream/KafkaState.java | 6 ++ .../internal/stream/ClientOffsetCommitIT.java | 18 ++++ .../find.then.describe.negotiated/server.rpt | 100 ++++++++++++++++++ 4 files changed, 125 insertions(+), 1 deletion(-) create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.then.describe.negotiated/server.rpt diff --git a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java b/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java index d5838fd72e..140ac2ec6d 100644 --- a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientApiFactory.java +++ b/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; diff --git a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaState.java b/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaState.java index 4bc9c48ee5..bf5c8c0c16 100644 --- a/runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaState.java +++ b/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) { diff --git a/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/ClientOffsetCommitIT.java b/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/ClientOffsetCommitIT.java index bad1c434a1..50ca5ba658 100644 --- a/runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/ClientOffsetCommitIT.java +++ b/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({ diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.then.describe.negotiated/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.then.describe.negotiated/server.rpt new file mode 100644 index 0000000000..5eb77df07c --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.then.describe.negotiated/server.rpt @@ -0,0 +1,100 @@ +# +# Copyright 2021-2026 Aklivity Inc. +# +# Aklivity 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. +# + +# Same find-then-describe chaining as find.then.describe/server.rpt, but with +# api.versions negotiation enabled (the production default), reproducing the +# find.then.describe pipelining over the single pooled KafkaApiClient +# connection while apiVersionRangeByApiKey is already populated from the +# leading ApiVersions exchange. + +property networkAcceptWindow 8192 + +accept "zilla://streams/net0" + option zilla:window ${networkAcceptWindow} + option zilla:transmission "duplex" + option zilla:byteorder "network" + +accepted + +connected + +read 15 # size + 18s # api_versions + 0s # v0 + (int:apiVersionsRequestId) + 5s "zilla" # client id + +write 22 # size + ${apiVersionsRequestId} + 0s # error code + 2 # api key count + 10s # find coordinator + 0s # min version + 3s # max version + 15s # describe groups + 0s # min version + 5s # max version + +read 27 # size + 10s # find coordinator + 3s # v3 + (int:newRequestId1) + 5s "zilla" # client id + [0x00] # request header tagged fields + [0x09] "my-group" # group id (key) + [0x00] # key type (group) + [0x00] # tagged fields + +write 29 # size + ${newRequestId1} + [0x00] # tagged fields + 0 # throttle time ms + 0s # error code + [0x00] # message (null) + 1 # node id + [0x08] "broker1" # host + 9092 # port + [0x00] # tagged fields + +write flush + +read 28 # size + 15s # describe groups + 5s # v5 + (int:newRequestId2) + 5s "zilla" # client id + [0x00] # request header tagged fields + [0x02] # group count + [0x09] "my-group" # group id + [0x00] # include authorized operations + [0x00] # tagged fields + +write 35 # size + ${newRequestId2} + [0x00] # tagged fields + 0 # throttle time ms + [0x02] # groups + 0s # error code + [0x09] "my-group" # group id + [0x05] "Dead" # group state (never had members) + [0x01] # protocol type (empty) + [0x01] # protocol data (empty) + [0x01] # members (none) + [0xff 0xff 0xff 0xff] # authorized operations (-1) + [0x00] # group tagged fields + [0x00] # tagged fields + +write flush