From 7e55acf9918d3fa5ecb9bf5d190f6d4fcbf678da Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 02:08:59 +0000 Subject: [PATCH 1/3] test(binding-kafka): cover bare offset-commit and chained api-request reuse Closes a gap toward diagnosing aklivity/zilla#2532 (mcp-kafka reset_offsets hangs against a Dead consumer group): binding-kafka had no coverage for KafkaClientOffsetCommitFactory's bare admin commit (generationId=-1, memberId="", instanceId="") or for chaining multiple apiRequest-kind streams (FindCoordinator, DescribeGroups) on one reused connection, which is exactly how binding-mcp-kafka's reset_offsets flow drives binding-kafka. Add ClientOffsetCommitIT scenarios exercising these paths directly against a live engine: - shouldCommitBareOffset: bare admin commit in isolation - shouldFindCoordinatorOnly / shouldChainFindCoordinatorThenDescribeGroups(NoWait): chained apiRequest-kind streams sharing one affinity/connection, both waiting for and not waiting for the prior stream's END before reopening - shouldResetOffsetsForDeadGroup: the full FindCoordinator -> DescribeGroups (Dead) -> bare OffsetCommit sequence end to end All pass, which rules out KafkaClientApiFactory/KafkaClientOffsetCommitFactory's state machines as the cause of the reported hang in the plaintext, no-SASL case. Root cause investigation continues upstream in binding-mcp-kafka. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Kk7te5cPXiBGfwnVH3cMiq --- .../internal/stream/ClientOffsetCommitIT.java | 52 +++++++ .../offset.commit/bare.commit/client.rpt | 45 ++++++ .../find.coordinator.only/client.rpt | 57 ++++++++ .../find.then.describe.no.wait/client.rpt | 107 ++++++++++++++ .../find.then.describe/client.rpt | 107 ++++++++++++++ .../reset.offsets.dead/client.rpt | 138 ++++++++++++++++++ .../offset.commit.v7/bare.commit/server.rpt | 65 +++++++++ .../find.coordinator.only/server.rpt | 49 +++++++ .../find.then.describe/server.rpt | 81 ++++++++++ .../reset.offsets.dead/server.rpt | 120 +++++++++++++++ 10 files changed, 821 insertions(+) create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/bare.commit/client.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.coordinator.only/client.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.then.describe.no.wait/client.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/find.then.describe/client.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/offset.commit/reset.offsets.dead/client.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/bare.commit/server.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.coordinator.only/server.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/find.then.describe/server.rpt create mode 100644 specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/offset.commit.v7/reset.offsets.dead/server.rpt 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/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 From 28a9fe536067f76893f99dd4f621fdf3c3c49dd2 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 02:38:54 +0000 Subject: [PATCH 2/3] fix(binding-mcp-kafka): stop reset_offsets hanging after a stale stage END Fixes aklivity/zilla#2532. Reproduced against a real Kafka broker (docker compose) with zilla dump/logs and an instrumented build: reset_offsets against a Dead (or Empty) consumer group correctly runs FindCoordinator then DescribeGroups, but the actual bare OffsetCommit result never reaches the MCP client - the broker commits successfully, yet the caller hangs until the MCP session's own inactivity timeout closes the connection. KafkaApiResetOffsetsClient.advanceToOffsetCommit() ends the just-finished DescribeGroups stream and opens the OffsetCommit stream synchronously, setting stage = STAGE_OFFSET_COMMIT before returning. The engine delivers the DescribeGroups stream's own reply-side END acknowledgment on a later turn, by which point onKafkaEnd() sees stage == STAGE_OFFSET_COMMIT and misattributes that stale END as the OffsetCommit stream's own completion, calling peer.doMcpEnd() and closing the MCP response before the real result is ever produced (the subsequent OffsetCommit request and its successful broker response are then orphaned). KafkaApiDescribeConsumerGroupLagClient.onKafkaEnd() already guards against exactly this hazard (comparing end.streamId() against the current kafkaReplyId) for its own two-stage OffsetFetch -> ListOffsets sequence; apply the same guard here. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Kk7te5cPXiBGfwnVH3cMiq --- .../internal/stream/McpKafkaProxyFactory.java | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) 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); + } } } From 8d73435db912bdb1adbdff23ba76dbaf2c6f3aca Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 03:10:29 +0000 Subject: [PATCH 3/3] test(binding-mcp-kafka): add end-to-end regression coverage for reset_offsets against a dead group Adds k3po scripts and IT coverage for the mcp-kafka -> kafka offset-commit handoff (find-coordinator -> describe-groups -> offset-commit) against a consumer group that has never had members, exercising the full 3-stage reset_offsets flow through a live engine (McpKafkaClientIT) as well as a peer-to-peer self-consistency check (KafkaIT), closing the gap that let the stale-END misattribution bug in KafkaApiResetOffsetsClient.onKafkaEnd ship undetected. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Kk7te5cPXiBGfwnVH3cMiq --- .../internal/stream/McpKafkaClientIT.java | 10 ++ .../kafka/offset.commit.for.reset/client.rpt | 49 +++++++ .../kafka/offset.commit.for.reset/server.rpt | 33 +++++ .../kafka/reset.offsets.dead/client.rpt | 138 ++++++++++++++++++ .../kafka/reset.offsets.dead/server.rpt | 123 ++++++++++++++++ .../streams/mcp/reset.offsets.dead/client.rpt | 64 ++++++++ .../binding/mcp/kafka/streams/KafkaIT.java | 18 +++ 7 files changed, 435 insertions(+) create mode 100644 specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/offset.commit.for.reset/client.rpt create mode 100644 specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/offset.commit.for.reset/server.rpt create mode 100644 specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/reset.offsets.dead/client.rpt create mode 100644 specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/kafka/reset.offsets.dead/server.rpt create mode 100644 specs/binding-mcp-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mcp/kafka/streams/mcp/reset.offsets.dead/client.rpt 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-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",