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 0057ef75a38..bad1c434a12 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 @@ -41,6 +41,8 @@ public class ClientOffsetCommitIT .directory("target/zilla-itests") .countersBufferCapacity(8192) .configurationRoot("io/aklivity/zilla/specs/binding/kafka/config") + .configure("zilla.binding.kafka.client.connection.pool", "false") + .configure("zilla.binding.kafka.client.api.versions", "false") .external("net0") .clean(); @@ -87,4 +89,54 @@ public void shouldHandleOffsetCommitError() throws Exception { k3po.finish(); } + + @Test + @Configuration("client.yaml") + @Specification({ + "${app}/bare.commit/client", + "${net}/bare.commit/server"}) + public void shouldCommitBareOffset() throws Exception + { + k3po.finish(); + } + + @Test + @Configuration("client.yaml") + @Specification({ + "${app}/reset.offsets.dead/client", + "${net}/reset.offsets.dead/server"}) + public void shouldResetOffsetsForDeadGroup() throws Exception + { + k3po.finish(); + } + + @Test + @Configuration("client.yaml") + @Specification({ + "${app}/find.then.describe/client", + "${net}/find.then.describe/server"}) + public void shouldChainFindCoordinatorThenDescribeGroups() throws Exception + { + k3po.finish(); + } + + @Test + @Configuration("client.yaml") + @Specification({ + "${app}/find.coordinator.only/client", + "${net}/find.coordinator.only/server"}) + public void shouldFindCoordinatorOnly() throws Exception + { + k3po.finish(); + } + + @Test + @Configuration("client.yaml") + @Specification({ + "${app}/find.then.describe.no.wait/client", + "${net}/find.then.describe/server"}) + public void shouldChainFindCoordinatorThenDescribeGroupsNoWait() throws Exception + { + k3po.finish(); + } } diff --git a/runtime/binding-mcp-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactory.java b/runtime/binding-mcp-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactory.java index 4a775c47232..655c9d17570 100644 --- a/runtime/binding-mcp-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactory.java +++ b/runtime/binding-mcp-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaProxyFactory.java @@ -9041,16 +9041,23 @@ private void advanceToOffsetCommit( traceId, authorization, affinity, kafkaBeginEx); } + /** + * Only the reply of the stage currently in flight closes the MCP reply; the previous stage's + * reply END arrives after {@link #advanceToOffsetCommit} has already opened the next stream. + */ private void onKafkaEnd( EndFW end) { final long traceId = end.traceId(); - state = McpKafkaState.closedReply(state); - - if (stage == STAGE_OFFSET_COMMIT && mcpBegun) + if (end.streamId() == kafkaReplyId) { - peer.doMcpEnd(traceId); + state = McpKafkaState.closedReply(state); + + if (stage == STAGE_OFFSET_COMMIT && mcpBegun) + { + peer.doMcpEnd(traceId); + } } } diff --git a/runtime/binding-mcp-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaClientIT.java b/runtime/binding-mcp-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaClientIT.java index 8727e2c1603..0c740f27c85 100644 --- a/runtime/binding-mcp-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaClientIT.java +++ b/runtime/binding-mcp-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mcp/kafka/internal/stream/McpKafkaClientIT.java @@ -365,6 +365,16 @@ public void shouldResetOffsetsCoordinatorNotFound() throws Exception k3po.finish(); } + @Test + @Configuration("client.reset.offsets.yaml") + @Specification({ + "${mcp}/reset.offsets.dead/client", + "${kafka}/reset.offsets.dead/server"}) + public void shouldResetOffsetsForDeadGroup() throws Exception + { + k3po.finish(); + } + @Test @Configuration("client.describe.consumer.group.lag.yaml") @Specification({ diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/bare.commit/client.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/bare.commit/client.rpt new file mode 100644 index 00000000000..d8f0e1dd7dd --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/bare.commit/client.rpt @@ -0,0 +1,45 @@ +# +# 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" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .offsetCommit() + .groupId("orders-analytics") + .memberId("") + .instanceId("") + .host("broker1.example.com") + .port(9092) + .build() + .build()} + +connected + +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .offsetCommit() + .topic("orders") + .progress(0, 100, "") + .generationId(-1) + .leaderEpoch(-1) + .build() + .build()} + +write zilla:data.empty +write flush diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.coordinator.only/client.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.coordinator.only/client.rpt new file mode 100644 index 00000000000..adf29651e0c --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.coordinator.only/client.rpt @@ -0,0 +1,57 @@ +# +# 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" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(12) + .api(10) + .version(3) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x09] "my-group" # group id (key) + [0x00] # key type (group) + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(25) + .version(3) + .build() + .build()} + +read [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 close +read closed diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.then.describe.no.wait/client.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.then.describe.no.wait/client.rpt new file mode 100644 index 00000000000..daacce9c6b9 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.then.describe.no.wait/client.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. +# + +# Mirrors McpKafkaProxyFactory's KafkaApiResetOffsetsClient: it ends the +# FindCoordinator stream and opens the DescribeGroups stream in the same +# synchronous call, without waiting for the END to be acknowledged first. + +connect "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(12) + .api(10) + .version(3) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x09] "my-group" # group id (key) + [0x00] # key type (group) + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(25) + .version(3) + .build() + .build()} + +read [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 close + +connect "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(13) + .api(15) + .version(5) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x02] # group count + [0x09] "my-group" # group id + [0x00] # include authorized operations + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(31) + .version(5) + .build() + .build()} + +read [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 close +read closed diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.then.describe/client.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.then.describe/client.rpt new file mode 100644 index 00000000000..48a7d8fbff4 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.then.describe/client.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. +# + +connect "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(12) + .api(10) + .version(3) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x09] "my-group" # group id (key) + [0x00] # key type (group) + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(25) + .version(3) + .build() + .build()} + +read [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 close +read closed + +read notify FIND_COORDINATOR_DONE + +connect await FIND_COORDINATOR_DONE + "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(13) + .api(15) + .version(5) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x02] # group count + [0x09] "my-group" # group id + [0x00] # include authorized operations + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(31) + .version(5) + .build() + .build()} + +read [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 close +read closed diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/reset.offsets.dead/client.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/reset.offsets.dead/client.rpt new file mode 100644 index 00000000000..ac67357bfe9 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/reset.offsets.dead/client.rpt @@ -0,0 +1,138 @@ +# +# 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. +# + +# Reproduces the reset_offsets flow driven by binding-mcp-kafka: +# FindCoordinator, then DescribeGroups (group state Dead), then a bare +# offsetCommit (generationId=-1, memberId="") against the coordinator. + +connect "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(12) + .api(10) + .version(3) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x09] "my-group" # group id (key) + [0x00] # key type (group) + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(25) + .version(3) + .build() + .build()} + +read [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 close +read closed + +connect "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(13) + .api(15) + .version(5) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x02] # group count + [0x09] "my-group" # group id + [0x00] # include authorized operations + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(31) + .version(5) + .build() + .build()} + +read [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 close +read closed + +connect "zilla://streams/app0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .offsetCommit() + .groupId("my-group") + .memberId("") + .instanceId("") + .host("broker1") + .port(9092) + .build() + .build()} + +connected + +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .offsetCommit() + .topic("orders") + .progress(0, 100, "") + .generationId(-1) + .leaderEpoch(-1) + .build() + .build()} + +write zilla:data.empty +write flush diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/bare.commit/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/bare.commit/server.rpt new file mode 100644 index 00000000000..03be7527d59 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/bare.commit/server.rpt @@ -0,0 +1,65 @@ +# +# 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 + +read zilla:begin.ext ${proxy:matchBeginEx() + .typeId(zilla:id("proxy")) + .addressInet() + .protocol("stream") + .source("0.0.0.0") + .destination("broker1.example.com") + .sourcePort(0) + .destinationPort(9092) + .build() + .info() + .authority("broker1.example.com") + .build() + .build()} +connected + +read 75 # size + 8s # offset commit + 7s # 7 + (int:newRequestId) + 5s "zilla" # client id + 16s "orders-analytics" # group id + -1 # generation id (bare admin commit) + 0s # member id (empty) + 0s # group instance id (empty) + 1 # topics + 6s "orders" # "orders" topic + 1 # partitions + 0 # partition 0 + 100L # committed offset + -1 # committed leader epoch + 0s # metadata (empty) + +write 30 # size + ${newRequestId} + 0 # throttle time ms + 1 # topics + 6s "orders" # "orders" topic + 1 # partitions + 0 # partition index + 0s # no error diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.coordinator.only/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.coordinator.only/server.rpt new file mode 100644 index 00000000000..d76b2b549da --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.coordinator.only/server.rpt @@ -0,0 +1,49 @@ +# +# 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 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 diff --git a/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.then.describe/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.then.describe/server.rpt new file mode 100644 index 00000000000..88b143a0593 --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.then.describe/server.rpt @@ -0,0 +1,81 @@ +# +# 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. +# + +# The underlying network connection is shared/reused by KafkaClientApiFactory +# across sequential apiRequest-kind streams (one KafkaApiClient per affinity), +# so both requests arrive pipelined on the same accepted connection. + +property networkAcceptWindow 8192 + +accept "zilla://streams/net0" + option zilla:window ${networkAcceptWindow} + option zilla:transmission "duplex" + option zilla:byteorder "network" + +accepted + +connected + +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/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/reset.offsets.dead/server.rpt b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/reset.offsets.dead/server.rpt new file mode 100644 index 00000000000..658fdfa425c --- /dev/null +++ b/specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/reset.offsets.dead/server.rpt @@ -0,0 +1,120 @@ +# +# 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 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 + +accepted + +read zilla:begin.ext ${proxy:matchBeginEx() + .typeId(zilla:id("proxy")) + .addressInet() + .protocol("stream") + .source("0.0.0.0") + .destination("broker1") + .sourcePort(0) + .destinationPort(9092) + .build() + .info() + .authority("broker1") + .build() + .build()} +connected + +read 67 # size + 8s # offset commit + 7s # 7 + (int:newRequestId3) + 5s "zilla" # client id + 8s "my-group" # group id + -1 # generation id (bare admin commit) + 0s # member id (empty) + 0s # group instance id (empty) + 1 # topics + 6s "orders" # "orders" topic + 1 # partitions + 0 # partition 0 + 100L # committed offset + -1 # committed leader epoch + 0s # metadata (empty) + +write 30 # size + ${newRequestId3} + 0 # throttle time ms + 1 # topics + 6s "orders" # "orders" topic + 1 # partitions + 0 # partition index + 0s # no error diff --git a/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/offset.commit.for.reset/client.rpt b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/offset.commit.for.reset/client.rpt new file mode 100644 index 00000000000..83cdd8284e7 --- /dev/null +++ b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/offset.commit.for.reset/client.rpt @@ -0,0 +1,49 @@ +# +# Copyright 2021-2026 Aklivity Inc +# +# Licensed under the Aklivity Community License (the "License"); you may not use +# this file except in compliance with the License. You may obtain a copy of the +# License at +# +# https://www.aklivity.io/aklivity-community-license/ +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. +# + +connect "zilla://streams/kafka0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .offsetCommit() + .groupId("my-group") + .memberId("") + .instanceId("") + .host("broker1") + .port(9092) + .build() + .build()} + +connected + +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .offsetCommit() + .topic("my-topic") + .progress(0, 100, "") + .generationId(-1) + .leaderEpoch(-1) + .build() + .build()} + +write zilla:data.empty +write flush + +write advise zilla:flush + +read closed +write close diff --git a/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/offset.commit.for.reset/server.rpt b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/offset.commit.for.reset/server.rpt new file mode 100644 index 00000000000..1c0cee149a7 --- /dev/null +++ b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/offset.commit.for.reset/server.rpt @@ -0,0 +1,33 @@ +# +# Copyright 2021-2026 Aklivity Inc +# +# Licensed under the Aklivity Community License (the "License"); you may not use +# this file except in compliance with the License. You may obtain a copy of the +# License at +# +# https://www.aklivity.io/aklivity-community-license/ +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. +# + +accept "zilla://streams/kafka0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + +accepted + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .build()} + +connected + +read zilla:data.empty + +read advised zilla:flush + +write close +read closed diff --git a/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/reset.offsets.dead/client.rpt b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/reset.offsets.dead/client.rpt new file mode 100644 index 00000000000..a61b2a67e71 --- /dev/null +++ b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/reset.offsets.dead/client.rpt @@ -0,0 +1,138 @@ +# +# Copyright 2021-2026 Aklivity Inc +# +# Licensed under the Aklivity Community License (the "License"); you may not use +# this file except in compliance with the License. You may obtain a copy of the +# License at +# +# https://www.aklivity.io/aklivity-community-license/ +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. +# + +connect "zilla://streams/kafka0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(12) + .api(10) + .version(3) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x09] "my-group" # group id (key) + [0x00] # key type (group) + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(25) + .version(3) + .build() + .build()} + +read [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 close +read closed + +connect "zilla://streams/kafka0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:byteorder "network" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(13) + .api(15) + .version(5) + .clientId("zilla") + .build() + .build()} + +connected + +write [0x00] # request header tagged fields + [0x02] # group count + [0x09] "my-group" # group id + [0x00] # include authorized operations + [0x00] # tagged fields + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(31) + .version(5) + .build() + .build()} + +read [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 close +read closed + +connect "zilla://streams/kafka0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .offsetCommit() + .groupId("my-group") + .memberId("") + .instanceId("") + .host("broker1") + .port(9092) + .build() + .build()} + +connected + +write zilla:data.ext ${kafka:dataEx() + .typeId(zilla:id("kafka")) + .offsetCommit() + .topic("my-topic") + .progress(0, 100, "") + .generationId(-1) + .leaderEpoch(-1) + .build() + .build()} + +write zilla:data.empty +write flush + +write advise zilla:flush + +read closed +write close diff --git a/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/reset.offsets.dead/server.rpt b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/reset.offsets.dead/server.rpt new file mode 100644 index 00000000000..0c298ff798d --- /dev/null +++ b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/reset.offsets.dead/server.rpt @@ -0,0 +1,123 @@ +# +# Copyright 2021-2026 Aklivity Inc +# +# Licensed under the Aklivity Community License (the "License"); you may not use +# this file except in compliance with the License. You may obtain a copy of the +# License at +# +# https://www.aklivity.io/aklivity-community-license/ +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES 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/kafka0" + +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(12) + .api(10) + .version(3) + .clientId("zilla") + .build() + .build()} + +connected + +read [0x00] # request header tagged fields + [0x09] "my-group" # group id (key) + [0x00] # key type (group) + [0x00] # tagged fields + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(25) + .version(3) + .build() + .build()} + +write [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 closed +write close + +accepted + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .apiRequest() + .length(13) + .api(15) + .version(5) + .clientId("zilla") + .build() + .build()} + +connected + +read [0x00] # request header tagged fields + [0x02] # group count + [0x09] "my-group" # group id + [0x00] # include authorized operations + [0x00] # tagged fields + +write zilla:begin.ext ${kafka:beginEx() + .typeId(zilla:id("kafka")) + .apiResponse() + .length(31) + .version(5) + .build() + .build()} + +write [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 + +read closed +write close + +accepted + +read zilla:begin.ext ${kafka:matchBeginEx() + .typeId(zilla:id("kafka")) + .build()} + +connected + +read zilla:data.empty + +read advised zilla:flush + +write close +read closed diff --git a/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/mcp/reset.offsets.dead/client.rpt b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/mcp/reset.offsets.dead/client.rpt new file mode 100644 index 00000000000..8328344e0f0 --- /dev/null +++ b/specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/mcp/reset.offsets.dead/client.rpt @@ -0,0 +1,64 @@ +# +# Copyright 2021-2026 Aklivity Inc +# +# Licensed under the Aklivity Community License (the "License"); you may not use +# this file except in compliance with the License. You may obtain a copy of the +# License at +# +# https://www.aklivity.io/aklivity-community-license/ +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OF ANY KIND, either express or implied. See the License for the +# specific language governing permissions and limitations under the License. +# + +property authorization 0L + +connect "zilla://streams/mcp0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:authorization ${authorization} + +write zilla:begin.ext ${mcp:beginEx() + .typeId(zilla:id("mcp")) + .lifecycle() + .build() + .build()} + +connected + +read zilla:begin.ext ${mcp:matchBeginEx() + .typeId(zilla:id("mcp")) + .lifecycle() + .sessionId("5ca1ab1e-c0de-4a11-5e55-000100000000") + .build() + .build()} + +read notify LIFECYCLE_INITIALIZED + +connect await LIFECYCLE_INITIALIZED + "zilla://streams/mcp0" + option zilla:window 8192 + option zilla:transmission "half-duplex" + option zilla:authorization ${authorization} + +write zilla:begin.ext ${mcp:beginEx() + .typeId(zilla:id("mcp")) + .toolsCall() + .sessionId("5ca1ab1e-c0de-4a11-5e55-000100000000") + .name("reset_offsets") + .contentLength(106) + .build() + .build()} + +connected + +write '{"name":"reset_offsets","arguments":{"group_id":"my-group","topic":"my-topic","partition":0,"offset":100}}' + +read '{"structuredContent":{"group_id":"my-group","topic":"my-topic","partition":0,"offset":100,"reset":true},' + '"content":[{"type":"text","text":"Reset offset for group my-group topic my-topic partition 0 to 100"}],' + '"isError":false}' +read closed + +write close diff --git a/specs/binding-mcp-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mcp/kafka/streams/KafkaIT.java b/specs/binding-mcp-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mcp/kafka/streams/KafkaIT.java index 51cb543cd8b..a48a9453934 100644 --- a/specs/binding-mcp-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mcp/kafka/streams/KafkaIT.java +++ b/specs/binding-mcp-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mcp/kafka/streams/KafkaIT.java @@ -378,6 +378,24 @@ public void shouldDescribeGroupsForReset() throws Exception k3po.finish(); } + @Test + @Specification({ + "${kafka}/offset.commit.for.reset/client", + "${kafka}/offset.commit.for.reset/server"}) + public void shouldCommitOffsetForReset() throws Exception + { + k3po.finish(); + } + + @Test + @Specification({ + "${kafka}/reset.offsets.dead/client", + "${kafka}/reset.offsets.dead/server"}) + public void shouldResetOffsetsForDeadGroup() throws Exception + { + k3po.finish(); + } + @Test @Specification({ "${kafka}/find.coordinator.error/client",