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..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; @@ -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,6 +2365,11 @@ private void cleanupNetPending( { stream.cleanupApp(traceId, kafkaResetEx); } + + doNetReset(traceId); + doNetAbort(traceId); + + apiVersionRangeByApiKey.clear(); } private void cleanupNet( @@ -2346,13 +2377,6 @@ 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); } private void cleanupDecodeSlot() 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/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/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/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/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 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(); + } } 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(); + } }