You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/10/15 15:58:14 UTC
(iggy) branch kafka-empty-groups updated (6068db0b4 -> 3cd1923ef)
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a change to branch kafka-empty-groups
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 6068db0b4 feat(gateways): report groups that only hold offsets as Empty
add b669d9669 fix(gateways): stop reporting cleanup.policy and accept -1 defaults (#4534)
add b815165c3 fix(ci): use an approved setup-uv revision (#4540)
add b90afc2ac chore(deps): Bump the csharp group with 4 updates (#4536)
add c5152a500 feat(server)!: route partition requests to their primaries (#4506)
add e9e5e10a3 chore(deps): upgrade Rust dependencies (#4542)
add f0d10d46b feat(swift): add the TCP and TLS client (#4503)
add b1166a590 fix(sdk): stop a reconnect sweep when the client shuts down (#4507)
add e06fb351d fix(csharp): report refusal when routed budget expires before resend (#4547)
add c70c286cd chore(deps): Bump the github-actions group with 2 updates (#4549)
new 3cd1923ef feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (6068db0b4)
\
N -- N -- N refs/heads/kafka-empty-groups (3cd1923ef)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../actions/python-maturin/pre-merge/action.yml | 2 +-
.github/workflows/_common.yml | 4 +-
.github/workflows/coverage-baseline.yml | 4 +-
Cargo.lock | 2480 ++--
Cargo.toml | 98 +-
bdd/go/tests/tcp_test/messages_feature_send.go | 4 +-
bdd/go/tests/tcp_test/offset_feature_delete.go | 2 +-
bdd/rust/tests/steps/leader_redirection.rs | 3 +-
core/ai/mcp/src/service/mod.rs | 6 +-
core/bench/dashboard/server/src/github/client.rs | 2 +-
core/bench/src/analytics/report_builder.rs | 13 +-
core/binary_protocol/src/codes.rs | 6 +-
core/binary_protocol/src/consensus/command.rs | 11 +-
core/binary_protocol/src/consensus/header.rs | 426 +-
core/binary_protocol/src/consensus/mod.rs | 7 +-
core/binary_protocol/src/consensus/request.rs | 19 +-
core/binary_protocol/src/dispatch.rs | 4 +-
core/binary_protocol/src/lib.rs | 20 +-
core/binary_protocol/src/primitives/identifier.rs | 3 +
.../src/primitives/partition_history.rs | 7 +-
...{get_send_context.rs => get_partition_route.rs} | 22 +-
core/binary_protocol/src/requests/messages/mod.rs | 4 +-
.../partitions/retire_consumer_group_owners.rs | 8 +-
core/binary_protocol/src/responses/messages/mod.rs | 2 +
.../src/responses/messages/partition_route.rs | 157 +
.../src/responses/messages/poll_routing.rs | 1 +
.../src/responses/system/get_cluster_metadata.rs | 45 +-
.../binary_cluster/get_cluster_metadata.rs | 7 +-
core/common/src/error/iggy_error.rs | 81 +-
core/common/src/lib.rs | 4 +-
core/common/src/traits/binary_impls/messages.rs | 19 +-
core/common/src/traits/binary_impls/mod.rs | 58 +-
.../traits/binary_impls/personal_access_tokens.rs | 39 +-
core/common/src/traits/binary_impls/users.rs | 39 +-
core/common/src/traits/binary_transport.rs | 71 +-
core/common/src/traits/client.rs | 28 +
core/common/src/traits/mod.rs | 2 +-
core/common/src/types/cluster/node.rs | 8 +-
core/common/src/wire_conversions.rs | 20 +
core/configs/src/server_config/cluster.rs | 11 +-
core/connectors/sinks/rabbitmq_sink/src/lib.rs | 42 +-
core/consensus/src/client_table.rs | 15 +-
core/consensus/src/impls.rs | 232 +-
core/consensus/src/lib.rs | 7 +
core/consensus/src/plane_helpers.rs | 285 +-
core/integration/src/harness/handle/mcp.rs | 4 +-
.../cli/system/test_cluster_metadata_command.rs | 4 +-
.../tests/cluster/client_table_adversarial.rs | 1 +
.../tests/cluster/client_table_restart.rs | 1 +
core/integration/tests/cluster/partition_dedup.rs | 1 +
.../tests/cluster/partition_primary_routing.rs | 570 +-
.../tests/cluster/partition_state_transfer.rs | 100 +-
.../connectors/fixtures/rabbitmq/container.rs | 14 +-
.../fixtures/redshift/redshift_mock/handler.rs | 4 +-
.../fixtures/redshift/redshift_mock/parser.rs | 8 +-
core/integration/tests/connectors/s3/s3_sink.rs | 1 +
.../verify_consumer_group_partition_assignment.rs | 89 +-
.../tests/sdk/consumer_group_membership.rs | 15 +-
core/integration/tests/server/http_client.rs | 17 +
.../tests/server/http_read_your_writes.rs | 19 +-
core/integration/tests/server/http_vsr.rs | 82 +-
core/integration/tests/server/mod.rs | 3 +
.../tests/server/pipelined_frames_vsr.rs | 303 +
core/integration/tests/server/raw_tcp.rs | 7 +
.../server/scenarios/authentication_scenario.rs | 6 +-
core/journal/src/lib.rs | 13 +-
core/journal/src/prepare_journal.rs | 75 +-
core/message_bus/src/config.rs | 21 +-
core/message_bus/src/lib.rs | 27 +
core/message_bus/src/transports/mod.rs | 8 +-
core/message_bus/src/transports/quic.rs | 860 +-
core/metadata/src/impls/metadata.rs | 293 +-
core/metadata/src/stm/consumer_group.rs | 346 +-
core/metadata/src/stm/lifecycle.rs | 389 +-
core/metadata/src/stm/mod.rs | 80 +-
core/metadata/src/stm/result.rs | 153 +-
core/metadata/src/stm/snapshot.rs | 28 +-
core/metadata/src/stm/stream.rs | 141 +-
core/partitions/src/consumer_offset_capacity.rs | 14 +
core/partitions/src/iggy_partition.rs | 11654 ++++++++++++-------
core/partitions/src/io.rs | 135 +-
core/partitions/src/journal.rs | 63 +-
core/partitions/src/lib.rs | 5 +-
core/partitions/src/log.rs | 7 +-
core/partitions/src/offset_recovery.rs | 62 +-
core/partitions/src/offset_storage.rs | 31 +
core/partitions/src/partition_storage.rs | 104 +-
core/partitions/src/segment_recovery.rs | 3 +-
core/partitions/src/state_transfer.rs | 162 +-
core/partitions/src/types.rs | 10 +-
core/sdk/src/client_wrappers/binary_client.rs | 21 +
core/sdk/src/clients/client.rs | 110 +-
core/sdk/src/clients/consumer.rs | 1 +
core/sdk/src/clients/producer.rs | 10 +-
core/sdk/src/driver.rs | 1325 +++
.../src/{poll_routing.rs => driver/poll_router.rs} | 185 +-
core/sdk/src/driver/route_cache.rs | 381 +
core/sdk/src/leader_aware.rs | 611 +-
core/sdk/src/lib.rs | 4 +-
core/sdk/src/{leader_aware.rs => links/connect.rs} | 524 +-
core/sdk/src/links/dial.rs | 324 +
core/sdk/src/links/fake_cluster.rs | 442 +
core/sdk/src/links/mod.rs | 448 +
core/sdk/src/links/mux.rs | 815 ++
core/sdk/src/links/quic.rs | 335 +
core/sdk/src/links/replicas.rs | 1571 +++
core/sdk/src/quic/quic_client.rs | 791 +-
core/sdk/src/session.rs | 16 +-
core/sdk/src/session_owner.rs | 601 +
core/sdk/src/session_owner/lanes.rs | 606 +
core/sdk/src/tcp/mod.rs | 4 -
core/sdk/src/tcp/tcp_client.rs | 1527 +--
core/sdk/src/tcp/tcp_connection_stream.rs | 108 -
core/sdk/src/tcp/tcp_connection_stream_kind.rs | 66 -
core/sdk/src/tcp/tcp_stream.rs | 27 -
core/sdk/src/tcp/tcp_tls_connection_stream.rs | 105 -
core/sdk/src/vsr.rs | 396 +-
core/sdk/src/websocket/mod.rs | 4 -
core/sdk/src/websocket/websocket_client.rs | 912 +-
.../src/websocket/websocket_connection_stream.rs | 212 -
core/sdk/src/websocket/websocket_stream.rs | 36 -
core/sdk/src/websocket/websocket_stream_kind.rs | 58 -
.../websocket/websocket_tls_connection_stream.rs | 217 -
core/server/README.md | 22 +-
core/server/config.toml | 33 +-
core/server/src/boot/mod.rs | 61 +-
core/server/src/boot/topology.rs | 12 +-
core/server/src/cluster_meta.rs | 39 +
core/server/src/consumer_group.rs | 9 +-
core/server/src/consumer_group/liveness.rs | 10 +-
core/server/src/dispatch/authz.rs | 10 +-
core/server/src/dispatch/failure.rs | 73 +-
core/server/src/dispatch/host.rs | 40 +-
core/server/src/dispatch/mod.rs | 1482 ++-
core/server/src/dispatch/partition.rs | 866 +-
core/server/src/dispatch/reads.rs | 93 +-
core/server/src/dispatch/submit.rs | 1386 ++-
core/server/src/dispatch/test_support.rs | 91 +-
core/server/src/http.rs | 38 +-
core/server/src/http/error.rs | 17 +-
core/server/src/http/forward.rs | 965 +-
core/server/src/http/handlers.rs | 268 +-
core/server/src/http/reply.rs | 8 +
core/server/src/http/state.rs | 7 +
core/server/src/http/submit.rs | 98 +-
core/server/src/lib.rs | 8 +-
core/server/src/namespace.rs | 34 +-
core/server/src/partition_helpers.rs | 5 +-
core/server/src/partition_reconciler.rs | 702 +-
core/server/src/reply_frame.rs | 12 +-
core/server/src/rewrite.rs | 42 +-
core/server_common/src/consensus_message.rs | 16 +-
core/shard/src/coordinator.rs | 5 +
core/shard/src/lib.rs | 1224 +-
core/shard/src/partition_io.rs | 2694 ++++-
core/shard/src/poll.rs | 87 +-
core/shard/src/poll/completion_tests.rs | 229 +-
core/simulator/src/bin/workload-fuzz.rs | 64 +-
core/simulator/src/client.rs | 27 +-
core/simulator/src/deps.rs | 21 +-
core/simulator/src/lib.rs | 2468 +++-
core/simulator/src/packet.rs | 3 +-
core/simulator/src/replica.rs | 6 +-
core/simulator/src/workload/actions.rs | 4 +
core/simulator/src/workload/auditor.rs | 18 +-
core/simulator/src/workload/invariants.rs | 30 +-
core/simulator/src/workload/mod.rs | 2481 +++-
.../src/workload/ops/join_fixture_group.rs | 96 +
...pdate_permissions.rs => leave_fixture_group.rs} | 50 +-
core/simulator/src/workload/ops/mod.rs | 12 +-
.../src/workload/ops/replace_fixture_topic.rs | 112 +
..._permissions.rs => store_stale_owner_offset.rs} | 66 +-
core/simulator/src/workload/options.rs | 53 +-
core/simulator/src/workload/oracle.rs | 384 +-
core/simulator/src/workload/shadow.rs | 550 +-
core/simulator/src/workload/state_checker.rs | 568 +-
foreign/cpp/Cargo.toml | 6 +-
foreign/cpp/src/lib.rs | 1 +
foreign/cpp/src/type_conversion.rs | 1 +
foreign/csharp/Directory.Packages.props | 8 +-
.../Fixtures/VsrCluster.cs | 9 +
.../Helpers/ConsumerGroupAssignment.cs | 4 +-
.../Vsr/VsrMetadataTests.cs | 3 +-
foreign/csharp/Iggy_SDK/Contracts/ClusterNode.cs | 3 +
.../SdkVersion.cs => Contracts/PartitionRoute.cs} | 13 +-
.../Exceptions/IggyInvalidStatusCodeException.cs | 7 +
.../TcpMessageStream.PollRouting.cs | 449 +-
.../Implementations/TcpMessageStream.Vsr.cs | 73 +-
.../IggyClient/Implementations/TcpMessageStream.cs | 101 +-
foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs | 21 +-
foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs | 2 +-
foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs | 61 +-
foreign/csharp/Iggy_SDK/Vsr/VsrError.cs | 13 +-
foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs | 4 +-
foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs | 2 +-
foreign/csharp/Iggy_SDK/Vsr/VsrReplyDecoder.cs | 3 +-
.../ConsumerTests/OffsetContinuationTests.cs | 9 +-
.../Iggy_SDK_Tests/MapperTests/BinaryMapper.cs | 59 +-
.../VsrTests/EndpointFailoverTests.cs | 249 +-
.../Iggy_SDK_Tests/VsrTests/GroupPollingTests.cs | 1 +
foreign/csharp/Iggy_SDK_Tests/VsrTests/MockNode.cs | 41 +-
.../VsrTests/PartitionContextTests.cs | 54 +-
.../Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs | 487 +-
.../VsrTests/VsrReplyDecoderTests.cs | 10 +-
.../Iggy_SDK_Tests/VsrTests/VsrTestPayloads.cs | 5 +
foreign/go/README.md | 2 +-
foreign/go/client/tcp/cluster.go | 12 +
foreign/go/client/tcp/tcp_core.go | 73 +-
foreign/go/client/tcp/tcp_core_review_test.go | 2 +-
foreign/go/client/tcp/tcp_core_test.go | 35 +-
foreign/go/client/tcp/tcp_failover_test.go | 4 +-
foreign/go/client/tcp/tcp_messaging.go | 174 +-
foreign/go/client/tcp/tcp_offset_management.go | 27 +-
foreign/go/client/tcp/tcp_partition_management.go | 3 +
foreign/go/client/tcp/tcp_poll_routing.go | 248 +-
foreign/go/client/tcp/tcp_poll_routing_test.go | 498 +-
foreign/go/client/tcp/tcp_stream_management.go | 1 +
foreign/go/client/tcp/tcp_testing_test.go | 19 +-
foreign/go/client/tcp/tcp_topic_cache.go | 44 +-
foreign/go/client/tcp/tcp_topic_cache_test.go | 71 +-
foreign/go/contracts/cluster_metadata_test.go | 56 +-
foreign/go/contracts/cluster_node.go | 10 +-
foreign/go/contracts/partitions.go | 26 +
foreign/go/errors/errors.yaml | 4 +
foreign/go/errors/errors_gen.go | 15 +
foreign/go/internal/command/code.go | 2 +-
foreign/go/internal/vsr/envelope.go | 5 +
foreign/go/internal/vsr/envelope_test.go | 4 +-
foreign/go/internal/vsr/session.go | 46 +-
foreign/go/internal/vsr/session_test.go | 40 +-
foreign/go/tests/e2e_test.go | 29 +-
.../iggy/client/async/tcp/AsyncIggyTcpClient.java | 69 +-
.../iggy/client/async/tcp/AsyncTcpConnection.java | 137 +-
.../iggy/client/async/tcp/ClientRoutingState.java | 36 +-
.../client/async/tcp/ConsumerOffsetsTcpClient.java | 49 +-
.../iggy/client/async/tcp/MessagesTcpClient.java | 146 +-
.../iggy/client/async/tcp/PartitionContexts.java | 191 -
.../apache/iggy/client/async/tcp/PollRouter.java | 429 +-
.../client/async/tcp/RawPartitionRequests.java | 12 +-
.../client/async/tcp/vsr/ConsensusSession.java | 21 +-
.../iggy/client/async/tcp/vsr/VsrHeaders.java | 2 +-
.../client/async/tcp/vsr/VsrRequestEncoder.java | 13 +-
.../java/org/apache/iggy/cluster/ClusterNode.java | 14 +-
.../org/apache/iggy/exception/IggyErrorCode.java | 1 +
.../org/apache/iggy/partition/PartitionRoute.java | 47 +
.../org/apache/iggy/serde/BytesDeserializer.java | 5 +-
.../java/org/apache/iggy/serde/CommandCode.java | 2 +-
.../AsyncIggyTcpClientEndpointFailoverTest.java | 1 +
.../AsyncIggyTcpClientPartitionContextTest.java | 141 +-
.../tcp/AsyncIggyTcpClientRawRequestTest.java | 192 +-
.../AsyncIggyTcpClientSendContextCacheTest.java | 99 +-
.../AsyncIggyTcpClientTransientFailoverTest.java | 128 +-
.../tcp/AsyncTcpConnectionLifecycleBusyTest.java | 52 +-
.../client/async/tcp/ClientRoutingStateTest.java | 40 -
.../iggy/client/async/tcp/LeaderAwarenessTest.java | 2 +-
.../iggy/client/async/tcp/PollRouterTest.java | 102 +-
.../async/tcp/vsr/VsrRequestEncoderTest.java | 35 +-
.../client/blocking/http/ObjectMapperTest.java | 19 +
.../apache/iggy/partition/PartitionRouteTest.java | 65 +
.../apache/iggy/serde/BytesDeserializerTest.java | 3 +
foreign/node/src/client/client.socket.test.ts | 521 +-
foreign/node/src/client/client.socket.ts | 461 +-
foreign/node/src/e2e/tcp.cluster.e2e.ts | 1 +
foreign/node/src/wire/cluster/cluster.utils.ts | 6 +-
foreign/node/src/wire/command.code.ts | 2 +-
foreign/node/src/wire/error.code.ts | 1 +
.../src/wire/message/send-messages.command.test.ts | 52 +-
.../node/src/wire/message/send-messages.command.ts | 56 +-
foreign/node/src/wire/offset/get-offset.command.ts | 2 +-
foreign/node/src/wire/offset/offset.utils.ts | 12 +-
.../src/wire/partition/partition-route.test.ts | 46 +
foreign/node/src/wire/partition/partition-route.ts | 51 +
foreign/node/src/wire/vsr/index.ts | 2 +-
foreign/node/src/wire/vsr/session.ts | 5 +-
foreign/node/src/wire/vsr/vsr.test.ts | 4 +-
foreign/python/Cargo.toml | 6 +-
foreign/swift/Package.resolved | 60 +
foreign/swift/Package.swift | 24 +-
foreign/swift/README.md | 74 +-
.../Sources/Iggy/Client/ClientConfiguration.swift | 142 +
foreign/swift/Sources/Iggy/Client/ClientCore.swift | 1082 ++
.../Sources/Iggy/Client/IggyClient+Resources.swift | 195 +
foreign/swift/Sources/Iggy/Client/IggyClient.swift | 239 +
foreign/swift/Sources/Iggy/Errors/IggyError.swift | 28 +
.../swift/Sources/Iggy/Errors/IggyErrorCode.swift | 8 +
foreign/swift/Sources/Iggy/Models/Resources.swift | 4 +-
.../swift/Sources/Iggy/Models/TopicOptions.swift | 3 +
.../swift/Sources/Iggy/Protocol/CommandCode.swift | 16 +-
foreign/swift/Sources/Iggy/Protocol/Requests.swift | 6 +-
.../swift/Sources/Iggy/Protocol/Responses.swift | 3 +-
.../swift/Sources/Iggy/Transport/Endpoint.swift | 155 +
.../Sources/Iggy/Transport/TCPConnection.swift | 295 +
.../swift/Sources/Iggy/Transport/TLSOptions.swift | 67 +
.../swift/Sources/Iggy/Utilities/AsyncMutex.swift | 201 +
.../Tests/IggyTests/ClientTransportTests.swift | 1168 ++
.../swift/Tests/IggyTests/ConcurrencyTests.swift | 209 +
.../swift/Tests/IggyTests/ConfigurationTests.swift | 120 +
foreign/swift/Tests/IggyTests/Fixtures/golden.json | 8 +-
foreign/swift/Tests/IggyTests/ProtocolTests.swift | 6 +-
.../Tests/IggyTests/ResponseVectorTests.swift | 2 +
.../Tests/IggyTests/Support/FakeVSRServer.swift | 331 +
.../Tests/IggyTests/Support/TestHelpers.swift | 64 +
foreign/swift/Tools/golden-vectors/src/main.rs | 2 +
gateways/kafka/README.md | 11 +-
gateways/kafka/docs/CONFIGS.md | 31 +-
gateways/kafka/docs/CONSUMER_GROUPS.md | 61 +-
gateways/kafka/docs/MANUAL_TESTING.md | 8 +-
gateways/kafka/docs/OFFSET_STORAGE.md | 10 +-
gateways/kafka/docs/SCOPE.md | 12 +-
gateways/kafka/docs/TEST_SUITE.md | 2 +-
gateways/kafka/docs/kafka_api_keys_reference.md | 8 +-
gateways/kafka/scripts/ci-wire-fixtures.sh | 2 +-
gateways/kafka/src/bridge/error.rs | 16 +-
.../kafka/src/bridge/iggy_bridge/group_index.rs | 80 +-
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 145 +-
gateways/kafka/src/bridge/iggy_bridge/topics.rs | 2 +-
gateways/kafka/src/group/mod.rs | 73 +-
gateways/kafka/src/group/state.rs | 246 +-
gateways/kafka/src/protocol/api.rs | 19 +-
gateways/kafka/src/protocol/bounds_guard.rs | 117 +-
.../kafka/src/protocol/handlers/create_topics.rs | 125 +-
.../kafka/src/protocol/handlers/delete_groups.rs | 400 +
.../src/protocol/handlers/describe_configs.rs | 42 +-
gateways/kafka/src/protocol/handlers/mod.rs | 43 +-
.../kafka/src/protocol/handlers/offset_delete.rs | 484 +
.../kafka/src/protocol/handlers/topic_config.rs | 55 +-
gateways/kafka/tests/common/scope.rs | 2 +
gateways/kafka/tests/common/wire.rs | 38 +-
gateways/kafka/tests/configs_java_client_tests.rs | 5 +-
gateways/kafka/tests/configs_real_bridge_tests.rs | 18 +-
.../kafka/tests/create_topics_real_bridge_tests.rs | 45 +
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 16 +-
gateways/kafka/tests/java/ConfigsProbe.java | 17 +-
.../kafka/tests/offset_groups_real_bridge_tests.rs | 314 +-
gateways/kafka/tests/response_negative_tests.rs | 11 +-
gateways/kafka/tests/version_firewall_tests.rs | 36 +-
gateways/kafka/tools/kafka-tool/src/main.rs | 21 +
gateways/kafka/tools/kafka-tool/src/response.rs | 1 +
338 files changed, 49540 insertions(+), 14671 deletions(-)
rename core/binary_protocol/src/requests/messages/{get_send_context.rs => get_partition_route.rs} (77%)
create mode 100644 core/binary_protocol/src/responses/messages/partition_route.rs
create mode 100644 core/integration/tests/server/pipelined_frames_vsr.rs
create mode 100644 core/sdk/src/driver.rs
rename core/sdk/src/{poll_routing.rs => driver/poll_router.rs} (91%)
create mode 100644 core/sdk/src/driver/route_cache.rs
copy core/sdk/src/{leader_aware.rs => links/connect.rs} (54%)
create mode 100644 core/sdk/src/links/dial.rs
create mode 100644 core/sdk/src/links/fake_cluster.rs
create mode 100644 core/sdk/src/links/mod.rs
create mode 100644 core/sdk/src/links/mux.rs
create mode 100644 core/sdk/src/links/quic.rs
create mode 100644 core/sdk/src/links/replicas.rs
create mode 100644 core/sdk/src/session_owner.rs
create mode 100644 core/sdk/src/session_owner/lanes.rs
delete mode 100644 core/sdk/src/tcp/tcp_connection_stream.rs
delete mode 100644 core/sdk/src/tcp/tcp_connection_stream_kind.rs
delete mode 100644 core/sdk/src/tcp/tcp_stream.rs
delete mode 100644 core/sdk/src/tcp/tcp_tls_connection_stream.rs
delete mode 100644 core/sdk/src/websocket/websocket_connection_stream.rs
delete mode 100644 core/sdk/src/websocket/websocket_stream.rs
delete mode 100644 core/sdk/src/websocket/websocket_stream_kind.rs
delete mode 100644 core/sdk/src/websocket/websocket_tls_connection_stream.rs
create mode 100644 core/simulator/src/workload/ops/join_fixture_group.rs
copy core/simulator/src/workload/ops/{update_permissions.rs => leave_fixture_group.rs} (55%)
create mode 100644 core/simulator/src/workload/ops/replace_fixture_topic.rs
copy core/simulator/src/workload/ops/{update_permissions.rs => store_stale_owner_offset.rs} (50%)
copy foreign/csharp/Iggy_SDK/{Utils/SdkVersion.cs => Contracts/PartitionRoute.cs} (73%)
delete mode 100644 foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PartitionContexts.java
create mode 100644 foreign/java/java-sdk/src/main/java/org/apache/iggy/partition/PartitionRoute.java
create mode 100644 foreign/java/java-sdk/src/test/java/org/apache/iggy/partition/PartitionRouteTest.java
create mode 100644 foreign/node/src/wire/partition/partition-route.test.ts
create mode 100644 foreign/node/src/wire/partition/partition-route.ts
create mode 100644 foreign/swift/Package.resolved
create mode 100644 foreign/swift/Sources/Iggy/Client/ClientConfiguration.swift
create mode 100644 foreign/swift/Sources/Iggy/Client/ClientCore.swift
create mode 100644 foreign/swift/Sources/Iggy/Client/IggyClient+Resources.swift
create mode 100644 foreign/swift/Sources/Iggy/Client/IggyClient.swift
create mode 100644 foreign/swift/Sources/Iggy/Transport/Endpoint.swift
create mode 100644 foreign/swift/Sources/Iggy/Transport/TCPConnection.swift
create mode 100644 foreign/swift/Sources/Iggy/Transport/TLSOptions.swift
create mode 100644 foreign/swift/Sources/Iggy/Utilities/AsyncMutex.swift
create mode 100644 foreign/swift/Tests/IggyTests/ClientTransportTests.swift
create mode 100644 foreign/swift/Tests/IggyTests/ConcurrencyTests.swift
create mode 100644 foreign/swift/Tests/IggyTests/ConfigurationTests.swift
create mode 100644 foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
create mode 100644 foreign/swift/Tests/IggyTests/Support/TestHelpers.swift
create mode 100644 gateways/kafka/src/protocol/handlers/delete_groups.rs
create mode 100644 gateways/kafka/src/protocol/handlers/offset_delete.rs
(iggy) 01/01: feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-empty-groups
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 3cd1923ef084e05bd90eb141b2042072a290c992
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Wed Oct 14 14:47:33 2026 +0530
feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups
Groups with offsets stay visible after their consumers leave, and operators can delete them.
---
gateways/kafka/README.md | 18 +-
gateways/kafka/docs/CONSUMER_GROUPS.md | 117 +++-
gateways/kafka/docs/MANUAL_TESTING.md | 9 +-
gateways/kafka/docs/OFFSET_STORAGE.md | 10 +-
gateways/kafka/docs/SCOPE.md | 12 +-
gateways/kafka/docs/TEST_SUITE.md | 1 +
gateways/kafka/docs/kafka_api_keys_reference.md | 9 +-
gateways/kafka/scripts/ci-wire-fixtures.sh | 2 +-
gateways/kafka/src/bridge/error.rs | 16 +-
.../kafka/src/bridge/iggy_bridge/group_index.rs | 392 ++++++++++++
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 250 +++++++-
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 7 +
gateways/kafka/src/bridge/iggy_bridge/topics.rs | 2 +-
gateways/kafka/src/bridge/mod.rs | 4 +-
gateways/kafka/src/group/mod.rs | 107 +++-
gateways/kafka/src/group/state.rs | 335 +++++++++-
gateways/kafka/src/protocol/api.rs | 19 +-
gateways/kafka/src/protocol/bounds_guard.rs | 117 +++-
.../kafka/src/protocol/handlers/delete_groups.rs | 400 ++++++++++++
.../kafka/src/protocol/handlers/describe_groups.rs | 74 ++-
.../kafka/src/protocol/handlers/list_groups.rs | 70 +-
gateways/kafka/src/protocol/handlers/mod.rs | 49 +-
.../kafka/src/protocol/handlers/offset_delete.rs | 484 ++++++++++++++
gateways/kafka/tests/common/scope.rs | 2 +
gateways/kafka/tests/common/wire.rs | 38 +-
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 16 +-
.../kafka/tests/offset_groups_real_bridge_tests.rs | 712 +++++++++++++++++++++
gateways/kafka/tests/version_firewall_tests.rs | 36 +-
gateways/kafka/tools/kafka-tool/src/main.rs | 21 +
gateways/kafka/tools/kafka-tool/src/response.rs | 1 +
30 files changed, 3190 insertions(+), 140 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 3c83aa194..030d20c18 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -15,16 +15,20 @@ Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/34
> `manage_topics`; without it every resource/name in the request fails
> `TOPIC_AUTHORIZATION_FAILED` (29) and nothing is written or deleted. `CreateTopics` has no such
> gate yet - SASL on does not protect it. Every client reads and writes as the bridge's Iggy user.
-> See [docs/SCOPE.md](docs/SCOPE.md).
+> DeleteGroups and OffsetDelete also run as that user, so with SASL off any client can delete
+> committed offsets. With SASL on, they also require `manage_topics`, and without it they answer
+> `GROUP_AUTHORIZATION_FAILED` (30). See [docs/SCOPE.md](docs/SCOPE.md).
>
> InitProducerId does real work too, with or without the bridge: it allocates a producer id, so a stock idempotent producer starts instead of failing at startup.
>
> Consumer group coordination is not a stub either: `FindCoordinator`, `JoinGroup`, `Heartbeat`,
> `LeaveGroup`, `SyncGroup`, `DescribeGroups` and `ListGroups` are real, with real membership,
> rebalances, graceful leave, session expiry and admin views of that membership
-> ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)). `kafka-consumer-groups.sh --list`,
-> `--describe --members` and `--describe --state` use those views. The default `--describe`
-> calls OffsetFetch (API key 9), which is not implemented, so that mode does not work. With the bridge off,
+> ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)). `kafka-consumer-groups.sh --list` and
+> `--describe` use those views. With the bridge on, they also show a group whose consumers stopped,
+> as `Empty`, while it holds offsets. The default `--describe` then prints each committed offset
+> and its lag. `--delete` deletes such a group with its offsets, and `--delete-offsets` deletes
+> single offsets. With the bridge off,
> Metadata reports every topic unknown, so a consumer joins a group and is assigned 0 partitions.
> With the bridge on, OffsetCommit and OffsetFetch keep each group's offsets in Iggy, so a group
> resumes from its last commit, also after a gateway restart
@@ -234,10 +238,8 @@ Limits:
- Classic protocol only. A consumer with `group.protocol=consumer` (KIP-848) fails.
- `group.instance.id` is accepted but not honoured. A static member that restarts causes two
rebalances.
-- No DescribeGroups or ListGroups yet ([#3548](https://github.com/apache/iggy/issues/3548)), so
- `kafka-consumer-groups.sh` cannot list or describe a group.
-- No DeleteGroups or OffsetDelete, and offsets never expire. An operator deletes a group's offsets
- in Iggy ([docs/OFFSET_STORAGE.md](docs/OFFSET_STORAGE.md#limits)).
+- Offsets never expire. `kafka-consumer-groups.sh --delete` and `--delete-offsets` delete them
+ ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md#deleting-groups-and-offsets)).
- Delivery is at-least-once. A consumer that stops before it commits reads those records again.
## Authentication ([#3549](https://github.com/apache/iggy/issues/3549))
diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md
index 166f6acac..e2a5460d4 100644
--- a/gateways/kafka/docs/CONSUMER_GROUPS.md
+++ b/gateways/kafka/docs/CONSUMER_GROUPS.md
@@ -4,8 +4,9 @@ What [#3541](https://github.com/apache/iggy/issues/3541) added: `FindCoordinator
`JoinGroup` (11), `Heartbeat` (12) and `SyncGroup` (14), backed by an in-memory coordinator in
`src/group/`. [#3543](https://github.com/apache/iggy/issues/3543) added `LeaveGroup` (13).
[#3548](https://github.com/apache/iggy/issues/3548) added `DescribeGroups` (15) and `ListGroups`
-(16). This is Kafka's *classic* group protocol. Offsets are a separate concern and live
-in Iggy ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)).
+(16). [#4511](https://github.com/apache/iggy/issues/4511) added `DeleteGroups` (42) and
+`OffsetDelete` (47). This is Kafka's *classic* group protocol. Offsets are a separate concern and
+live in Iggy ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)).
| API key | Name | Versions | Notes |
| --- | --- | --- | --- |
@@ -14,8 +15,10 @@ in Iggy ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)).
| 12 | Heartbeat | 0-4 | Refreshes a session; `REBALANCE_IN_PROGRESS` is how a follower learns to rejoin |
| 13 | LeaveGroup | 0-5 | Removes members; survivors' parked joins and syncs are released at once |
| 14 | SyncGroup | 0-5 | Relays the leader's assignment blobs; a follower parks until the leader syncs |
-| 15 | DescribeGroups | 0-6 | Members, assignment and state. A group that is not here is `Dead`; v6 also returns `GROUP_ID_NOT_FOUND` (69) |
-| 16 | ListGroups | 0-5 | Groups currently on this coordinator. v4+ can filter by state, v5 by type `classic` |
+| 15 | DescribeGroups | 0-6 | Members, assignment and state. A group that is not here but holds offsets in Iggy is `Empty`. Otherwise it is `Dead`, and v6 also returns `GROUP_ID_NOT_FOUND` (69) |
+| 16 | ListGroups | 0-5 | Groups currently on this coordinator, plus the groups that hold offsets in Iggy as `Empty`. v4+ can filter by state, v5 by type `classic` |
+| 42 | DeleteGroups | 0-2 | Deletes a group without members, with its offsets. While it has members: `NON_EMPTY_GROUP` (68). When it does not exist: `GROUP_ID_NOT_FOUND` (69) |
+| 47 | OffsetDelete | 0 | Deletes offsets of a group. `GROUP_SUBSCRIBED_TO_TOPIC` (86) for a topic that a member subscribes to |
`kafka-protocol` can encode FindCoordinator v5 and v6 as well, and they are byte-identical to v4.
They are not advertised because `SCOPE.md`'s governance model only admits a version once it has
@@ -25,16 +28,14 @@ been manually tested.
`DescribeGroups` and `ListGroups` read the same in-memory map the join path writes. State strings
are `Empty`, `PreparingRebalance`, `CompletingRebalance` and `Stable`. `Empty` means the group
-has no joined members: a v4+ client has been handed a member id and has not rejoined with it yet.
-`Dead` is only the describe answer for an id that is not in the map.
-
-A group with neither members nor pending member ids is removed. Kafka keeps an `Empty` group
-that only has committed offsets. Offsets are not stored here yet.
-TODO(#3542): once offset storage lands, keep those groups and report `Empty`.
+has no joined members. Either a v4+ client got a member id and did not rejoin with it yet, or
+the group only holds offsets
+([below](#groups-that-only-hold-offsets)). `Dead` is only the describe answer for an id
+that is not in the map and holds no offsets.
`ListGroups` does not tick. It leaves sessions, pending ids and phases untouched, and it omits a
group whose member sessions and pending-id deadlines are all already past. `DescribeGroups` does
-tick each named group, so that same id is then `Dead`.
+tick each named group, so that same id is then `Dead`, or `Empty` for a group that holds offsets.
While a group is not `Stable`, `DescribeGroups` leaves `protocol_data`, member metadata and
member assignment empty. Kafka does the same: during `PreparingRebalance` the stored assignment
@@ -48,19 +49,94 @@ second snapshot of the same members would otherwise multiply the encoded respons
count.
`kafka-consumer-groups.sh --list`, `--describe --members` and `--describe --state` use these
-two APIs. The default `--describe` calls `OffsetFetch` (API key 9), which is not implemented,
-so that mode does not work. `--reset-offsets` needs the same API.
+two APIs. With a bridge, the default `--describe` also reads each committed offset with
+`OffsetFetch` (9) and each end offset with `ListOffsets` (2), and prints the lag.
+`--reset-offsets` writes the new offsets with `OffsetCommit` (8). Both work for a group whose
+consumers stopped, because that group is `Empty` while it holds offsets. `--delete` and
+`--delete-offsets` use `DeleteGroups` and `OffsetDelete`
+([below](#deleting-groups-and-offsets)).
Member `client_id` and `client_host` are empty. The coordinator does not keep the request
header's client id or the connection's peer address. `include_authorized_operations` is answered
with the omitted sentinel. There is no group ACL bitfield. `ListGroups` v5 sets `group_type` to
`classic`, which is the only protocol this coordinator runs.
+## Groups that only hold offsets
+
+A group with neither members nor pending member ids is removed from the map. Its offsets stay in
+Iggy, and Kafka keeps such a group and reports it as `Empty`. So both APIs also report each group
+that holds offsets in Iggy and has no members here. Such a group is `Empty`, with protocol type
+`consumer` and no members. Iggy stores no protocol type. Kafka keeps `consumer` for a group whose
+consumers left, and the Java admin client takes `consumer` or an empty type as a consumer group.
+
+The gateway finds these groups with a scan. A scan lists the Kafka topics, then calls
+`get_consumer_groups` once per topic and keeps the Iggy groups named `kafka.cg.<group>`
+([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). It runs on the offset read slots, so it never waits
+behind Produce. One scan answers every request for 10 seconds, because kafka-ui lists groups every
+30 seconds and a scan costs one Iggy call per topic. At most one scan runs at a time, and a request
+that comes while it runs waits for it. A commit through this gateway adds its group at once, so a
+new group does not wait for the next scan. A group that another gateway created appears with the
+next scan.
+
+A request waits for a scan until its offset deadline, 3 seconds by default. The scan keeps running
+after that, for up to 30 seconds, and serves the next request. If a scan fails, the groups of the
+last scan that found them stay the answer. If no scan has found them yet, `ListGroups` answers
+`COORDINATOR_LOAD_IN_PROGRESS` (14), and `DescribeGroups` answers 14 for each named group that is
+not in the map. The Java admin client retries that code. A `ListGroups` state filter without
+`Empty`, or a type filter without `classic`, needs no scan. `DescribeGroups` scans only for a named
+group that is not in the map.
+
+A negative commit, or `OffsetDelete`, deletes an offset but not the Iggy group. So a group whose
+offsets were all deleted that way still shows as `Empty`, until `DeleteGroups` deletes it. Without
+a bridge, no group holds offsets, and both APIs report only the groups in the map.
+
+## Deleting groups and offsets
+
+`DeleteGroups` and `OffsetDelete` follow Kafka's rules for a classic group
+(`ClassicGroup.validateDeleteGroup` and `validateOffsetDelete`).
+
+| Group | `DeleteGroups` | `OffsetDelete` |
+| --- | --- | --- |
+| Has members of protocol type `consumer` | `NON_EMPTY_GROUP` (68) | Per partition: `GROUP_SUBSCRIBED_TO_TOPIC` (86) for a topic that a member subscribes to, else deleted |
+| Has members of another protocol type | `NON_EMPTY_GROUP` (68) | `NON_EMPTY_GROUP` (68) |
+| Has no members here, holds offsets in Iggy | Deleted | Deleted |
+| Only has ids that wait to join | Deleted | Deleted |
+| Not here, holds no offsets | `GROUP_ID_NOT_FOUND` (69) | `GROUP_ID_NOT_FOUND` (69) |
+| Empty id, or longer than 246 bytes | `INVALID_GROUP_ID` (24) | `INVALID_GROUP_ID` (24) |
+
+`DeleteGroups` deletes the Iggy group `kafka.cg.<group>` on every Kafka topic, and Iggy deletes its
+offsets with it. It first lists the Kafka topics and the Iggy groups of each topic. So it costs one
+call per topic, and at least two per deletion. It answers a repeated group id once.
+
+Iggy deletes a group on each partition of the topic, and lists it until every partition did. So the
+gateway looks up the group on the topic every 5 to 100 milliseconds. When the group is gone, the
+gateway answers. Then `OffsetFetch` returns -1 for the group, and the admin APIs no longer find it.
+
+`OffsetDelete` reads the topics that the members subscribe to from their metadata for the selected
+protocol, as Kafka does. It reads only the topic list of the consumer protocol's v0 prefix, which
+every later version keeps. If a member's metadata does not decode, or the group has not chosen a
+protocol yet, every topic counts as subscribed. A topic or partition that does not exist gets
+`UNKNOWN_TOPIC_OR_PARTITION` (3). Any other partition gets 0. A partition that held no offset gets
+0 too.
+
+A call to Iggy that a retry can fix answers `COORDINATOR_LOAD_IN_PROGRESS` (14), and the Java admin
+client retries it. `DeleteGroups` answers it for the group, and `OffsetDelete` for the whole
+request. A second delete of an offset or a group does no harm. A deletion that does not end by the
+offset deadline also answers 14. When Iggy refuses the bridge user, `DeleteGroups` answers
+`GROUP_AUTHORIZATION_FAILED` (30).
+
+With SASL on, both APIs require the authenticated principal to hold the Iggy permission
+`manage_topics` or `manage_streams`, as `DeleteTopics` does. Kafka asks for `DELETE` on the group,
+and Iggy has no group permission. Without it, `DeleteGroups` answers `GROUP_AUTHORIZATION_FAILED`
+(30) for every group, `OffsetDelete` answers it for the request, and the gateway deletes nothing.
+With SASL off, any client that reaches the gateway can delete groups and offsets.
+
## Assignment is the client's job
The gateway elects a leader, hands it every member's subscription metadata, and fans the blobs the
leader returns back out - one per member, exactly the bytes filed under that member's id, in the
-same generation. It never decodes `ConsumerProtocolSubscription` or `ConsumerProtocolAssignment`.
+same generation. It never decodes `ConsumerProtocolAssignment`, and it reads only the topic list of
+a `ConsumerProtocolSubscription`, for `OffsetDelete`.
The practical consequence: **whatever assignor the client ships is the assignor** - range,
round-robin, sticky, cooperative-sticky, or a custom one. There is no supported subset to
@@ -120,12 +196,13 @@ survivors keep the current generation and learn about the rebalance through
`REBALANCE_IN_PROGRESS` (27) on their next heartbeat, which the Java client handles as a graceful
rejoin rather than as lost partitions.
-When a group empties, by leave or by expiry, it is dropped, so the next group under that name
-starts again at generation 1. The pending-member-id window above is the one case that stays and
-reports `Empty`. Kafka also keeps an `Empty` group that only has committed offsets, and its
-generation with it. This coordinator does not, until offset storage lands (#3542). This is
-safe because member ids carry a UUID and every generation check is preceded by a membership
-check, so a stale client is told `UNKNOWN_MEMBER_ID` before its generation is ever compared.
+When a group empties, by leave or by expiry, it is dropped, so the next group under that name starts
+again at generation 1. The pending-member-id window above is the one case that stays and reports
+`Empty`. Kafka also keeps an `Empty` group that only holds committed offsets, and its generation
+with it. This gateway reports that group as `Empty` from its offsets in Iggy ([Groups that only hold
+offsets](#groups-that-only-hold-offsets)), but keeps no generation for it. This is safe because
+member ids carry a UUID and every generation check is preceded by a membership check. So a stale
+client is told `UNKNOWN_MEMBER_ID` before its generation is ever compared.
## Leaving a group
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index 21c88dd13..b698fb147 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -326,9 +326,12 @@ success. This category predates the bridge landing: G2/G3's "fails at metadata"
| G9 | Group resumes from its commit | Bridge on and topic `orders` made as in Category I. Run `kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --group g9 --topic orders --from-beginning`, stop it, produce one record, run it again | The second run prints only the new record. `--from-beginning` applies only to a group with no committed offset |
| G10 | Resume after a gateway restart | G9 with a gateway restart before the second run | Same as G9, because the offsets live in Iggy ([#3542](https://github.com/apache/iggy/issues/3542)) |
| G11 | Two members share a topic | Bridge on and a topic with 2 partitions. Run `kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --group g11 --topic <topic> --property print.partition=true --property print.offset=true` in two terminals, produce to both partitions, stop one, produce again | Each prints one partition. After the stop, the other prints the new records of both partitions, and none twice |
-| G12 | List groups | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --list` after G5 has joined | Prints the group. `--list --state Empty` matches a group that was issued a member id and has not joined yet. A group whose members have all expired is absent |
+| G12 | List groups | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --list` after G5 joined | Prints the group. `--list --state Empty` matches a group that was issued a member id and did not join yet. With the bridge on, it also matches a group whose members left or expired and that holds offsets. A group with neither members nor offsets is absent |
| G13 | Describe members | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --describe --members --group g2` | Prints member ids. Assignment and protocol are filled only while the group is Stable |
-| G14 | Describe state | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --describe --state --group g2` | Prints the state string (`Stable`, `PreparingRebalance`, `CompletingRebalance`, or `Empty`). The default `--describe` (no `--members` or `--state`) calls OffsetFetch and fails, because that API is not implemented |
+| G14 | Describe state | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --describe --state --group g2` | Prints the state string (`Stable`, `PreparingRebalance`, `CompletingRebalance`, or `Empty`). With the bridge on, the default `--describe` (no `--members` or `--state`) prints each committed offset and its lag |
+| G15 | Group with no live members | Bridge on. After G9's second run ends, `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --list --state`, then `--describe --group g9` | The group is listed as `Empty`. `--describe` prints `Consumer group 'g9' has no active members.` and then each committed offset and its lag |
+| G16 | Delete offsets | Bridge on. While a G9 consumer runs, `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --delete-offsets --group g9 --topic orders`. Stop the consumer and run the command again | The first run fails for each partition with `GroupSubscribedToTopicException`. The second prints `Successful` for each partition, and the default `--describe` prints no offsets for `orders` |
+| G17 | Delete a group | Bridge on. While a G9 consumer runs, `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --delete --group g9`. Stop the consumer and run the command again, then `--list` and `--describe --group g9` | The first run fails with `GroupNotEmptyException`. The second prints `Deletion of requested consumer groups ('g9') was successful.`. `--list` leaves out `g9`, and `--describe` prints `Consumer group 'g9' does not exist.` |
Record kcat version and exact error strings in your test log. G1 passing is the minimum bar for client compatibility smoke.
@@ -458,7 +461,7 @@ kcat version (if used): ___________
[ ] D1–D10 Flexible vs legacy encoding
[ ] E1–E4 Metadata stub semantics
[ ] F1–F6 TCP / connection behavior
-[ ] G1–G11 kcat / Java client / kafka-consumer-groups.sh (record errors for G2/G3; G11's default `--describe` is expected to fail)
+[ ] G1–G17 kcat / Java client / kafka-consumer-groups.sh (record errors for G2/G3)
[ ] H1–H6 Adversarial input
[ ] I1–I6 Docker Compose quick start (record wall-clock time to I3; I5 is restart durability)
diff --git a/gateways/kafka/docs/OFFSET_STORAGE.md b/gateways/kafka/docs/OFFSET_STORAGE.md
index 663135d28..58b55d403 100644
--- a/gateways/kafka/docs/OFFSET_STORAGE.md
+++ b/gateways/kafka/docs/OFFSET_STORAGE.md
@@ -3,7 +3,8 @@
Status: accepted. Replaces the first design ([#4205](https://github.com/apache/iggy/pull/4205),
[#4289](https://github.com/apache/iggy/pull/4289)). Answers
[#3540](https://github.com/apache/iggy/issues/3540). Used by OffsetCommit and OffsetFetch
-([#3542](https://github.com/apache/iggy/issues/3542)).
+([#3542](https://github.com/apache/iggy/issues/3542)), and by DeleteGroups and OffsetDelete
+([#4511](https://github.com/apache/iggy/issues/4511)).
## Decision
@@ -88,8 +89,11 @@ leaves it behind.
| Keys per partition, per kind | 4096, at most 262144: `partition.consumer_offsets_max` |
| Past the limit | `TooManyConsumerOffsets` (3024). Sent as `UNKNOWN_SERVER_ERROR`, logged at `error!` |
-The gateway never deletes a `kafka.cg.<group>` group, and does not serve DeleteGroups (42). A group
-nobody uses keeps its keys until an operator deletes its Iggy group on each topic:
+DeleteGroups (42) deletes the `kafka.cg.<group>` group on every Kafka topic, and its keys go with
+it. OffsetDelete (47) deletes single keys
+([`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md#deleting-groups-and-offsets)). The gateway deletes
+nothing on its own. So a group nobody uses keeps its keys until an admin tool deletes the group, or
+an operator deletes its Iggy group on each topic:
```bash
iggy consumer-group list <stream> <topic>
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index 5a3ef4695..1f87f39c4 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -63,11 +63,13 @@ it knows the server supports flexible encoding.
| 12 | Heartbeat | 0 | 4 | 0, 1, 2, 3, 4 | Refreshes a session; `REBALANCE_IN_PROGRESS` (27) drives a rejoin; flexible encoding at v4+ |
| 13 | LeaveGroup | 0 | 5 | 0, 1, 2, 3, 4, 5 | Removes members, per-member errors from v3; flexible encoding at v4+ |
| 14 | SyncGroup | 0 | 5 | 0, 1, 2, 3, 4, 5 | Relays the leader's assignment blobs; flexible encoding at v4+ |
-| 15 | DescribeGroups | 0 | 6 | 0, 1, 2, 3, 4, 5, 6 | Members, assignment and state of groups on this coordinator; a missing group is `Dead`, and v6 also returns `GROUP_ID_NOT_FOUND` (69); flexible encoding at v5+ |
-| 16 | ListGroups | 0 | 5 | 0, 1, 2, 3, 4, 5 | Groups on this coordinator; v4+ state filter and v5 type filter; flexible encoding at v3+ |
+| 15 | DescribeGroups | 0 | 6 | 0, 1, 2, 3, 4, 5, 6 | Members, assignment and state of groups on this coordinator. A group that is not there but holds offsets in Iggy is `Empty`. Any other missing group is `Dead`, and v6 also returns `GROUP_ID_NOT_FOUND` (69). Flexible encoding at v5+ |
+| 16 | ListGroups | 0 | 5 | 0, 1, 2, 3, 4, 5 | Groups on this coordinator, plus the groups that hold offsets in Iggy as `Empty` (see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)). v4+ state filter and v5 type filter. Flexible encoding at v3+ |
| 22 | InitProducerId | 0 | 5 | 0, 1, 2, 3, 4, 5 | Allocate a producer id (epoch 0); a `transactional_id` gets `UNSUPPORTED_VERSION` (35); flexible encoding at v2+ |
| 32 | DescribeConfigs | 1 | 4 | 1, 2, 3, 4 | Topic resources only. Returns `retention.ms` and leaves out `cleanup.policy`. Flexible encoding at v4. See [`CONFIGS.md`](CONFIGS.md) |
| 33 | AlterConfigs | 0 | 2 | 0, 1, 2 | Persists `retention.ms` only. Flexible encoding at v2. See [`CONFIGS.md`](CONFIGS.md) |
+| 42 | DeleteGroups | 0 | 2 | 0, 1, 2 | With a bridge: deletes a group without members, with its offsets in Iggy. While it has members: `NON_EMPTY_GROUP` (68). When it does not exist: `GROUP_ID_NOT_FOUND` (69). Flexible encoding at v2. v3 adds an error message that `kafka-protocol` cannot encode yet |
+| 47 | OffsetDelete | 0 | 0 | 0 | With a bridge: deletes offsets of a group. `GROUP_SUBSCRIBED_TO_TOPIC` (86) for a topic that a member subscribes to. See [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md). v1 is not in a Kafka release |
A request is accepted when `min_version ≤ api_version ≤ max_version` for that API key. Any other version for a listed key closes the connection (ApiVersions excepted - see Governance model above).
@@ -98,6 +100,8 @@ Use this table when configuring clients or generating wire fixtures with `kafka-
| 22 | InitProducerId | 0–5 | v2 |
| 32 | DescribeConfigs | 1–4 | v4 |
| 33 | AlterConfigs | 0–2 | v2 |
+| 42 | DeleteGroups | 0–2 | v2 |
+| 47 | OffsetDelete | 0 | never |
---
@@ -111,7 +115,7 @@ All API keys not listed above close the connection (see Governance model above)
| 29 | DescribeAcls | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised only while it is on ([`ACL_MAPPING.md`](ACL_MAPPING.md)) |
| 36 | SaslAuthenticate | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised only while it is on ([`AUTHENTICATION.md`](AUTHENTICATION.md)) |
| 24, 25, 26, 28 | AddPartitionsToTxn, AddOffsetsToTxn, EndTxn, TxnOffsetCommit | Transactions - not supported, see below |
-| 23, 27, 30, 31, 34, 35, 37+ | `OffsetForLeaderEpoch`, `WriteTxnMarkers`, `CreateAcls`/`DeleteAcls`, IncrementalAlterConfigs (44), etc. | Later issues |
+| 23, 27, 30, 31, 34, 35, 37-41, 43-46, 48+ | `OffsetForLeaderEpoch`, `WriteTxnMarkers`, `CreateAcls`/`DeleteAcls`, IncrementalAlterConfigs (44), etc. | Later issues |
| 21 | DeleteRecords | Not advertised on purpose, see below |
| 68 | ConsumerGroupHeartbeat | KIP-848 protocol, opt-in via `group.protocol=consumer`; the 4.0 default is still `classic` |
@@ -401,6 +405,8 @@ Offset persistence design ([#3540](https://github.com/apache/iggy/issues/3540)):
[#3548](https://github.com/apache/iggy/issues/3548); see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)
- [x] OffsetCommit (8), OffsetFetch (9) - [#3542](https://github.com/apache/iggy/issues/3542);
offsets live in Iggy, see [`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)
+- [x] DeleteGroups (42), OffsetDelete (47) - [#4511](https://github.com/apache/iggy/issues/4511);
+ see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)
### Phase 3+ — Auth, admin, tuning
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index 0a62468fd..76ae9dd60 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -84,6 +84,7 @@ file under `tests/` anymore.
| [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) |
| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition | No (needs the `iggy-server` binary - see Prerequisites) |
| [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) |
+| [`offset_groups_real_bridge_tests.rs`](../tests/offset_groups_real_bridge_tests.rs) | ListGroups (key 16), DescribeGroups (key 15), DeleteGroups (key 42) and OffsetDelete (key 47) through the whole handler against a real, spawned `iggy-server`. A group with offsets and no members is `Empty`, and a live group is listed once. Also the filters, a commit after the last scan, a new gateway, a native Iggy group, and `COORDINATOR_LOAD_IN_PROGRESS` (14) before any scan. DeleteGroups refuses a [...]
| [`phase1_e2e_tests.rs`](../tests/phase1_e2e_tests.rs) | [#3539](https://github.com/apache/iggy/issues/3539) acceptance: CreateTopics → Metadata → Produce → ListOffsets over real TCP against the real compiled `iggy-gateway-kafka` binary bridged to a real spawned `iggy-server` - the only suite that spawns the gateway as a process rather than calling `KafkaGateway::run` in-process | No (needs both binaries - see Prerequisites) |
`tests/common/` holds shared helpers (`codec.rs`, `fixtures.rs`, `gateway_process.rs`, `scope.rs`,
diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md
index 59f4f2f9d..33049f635 100644
--- a/gateways/kafka/docs/kafka_api_keys_reference.md
+++ b/gateways/kafka/docs/kafka_api_keys_reference.md
@@ -90,10 +90,11 @@ Key new minimums:
| 14 | **SyncGroup** | 1 | 5 | v4 | 🟠 Required Stub |
| 15 | **DescribeGroups** | 0 | 6 | v5 | Implemented |
| 16 | **ListGroups** | 1 | 5 | v3 | Implemented |
-| 42 | **DeleteGroups** | 1 | 2 | v2 | 🟡 Optional Stub |
+| 42 | **DeleteGroups** | 1 | 2 | v2 | Implemented |
-> DescribeGroups and ListGroups are implemented. They read this coordinator's in-memory groups.
-> See [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md).
+> DescribeGroups and ListGroups are implemented. They read this coordinator's in-memory groups, and
+> report a group that only holds offsets in Iggy as `Empty`. DeleteGroups, and OffsetDelete (47)
+> below, delete those offsets. See [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md).
---
@@ -183,7 +184,7 @@ Key new minimums:
| 43 | **ElectLeaders** | 0 | 2 | v2 | 🟡 Optional Stub |
| 45 | **AlterPartitionReassignments** | 0 | 0 | v0 | 🟡 Optional Stub |
| 46 | **ListPartitionReassignments** | 0 | 0 | v0 | 🟡 Optional Stub |
-| 47 | **OffsetDelete** | 0 | 0 | never | 🟡 Optional Stub |
+| 47 | **OffsetDelete** | 0 | 0 | never | Implemented |
| 57 | **UpdateFeatures** | 0 | 1 | v1 | 🟡 Optional Stub |
---
diff --git a/gateways/kafka/scripts/ci-wire-fixtures.sh b/gateways/kafka/scripts/ci-wire-fixtures.sh
index 636902224..ef1c32806 100755
--- a/gateways/kafka/scripts/ci-wire-fixtures.sh
+++ b/gateways/kafka/scripts/ci-wire-fixtures.sh
@@ -24,7 +24,7 @@ set -euo pipefail
FIXTURES_DIR="gateways/kafka/tools/kafka-tool/kafka_messages"
# API keys requested by api_handler_tests, version_firewall_tests, and server_e2e_tests.
-FIXTURE_API_KEYS=(0 1 2 8 9 10 11 12 13 14 15 16 19 20 22 32 33)
+FIXTURE_API_KEYS=(0 1 2 8 9 10 11 12 13 14 15 16 19 20 22 32 33 42 47)
usage() {
echo "Usage: $0 {generate|cleanup}" >&2
diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs
index b7108d135..95a4d65dd 100644
--- a/gateways/kafka/src/bridge/error.rs
+++ b/gateways/kafka/src/bridge/error.rs
@@ -124,14 +124,18 @@ impl BridgeError {
}
}
- /// [`Self::to_kafka_error_code`] for `OffsetCommit` and `OffsetFetch`. A failure that a retry
- /// can fix becomes `COORDINATOR_LOAD_IN_PROGRESS` (14): the Java client retries 14 in both
- /// APIs, but fails the call on 6, and on 7 in `OffsetFetch`. Offset calls are idempotent, so
- /// the unknown outcome behind 7 does no harm on a retry. Nor does `RequestTooOld`'s: the retry
- /// is a new request, which Iggy runs.
+ /// [`Self::to_kafka_error_code`] for the group offset APIs. A failure that a retry can fix
+ /// becomes `COORDINATOR_LOAD_IN_PROGRESS` (14): the Java client retries 14 in `OffsetCommit`
+ /// and `OffsetFetch`, but fails the call on 6, and on 7 in `OffsetFetch`. Offset calls are
+ /// idempotent, so the unknown outcome behind 7 does no harm on a retry. Nor does
+ /// `RequestTooOld`'s: the retry is a new request, which Iggy runs. `LifecycleBusy` ends when
+ /// the delete of a group on the topic ends.
#[must_use]
pub const fn to_offset_error_code(&self) -> i16 {
- if matches!(self, Self::Iggy(IggyError::RequestTooOld)) {
+ if matches!(
+ self,
+ Self::Iggy(IggyError::RequestTooOld | IggyError::LifecycleBusy)
+ ) {
return ERROR_COORDINATOR_LOAD_IN_PROGRESS;
}
match self.to_kafka_error_code() {
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_index.rs b/gateways/kafka/src/bridge/iggy_bridge/group_index.rs
new file mode 100644
index 000000000..c7898b5dc
--- /dev/null
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_index.rs
@@ -0,0 +1,392 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF 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 Kafka groups that hold offsets in Iggy, for the group admin APIs. Finding them costs one
+//! Iggy call per topic, so one scan answers every request for a while.
+
+use std::collections::BTreeSet;
+use std::future::Future;
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::{Semaphore, watch};
+use tokio::time::{Instant, timeout_at};
+use tracing::warn;
+
+use crate::bridge::error::BridgeError;
+
+/// How long one scan answers for. kafka-ui lists groups every 30 s, and a scan costs one Iggy call
+/// per topic.
+const SCAN_MAX_AGE: Duration = Duration::from_secs(10);
+
+/// The Kafka groups one scan found holding offsets in Iggy.
+#[derive(Debug, Clone)]
+pub struct OffsetGroups {
+ /// When the scan began.
+ started: Instant,
+ groups: BTreeSet<String>,
+}
+
+impl OffsetGroups {
+ #[must_use]
+ pub fn contains(&self, group_id: &str) -> bool {
+ self.groups.contains(group_id)
+ }
+
+ /// The group ids, in order.
+ pub fn iter(&self) -> impl Iterator<Item = &str> {
+ self.groups.iter().map(String::as_str)
+ }
+
+ fn is_fresh(&self) -> bool {
+ self.started.elapsed() < SCAN_MAX_AGE
+ }
+}
+
+/// The groups of the last scan that ended, and at most one scan that runs.
+pub(super) struct GroupIndex {
+ published: Arc<watch::Sender<Published>>,
+ /// One permit, held by the scan that runs.
+ scanning: Arc<Semaphore>,
+}
+
+#[derive(Default)]
+struct Published {
+ /// Scans that ended, with groups or without. A caller waits for this count to move.
+ ended: u64,
+ /// The groups of the last scan that found them.
+ groups: Option<Arc<OffsetGroups>>,
+ /// The groups this gateway stored offsets for (`true`) or deleted (`false`) while a scan
+ /// runs, in order. `None` while none runs. The scan can list a topic before such a change,
+ /// so the changes go on top of what it finds.
+ changes: Option<Vec<(String, bool)>>,
+}
+
+impl GroupIndex {
+ pub(super) fn new() -> Self {
+ Self {
+ published: Arc::new(watch::Sender::new(Published::default())),
+ scanning: Arc::new(Semaphore::new(1)),
+ }
+ }
+
+ /// The groups of a scan that began less than `SCAN_MAX_AGE` ago. Otherwise starts a scan with
+ /// `scan`, unless one runs, and waits for the scan that runs until `deadline`.
+ ///
+ /// A scan that fails, or that does not end by `deadline`, leaves the groups of the last scan
+ /// that found them, and those are the answer. `None` when no scan found groups yet.
+ pub(super) async fn groups<F, S>(&self, deadline: Instant, scan: F) -> Option<Arc<OffsetGroups>>
+ where
+ F: FnOnce() -> S,
+ S: Future<Output = Result<BTreeSet<String>, BridgeError>> + Send + 'static,
+ {
+ let mut published = self.published.subscribe();
+ let ended = {
+ let current = published.borrow_and_update();
+ if let Some(groups) = current.groups.as_ref().filter(|groups| groups.is_fresh()) {
+ return Some(Arc::clone(groups));
+ }
+ current.ended
+ };
+ self.start(scan);
+ // A timeout leaves the groups of the last scan as the answer.
+ let _ = timeout_at(
+ deadline,
+ published.wait_for(|current| current.ended > ended),
+ )
+ .await;
+ published.borrow().groups.clone()
+ }
+
+ /// Adds `group` to the groups of the last scan. A commit of this gateway created its offsets,
+ /// so it shows before the next scan.
+ pub(super) fn note(&self, group: &str) {
+ self.change(group, true);
+ }
+
+ /// Removes `group` from the groups of the last scan. This gateway deleted its offsets, so it
+ /// goes before the next scan.
+ pub(super) fn forget(&self, group: &str) {
+ self.change(group, false);
+ }
+
+ fn change(&self, group: &str, holds: bool) {
+ self.published.send_if_modified(|current| {
+ if let Some(changes) = current.changes.as_mut() {
+ changes.push((group.to_owned(), holds));
+ }
+ let Some(groups) = current.groups.as_mut() else {
+ return false;
+ };
+ if groups.contains(group) == holds {
+ return false;
+ }
+ apply(&mut Arc::make_mut(groups).groups, group.to_owned(), holds);
+ true
+ });
+ }
+
+ /// Runs `scan` in a task of its own, so it ends even when no caller waits for it any more.
+ /// Does nothing while another scan runs.
+ fn start<F, S>(&self, scan: F)
+ where
+ F: FnOnce() -> S,
+ S: Future<Output = Result<BTreeSet<String>, BridgeError>> + Send + 'static,
+ {
+ let Ok(permit) = Arc::clone(&self.scanning).try_acquire_owned() else {
+ return;
+ };
+ let published = Arc::clone(&self.published);
+ published.send_if_modified(|current| {
+ current.changes = Some(Vec::new());
+ false
+ });
+ let started = Instant::now();
+ let scanning = scan();
+ tokio::spawn(async move {
+ let found = scanning.await;
+ if let Err(error) = &found {
+ warn!(%error, "could not list the Kafka groups that hold offsets in Iggy");
+ }
+ published.send_modify(|current| {
+ current.ended += 1;
+ let changes = current.changes.take().unwrap_or_default();
+ if let Ok(mut groups) = found {
+ for (group, holds) in changes {
+ apply(&mut groups, group, holds);
+ }
+ current.groups = Some(Arc::new(OffsetGroups { started, groups }));
+ }
+ });
+ drop(permit);
+ });
+ }
+}
+
+fn apply(groups: &mut BTreeSet<String>, group: String, holds: bool) {
+ if holds {
+ groups.insert(group);
+ } else {
+ groups.remove(&group);
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::atomic::{AtomicUsize, Ordering};
+
+ use iggy::prelude::IggyError;
+ use tokio::sync::oneshot;
+
+ use super::*;
+
+ fn later() -> Instant {
+ Instant::now() + Duration::from_secs(5)
+ }
+
+ fn found(groups: &[&str]) -> BTreeSet<String> {
+ groups.iter().map(|group| (*group).to_owned()).collect()
+ }
+
+ fn names(groups: Option<Arc<OffsetGroups>>) -> Vec<String> {
+ groups
+ .expect("a scan found groups")
+ .iter()
+ .map(str::to_owned)
+ .collect()
+ }
+
+ /// A scan that counts its runs and finds `groups`.
+ fn counted(
+ runs: &Arc<AtomicUsize>,
+ groups: &'static [&'static str],
+ ) -> impl FnOnce() -> std::future::Ready<Result<BTreeSet<String>, BridgeError>> {
+ let runs = Arc::clone(runs);
+ move || {
+ runs.fetch_add(1, Ordering::SeqCst);
+ std::future::ready(Ok(found(groups)))
+ }
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_fresh_scan_should_answer_from_it_without_scanning_again() {
+ let index = GroupIndex::new();
+ let runs = Arc::new(AtomicUsize::new(0));
+
+ let first = index.groups(later(), counted(&runs, &["orders"])).await;
+ tokio::time::advance(SCAN_MAX_AGE.saturating_sub(Duration::from_millis(1))).await;
+ let second = index.groups(later(), counted(&runs, &["payments"])).await;
+
+ assert_eq!(names(first), ["orders"]);
+ assert_eq!(names(second), ["orders"]);
+ assert_eq!(runs.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_scan_older_than_its_max_age_should_scan_again() {
+ let index = GroupIndex::new();
+ let runs = Arc::new(AtomicUsize::new(0));
+
+ index.groups(later(), counted(&runs, &["orders"])).await;
+ tokio::time::advance(SCAN_MAX_AGE).await;
+ let second = index.groups(later(), counted(&runs, &["payments"])).await;
+
+ assert_eq!(names(second), ["payments"]);
+ assert_eq!(runs.load(Ordering::SeqCst), 2);
+ }
+
+ /// Callers that come while a scan runs wait for that scan rather than start their own.
+ #[tokio::test(start_paused = true)]
+ async fn given_callers_during_a_scan_should_share_that_scan() {
+ let index = GroupIndex::new();
+ let runs = Arc::new(AtomicUsize::new(0));
+ let (release, released) = oneshot::channel::<()>();
+ let first_runs = Arc::clone(&runs);
+ let blocked = move || {
+ first_runs.fetch_add(1, Ordering::SeqCst);
+ async move {
+ let _ = released.await;
+ Ok(found(&["orders"]))
+ }
+ };
+
+ let (first, second, ()) = tokio::join!(
+ index.groups(later(), blocked),
+ index.groups(later(), counted(&runs, &["payments"])),
+ async {
+ tokio::task::yield_now().await;
+ release.send(()).expect("the scan waits for its release");
+ },
+ );
+
+ assert_eq!(names(first), ["orders"]);
+ assert_eq!(names(second), ["orders"]);
+ assert_eq!(runs.load(Ordering::SeqCst), 1);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_failed_scan_should_answer_the_groups_of_the_last_scan_that_found_them() {
+ let index = GroupIndex::new();
+ let runs = Arc::new(AtomicUsize::new(0));
+ index.groups(later(), counted(&runs, &["orders"])).await;
+ tokio::time::advance(SCAN_MAX_AGE).await;
+
+ let failed = index
+ .groups(later(), || async {
+ Err(BridgeError::Iggy(IggyError::CannotEstablishConnection))
+ })
+ .await;
+
+ assert_eq!(names(failed), ["orders"]);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_no_scan_that_found_groups_when_a_scan_fails_should_answer_none() {
+ let index = GroupIndex::new();
+
+ let failed = index
+ .groups(later(), || async {
+ Err(BridgeError::Iggy(IggyError::CannotEstablishConnection))
+ })
+ .await;
+
+ assert!(failed.is_none());
+ }
+
+ /// A scan past the deadline of its caller keeps running, and the next caller gets its groups
+ /// without another scan.
+ #[tokio::test(start_paused = true)]
+ async fn given_a_scan_past_the_deadline_should_finish_for_the_next_caller() {
+ let index = GroupIndex::new();
+ let runs = Arc::new(AtomicUsize::new(0));
+ let (release, released) = oneshot::channel::<()>();
+
+ let late = index
+ .groups(
+ Instant::now() + Duration::from_millis(20),
+ move || async move {
+ let _ = released.await;
+ Ok(found(&["orders"]))
+ },
+ )
+ .await;
+ assert!(late.is_none(), "no scan found groups by the deadline");
+
+ release.send(()).expect("the scan still runs");
+ tokio::task::yield_now().await;
+ let next = index.groups(later(), counted(&runs, &["payments"])).await;
+
+ assert_eq!(names(next), ["orders"]);
+ assert_eq!(
+ runs.load(Ordering::SeqCst),
+ 0,
+ "the scan that ended is fresh"
+ );
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_forgotten_group_should_leave_it_out_before_the_next_scan() {
+ let index = GroupIndex::new();
+ let runs = Arc::new(AtomicUsize::new(0));
+ index
+ .groups(later(), counted(&runs, &["orders", "payments"]))
+ .await;
+
+ index.forget("orders");
+ index.forget("never-scanned");
+ let groups = index.groups(later(), counted(&runs, &[])).await;
+
+ assert_eq!(names(groups), ["payments"]);
+ assert_eq!(runs.load(Ordering::SeqCst), 1);
+ }
+
+ /// A scan can list a topic before this gateway deletes or stores offsets there. What it finds
+ /// must not undo those changes.
+ #[tokio::test(start_paused = true)]
+ async fn given_changes_while_a_scan_runs_should_apply_them_to_what_it_finds() {
+ let index = GroupIndex::new();
+ let (release, released) = oneshot::channel::<()>();
+ let stale = move || async move {
+ let _ = released.await;
+ Ok(found(&["deleted", "kept"]))
+ };
+
+ let (groups, ()) = tokio::join!(index.groups(later(), stale), async {
+ tokio::task::yield_now().await;
+ index.forget("deleted");
+ index.note("stored");
+ release.send(()).expect("the scan waits for its release");
+ });
+
+ assert_eq!(names(groups), ["kept", "stored"]);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_noted_group_should_answer_it_before_the_next_scan() {
+ let index = GroupIndex::new();
+ let runs = Arc::new(AtomicUsize::new(0));
+ index.note("before-any-scan");
+ index.groups(later(), counted(&runs, &["orders"])).await;
+
+ index.note("payments");
+ index.note("orders");
+ let groups = index.groups(later(), counted(&runs, &[])).await;
+
+ assert_eq!(names(groups), ["orders", "payments"]);
+ assert_eq!(runs.load(Ordering::SeqCst), 1);
+ }
+}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
index 3f71a0949..6c54d9942 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
@@ -17,9 +17,11 @@
//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+use std::collections::BTreeSet;
use std::future::Future;
use std::hash::{DefaultHasher, Hash, Hasher};
use std::sync::Arc;
+use std::time::Duration;
use futures::future::join_all;
use iggy::prelude::{
@@ -27,17 +29,28 @@ use iggy::prelude::{
};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::time::{Instant, timeout, timeout_at};
-use tracing::info;
+use tracing::{debug, info};
use super::{
- IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget, in_slot, permit_by,
+ IggyBridge, KafkaTopicMetadata, LazyClient, OffsetGroups, SLOT_LIMIT, TopicTarget, in_slot,
+ permit_by,
};
use crate::bridge::error::BridgeError;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::ERROR_UNKNOWN_TOPIC_OR_PARTITION;
/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
/// apart from a native Iggy group of the same name.
pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+/// How long one scan for the groups that hold offsets may take. It keeps running after the request
+/// that started it stops waiting.
+const GROUP_SCAN_LIMIT: Duration = Duration::from_secs(30);
+
+/// The first pause, and the longest, between two checks of whether Iggy finished deleting a group.
+const DELETE_CHECK_FIRST: Duration = Duration::from_millis(5);
+const DELETE_CHECK_MAX: Duration = Duration::from_millis(100);
+
/// Offset calls that run at once, each slot on clients of its own, so offset calls never wait
/// behind a Produce send.
const OFFSET_SLOTS: usize = 4;
@@ -90,7 +103,8 @@ impl IggyBridge {
/// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
/// applied counts as done.
///
- /// Commits on one slot start one at a time, in the order given.
+ /// Commits on one slot start one at a time, in the order given. A stored offset adds `group`
+ /// to [`Self::offset_groups`] at once.
///
/// # Errors
///
@@ -109,7 +123,7 @@ impl IggyBridge {
};
// `join_all` polls each call once, in order, before any again, so calls on one slot
// queue for it in order.
- join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let results = join_all(commits.iter().map(|&(topic, partition, offset)| {
let (names, stream_id, topic_id) = (
Arc::clone(&names),
topic.stream_id.clone(),
@@ -120,7 +134,15 @@ impl IggyBridge {
};
self.commit_in_slot(slot_of(topic, partition), calls, commit)
}))
- .await
+ .await;
+ let stored = commits
+ .iter()
+ .zip(&results)
+ .any(|(&(_, _, offset), result)| offset >= 0 && result.is_ok());
+ if stored {
+ self.group_index.note(group);
+ }
+ results
}
/// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
@@ -185,6 +207,89 @@ impl IggyBridge {
self.read_in_slot(slot_of(topic, 0), calls, lookup).await
}
+ /// Deletes the Iggy group that holds the offsets of each Kafka group of `groups`, on every
+ /// Kafka topic. Iggy deletes the offsets with it. One result per group, in order: whether any
+ /// topic held the group.
+ ///
+ /// Lists the topics, then the Iggy groups of each topic, so it costs one call per topic, plus
+ /// at least two per deletion. Iggy deletes a group on each partition of the topic, and lists
+ /// it until every partition did, so a deletion waits for that. A topic that the bridge user
+ /// may not read, or that was deleted since the listing, holds none. A group whose deletions
+ /// all ended leaves [`Self::offset_groups`] at once.
+ ///
+ /// # Errors
+ ///
+ /// The outer error holds for every group: the listing failed, and nothing was deleted. Per
+ /// group, as [`Self::commit_group_offsets`], and [`BridgeError::Timeout`] also when Iggy has
+ /// not finished a deletion by the deadline of `calls`.
+ pub async fn delete_offset_groups(
+ &self,
+ groups: &[&str],
+ calls: &OffsetCalls,
+ ) -> Result<Vec<Result<bool, BridgeError>>, BridgeError> {
+ let topics = self.list_kafka_topics_in_slot(calls).await?;
+ let listings = join_all(
+ topics
+ .iter()
+ .map(|topic| self.offset_groups_on(&topic.kafka_topic, calls)),
+ )
+ .await;
+ // The index in `groups` and the Kafka topic of each Iggy group to delete.
+ let mut deletions = Vec::new();
+ for (topic, listing) in topics.iter().zip(listings) {
+ let found = match listing {
+ Ok(found) => found,
+ Err(error) if holds_no_offset_groups(&error) => continue,
+ Err(error) => return Err(error),
+ };
+ for (at, group) in groups.iter().enumerate() {
+ if found.iter().any(|name| name == group) {
+ deletions.push((at, topic.kafka_topic.as_str()));
+ }
+ }
+ }
+ let deleted =
+ join_all(deletions.iter().map(|&(at, kafka_topic)| {
+ self.delete_offset_group_on(groups[at], kafka_topic, calls)
+ }))
+ .await;
+ let mut results: Vec<Result<bool, BridgeError>> =
+ groups.iter().map(|_| Ok(false)).collect();
+ for (&(at, _), result) in deletions.iter().zip(deleted) {
+ match result {
+ Ok(()) if results[at].is_ok() => results[at] = Ok(true),
+ Ok(()) => {}
+ Err(error) => results[at] = Err(error),
+ }
+ }
+ for (group, result) in groups.iter().zip(&results) {
+ if result.is_ok() {
+ self.group_index.forget(group);
+ }
+ }
+ Ok(results)
+ }
+
+ /// The partition count of `kafka_topic`, read on an offset slot. `None` when the topic does
+ /// not exist.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::get_kafka_topic`]. [`BridgeError::Timeout`] past the deadline of `calls`.
+ pub async fn kafka_topic_partitions(
+ &self,
+ kafka_topic: &str,
+ calls: &OffsetCalls,
+ ) -> Result<Option<u32>, BridgeError> {
+ let topic = self.topic_target(kafka_topic)?;
+ let lookup = |client: Arc<IggyClient>| async move {
+ let details = self.get_kafka_topic_on(&client, kafka_topic).await?;
+ Ok(details.map(|details| details.partitions_count))
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(&topic, 0), calls, lookup).await
+ }
+
/// [`Self::list_kafka_topics`] on an offset slot's read client, so it never waits behind
/// Produce.
///
@@ -200,6 +305,120 @@ impl IggyBridge {
self.read_in_slot(LISTING_SLOT, calls, listing).await
}
+ /// The Kafka groups that hold offsets in Iggy, from a scan that began less than 10 s ago, plus
+ /// the groups that this gateway stored offsets for since. A scan lists the Kafka topics, then
+ /// the Iggy groups of each topic on the offset read slots, so it costs one call per topic.
+ ///
+ /// Callers that come while a scan runs wait for it, until `deadline`. The scan keeps running
+ /// after that and serves the next caller. A scan that fails, or does not end in time, leaves
+ /// the groups of the last scan that found them as the answer. `None` when no scan found them
+ /// yet.
+ pub async fn offset_groups(self: &Arc<Self>, deadline: Instant) -> Option<Arc<OffsetGroups>> {
+ let bridge = Arc::clone(self);
+ let scan = move || async move {
+ let calls = OffsetCalls::new(Instant::now() + GROUP_SCAN_LIMIT);
+ bridge.scan_offset_groups(&calls).await
+ };
+ self.group_index.groups(deadline, scan).await
+ }
+
+ /// Every Kafka group with offsets on a Kafka topic. A topic deleted since the listing holds
+ /// none, and so does one the bridge user may not read.
+ async fn scan_offset_groups(
+ &self,
+ calls: &OffsetCalls,
+ ) -> Result<BTreeSet<String>, BridgeError> {
+ let topics = self.list_kafka_topics_in_slot(calls).await?;
+ let listings = join_all(
+ topics
+ .iter()
+ .map(|topic| self.offset_groups_on(&topic.kafka_topic, calls)),
+ )
+ .await;
+ let mut groups = BTreeSet::new();
+ for (topic, listing) in topics.iter().zip(listings) {
+ match listing {
+ Ok(found) => groups.extend(found),
+ Err(error) if holds_no_offset_groups(&error) => {
+ debug!(
+ topic = topic.kafka_topic.as_str(),
+ %error,
+ "skipping a topic in the scan for groups"
+ );
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ Ok(groups)
+ }
+
+ /// The Kafka groups with offsets on `kafka_topic`.
+ async fn offset_groups_on(
+ &self,
+ kafka_topic: &str,
+ calls: &OffsetCalls,
+ ) -> Result<Vec<String>, BridgeError> {
+ let topic = self.topic_target(kafka_topic)?;
+ // Any slot will do. Partition 0's spreads the topics over the slots.
+ let slot = slot_of(&topic, 0);
+ let listing = |client: Arc<IggyClient>| async move {
+ let groups = client
+ .get_consumer_groups(&topic.stream_id, &topic.topic_id)
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(groups
+ .iter()
+ .filter_map(|group| kafka_group_of(&group.name))
+ .collect())
+ };
+ self.read_in_slot(slot, calls, listing).await
+ }
+
+ /// Deletes the Iggy group of Kafka group `group` on `kafka_topic`, then waits until the topic
+ /// no longer lists it. A group that is not there counts as deleted.
+ async fn delete_offset_group_on(
+ &self,
+ group: &str,
+ kafka_topic: &str,
+ calls: &OffsetCalls,
+ ) -> Result<(), BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let topic = self.topic_target(kafka_topic)?;
+ let (stream_id, topic_id) = (topic.stream_id.clone(), topic.topic_id.clone());
+ let delete = move |client: Arc<IggyClient>| async move {
+ match client
+ .delete_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ {
+ Err(
+ IggyError::ConsumerGroupIdNotFound(..)
+ | IggyError::ConsumerGroupNameNotFound(..),
+ ) => Ok(()),
+ deleted => landed(deleted),
+ }
+ };
+ // Partition 0's slot, so the delete follows the commits of that partition.
+ self.commit_in_slot(slot_of(&topic, 0), calls, delete)
+ .await?;
+ let mut pause = DELETE_CHECK_FIRST;
+ loop {
+ match self.holds_offset_group(group, &topic, calls).await {
+ Ok(false) => return Ok(()),
+ Ok(true) => {}
+ Err(error) if holds_no_offset_groups(&error) => return Ok(()),
+ Err(error) => return Err(error),
+ }
+ if Instant::now() + pause >= calls.deadline() {
+ return Err(BridgeError::Timeout);
+ }
+ tokio::time::sleep(pause).await;
+ pause = (pause * 2).min(DELETE_CHECK_MAX);
+ }
+ }
+
/// Runs `call` with the commit client of `slot`, and waits for it until the deadline of
/// `calls`. The call keeps the slot until it ends, or for at most `SLOT_LIMIT`, even after its
/// caller stops waiting. A call that has not started by the deadline never starts.
@@ -310,6 +529,13 @@ fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
(first + step) % OFFSET_SLOTS
}
+/// Whether a failed listing of the Iggy groups of a topic means that the topic holds no Kafka
+/// offsets here: the bridge user may not read it, or it was deleted since the topic listing.
+const fn holds_no_offset_groups(error: &BridgeError) -> bool {
+ matches!(error, BridgeError::Iggy(IggyError::Unauthorized))
+ || error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION
+}
+
/// One refusal per call, for a group id that names no valid Iggy group.
fn refused_all<T>(count: usize, error: &IggyError) -> Vec<Result<T, BridgeError>> {
(0..count)
@@ -336,6 +562,13 @@ impl OffsetNames {
}
}
+/// The Kafka group whose offsets the Iggy group `name` holds. `None` for any other Iggy group.
+fn kafka_group_of(name: &str) -> Option<String> {
+ name.strip_prefix(OFFSET_GROUP_PREFIX)
+ .filter(|group| is_valid_group_id(group))
+ .map(str::to_owned)
+}
+
/// One commit on `client`. A store that finds no Iggy group creates it and stores again.
async fn commit_on(
client: &IggyClient,
@@ -481,6 +714,13 @@ mod tests {
);
}
+ #[test]
+ fn given_an_iggy_group_name_only_an_offset_group_should_name_a_kafka_group() {
+ assert_eq!(kafka_group_of("kafka.cg.orders").as_deref(), Some("orders"));
+ assert_eq!(kafka_group_of("orders"), None, "a native Iggy group");
+ assert_eq!(kafka_group_of("kafka.cg."), None, "an empty Kafka group id");
+ }
+
#[test]
fn given_a_kafka_group_the_key_should_be_its_prefixed_external_group() {
let names = OffsetNames::new("orders").unwrap();
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index fc068e8ec..d9bbc94c0 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -32,6 +32,7 @@ use crate::bridge::error::BridgeError;
use crate::bridge::topic_map::validate_kafka_topic_name;
mod fetch;
+mod group_index;
mod group_offsets;
mod offsets;
mod produce;
@@ -39,6 +40,8 @@ mod topics;
use fetch::{FetchPool, LazyClient};
pub(crate) use fetch::{FetchSlot, PartitionProbe, TopicProbe};
+use group_index::GroupIndex;
+pub use group_index::OffsetGroups;
use group_offsets::OffsetPool;
pub use group_offsets::{OFFSET_GROUP_PREFIX, OffsetCalls};
use produce::ProducePool;
@@ -150,6 +153,8 @@ pub struct IggyBridge {
probe_client: LazyClient,
/// Group offset slots, each with a commit client and a read client. See `OffsetPool`.
offset_pool: OffsetPool,
+ /// The Kafka groups that hold offsets. See `offset_groups`.
+ group_index: GroupIndex,
}
/// The Iggy stream and topic one Kafka topic maps to. Resolve once per topic, use many times.
@@ -190,6 +195,7 @@ impl IggyBridge {
fetch_pool: Arc::new(FetchPool::new()),
probe_client: LazyClient::default(),
offset_pool: OffsetPool::new(),
+ group_index: GroupIndex::new(),
})
}
@@ -289,6 +295,7 @@ impl IggyBridge {
fetch_pool: Arc::new(FetchPool::new()),
probe_client: LazyClient::default(),
offset_pool: OffsetPool::new(),
+ group_index: GroupIndex::new(),
}
}
}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/topics.rs b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
index b860af435..d8df45985 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/topics.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
@@ -305,7 +305,7 @@ impl IggyBridge {
}
/// [`Self::get_kafka_topic`] on `client`.
- async fn get_kafka_topic_on(
+ pub(super) async fn get_kafka_topic_on(
&self,
client: &IggyClient,
kafka_topic: &str,
diff --git a/gateways/kafka/src/bridge/mod.rs b/gateways/kafka/src/bridge/mod.rs
index 8b25a8ae7..5bb822400 100644
--- a/gateways/kafka/src/bridge/mod.rs
+++ b/gateways/kafka/src/bridge/mod.rs
@@ -29,7 +29,7 @@ pub mod topic_map;
pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, ProduceSessionLimits};
pub use error::BridgeError;
pub use iggy_bridge::{
- IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, ProduceTurn, Producer,
- StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget,
+ IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, OffsetGroups, ProduceTurn,
+ Producer, StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget,
};
pub use topic_map::{TopicMapping, TopicOverride};
diff --git a/gateways/kafka/src/group/mod.rs b/gateways/kafka/src/group/mod.rs
index 4af5d481a..1e27ec003 100644
--- a/gateways/kafka/src/group/mod.rs
+++ b/gateways/kafka/src/group/mod.rs
@@ -19,8 +19,9 @@
//!
//! [`GroupCoordinator`] owns every group this gateway instance coordinates and is the only
//! module that awaits: `FindCoordinator`/`JoinGroup`/`Heartbeat`/`LeaveGroup`/`SyncGroup`/
-//! `DescribeGroups`/`ListGroups`/`OffsetCommit` handlers translate wire messages into the request types here,
-//! and `state` holds the synchronous state machine those requests drive.
+//! `DescribeGroups`/`ListGroups`/`OffsetCommit`/`DeleteGroups`/`OffsetDelete` handlers translate
+//! wire messages into the request types here, and `state` holds the synchronous state machine
+//! those requests drive.
//!
//! Membership is process memory, not Iggy state. Two gateway instances fronting one Iggy cluster
//! therefore coordinate two independent groups under one name; see `docs/CONSUMER_GROUPS.md`.
@@ -29,7 +30,7 @@ mod state;
pub use state::{MAX_GROUP_ID_BYTES, is_valid_group_id};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::time::Duration;
use bytes::Bytes;
@@ -53,6 +54,16 @@ pub const GROUP_STATE_COMPLETING_REBALANCE: &str = "CompletingRebalance";
/// `DescribeGroups` / `ListGroups` state string for a group whose assignment is in effect.
pub const GROUP_STATE_STABLE: &str = "Stable";
+/// Kafka's `ConsumerProtocol.PROTOCOL_TYPE`, the protocol type of a consumer group.
+pub const CONSUMER_PROTOCOL_TYPE: &str = "consumer";
+
+/// Protocol type of a group that only holds offsets in Iggy.
+///
+/// Kafka keeps `consumer` for a group whose consumers left, and Iggy stores no protocol type to say
+/// otherwise. Kafka's admin client takes a group as a consumer group with `consumer` or an empty
+/// type.
+pub const OFFSETS_ONLY_PROTOCOL_TYPE: &str = CONSUMER_PROTOCOL_TYPE;
+
/// v6 `error_message` when one `DescribeGroups` group does not fit the response budget.
///
/// `UNKNOWN_SERVER_ERROR` is not retried by `kafka-consumer-groups.sh`, so the tool prints this
@@ -358,6 +369,19 @@ pub struct GroupListing {
pub state: &'static str,
}
+impl GroupListing {
+ /// A group with no members here that holds offsets in Iggy. Kafka keeps a group whose
+ /// consumers left, and lists it as `Empty`.
+ #[must_use]
+ pub const fn offsets_only(group_id: StrBytes) -> Self {
+ Self {
+ group_id,
+ protocol_type: StrBytes::from_static_str(OFFSETS_ONLY_PROTOCOL_TYPE),
+ state: GROUP_STATE_EMPTY,
+ }
+ }
+}
+
/// One group the coordinator currently holds, as `DescribeGroups` reports it.
#[derive(Debug, Clone)]
pub struct GroupDescription {
@@ -374,6 +398,66 @@ pub struct GroupDescription {
pub members: Vec<MemberDescription>,
}
+impl GroupDescription {
+ /// [`GroupListing::offsets_only`] as `DescribeGroups` reports it: `Empty`, with no protocol
+ /// and no members.
+ #[must_use]
+ pub const fn offsets_only(group_id: StrBytes) -> Self {
+ Self {
+ group_id,
+ error: ERROR_NONE,
+ state: GROUP_STATE_EMPTY,
+ protocol_type: StrBytes::from_static_str(OFFSETS_ONLY_PROTOCOL_TYPE),
+ protocol_name: None,
+ members: Vec::new(),
+ }
+ }
+}
+
+/// What `DeleteGroups` found for one group on this coordinator, by Kafka's
+/// `ClassicGroup.validateDeleteGroup`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum GroupRemoval {
+ /// Not on this coordinator.
+ Absent,
+ /// Has joined members. Kafka refuses with `NON_EMPTY_GROUP`.
+ Live,
+ /// Had no joined members, only ids that wait to join, and is removed.
+ Removed,
+}
+
+/// What `OffsetDelete` may do with the offsets of one group on this coordinator, by Kafka's
+/// `ClassicGroup.validateOffsetDelete`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum OffsetDeletion {
+ /// Not on this coordinator.
+ Absent,
+ /// Has members of a protocol other than `consumer`. Kafka refuses with `NON_EMPTY_GROUP`.
+ Refused,
+ /// On this coordinator. The offsets of a topic that a member subscribes to stay.
+ Allowed(Subscribed),
+}
+
+/// The topics that the members of a group subscribe to.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Subscribed {
+ /// Empty for a group with no joined members.
+ Topics(HashSet<String>),
+ /// Every topic. A member's subscription does not decode, or the group has not chosen a
+ /// protocol yet. Kafka's `ClassicGroup.isSubscribedToTopic` does the same.
+ All,
+}
+
+impl Subscribed {
+ #[must_use]
+ pub fn contains(&self, topic: &str) -> bool {
+ match self {
+ Self::Topics(topics) => topics.contains(topic),
+ Self::All => true,
+ }
+ }
+}
+
/// The answer for one `LeavingMember`.
#[derive(Debug, Clone)]
pub struct LeftMember {
@@ -571,6 +655,23 @@ impl GroupCoordinator {
state::commit_step(&mut groups, request, Instant::now())
}
+ /// Removes each group of `group_ids` that has no joined members, as `DeleteGroups` does. One
+ /// answer per id, in order. Never parks.
+ pub async fn remove_empty(&self, group_ids: &[StrBytes]) -> Vec<GroupRemoval> {
+ let mut groups = self.groups.lock().await;
+ let now = Instant::now();
+ group_ids
+ .iter()
+ .map(|group_id| state::remove_empty(&mut groups, group_id, now))
+ .collect()
+ }
+
+ /// What `OffsetDelete` may do with the offsets of `group_id`. Never parks.
+ pub async fn offset_deletion(&self, group_id: &StrBytes) -> OffsetDeletion {
+ let mut groups = self.groups.lock().await;
+ state::offset_deletion(&mut groups, group_id, Instant::now())
+ }
+
/// How long an `OffsetCommit` or `OffsetFetch` may hold its connection. A heartbeat there
/// waits behind it. So it takes half the shortest session, and at most half of what is left
/// of the session of `member`, the member that heartbeats on that connection. Never parks.
diff --git a/gateways/kafka/src/group/state.rs b/gateways/kafka/src/group/state.rs
index 9106e6daa..3937552e0 100644
--- a/gateways/kafka/src/group/state.rs
+++ b/gateways/kafka/src/group/state.rs
@@ -23,8 +23,8 @@
//! across an `.await`.
//!
//! A group with no joined members is reported as Kafka's `Empty`. One with neither members nor
-//! pending member ids is dropped: offsets live in Iggy, so nothing here is worth keeping.
-//! TODO(#3542): keep an offsets-only group and report `Empty` once those offsets are stored here.
+//! pending member ids is dropped. Its offsets stay in Iggy, and the `ListGroups` and
+//! `DescribeGroups` handlers report it as `Empty` from there.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::time::Duration;
@@ -36,11 +36,12 @@ use tokio::time::Instant;
use uuid::Uuid;
use crate::group::{
- CommitRequest, DESCRIBE_RESPONSE_TOO_LARGE, GROUP_STATE_COMPLETING_REBALANCE,
- GROUP_STATE_EMPTY, GROUP_STATE_PREPARING_REBALANCE, GROUP_STATE_STABLE, GroupCoordinatorConfig,
- GroupDescription, GroupListing, GroupMember, JoinRequest, JoinResult, JoinedMember,
- LeaveRequest, LeaveResult, LeavingMember, LeftMember, MemberDescription, SyncRequest,
- SyncResult, owned_str,
+ CONSUMER_PROTOCOL_TYPE, CommitRequest, DESCRIBE_RESPONSE_TOO_LARGE,
+ GROUP_STATE_COMPLETING_REBALANCE, GROUP_STATE_EMPTY, GROUP_STATE_PREPARING_REBALANCE,
+ GROUP_STATE_STABLE, GroupCoordinatorConfig, GroupDescription, GroupListing, GroupMember,
+ GroupRemoval, JoinRequest, JoinResult, JoinedMember, LeaveRequest, LeaveResult, LeavingMember,
+ LeftMember, MemberDescription, OFFSETS_ONLY_PROTOCOL_TYPE, OffsetDeletion, Subscribed,
+ SyncRequest, SyncResult, owned_str,
};
use crate::protocol::api::{
API_KEY_DESCRIBE_GROUPS, ERROR_COORDINATOR_NOT_AVAILABLE, ERROR_FENCED_INSTANCE_ID,
@@ -809,6 +810,22 @@ impl GroupState {
}
total
}
+
+ /// The topics in each member's subscription for the selected protocol.
+ fn subscribed(&self) -> Subscribed {
+ let Some(protocol_name) = &self.protocol_name else {
+ return Subscribed::All;
+ };
+ let mut topics = HashSet::new();
+ for member in self.members.values() {
+ let Some(subscription) = subscription_topics(&member.metadata_for(protocol_name))
+ else {
+ return Subscribed::All;
+ };
+ topics.extend(subscription);
+ }
+ Subscribed::Topics(topics)
+ }
}
/// Tick every group and drop the ones that emptied, returning how many were reclaimed.
@@ -842,8 +859,7 @@ fn tick_group(groups: &mut Groups, group_id: &StrBytes, now: Instant) -> bool {
};
group.tick(now);
if group.is_empty() {
- // TODO(#3542): keep a group that only holds committed offsets and report Empty.
- // Offsets are not stored here yet, so nothing remains that a later join would need.
+ // Its offsets stay in Iggy, and the admin APIs find the group there.
groups.remove(group_id);
return false;
}
@@ -1332,10 +1348,85 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
+/// Kafka's `ClassicGroup.validateDeleteGroup`. A group with no joined members is `Empty`, so it
+/// goes, with the ids that wait to join it. A join with one of those ids gets
+/// `UNKNOWN_MEMBER_ID`, and the client joins again from scratch.
+pub fn remove_empty(groups: &mut Groups, group_id: &StrBytes, now: Instant) -> GroupRemoval {
+ if !tick_group(groups, group_id, now) {
+ return GroupRemoval::Absent;
+ }
+ if groups
+ .get(group_id)
+ .is_some_and(|group| !group.members.is_empty())
+ {
+ return GroupRemoval::Live;
+ }
+ groups.remove(group_id);
+ GroupRemoval::Removed
+}
+
+/// Kafka's `ClassicGroup.validateOffsetDelete` and `isSubscribedToTopic`.
+pub fn offset_deletion(groups: &mut Groups, group_id: &StrBytes, now: Instant) -> OffsetDeletion {
+ if !tick_group(groups, group_id, now) {
+ return OffsetDeletion::Absent;
+ }
+ let Some(group) = groups.get(group_id) else {
+ return OffsetDeletion::Absent;
+ };
+ if group.members.is_empty() {
+ return OffsetDeletion::Allowed(Subscribed::Topics(HashSet::new()));
+ }
+ if group.protocol_type.as_str() != CONSUMER_PROTOCOL_TYPE {
+ return OffsetDeletion::Refused;
+ }
+ OffsetDeletion::Allowed(group.subscribed())
+}
+
+/// The topics of a consumer's `ConsumerProtocolSubscription`: a version, then the topics and the
+/// user data of v0, which every later version keeps as its prefix. Kafka reads the same prefix.
+/// `None` when the bytes do not hold it.
+///
+/// Read by hand, not by `kafka_protocol`, which sizes a `Vec` from a declared count before it
+/// checks the bytes (see `protocol::bounds_guard`), and a member chose these bytes.
+fn subscription_topics(metadata: &[u8]) -> Option<Vec<String>> {
+ let mut rest = metadata;
+ let _version = take_i16(&mut rest)?;
+ let count = usize::try_from(take_i32(&mut rest)?).ok()?;
+ // Each topic takes at least its 2 length bytes.
+ if count > rest.len() / 2 {
+ return None;
+ }
+ let mut topics = Vec::with_capacity(count);
+ for _ in 0..count {
+ let len = usize::try_from(take_i16(&mut rest)?).ok()?;
+ let (name, after) = rest.split_at_checked(len)?;
+ topics.push(String::from_utf8_lossy(name).into_owned());
+ rest = after;
+ }
+ // A negative length is null user data.
+ if let Ok(user_data_len) = usize::try_from(take_i32(&mut rest)?) {
+ rest.get(..user_data_len)?;
+ }
+ Some(topics)
+}
+
+fn take_i16(rest: &mut &[u8]) -> Option<i16> {
+ let (bytes, after) = rest.split_first_chunk::<2>()?;
+ *rest = after;
+ Some(i16::from_be_bytes(*bytes))
+}
+
+fn take_i32(rest: &mut &[u8]) -> Option<i32> {
+ let (bytes, after) = rest.split_first_chunk::<4>()?;
+ *rest = after;
+ Some(i32::from_be_bytes(*bytes))
+}
+
/// How one requested id is answered. Priced before [`GroupState::description`] copies blobs.
#[derive(Clone, Copy)]
enum DescribedChoice {
- /// Not in the map after the tick. Encoded as `Dead`.
+ /// Not in the map after the tick. Encoded as `Empty` when the group holds offsets in Iggy,
+ /// and as `Dead` otherwise.
Missing,
/// `summary` is Kafka's `summaryNoMetadata` shape.
Present { summary: bool },
@@ -1519,9 +1610,16 @@ fn too_large_described_len(version: i16, group_id: &StrBytes) -> usize {
total
}
-/// A group that is not in the map. v6 carries `GROUP_ID_NOT_FOUND` and `Group {id} not found.`;
-/// older versions encode error 0 and omit `error_message`. Matches `missing_group` in the handler.
+/// A group that is not in the map. The handler answers it after this lock is released, `Empty`
+/// when it holds offsets in Iggy and `Dead` otherwise, so it is priced as the larger answer. A
+/// `COORDINATOR_LOAD_IN_PROGRESS` answer, for when the handler cannot tell, is smaller than both.
fn missing_described_len(version: i16, group_id: &StrBytes) -> usize {
+ dead_described_len(version, group_id).max(offsets_only_described_len(version, group_id))
+}
+
+/// v6 carries `GROUP_ID_NOT_FOUND` and `Group {id} not found.`. Older versions encode error 0 and
+/// omit `error_message`. Matches `missing_group` in the handler.
+fn dead_described_len(version: i16, group_id: &StrBytes) -> usize {
let flexible = version >= 5;
let mut total = 2;
if version >= 6 {
@@ -1542,6 +1640,28 @@ fn missing_described_len(version: i16, group_id: &StrBytes) -> usize {
total
}
+/// Error 0 with a null v6 `error_message`, `Empty`, [`OFFSETS_ONLY_PROTOCOL_TYPE`], no protocol
+/// and no members. Matches [`GroupDescription::offsets_only`] through `described_group`.
+fn offsets_only_described_len(version: i16, group_id: &StrBytes) -> usize {
+ let flexible = version >= 5;
+ let mut total = 2;
+ if version >= 6 {
+ total += kafka_nullable_string_len(true, None);
+ }
+ total += kafka_string_len(flexible, group_id.len());
+ total += kafka_string_len(flexible, GROUP_STATE_EMPTY.len());
+ total += kafka_string_len(flexible, OFFSETS_ONLY_PROTOCOL_TYPE.len());
+ total += kafka_string_len(flexible, 0);
+ total += kafka_array_prefix(flexible, 0);
+ if version >= 3 {
+ total += 4;
+ }
+ if flexible {
+ total += 1;
+ }
+ total
+}
+
fn kafka_response_header_len(api_key: i16, version: i16) -> usize {
if response_header_version(api_key, version) >= 1 {
5
@@ -1714,6 +1834,11 @@ fn generate_member_id(group_instance_id: Option<&StrBytes>) -> StrBytes {
#[cfg(test)]
mod tests {
+ use bytes::{BufMut, BytesMut};
+ use kafka_protocol::messages::ConsumerProtocolSubscription;
+ use kafka_protocol::messages::describe_groups_response::DescribedGroup;
+ use kafka_protocol::protocol::Encodable;
+
use super::*;
use crate::protocol::handlers::describe_groups::{
described_group, encode_response, missing_group,
@@ -2335,6 +2460,145 @@ mod tests {
assert!(groups.contains_key(&group_id()));
}
+ /// A `ConsumerProtocolSubscription` as a consumer sends it: its version, then its fields.
+ fn subscription(version: i16, topics: &[&str]) -> Bytes {
+ let mut buf = BytesMut::new();
+ buf.put_i16(version);
+ ConsumerProtocolSubscription::default()
+ .with_topics(
+ topics
+ .iter()
+ .map(|topic| StrBytes::from_string((*topic).to_owned()))
+ .collect(),
+ )
+ .encode(&mut buf, version)
+ .expect("a subscription encodes");
+ buf.freeze()
+ }
+
+ fn topics(names: &[&str]) -> Subscribed {
+ Subscribed::Topics(names.iter().map(|name| (*name).to_owned()).collect())
+ }
+
+ /// Every version of the consumer protocol starts with the v0 fields, so the topics read the
+ /// same from each.
+ #[test]
+ fn given_a_consumer_subscription_should_read_its_topics_at_every_version() {
+ for version in 0..=3 {
+ assert_eq!(
+ subscription_topics(&subscription(version, &["orders", "payments"])),
+ Some(vec!["orders".to_owned(), "payments".to_owned()]),
+ "v{version}"
+ );
+ }
+ }
+
+ #[test]
+ fn given_bytes_that_are_not_a_subscription_should_read_none() {
+ let whole = subscription(0, &["orders"]);
+ for cut in 0..whole.len() {
+ assert_eq!(subscription_topics(&whole[..cut]), None, "cut at {cut}");
+ }
+ assert_eq!(
+ subscription_topics(&[0, 0, 0x7F, 0xFF, 0xFF, 0xFF]),
+ None,
+ "a count that six bytes cannot hold"
+ );
+ assert_eq!(
+ subscription_topics(&[0, 0, 0xFF, 0xFF, 0xFF, 0xFF]),
+ None,
+ "a null topic list"
+ );
+ assert_eq!(subscription_topics(b"subscription"), None);
+ }
+
+ /// Kafka's `ClassicGroup.validateDeleteGroup`: only a group without joined members goes.
+ #[test]
+ fn given_each_kind_of_group_when_deleting_it_should_remove_only_a_group_without_members() {
+ let config = config();
+ let now = Instant::now();
+ let mut groups = Groups::new();
+ assert_eq!(
+ remove_empty(&mut groups, &group_id(), now),
+ GroupRemoval::Absent
+ );
+
+ let _ = join_step(&mut groups, &config, &request("", &["range"]), now);
+ assert_eq!(
+ remove_empty(&mut groups, &group_id(), now),
+ GroupRemoval::Live
+ );
+ assert!(groups.contains_key(&group_id()));
+
+ let mut groups = Groups::new();
+ let pending = member_id_of(&join_step(&mut groups, &config, &pending_request(), now));
+ assert_eq!(
+ remove_empty(&mut groups, &group_id(), now),
+ GroupRemoval::Removed
+ );
+ assert!(groups.is_empty());
+ let rejoin = JoinRequest {
+ require_known_member_id: true,
+ ..request(pending.as_str(), &["x"])
+ };
+ assert_eq!(
+ error_of(&join_step(&mut groups, &config, &rejoin, now)),
+ ERROR_UNKNOWN_MEMBER_ID,
+ "the id of a deleted group no longer joins, so the client starts over"
+ );
+ }
+
+ /// Kafka's `ClassicGroup.validateOffsetDelete` and `isSubscribedToTopic`.
+ #[test]
+ fn given_each_kind_of_group_when_deleting_offsets_should_answer_as_kafka() {
+ let config = config();
+ let now = Instant::now();
+ let mut groups = Groups::new();
+ assert_eq!(
+ offset_deletion(&mut groups, &group_id(), now),
+ OffsetDeletion::Absent
+ );
+
+ let step = join_step(&mut groups, &config, &pending_request(), now);
+ assert_eq!(error_of(&step), ERROR_MEMBER_ID_REQUIRED);
+ assert_eq!(
+ offset_deletion(&mut groups, &group_id(), now),
+ OffsetDeletion::Allowed(topics(&[])),
+ "only an id that waits to join: Empty, so nothing is subscribed"
+ );
+
+ let mut groups = Groups::new();
+ let mut member = request("", &["range"]);
+ member.protocols = vec![(
+ StrBytes::from_static_str("range"),
+ subscription(3, &["orders", "payments"]),
+ )];
+ let _ = join_step(&mut groups, &config, &member, now);
+ assert_eq!(
+ offset_deletion(&mut groups, &group_id(), now),
+ OffsetDeletion::Allowed(topics(&["orders", "payments"]))
+ );
+
+ let mut groups = Groups::new();
+ let _ = join_step(&mut groups, &config, &request("", &["range"]), now);
+ assert_eq!(
+ offset_deletion(&mut groups, &group_id(), now),
+ OffsetDeletion::Allowed(Subscribed::All),
+ "metadata that is not a subscription subscribes to every topic"
+ );
+
+ let mut groups = Groups::new();
+ let connect = JoinRequest {
+ protocol_type: StrBytes::from_static_str("connect"),
+ ..request("", &["sessioned"])
+ };
+ let _ = join_step(&mut groups, &config, &connect, now);
+ assert_eq!(
+ offset_deletion(&mut groups, &group_id(), now),
+ OffsetDeletion::Refused
+ );
+ }
+
/// List skips a fully expired group and does not tick it. Describe then ticks and the group
/// is gone, which the handler reports as `Dead`.
#[test]
@@ -2375,6 +2639,9 @@ mod tests {
}
}
+ /// A missing id is answered `Dead`, or `Empty` when it holds offsets in Iggy. One of the two
+ /// is larger for every id at a given version, `Empty` up to v5 and `Dead` at v6, where it
+ /// carries a message. So the price is the larger of the two whole responses.
fn assert_priced_response(
version: i16,
groups: &mut Groups,
@@ -2384,28 +2651,35 @@ mod tests {
) {
let priced = describe_frame_len(version, groups, ids, max_frame_size);
let described = describe_groups(groups, ids, version, max_frame_size, now);
- let encoded = encode_described(version, &dedup_first_seen(ids), &described);
+ let ids = dedup_first_seen(ids);
let header = kafka_response_header_len(API_KEY_DESCRIBE_GROUPS, version);
+ let dead = header + encode_described(version, &ids, &described, missing_group).len();
+ let offsets_only =
+ header + encode_described(version, &ids, &described, offsets_only_group).len();
assert_eq!(
priced,
- header + encoded.len(),
- "version {version} ids {} priced {priced} encoded {}",
+ dead.max(offsets_only),
+ "version {version} ids {} priced {priced} dead {dead} offsets only {offsets_only}",
ids.len(),
- header + encoded.len()
);
}
+ fn offsets_only_group(version: i16, group_id: &StrBytes) -> DescribedGroup {
+ described_group(version, GroupDescription::offsets_only(group_id.clone()))
+ }
+
fn encode_described(
version: i16,
ids: &[StrBytes],
described: &[Option<GroupDescription>],
+ missing: fn(i16, &StrBytes) -> DescribedGroup,
) -> Bytes {
let groups = ids
.iter()
.zip(described)
.map(|(group_id, group)| {
group.clone().map_or_else(
- || missing_group(version, group_id),
+ || missing(version, group_id),
|group| described_group(version, group),
)
})
@@ -2413,6 +2687,33 @@ mod tests {
encode_response(version, groups).expect("described group encodes")
}
+ /// Each answer to a missing id on its own, so neither hides behind the other in the maximum.
+ /// The id is past the compact varint boundary.
+ #[test]
+ fn given_a_missing_group_when_pricing_each_answer_should_match_the_encoder() {
+ let group_id = StrBytes::from_string("m".repeat(200));
+ for version in 0..=6 {
+ let header = kafka_response_header_len(API_KEY_DESCRIBE_GROUPS, version);
+ let fixed = describe_fixed_len(version, 1);
+ let encoded = |group: DescribedGroup| {
+ header
+ + encode_response(version, vec![group])
+ .expect("described group encodes")
+ .len()
+ };
+ assert_eq!(
+ fixed + dead_described_len(version, &group_id),
+ encoded(missing_group(version, &group_id)),
+ "version {version} dead"
+ );
+ assert_eq!(
+ fixed + offsets_only_described_len(version, &group_id),
+ encoded(offsets_only_group(version, &group_id)),
+ "version {version} offsets only"
+ );
+ }
+ }
+
fn join_wide(groups: &mut Groups, config: &GroupCoordinatorConfig, now: Instant) -> JoinResult {
let wide = "x".repeat(130);
let mut request = request("", &["range"]);
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index e6038a06d..2025d05cb 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -40,10 +40,10 @@ use crate::protocol::bounds_guard::{
};
use crate::protocol::handlers::init_producer_id::ProducerIdAllocator;
use crate::protocol::handlers::{
- alter_configs, api_versions, create_topics, decode_guarded, delete_topics, describe_configs,
- describe_groups, dispatch, fetch, find_coordinator, heartbeat, init_producer_id, join_group,
- leave_group, list_groups, list_offsets, metadata, offset_commit, offset_fetch, produce,
- respond_or_close, sync_group,
+ alter_configs, api_versions, create_topics, decode_guarded, delete_groups, delete_topics,
+ describe_configs, describe_groups, dispatch, fetch, find_coordinator, heartbeat,
+ init_producer_id, join_group, leave_group, list_groups, list_offsets, metadata, offset_commit,
+ offset_delete, offset_fetch, produce, respond_or_close, sync_group,
};
use crate::protocol::probe_board::ProbeBoard;
use crate::protocol::sasl::{
@@ -72,6 +72,8 @@ pub const API_KEY_DESCRIBE_ACLS: i16 = 29;
pub const API_KEY_DESCRIBE_CONFIGS: i16 = 32;
pub const API_KEY_ALTER_CONFIGS: i16 = 33;
pub const API_KEY_SASL_AUTHENTICATE: i16 = 36;
+pub const API_KEY_DELETE_GROUPS: i16 = 42;
+pub const API_KEY_OFFSET_DELETE: i16 = 47;
pub const DEFAULT_KAFKA_PORT: u16 = 9093;
@@ -124,6 +126,13 @@ pub const ERROR_GROUP_ID_NOT_FOUND: i16 = ResponseError::GroupIdNotFound.code();
pub const ERROR_INVALID_SESSION_TIMEOUT: i16 = 26;
/// How a follower learns to rejoin: its heartbeat is answered with this while the group prepares.
pub const ERROR_REBALANCE_IN_PROGRESS: i16 = 27;
+/// `DeleteGroups`: the bridge user may not delete the Iggy group that holds the offsets.
+pub const ERROR_GROUP_AUTHORIZATION_FAILED: i16 = ResponseError::GroupAuthorizationFailed.code();
+/// `DeleteGroups`: the group has members. `OffsetDelete`: it has members of a protocol other than
+/// `consumer`.
+pub const ERROR_NON_EMPTY_GROUP: i16 = ResponseError::NonEmptyGroup.code();
+/// `OffsetDelete`: a member subscribes to the topic, so its offsets stay.
+pub const ERROR_GROUP_SUBSCRIBED_TO_TOPIC: i16 = ResponseError::GroupSubscribedToTopic.code();
/// Closest fit for an Iggy permission/credential rejection in `bridge`'s error mapping.
///
/// Still not `SASL_AUTHENTICATION_FAILED`, and now for a firmer reason than when this was written:
@@ -307,6 +316,8 @@ static SUPPORTED_RANGES: &[ApiVersionRange] = &[
sync_group::RANGE,
describe_groups::RANGE,
list_groups::RANGE,
+ delete_groups::RANGE,
+ offset_delete::RANGE,
];
#[must_use]
diff --git a/gateways/kafka/src/protocol/bounds_guard.rs b/gateways/kafka/src/protocol/bounds_guard.rs
index 0d376d987..32d1faba2 100644
--- a/gateways/kafka/src/protocol/bounds_guard.rs
+++ b/gateways/kafka/src/protocol/bounds_guard.rs
@@ -1269,6 +1269,66 @@ pub fn validate_list_groups_shape(version: i16, body: &Bytes, max_frame_size: us
Ok(())
}
+/// Mirrors the field order `DeleteGroupsRequest::decode` walks.
+///
+/// Group ids are echoed, so their lengths count toward the projected response.
+///
+/// # Errors
+///
+/// Returns an error when a declared array/string length cannot fit in the bytes remaining in the
+/// frame, or the body is truncated or malformed in a way that cannot be walked.
+pub fn validate_delete_groups_shape(
+ version: i16,
+ body: &Bytes,
+ max_frame_size: usize,
+) -> Result<()> {
+ let mut c = ShapeCursor::new(body.clone(), max_frame_size);
+ let flexible = version >= 2;
+ let groups_count = if flexible {
+ c.compact_array_count()?
+ } else {
+ c.legacy_array_count()?
+ };
+ for _ in 0..groups_count {
+ if flexible {
+ c.compact_string(false)?;
+ } else {
+ c.legacy_string(false)?;
+ }
+ }
+ if flexible {
+ c.tagged_fields()?;
+ }
+ Ok(())
+}
+
+/// Mirrors the field order `OffsetDeleteRequest::decode` walks at v0, the only version this
+/// gateway advertises.
+///
+/// Topic names are echoed, so their lengths count toward the projected response.
+///
+/// # Errors
+///
+/// Returns an error when a declared array/string length cannot fit in the bytes remaining in the
+/// frame, or the body is truncated or malformed in a way that cannot be walked.
+pub fn validate_offset_delete_shape(
+ _version: i16,
+ body: &Bytes,
+ max_frame_size: usize,
+) -> Result<()> {
+ let mut c = ShapeCursor::new(body.clone(), max_frame_size);
+ c.legacy_string(false)?;
+ let topics_count = c.legacy_array_count()?;
+ for _ in 0..topics_count {
+ c.legacy_string(false)?;
+ let partitions_count = c.legacy_array_count()?;
+ for _ in 0..partitions_count {
+ let _partition_index = c.read_i32()?;
+ }
+ }
+ Ok(())
+}
+
/// Cap on a whole `SaslAuthenticate` body.
///
/// The body is one length-prefixed blob, so capping it is the same as capping `auth_bytes`.
@@ -1487,8 +1547,8 @@ fn walk_offset_fetch_topics(
mod tests {
use bytes::BytesMut;
use kafka_protocol::messages::{
- DescribeGroupsRequest, LeaveGroupRequest, ListGroupsRequest, OffsetCommitRequest,
- OffsetFetchRequest,
+ DeleteGroupsRequest, DescribeGroupsRequest, LeaveGroupRequest, ListGroupsRequest,
+ OffsetCommitRequest, OffsetDeleteRequest, OffsetFetchRequest,
};
use super::*;
@@ -2105,6 +2165,59 @@ mod tests {
assert!(validate_list_groups_shape(4, &body, TEST_MAX_FRAME_SIZE).is_err());
}
+ #[test]
+ fn delete_groups_v0_two_groups_accepted() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x00, 0x00, 0x02, // groups_names: 2
+ 0x00, 0x01, b'g', // first
+ 0x00, 0x02, b'g', b'2', // second
+ ]);
+ assert!(validate_delete_groups_shape(0, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<DeleteGroupsRequest>(0, body).expect("body matches the schema");
+ }
+
+ #[test]
+ fn delete_groups_v2_one_group_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, // groups_names: 1
+ 0x02, b'g', // group id
+ 0x00, // tagged fields
+ ]);
+ assert!(validate_delete_groups_shape(2, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<DeleteGroupsRequest>(2, body).expect("body matches the schema");
+ }
+
+ #[test]
+ fn delete_groups_v2_huge_group_count_rejected() {
+ let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF, 0xFF, 0x0F]);
+ assert!(validate_delete_groups_shape(2, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ #[test]
+ fn offset_delete_v0_topic_with_partitions_accepted() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x01, b'g', // group_id
+ 0x00, 0x00, 0x00, 0x01, // topics: 1
+ 0x00, 0x01, b't', // name
+ 0x00, 0x00, 0x00, 0x02, // partitions: 2
+ 0x00, 0x00, 0x00, 0x00, // 0
+ 0x00, 0x00, 0x00, 0x01, // 1
+ ]);
+ assert!(validate_offset_delete_shape(0, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetDeleteRequest>(0, body).expect("body matches the schema");
+ }
+
+ #[test]
+ fn offset_delete_v0_huge_partitions_count_rejected() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x01, b'g', // group_id
+ 0x00, 0x00, 0x00, 0x01, // topics: 1
+ 0x00, 0x01, b't', // name
+ 0x7F, 0xFF, 0xFF, 0xFF, // partitions count
+ ]);
+ assert!(validate_offset_delete_shape(0, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
#[test]
fn offset_commit_v2_null_metadata_accepted() {
let body = Bytes::from_static(&[
diff --git a/gateways/kafka/src/protocol/handlers/delete_groups.rs b/gateways/kafka/src/protocol/handlers/delete_groups.rs
new file mode 100644
index 000000000..226022a23
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/delete_groups.rs
@@ -0,0 +1,400 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF 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.
+
+//! `DeleteGroups` (API key 42).
+//!
+//! Deletes each named group that has no members here, with its offsets, as
+//! `kafka-consumer-groups.sh --delete` asks. Kafka's rules (`ClassicGroup.validateDeleteGroup`):
+//! a group with members gets `NON_EMPTY_GROUP` (68), and a group that does not exist gets
+//! `GROUP_ID_NOT_FOUND` (69). A group exists here while ids wait to join it, or while it holds
+//! offsets in Iggy. The gateway deletes the Iggy group `kafka.cg.<group>` on every Kafka topic,
+//! and Iggy deletes the offsets with it (`IggyBridge::delete_offset_groups`).
+//!
+//! With SASL on, a principal without `manage_topics` gets `GROUP_AUTHORIZATION_FAILED` (30) for
+//! every group, and nothing is deleted.
+//!
+//! An id repeated in one request is answered once, as in `DescribeGroups`. The response has no
+//! top-level error, so a request that cannot be decoded closes the connection.
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_groups_response::DeletableGroupResult;
+use kafka_protocol::messages::{DeleteGroupsRequest, DeleteGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::auth::AuthenticatedPrincipal;
+use crate::bridge::{BridgeError, IggyBridge, OffsetCalls};
+use crate::error::Result;
+use crate::group::{GroupRemoval, is_valid_group_id};
+use crate::protocol::api::{
+ API_KEY_DELETE_GROUPS, ApiVersionRange, ConnectionState, ERROR_GROUP_AUTHORIZATION_FAILED,
+ ERROR_GROUP_ID_NOT_FOUND, ERROR_INVALID_GROUP_ID, ERROR_NON_EMPTY_GROUP, ERROR_NONE,
+ ERROR_TOPIC_AUTHORIZATION_FAILED, ERROR_UNKNOWN_SERVER_ERROR, GatewayState, HandleOutcome,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_delete_groups_shape;
+use crate::protocol::dedup::dedup_first_seen;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+/// v3 adds a per-group `error_message`, which `kafka-protocol` cannot encode yet.
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_GROUPS,
+ min_version: 0,
+ max_version: 2,
+};
+
+/// Group id length prefix, error code and tagged fields of one result, past the id itself.
+const PER_RESULT_BYTES: usize = 5;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DELETE_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request = match decode_guarded::<DeleteGroupsRequest>(api_version, body, |version, body| {
+ validate_delete_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode DeleteGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups_names
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let group_ids = dedup_first_seen(&requested);
+ let codes = if may_delete_groups(principal, "DeleteGroups") {
+ delete(state, connection, &group_ids).await
+ } else {
+ vec![ERROR_GROUP_AUTHORIZATION_FAILED; group_ids.len()]
+ };
+ let results = group_ids
+ .into_iter()
+ .zip(codes)
+ .map(|(group_id, code)| {
+ DeletableGroupResult::default()
+ .with_group_id(GroupId(group_id))
+ .with_error_code(code)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, results), "DeleteGroups")
+}
+
+/// One code per group of `group_ids`, in order.
+async fn delete(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ group_ids: &[StrBytes],
+) -> Vec<i16> {
+ let valid: Vec<StrBytes> = group_ids
+ .iter()
+ .filter(|group_id| is_valid_group_id(group_id))
+ .cloned()
+ .collect();
+ let mut removals = state.groups.remove_empty(&valid).await.into_iter();
+ // Per group: its answer, or whether the coordinator held it when its offsets are to go.
+ let plans: Vec<core::result::Result<bool, i16>> = group_ids
+ .iter()
+ .map(|group_id| {
+ if !is_valid_group_id(group_id) {
+ return Err(ERROR_INVALID_GROUP_ID);
+ }
+ match removals.next() {
+ Some(GroupRemoval::Live) => Err(ERROR_NON_EMPTY_GROUP),
+ removal => Ok(removal == Some(GroupRemoval::Removed)),
+ }
+ })
+ .collect();
+ let deletable: Vec<&str> = group_ids
+ .iter()
+ .zip(&plans)
+ .filter(|(_, plan)| plan.is_ok())
+ .map(|(group_id, _)| group_id.as_str())
+ .collect();
+ let mut deleted = match &state.bridge {
+ Some(bridge) if !deletable.is_empty() => {
+ let calls = OffsetCalls::new(offset_deadline(state, connection).await);
+ delete_offsets(bridge, &deletable, &calls).await
+ }
+ _ => Vec::new(),
+ }
+ .into_iter();
+ plans
+ .into_iter()
+ .map(|plan| match plan {
+ Err(code) => code,
+ Ok(held) => match deleted.next().unwrap_or(Ok(false)) {
+ Ok(found) if found || held => ERROR_NONE,
+ Ok(_) => ERROR_GROUP_ID_NOT_FOUND,
+ Err(code) => code,
+ },
+ })
+ .collect()
+}
+
+/// Deletes the offsets of each group of `groups` in Iggy. Whether any topic held the group, or
+/// the code of the failure. A listing failure goes on every group.
+async fn delete_offsets(
+ bridge: &IggyBridge,
+ groups: &[&str],
+ calls: &OffsetCalls,
+) -> Vec<core::result::Result<bool, i16>> {
+ let results = match bridge.delete_offset_groups(groups, calls).await {
+ Ok(results) => results,
+ Err(error) => {
+ log_refusal(groups.first().copied().unwrap_or_default(), &error);
+ let code = deletion_error_code(&error);
+ return groups.iter().map(|_| Err(code)).collect();
+ }
+ };
+ let mut logged = false;
+ groups
+ .iter()
+ .zip(results)
+ .map(|(group, result)| {
+ result.map_err(|error| {
+ let code = deletion_error_code(&error);
+ // One line per request, as in `OffsetCommit`.
+ if code == ERROR_UNKNOWN_SERVER_ERROR && !logged {
+ logged = true;
+ log_refusal(group, &error);
+ }
+ code
+ })
+ })
+ .collect()
+}
+
+/// [`BridgeError::to_offset_error_code`], with the group's code where Iggy refused the bridge
+/// user: the request deletes a group, not a topic.
+const fn deletion_error_code(error: &BridgeError) -> i16 {
+ match error.to_offset_error_code() {
+ ERROR_TOPIC_AUTHORIZATION_FAILED => ERROR_GROUP_AUTHORIZATION_FAILED,
+ code => code,
+ }
+}
+
+fn log_refusal(group: &str, error: &BridgeError) {
+ if error.to_offset_error_code() != ERROR_UNKNOWN_SERVER_ERROR {
+ return;
+ }
+ tracing::error!(
+ group,
+ %error,
+ cause = offset_refusal_cause(error),
+ "Iggy refused to delete the offsets of a Kafka group"
+ );
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, results: Vec<DeletableGroupResult>) -> Result<Bytes> {
+ let echoed: usize = results
+ .iter()
+ .map(|result| result.group_id.0.len() + PER_RESULT_BYTES)
+ .sum();
+ encode_message(
+ &DeleteGroupsResponse::default().with_results(results),
+ version,
+ 16 + echoed,
+ )
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use bytes::BytesMut;
+ use kafka_protocol::protocol::{Decodable, Encodable};
+ use tokio_util::sync::CancellationToken;
+
+ use super::*;
+ use crate::group::{GroupCoordinator, GroupCoordinatorConfig, JoinRequest};
+ use crate::protocol::acl::PrincipalPermissions;
+ use crate::protocol::api::{BrokerAdvertise, ERROR_MEMBER_ID_REQUIRED};
+
+ /// No bridge, and a first join that completes at once.
+ fn state() -> GatewayState {
+ let groups = GroupCoordinatorConfig {
+ initial_rebalance_delay: Duration::ZERO,
+ ..GroupCoordinatorConfig::default()
+ };
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ None,
+ 8 * 1024 * 1024,
+ false,
+ 0,
+ GroupCoordinator::new(groups, CancellationToken::new()),
+ )
+ }
+
+ fn join_request(group: &str, require_known_member_id: bool) -> JoinRequest {
+ JoinRequest {
+ group_id: StrBytes::from_string(group.to_owned()),
+ session_timeout: Duration::from_secs(10),
+ rebalance_timeout: Duration::from_secs(10),
+ member_id: StrBytes::new(),
+ group_instance_id: None,
+ protocol_type: StrBytes::from_static_str("consumer"),
+ protocols: vec![(StrBytes::from_static_str("range"), Bytes::new())],
+ require_known_member_id,
+ }
+ }
+
+ fn body(version: i16, groups: &[&str]) -> Bytes {
+ let request = DeleteGroupsRequest::default().with_groups_names(
+ groups
+ .iter()
+ .map(|group| GroupId(StrBytes::from_string((*group).to_owned())))
+ .collect(),
+ );
+ let mut buf = BytesMut::new();
+ request.encode(&mut buf, version).unwrap();
+ buf.freeze()
+ }
+
+ fn principal(manage_topics: bool, permissions_known: bool) -> AuthenticatedPrincipal {
+ AuthenticatedPrincipal {
+ username: "alice".to_string(),
+ permissions: PrincipalPermissions {
+ manage_topics,
+ ..PrincipalPermissions::default()
+ },
+ permissions_known,
+ }
+ }
+
+ async fn codes(
+ state: &GatewayState,
+ principal: Option<&AuthenticatedPrincipal>,
+ version: i16,
+ groups: &[&str],
+ ) -> Vec<(String, i16)> {
+ let connection = ConnectionState::default();
+ let mut response = handle(
+ state,
+ &connection,
+ principal,
+ version,
+ body(version, groups),
+ )
+ .await
+ .expect_response("DeleteGroups answers");
+ let response = DeleteGroupsResponse::decode(&mut response, version).unwrap();
+ response
+ .results
+ .iter()
+ .map(|result| (result.group_id.0.to_string(), result.error_code))
+ .collect()
+ }
+
+ /// Without a bridge no group holds offsets, so only the coordinator's groups exist.
+ #[tokio::test]
+ async fn given_each_kind_of_group_without_a_bridge_should_answer_as_kafka() {
+ for version in 0..=2 {
+ let state = state();
+ let joined = state.groups.join(&join_request("live", false)).await;
+ assert_eq!(joined.error, ERROR_NONE);
+ let pending = state.groups.join(&join_request("waiting", true)).await;
+ assert_eq!(pending.error, ERROR_MEMBER_ID_REQUIRED);
+ let too_long = "g".repeat(crate::group::MAX_GROUP_ID_BYTES + 1);
+
+ assert_eq!(
+ codes(
+ &state,
+ None,
+ version,
+ &["live", "waiting", "unknown", "", &too_long, "unknown"],
+ )
+ .await,
+ vec![
+ ("live".to_string(), ERROR_NON_EMPTY_GROUP),
+ ("waiting".to_string(), ERROR_NONE),
+ ("unknown".to_string(), ERROR_GROUP_ID_NOT_FOUND),
+ (String::new(), ERROR_INVALID_GROUP_ID),
+ (too_long.clone(), ERROR_INVALID_GROUP_ID),
+ ],
+ "v{version}: a repeated id is answered once"
+ );
+ assert_eq!(
+ codes(&state, None, version, &["waiting"]).await,
+ vec![("waiting".to_string(), ERROR_GROUP_ID_NOT_FOUND)],
+ "v{version}: the deleted group is gone"
+ );
+ }
+ }
+
+ /// Kafka asks for `DELETE` on each group, which needs `manage_topics` here.
+ #[tokio::test]
+ async fn given_a_principal_without_manage_topics_should_refuse_every_group() {
+ let state = state();
+ let pending = state.groups.join(&join_request("waiting", true)).await;
+ assert_eq!(pending.error, ERROR_MEMBER_ID_REQUIRED);
+
+ for refused in [principal(false, true), principal(true, false)] {
+ assert_eq!(
+ codes(&state, Some(&refused), 2, &["waiting", ""]).await,
+ vec![
+ ("waiting".to_string(), ERROR_GROUP_AUTHORIZATION_FAILED),
+ (String::new(), ERROR_GROUP_AUTHORIZATION_FAILED),
+ ],
+ "known={}",
+ refused.permissions_known
+ );
+ }
+ assert_eq!(
+ codes(&state, Some(&principal(true, true)), 2, &["waiting"]).await,
+ vec![("waiting".to_string(), ERROR_NONE)],
+ "a refused request deletes nothing"
+ );
+ }
+
+ /// The response has no top-level error, so an unreadable request has no answer.
+ #[tokio::test]
+ async fn given_an_unreadable_request_should_close_the_connection() {
+ let state = state();
+ let connection = ConnectionState::default();
+
+ let truncated = body(2, &["orders-app"]).slice(..3);
+ assert!(
+ handle(&state, &connection, None, 2, truncated)
+ .await
+ .is_close()
+ );
+ assert!(
+ handle(&state, &connection, None, 3, body(2, &["orders-app"]))
+ .await
+ .is_close()
+ );
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/describe_groups.rs b/gateways/kafka/src/protocol/handlers/describe_groups.rs
index e8f59fd57..c4aa0ba63 100644
--- a/gateways/kafka/src/protocol/handlers/describe_groups.rs
+++ b/gateways/kafka/src/protocol/handlers/describe_groups.rs
@@ -18,28 +18,35 @@
//! `DescribeGroups` (API key 15).
//!
//! Reports the members, assignment and state of groups this coordinator holds. A named group
-//! that is not here is `Dead`: v6 also carries `GROUP_ID_NOT_FOUND`, and earlier versions do
-//! not, matching Kafka 4.0's `describeGroups`. A v6 client treats error 0 as a real group.
+//! that is not here is `Empty` when it holds offsets in Iggy, as Kafka keeps a group whose
+//! consumers left. Otherwise it is `Dead`: v6 also carries `GROUP_ID_NOT_FOUND`, and earlier
+//! versions do not, matching Kafka 4.0's `describeGroups`. A v6 client treats error 0 as a real
+//! group. When no scan has found the groups that hold offsets yet (`IggyBridge::offset_groups`),
+//! a group that is not here gets `COORDINATOR_LOAD_IN_PROGRESS` (14), which clients retry.
//!
//! `include_authorized_operations` is not answered. The field stays at the omitted sentinel
//! (`i32::MIN`): there is no group ACL bitfield to report, and any other value fails the
//! encoder below v3.
+use std::sync::Arc;
+
use bytes::Bytes;
use kafka_protocol::messages::describe_groups_response::{DescribedGroup, DescribedGroupMember};
use kafka_protocol::messages::{DescribeGroupsRequest, DescribeGroupsResponse, GroupId};
use kafka_protocol::protocol::StrBytes;
+use crate::bridge::OffsetGroups;
use crate::error::Result;
use crate::group::{DESCRIBE_RESPONSE_TOO_LARGE, GroupDescription, MemberDescription};
use crate::protocol::api::{
- API_KEY_DESCRIBE_GROUPS, ApiVersionRange, ERROR_GROUP_ID_NOT_FOUND, ERROR_NONE,
- ERROR_UNKNOWN_SERVER_ERROR, GatewayState, HandleOutcome, is_supported_version,
+ API_KEY_DESCRIBE_GROUPS, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_GROUP_ID_NOT_FOUND, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR, GatewayState, HandleOutcome,
+ is_supported_version,
};
use crate::protocol::bounds_guard::validate_describe_groups_shape;
use crate::protocol::dedup::dedup_first_seen;
use crate::protocol::handlers::{
- decode_guarded, encode_message, respond_or_close, unsupported_version_response,
+ decode_guarded, encode_message, offset_deadline, respond_or_close, unsupported_version_response,
};
pub const RANGE: ApiVersionRange = ApiVersionRange {
@@ -54,7 +61,12 @@ const DEAD: &str = "Dead";
/// `GROUP_ID_NOT_FOUND` for a missing group.
const FIRST_ERROR_MESSAGE_VERSION: i16 = 6;
-pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
if !is_supported_version(API_KEY_DESCRIBE_GROUPS, api_version) {
// No top-level error field to carry a code on, so this answers with an empty group
// list rather than a dedicated error response. Versions outside 0-6 fail in the
@@ -90,12 +102,17 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
.groups
.describe_groups(&group_ids, api_version, state.max_frame_size)
.await;
+ let absent = if described.iter().any(Option::is_none) {
+ Absent::look_up(state, connection).await
+ } else {
+ Absent::Dead
+ };
let groups = group_ids
.iter()
.zip(described)
.map(|(group_id, group)| {
group.map_or_else(
- || missing_group(api_version, group_id),
+ || absent.answer(api_version, group_id),
|group| described_group(api_version, group),
)
})
@@ -103,6 +120,39 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
respond_or_close(encode_response(api_version, groups), "DescribeGroups")
}
+/// How one request answers a group that is not on this coordinator.
+enum Absent {
+ /// Without a bridge no group holds offsets, so each one is `Dead`.
+ Dead,
+ /// `Empty` for a group that holds offsets, `Dead` for any other.
+ Known(Arc<OffsetGroups>),
+ /// No scan has found the groups that hold offsets yet.
+ Unknown,
+}
+
+impl Absent {
+ async fn look_up(state: &GatewayState, connection: &ConnectionState) -> Self {
+ let Some(bridge) = &state.bridge else {
+ return Self::Dead;
+ };
+ let deadline = offset_deadline(state, connection).await;
+ bridge
+ .offset_groups(deadline)
+ .await
+ .map_or(Self::Unknown, Self::Known)
+ }
+
+ fn answer(&self, version: i16, group_id: &StrBytes) -> DescribedGroup {
+ match self {
+ Self::Known(offset_groups) if offset_groups.contains(group_id.as_str()) => {
+ described_group(version, GroupDescription::offsets_only(group_id.clone()))
+ }
+ Self::Dead | Self::Known(_) => missing_group(version, group_id),
+ Self::Unknown => loading_group(group_id),
+ }
+ }
+}
+
pub(crate) fn described_group(version: i16, group: GroupDescription) -> DescribedGroup {
if group.error != ERROR_NONE {
return too_large_group(version, &group.group_id);
@@ -131,6 +181,16 @@ fn too_large_group(version: i16, group_id: &StrBytes) -> DescribedGroup {
group
}
+/// The same empty shape as [`too_large_group`], with a code the Java admin retries.
+fn loading_group(group_id: &StrBytes) -> DescribedGroup {
+ DescribedGroup::default()
+ .with_error_code(ERROR_COORDINATOR_LOAD_IN_PROGRESS)
+ .with_group_id(GroupId(group_id.clone()))
+ .with_group_state(StrBytes::from_static_str(""))
+ .with_protocol_type(StrBytes::from_static_str(""))
+ .with_protocol_data(StrBytes::from_static_str(""))
+}
+
fn described_member(member: MemberDescription) -> DescribedGroupMember {
// `client_id` and `client_host` stay empty: the coordinator does not retain the request
// header's client id, and handlers are not given the connection's peer address.
diff --git a/gateways/kafka/src/protocol/handlers/list_groups.rs b/gateways/kafka/src/protocol/handlers/list_groups.rs
index bf29264cc..81828c279 100644
--- a/gateways/kafka/src/protocol/handlers/list_groups.rs
+++ b/gateways/kafka/src/protocol/handlers/list_groups.rs
@@ -17,26 +17,36 @@
//! `ListGroups` (API key 16).
//!
-//! Lists the groups this coordinator holds. v4+ `states_filter` and v5 `types_filter` keep a
-//! group only when the filter is empty or one entry matches, case-insensitively, after trim.
-//! A filter value this coordinator does not use matches nothing, so the group is left out and
-//! the response error stays 0. Kafka 4.0's `listGroups` does the same ("if invalid, no groups
-//! are returned").
+//! Lists the groups this coordinator holds, and as `Empty` each group with no members here that
+//! holds offsets in Iggy. Kafka keeps a group whose consumers left while it has offsets. v4+
+//! `states_filter` and v5 `types_filter` keep a group only when the filter is empty or one entry
+//! matches, case-insensitively, after trim. A filter value this coordinator does not use matches
+//! nothing, so the group is left out and the response error stays 0. Kafka 4.0's `listGroups`
+//! does the same ("if invalid, no groups are returned").
+//!
+//! The groups that hold offsets come from a scan that costs one Iggy call per topic, and one scan
+//! answers for 10 s (`IggyBridge::offset_groups`). A filter that leaves out `Empty` or `classic`
+//! needs no scan. When no scan has found the groups yet, the answer is
+//! `COORDINATOR_LOAD_IN_PROGRESS` (14), which clients retry.
+
+use std::collections::HashSet;
use bytes::Bytes;
use kafka_protocol::messages::list_groups_response::ListedGroup;
use kafka_protocol::messages::{GroupId, ListGroupsRequest, ListGroupsResponse};
use kafka_protocol::protocol::StrBytes;
+use crate::bridge::OffsetGroups;
use crate::error::Result;
-use crate::group::GroupListing;
+use crate::group::{GROUP_STATE_EMPTY, GroupListing};
use crate::protocol::api::{
- API_KEY_LIST_GROUPS, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION,
- GatewayState, HandleOutcome, is_supported_version,
+ API_KEY_LIST_GROUPS, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
+ is_supported_version,
};
use crate::protocol::bounds_guard::validate_list_groups_shape;
use crate::protocol::handlers::{
- decode_guarded, encode_message, respond_or_close, unsupported_version_response,
+ decode_guarded, encode_message, offset_deadline, respond_or_close, unsupported_version_response,
};
use crate::protocol::header::response_header_version;
use crate::protocol::wire_len::{kafka_compact_prefix, kafka_string_len};
@@ -50,7 +60,12 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
/// `ListGroups` v5 `group_type` for the classic protocol this coordinator runs.
const CLASSIC_GROUP_TYPE: &str = "classic";
-pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
if !is_supported_version(API_KEY_LIST_GROUPS, api_version) {
return unsupported_version_response(API_KEY_LIST_GROUPS, api_version, |version| {
encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
@@ -70,10 +85,21 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
}
};
- let listings: Vec<GroupListing> = state
- .groups
- .list_groups()
- .await
+ let mut listings = state.groups.list_groups().await;
+ let lists_offsets_only = selected(&request.states_filter, GROUP_STATE_EMPTY)
+ && selected(&request.types_filter, CLASSIC_GROUP_TYPE);
+ if lists_offsets_only && let Some(bridge) = &state.bridge {
+ let deadline = offset_deadline(state, connection).await;
+ let Some(offset_groups) = bridge.offset_groups(deadline).await else {
+ tracing::debug!("ListGroups found no scan of the groups that hold offsets yet");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ "ListGroups",
+ );
+ };
+ add_offsets_only(&mut listings, &offset_groups);
+ }
+ let listings: Vec<GroupListing> = listings
.into_iter()
.filter(|group| {
selected(&request.states_filter, group.state)
@@ -96,6 +122,22 @@ fn selected(filter: &[StrBytes], value: &str) -> bool {
.any(|wanted| wanted.as_str().trim().eq_ignore_ascii_case(value))
}
+/// Adds each group of `offset_groups` that is not live here, as `Empty`. Keeps the list in group
+/// id order.
+fn add_offsets_only(listings: &mut Vec<GroupListing>, offset_groups: &OffsetGroups) {
+ let live: HashSet<&str> = listings
+ .iter()
+ .map(|group| group.group_id.as_str())
+ .collect();
+ let offsets_only: Vec<GroupListing> = offset_groups
+ .iter()
+ .filter(|group_id| !live.contains(group_id))
+ .map(|group_id| GroupListing::offsets_only(StrBytes::from_string(group_id.to_owned())))
+ .collect();
+ listings.extend(offsets_only);
+ listings.sort_by(|left, right| left.group_id.cmp(&right.group_id));
+}
+
/// Keeps as many listed groups as fit `max_frame_size`, dropping the rest rather than closing
/// the connection - same approach as `metadata::truncate_all_topics_to_frame_budget` for the
/// identical problem: this coordinator's own state, not a request the client chose, is what
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs
index 941fe8e74..8502626a1 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -27,6 +27,7 @@
pub mod alter_configs;
pub mod api_versions;
pub mod create_topics;
+pub mod delete_groups;
pub mod delete_topics;
pub mod describe_configs;
pub mod describe_groups;
@@ -40,6 +41,7 @@ pub mod list_groups;
pub mod list_offsets;
pub mod metadata;
pub mod offset_commit;
+pub mod offset_delete;
pub mod offset_fetch;
pub mod produce;
pub mod sync_group;
@@ -58,13 +60,13 @@ use crate::auth::AuthenticatedPrincipal;
use crate::bridge::BridgeError;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
- API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_DELETE_TOPICS,
- API_KEY_DESCRIBE_CONFIGS, API_KEY_DESCRIBE_GROUPS, API_KEY_FETCH, API_KEY_FIND_COORDINATOR,
- API_KEY_HEARTBEAT, API_KEY_INIT_PRODUCER_ID, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
- API_KEY_LIST_GROUPS, API_KEY_LIST_OFFSETS, API_KEY_METADATA, API_KEY_OFFSET_COMMIT,
- API_KEY_OFFSET_FETCH, API_KEY_PRODUCE, API_KEY_SYNC_GROUP, ConnectionState,
- ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
- is_supported_version, supported_max_version,
+ API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_DELETE_GROUPS,
+ API_KEY_DELETE_TOPICS, API_KEY_DESCRIBE_CONFIGS, API_KEY_DESCRIBE_GROUPS, API_KEY_FETCH,
+ API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, API_KEY_INIT_PRODUCER_ID, API_KEY_JOIN_GROUP,
+ API_KEY_LEAVE_GROUP, API_KEY_LIST_GROUPS, API_KEY_LIST_OFFSETS, API_KEY_METADATA,
+ API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_DELETE, API_KEY_OFFSET_FETCH, API_KEY_PRODUCE,
+ API_KEY_SYNC_GROUP, ConnectionState, ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version, supported_max_version,
};
/// Record encodes and decodes of this many bytes or more run off the async worker.
@@ -127,8 +129,16 @@ pub async fn dispatch(
API_KEY_HEARTBEAT => heartbeat::handle(state, connection, api_version, body).await,
API_KEY_LEAVE_GROUP => leave_group::handle(state, api_version, body).await,
API_KEY_SYNC_GROUP => sync_group::handle(state, api_version, body).await,
- API_KEY_DESCRIBE_GROUPS => describe_groups::handle(state, api_version, body).await,
- API_KEY_LIST_GROUPS => list_groups::handle(state, api_version, body).await,
+ API_KEY_DESCRIBE_GROUPS => {
+ describe_groups::handle(state, connection, api_version, body).await
+ }
+ API_KEY_LIST_GROUPS => list_groups::handle(state, connection, api_version, body).await,
+ API_KEY_DELETE_GROUPS => {
+ delete_groups::handle(state, connection, principal, api_version, body).await
+ }
+ API_KEY_OFFSET_DELETE => {
+ offset_delete::handle(state, connection, principal, api_version, body).await
+ }
API_KEY_INIT_PRODUCER_ID => init_producer_id::handle(state, api_version, body).await,
_ => HandleOutcome::Close,
}
@@ -143,6 +153,27 @@ pub(crate) fn is_transactional(transactional_id: Option<&TransactionalId>) -> bo
transactional_id.is_some_and(|id| !id.is_empty())
}
+/// Whether `principal` may delete a group or its offsets. `None` means SASL is off, with no
+/// principal to enforce against, as in `DeleteTopics`.
+///
+/// Kafka asks for `DELETE` on the group. Iggy has no group permission, and lets any reader of a
+/// topic delete a group on it, so the gate asks for `manage_topics`, as `DeleteTopics` does.
+fn may_delete_groups(principal: Option<&AuthenticatedPrincipal>, api: &str) -> bool {
+ let Some(principal) = principal else {
+ return true;
+ };
+ if !principal.permissions_known {
+ // Denying is the fail-closed choice: granting would authorize an irreversible delete off
+ // a value nothing ever actually read.
+ tracing::warn!(
+ principal = %principal.username,
+ "{api} asked on a connection whose permissions were never read; denying"
+ );
+ return false;
+ }
+ principal.permissions.manage_topics
+}
+
/// When an offset call on `connection` gives up. See `GroupCoordinator::offset_hold`.
async fn offset_deadline(state: &GatewayState, connection: &ConnectionState) -> Instant {
let member = connection.heartbeat_member();
diff --git a/gateways/kafka/src/protocol/handlers/offset_delete.rs b/gateways/kafka/src/protocol/handlers/offset_delete.rs
new file mode 100644
index 000000000..a4af2d5d0
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/offset_delete.rs
@@ -0,0 +1,484 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF 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.
+
+//! `OffsetDelete` (API key 47).
+//!
+//! Deletes committed offsets of one group, as `kafka-consumer-groups.sh --delete-offsets` asks.
+//! Kafka's rules (`ClassicGroup.validateOffsetDelete`, `OffsetMetadataManager.deleteOffsets`):
+//!
+//! - A group that has no members here and holds no offsets in Iggy gets `GROUP_ID_NOT_FOUND` (69).
+//! - A group with members of a protocol other than `consumer` gets `NON_EMPTY_GROUP` (68).
+//! - Each partition of a topic that a member subscribes to gets `GROUP_SUBSCRIBED_TO_TOPIC` (86).
+//! - A partition that does not exist gets `UNKNOWN_TOPIC_OR_PARTITION` (3).
+//! - Any other partition gets 0, also one that held no offset.
+//!
+//! A group error leaves the topics out. A deletion that a retry can fix answers
+//! `COORDINATOR_LOAD_IN_PROGRESS` (14) for the whole group, which the Java admin client retries.
+//! Deleting an offset twice does no harm. v1 is not in a Kafka release, so only v0 is advertised.
+//!
+//! With SASL on, a principal without `manage_topics` gets `GROUP_AUTHORIZATION_FAILED` (30), and
+//! nothing is deleted.
+
+use std::collections::{HashMap, HashSet};
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::offset_delete_response::{
+ OffsetDeleteResponsePartition, OffsetDeleteResponseTopic,
+};
+use kafka_protocol::messages::{OffsetDeleteRequest, OffsetDeleteResponse};
+
+use crate::auth::AuthenticatedPrincipal;
+use crate::bridge::{BridgeError, IggyBridge, OffsetCalls};
+use crate::error::Result;
+use crate::group::{OffsetDeletion, Subscribed, is_valid_group_id};
+use crate::protocol::api::{
+ API_KEY_OFFSET_DELETE, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_GROUP_AUTHORIZATION_FAILED, ERROR_GROUP_ID_NOT_FOUND, ERROR_GROUP_SUBSCRIBED_TO_TOPIC,
+ ERROR_INVALID_GROUP_ID, ERROR_INVALID_REQUEST, ERROR_NON_EMPTY_GROUP, ERROR_NONE,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_delete_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_DELETE,
+ min_version: 0,
+ max_version: 0,
+};
+
+/// Partition index and error code of one echoed partition.
+const PER_PARTITION_BYTES: usize = 6;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_OFFSET_DELETE, api_version) {
+ return unsupported_version_response(API_KEY_OFFSET_DELETE, api_version, |version| {
+ encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let request = match decode_guarded::<OffsetDeleteRequest>(api_version, body, |version, body| {
+ validate_offset_delete_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetDelete request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "OffsetDelete",
+ );
+ }
+ };
+
+ if !may_delete_groups(principal, "OffsetDelete") {
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_GROUP_AUTHORIZATION_FAILED),
+ "OffsetDelete",
+ );
+ }
+ let encoded = match delete_offsets(state, connection, &request).await {
+ Ok(topics) => encode_response(api_version, topics),
+ Err(code) => encode_error_response(api_version, code),
+ };
+ respond_or_close(encoded, "OffsetDelete")
+}
+
+/// The answer for each requested partition, or the group's error.
+async fn delete_offsets(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ request: &OffsetDeleteRequest,
+) -> core::result::Result<Vec<OffsetDeleteResponseTopic>, i16> {
+ let group = &request.group_id.0;
+ if !is_valid_group_id(group) {
+ return Err(ERROR_INVALID_GROUP_ID);
+ }
+ let deletion = state.groups.offset_deletion(group).await;
+ let Some(bridge) = &state.bridge else {
+ // No group holds offsets, and no topic exists.
+ return match deletion {
+ OffsetDeletion::Absent => Err(ERROR_GROUP_ID_NOT_FOUND),
+ OffsetDeletion::Refused => Err(ERROR_NON_EMPTY_GROUP),
+ OffsetDeletion::Allowed(_) => Ok(same_code(request, ERROR_UNKNOWN_TOPIC_OR_PARTITION)),
+ };
+ };
+ let calls = OffsetCalls::new(offset_deadline(state, connection).await);
+ let subscribed = match deletion {
+ OffsetDeletion::Refused => return Err(ERROR_NON_EMPTY_GROUP),
+ OffsetDeletion::Allowed(subscribed) => subscribed,
+ // No members here, so the group exists only while it holds offsets.
+ OffsetDeletion::Absent => match bridge.offset_groups(calls.deadline()).await {
+ None => return Err(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ Some(groups) if groups.contains(group) => Subscribed::Topics(HashSet::new()),
+ Some(_) => return Err(ERROR_GROUP_ID_NOT_FOUND),
+ },
+ };
+ let partitions = partition_counts(bridge, request, &calls).await?;
+ delete_unsubscribed(bridge, request, &partitions, &subscribed, &calls).await
+}
+
+/// The partition count of each requested topic, or the code for each of its partitions. A lookup
+/// that a retry can fix fails the group.
+async fn partition_counts<'a>(
+ bridge: &IggyBridge,
+ request: &'a OffsetDeleteRequest,
+ calls: &OffsetCalls,
+) -> core::result::Result<HashMap<&'a str, core::result::Result<u32, i16>>, i16> {
+ let mut names: Vec<&str> = request
+ .topics
+ .iter()
+ .map(|topic| topic.name.0.as_str())
+ .collect();
+ names.sort_unstable();
+ names.dedup();
+ let counts = join_all(
+ names
+ .iter()
+ .map(|name| bridge.kafka_topic_partitions(name, calls)),
+ )
+ .await;
+ let mut found = HashMap::with_capacity(names.len());
+ for (name, count) in names.into_iter().zip(counts) {
+ let count = match count {
+ Ok(Some(count)) => Ok(count),
+ // Kafka answers a topic that it does not know with 3, whatever its name.
+ Ok(None) | Err(BridgeError::InvalidKafkaTopicName { .. }) => {
+ Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION)
+ }
+ Err(error) => match error.to_offset_error_code() {
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS => {
+ return Err(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ code => Err(code),
+ },
+ };
+ found.insert(name, count);
+ }
+ Ok(found)
+}
+
+/// Deletes the offset of each requested partition that exists and whose topic no member
+/// subscribes to.
+async fn delete_unsubscribed(
+ bridge: &IggyBridge,
+ request: &OffsetDeleteRequest,
+ partitions: &HashMap<&str, core::result::Result<u32, i16>>,
+ subscribed: &Subscribed,
+ calls: &OffsetCalls,
+) -> core::result::Result<Vec<OffsetDeleteResponseTopic>, i16> {
+ let group = request.group_id.0.as_str();
+ let targets: Vec<_> = request
+ .topics
+ .iter()
+ .map(|topic| bridge.topic_target(topic.name.0.as_str()))
+ .collect();
+ let mut topics = same_code(request, ERROR_NONE);
+ let mut deletions = Vec::new();
+ // Where each deletion's code goes: (topic, partition) in `topics`.
+ let mut positions = Vec::new();
+ for (topic_at, (topic, target)) in request.topics.iter().zip(&targets).enumerate() {
+ let name = topic.name.0.as_str();
+ let count = partitions
+ .get(name)
+ .copied()
+ .unwrap_or(Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION));
+ for (partition_at, partition) in topic.partitions.iter().enumerate() {
+ let index = u32::try_from(partition.partition_index).ok();
+ let code = match (count, index, target) {
+ (Err(code), _, _) => code,
+ (Ok(count), Some(index), Ok(target)) if index < count => {
+ if subscribed.contains(name) {
+ ERROR_GROUP_SUBSCRIBED_TO_TOPIC
+ } else {
+ deletions.push((target, index, -1));
+ positions.push((topic_at, partition_at));
+ continue;
+ }
+ }
+ (Ok(_), _, Err(error)) => error.to_offset_error_code(),
+ (Ok(_), _, Ok(_)) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ };
+ topics[topic_at].partitions[partition_at].error_code = code;
+ }
+ }
+
+ let results = bridge.commit_group_offsets(group, &deletions, calls).await;
+ let mut first_refusal = None;
+ for ((topic_at, partition_at), result) in positions.into_iter().zip(results) {
+ let Err(error) = result else {
+ continue;
+ };
+ let code = error.to_offset_error_code();
+ if code == ERROR_COORDINATOR_LOAD_IN_PROGRESS {
+ return Err(code);
+ }
+ topics[topic_at].partitions[partition_at].error_code = code;
+ if code == ERROR_UNKNOWN_SERVER_ERROR && first_refusal.is_none() {
+ first_refusal = Some((topic_at, partition_at, error));
+ }
+ }
+ // One line per request, as in `OffsetCommit`.
+ if let Some((topic_at, partition_at, error)) = first_refusal {
+ let topic = &topics[topic_at];
+ tracing::error!(
+ group,
+ kafka_topic = topic.name.0.as_str(),
+ partition = topic.partitions[partition_at].partition_index,
+ %error,
+ cause = offset_refusal_cause(&error),
+ "Iggy refused to delete Kafka offsets"
+ );
+ }
+ Ok(topics)
+}
+
+/// Every requested topic and partition, in request order, with `code`.
+fn same_code(request: &OffsetDeleteRequest, code: i16) -> Vec<OffsetDeleteResponseTopic> {
+ request
+ .topics
+ .iter()
+ .map(|topic| {
+ OffsetDeleteResponseTopic::default()
+ .with_name(topic.name.clone())
+ .with_partitions(
+ topic
+ .partitions
+ .iter()
+ .map(|partition| {
+ OffsetDeleteResponsePartition::default()
+ .with_partition_index(partition.partition_index)
+ .with_error_code(code)
+ })
+ .collect(),
+ )
+ })
+ .collect()
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, topics: Vec<OffsetDeleteResponseTopic>) -> Result<Bytes> {
+ let echoed: usize = topics
+ .iter()
+ .map(|topic| topic.name.0.len() + topic.partitions.len() * PER_PARTITION_BYTES)
+ .sum();
+ encode_message(
+ &OffsetDeleteResponse::default().with_topics(topics),
+ version,
+ 16 + echoed,
+ )
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_error_response(version: i16, error_code: i16) -> Result<Bytes> {
+ encode_message(
+ &OffsetDeleteResponse::default().with_error_code(error_code),
+ version,
+ 16,
+ )
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use bytes::BytesMut;
+ use kafka_protocol::messages::offset_delete_request::{
+ OffsetDeleteRequestPartition, OffsetDeleteRequestTopic,
+ };
+ use kafka_protocol::messages::{GroupId, TopicName};
+ use kafka_protocol::protocol::{Decodable, Encodable, StrBytes};
+ use tokio_util::sync::CancellationToken;
+
+ use super::*;
+ use crate::group::{GroupCoordinator, GroupCoordinatorConfig, JoinRequest};
+ use crate::protocol::acl::PrincipalPermissions;
+ use crate::protocol::api::BrokerAdvertise;
+
+ /// No bridge, and a first join that completes at once.
+ fn state() -> GatewayState {
+ let groups = GroupCoordinatorConfig {
+ initial_rebalance_delay: Duration::ZERO,
+ ..GroupCoordinatorConfig::default()
+ };
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ None,
+ 8 * 1024 * 1024,
+ false,
+ 0,
+ GroupCoordinator::new(groups, CancellationToken::new()),
+ )
+ }
+
+ fn join_request(group: &str, protocol_type: &str) -> JoinRequest {
+ JoinRequest {
+ group_id: StrBytes::from_string(group.to_owned()),
+ session_timeout: Duration::from_secs(10),
+ rebalance_timeout: Duration::from_secs(10),
+ member_id: StrBytes::new(),
+ group_instance_id: None,
+ protocol_type: StrBytes::from_string(protocol_type.to_owned()),
+ protocols: vec![(StrBytes::from_static_str("range"), Bytes::new())],
+ require_known_member_id: false,
+ }
+ }
+
+ fn body(group: &str) -> Bytes {
+ let request = OffsetDeleteRequest::default()
+ .with_group_id(GroupId(StrBytes::from_string(group.to_owned())))
+ .with_topics(vec![
+ OffsetDeleteRequestTopic::default()
+ .with_name(TopicName(StrBytes::from_static_str("orders")))
+ .with_partitions(vec![
+ OffsetDeleteRequestPartition::default().with_partition_index(0),
+ OffsetDeleteRequestPartition::default().with_partition_index(1),
+ ]),
+ ]);
+ let mut buf = BytesMut::new();
+ request.encode(&mut buf, 0).unwrap();
+ buf.freeze()
+ }
+
+ fn principal(manage_topics: bool, permissions_known: bool) -> AuthenticatedPrincipal {
+ AuthenticatedPrincipal {
+ username: "alice".to_string(),
+ permissions: PrincipalPermissions {
+ manage_topics,
+ ..PrincipalPermissions::default()
+ },
+ permissions_known,
+ }
+ }
+
+ /// The top-level error and each partition's `(topic, partition, error)`.
+ async fn answer(
+ state: &GatewayState,
+ principal: Option<&AuthenticatedPrincipal>,
+ body: Bytes,
+ ) -> (i16, Vec<(String, i32, i16)>) {
+ let connection = ConnectionState::default();
+ let mut response = handle(state, &connection, principal, 0, body)
+ .await
+ .expect_response("OffsetDelete answers");
+ let response = OffsetDeleteResponse::decode(&mut response, 0).unwrap();
+ let partitions = response
+ .topics
+ .iter()
+ .flat_map(|topic| {
+ topic.partitions.iter().map(|partition| {
+ (
+ topic.name.0.to_string(),
+ partition.partition_index,
+ partition.error_code,
+ )
+ })
+ })
+ .collect();
+ (response.error_code, partitions)
+ }
+
+ /// Without a bridge no group holds offsets and no topic exists.
+ #[tokio::test]
+ async fn given_each_kind_of_group_without_a_bridge_should_answer_as_kafka() {
+ let state = state();
+ let joined = state
+ .groups
+ .join(&join_request("consumers", "consumer"))
+ .await;
+ assert_eq!(joined.error, ERROR_NONE);
+ let joined = state.groups.join(&join_request("workers", "connect")).await;
+ assert_eq!(joined.error, ERROR_NONE);
+
+ assert_eq!(
+ answer(&state, None, body("")).await,
+ (ERROR_INVALID_GROUP_ID, vec![])
+ );
+ assert_eq!(
+ answer(&state, None, body("unknown")).await,
+ (ERROR_GROUP_ID_NOT_FOUND, vec![])
+ );
+ assert_eq!(
+ answer(&state, None, body("workers")).await,
+ (ERROR_NON_EMPTY_GROUP, vec![])
+ );
+ assert_eq!(
+ answer(&state, None, body("consumers")).await,
+ (
+ ERROR_NONE,
+ vec![
+ ("orders".to_string(), 0, ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ ("orders".to_string(), 1, ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ ]
+ )
+ );
+ }
+
+ /// Kafka asks for `DELETE` on the group, which needs `manage_topics` here.
+ #[tokio::test]
+ async fn given_a_principal_without_manage_topics_should_refuse_the_group() {
+ let state = state();
+
+ for refused in [principal(false, true), principal(true, false)] {
+ assert_eq!(
+ answer(&state, Some(&refused), body("unknown")).await,
+ (ERROR_GROUP_AUTHORIZATION_FAILED, vec![]),
+ "known={}",
+ refused.permissions_known
+ );
+ }
+ assert_eq!(
+ answer(&state, Some(&principal(true, true)), body("unknown")).await,
+ (ERROR_GROUP_ID_NOT_FOUND, vec![])
+ );
+ }
+
+ #[tokio::test]
+ async fn given_an_unreadable_request_should_answer_invalid_request() {
+ let state = state();
+
+ assert_eq!(
+ answer(&state, None, body("consumers").slice(..5)).await,
+ (ERROR_INVALID_REQUEST, vec![])
+ );
+ assert!(
+ handle(
+ &state,
+ &ConnectionState::default(),
+ None,
+ 1,
+ body("consumers")
+ )
+ .await
+ .is_close(),
+ "v1 is not advertised"
+ );
+ }
+}
diff --git a/gateways/kafka/tests/common/scope.rs b/gateways/kafka/tests/common/scope.rs
index b263570cf..599bcf25c 100644
--- a/gateways/kafka/tests/common/scope.rs
+++ b/gateways/kafka/tests/common/scope.rs
@@ -45,6 +45,8 @@ pub const SCOPED_API_KEYS: &[(i16, &str, i16, i16)] = &[
(22, "InitProducerId", 0, 5),
(32, "DescribeConfigs", 1, 4),
(33, "AlterConfigs", 0, 2),
+ (42, "DeleteGroups", 0, 2),
+ (47, "OffsetDelete", 0, 0),
];
pub fn default_broker() -> BrokerAdvertise {
diff --git a/gateways/kafka/tests/common/wire.rs b/gateways/kafka/tests/common/wire.rs
index 9add5afa3..55b125927 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -44,12 +44,9 @@ pub fn decode<M: Decodable>(body: Bytes, version: i16) -> M {
M::decode(&mut buf, version).expect("decode response")
}
-/// Consumer-group and admin keys explicitly out of scope in SCOPE.md.
-pub const OUT_OF_SCOPE_API_KEYS: &[(i16, &str)] = &[
- (21, "DeleteRecords"),
- (42, "DeleteGroups"),
- (44, "IncrementalAlterConfigs"),
-];
+/// Admin keys explicitly out of scope in SCOPE.md.
+pub const OUT_OF_SCOPE_API_KEYS: &[(i16, &str)] =
+ &[(21, "DeleteRecords"), (44, "IncrementalAlterConfigs")];
/// Append Metadata request fields that follow the topics array for `version`.
fn write_metadata_request_trailer(enc: &mut Encoder, version: i16) {
@@ -574,6 +571,35 @@ pub fn build_describe_groups_request(
enc.freeze()
}
+/// `DeleteGroups` request naming `groups`.
+pub fn build_delete_groups_request(version: i16, groups: &[&str]) -> Bytes {
+ let flexible = version >= 2;
+ let mut enc = Encoder::with_capacity(64);
+ write_array_count(&mut enc, flexible, groups.len());
+ for group in groups {
+ write_string(&mut enc, flexible, Some(group));
+ }
+ if flexible {
+ enc.write_empty_tagged_fields();
+ }
+ enc.freeze()
+}
+
+/// `OffsetDelete` v0 request for `partitions` of each topic.
+pub fn build_offset_delete_request(group_id: &str, topics: &[(&str, &[i32])]) -> Bytes {
+ let mut enc = Encoder::with_capacity(64);
+ write_string(&mut enc, false, Some(group_id));
+ write_array_count(&mut enc, false, topics.len());
+ for (topic, partitions) in topics {
+ write_string(&mut enc, false, Some(topic));
+ write_array_count(&mut enc, false, partitions.len());
+ for partition in *partitions {
+ enc.write_i32(*partition);
+ }
+ }
+ enc.freeze()
+}
+
/// `ListGroups` request. `states` is written from v4 and `group_types` from v5.
pub fn build_list_groups_request(version: i16, states: &[&str], group_types: &[&str]) -> Bytes {
let mut enc = Encoder::with_capacity(64);
diff --git a/gateways/kafka/tests/golden_wire_fixtures_tests.rs b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
index e5cc16210..d9501878e 100644
--- a/gateways/kafka/tests/golden_wire_fixtures_tests.rs
+++ b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
@@ -41,11 +41,11 @@ async fn golden_apiversions_v3_flexible_response_fixture() {
.await
.expect_response("test request has acks != 0 and expects a response");
- // error_code=0, api_count=19 (compact array: N+1=20)
+ // error_code=0, api_count=21 (compact array: N+1=22)
// each entry followed by an empty tagged-fields byte; throttle_ms=0; top-level tagged fields
- let expected: [u8; 141] = [
+ let expected: [u8; 155] = [
0x00, 0x00, // error_code
- 0x14, // compact array count (19+1)
+ 0x16, // compact array count (21+1)
0x00, 0x00, 0x00, 0x00, 0x00, 0x09, 0x00, // key 0: Produce 0-9 (advertised)
0x00, 0x01, 0x00, 0x04, 0x00, 0x0C, 0x00, // key 1: Fetch 4-12
0x00, 0x02, 0x00, 0x01, 0x00, 0x06, 0x00, // key 2: ListOffsets 1-6
@@ -65,6 +65,8 @@ async fn golden_apiversions_v3_flexible_response_fixture() {
0x00, 0x16, 0x00, 0x00, 0x00, 0x05, 0x00, // key 22: InitProducerId 0-5
0x00, 0x20, 0x00, 0x01, 0x00, 0x04, 0x00, // key 32: DescribeConfigs 1-4
0x00, 0x21, 0x00, 0x00, 0x00, 0x02, 0x00, // key 33: AlterConfigs 0-2
+ 0x00, 0x2A, 0x00, 0x00, 0x00, 0x02, 0x00, // key 42: DeleteGroups 0-2
+ 0x00, 0x2F, 0x00, 0x00, 0x00, 0x00, 0x00, // key 47: OffsetDelete 0-0
0x00, 0x00, 0x00, 0x00, // throttle_ms
0x00, // top-level tagged fields
];
@@ -78,10 +80,10 @@ async fn golden_apiversions_v1_response_fixture() {
.await
.expect_response("test request has acks != 0 and expects a response");
- // error_code=0, api_count=19; Produce advertises min=0 per KAFKA-18659; throttle_ms=0
- let expected: [u8; 124] = [
+ // error_code=0, api_count=21; Produce advertises min=0 per KAFKA-18659; throttle_ms=0
+ let expected: [u8; 136] = [
0x00, 0x00, // error_code
- 0x00, 0x00, 0x00, 0x13, // api count = 19
+ 0x00, 0x00, 0x00, 0x15, // api count = 21
0x00, 0x00, 0x00, 0x00, 0x00, 0x09, // key 0: Produce 0-9 (advertised)
0x00, 0x01, 0x00, 0x04, 0x00, 0x0C, // key 1: Fetch 4-12
0x00, 0x02, 0x00, 0x01, 0x00, 0x06, // key 2: ListOffsets 1-6
@@ -101,6 +103,8 @@ async fn golden_apiversions_v1_response_fixture() {
0x00, 0x16, 0x00, 0x00, 0x00, 0x05, // key 22: InitProducerId 0-5
0x00, 0x20, 0x00, 0x01, 0x00, 0x04, // key 32: DescribeConfigs 1-4
0x00, 0x21, 0x00, 0x00, 0x00, 0x02, // key 33: AlterConfigs 0-2
+ 0x00, 0x2A, 0x00, 0x00, 0x00, 0x02, // key 42: DeleteGroups 0-2
+ 0x00, 0x2F, 0x00, 0x00, 0x00, 0x00, // key 47: OffsetDelete 0-0
0x00, 0x00, 0x00, 0x00, // throttle_ms
];
assert_eq!(actual.as_ref(), &expected);
diff --git a/gateways/kafka/tests/offset_groups_real_bridge_tests.rs b/gateways/kafka/tests/offset_groups_real_bridge_tests.rs
new file mode 100644
index 000000000..bd7c091c1
--- /dev/null
+++ b/gateways/kafka/tests/offset_groups_real_bridge_tests.rs
@@ -0,0 +1,712 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF 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 group admin APIs against a real `iggy-server` process. `ListGroups` and `DescribeGroups`
+//! report a group with no members on the coordinator whose offsets are in Iggy, as Kafka keeps
+//! such a group and reports it as `Empty`. `DeleteGroups` and `OffsetDelete` delete those offsets.
+//! Responses are decoded by hand for the reason that `create_topics_real_bridge_tests.rs` gives.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use iggy::prelude::{ConsumerGroupClient, Identifier};
+use serial_test::serial;
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
+use iggy_gateway_kafka::protocol::api::{
+ API_KEY_DELETE_GROUPS, API_KEY_DESCRIBE_GROUPS, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
+ API_KEY_LIST_GROUPS, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_DELETE, API_KEY_OFFSET_FETCH,
+ BrokerAdvertise, ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_GROUP_ID_NOT_FOUND,
+ ERROR_GROUP_SUBSCRIBED_TO_TOPIC, ERROR_NON_EMPTY_GROUP, ERROR_NONE,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, handle_request_bounded,
+};
+
+use codec::Decoder;
+use iggy_server::TestServer;
+use wire::{
+ JoinGroupParams, build_delete_groups_request, build_describe_groups_request,
+ build_join_group_request, build_leave_group_request, build_list_groups_request,
+ build_offset_commit_request, build_offset_delete_request, build_offset_fetch_request,
+};
+
+const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+const COMMIT_VERSION: i16 = 8;
+const FETCH_VERSION: i16 = 8;
+/// The first flexible version.
+const DELETE_GROUPS_VERSION: i16 = 2;
+/// The first version with both filters.
+const LIST_VERSION: i16 = 5;
+/// The last version without `error_message`, so a missing group is `Dead` with error 0.
+const DESCRIBE_VERSION: i16 = 5;
+/// v3 admits a member without the `MEMBER_ID_REQUIRED` round trip.
+const JOIN_VERSION: i16 = 3;
+const LEAVE_VERSION: i16 = 1;
+const GROUP: &str = "orders-app";
+
+/// One `ListGroups` entry: `(group_id, protocol_type, state, group_type)`.
+type Listed = (String, String, String, String);
+
+/// One `DescribeGroups` entry with the fields an offsets-only group sets.
+#[derive(Debug, PartialEq, Eq)]
+struct Described {
+ error: i16,
+ error_message: Option<String>,
+ group_id: String,
+ state: String,
+ protocol_type: String,
+ protocol_data: String,
+ members: usize,
+}
+
+async fn connected_state(server: &TestServer) -> GatewayState {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ // A new group's first join completes at once, so a test can hold a live member.
+ let groups = GroupCoordinatorConfig {
+ initial_rebalance_delay: Duration::ZERO,
+ ..GroupCoordinatorConfig::default()
+ };
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ TEST_MAX_FRAME_SIZE,
+ false,
+ 0,
+ GroupCoordinator::new(groups, CancellationToken::new()),
+ )
+}
+
+async fn seed_topic(server: &TestServer, topic: &str) {
+ IggyBridge::connect(server.test_config())
+ .await
+ .expect("seed bridge should connect")
+ .ensure_stream_and_topic(topic, 1)
+ .await
+ .expect("seed the topic");
+}
+
+/// Commits `offset` for partition 0 of `topic` outside the group protocol, as admin tools do.
+async fn commit(state: &GatewayState, group: &str, topic: &str, offset: i64) {
+ let body = build_offset_commit_request(COMMIT_VERSION, group, topic, offset);
+ let response = handle_request_bounded(state, API_KEY_OFFSET_COMMIT, COMMIT_VERSION, body)
+ .await
+ .expect_response("OffsetCommit answers");
+ let mut d = Decoder::new(response);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ assert_eq!(d.read_varint().expect("topics"), 2, "one topic");
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ assert_eq!(d.read_varint().expect("partitions"), 2, "one partition");
+ let _index = d.read_i32().expect("partition_index");
+ assert_eq!(d.read_i16().expect("error_code"), ERROR_NONE, "the commit");
+}
+
+/// The top-level error and every group of a v5 `ListGroups` answer.
+async fn list(state: &GatewayState, states: &[&str], group_types: &[&str]) -> (i16, Vec<Listed>) {
+ let body = build_list_groups_request(LIST_VERSION, states, group_types);
+ let response = handle_request_bounded(state, API_KEY_LIST_GROUPS, LIST_VERSION, body)
+ .await
+ .expect_response("ListGroups answers");
+ let mut d = Decoder::new(response);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let error = d.read_i16().expect("error_code");
+ let mut groups = Vec::new();
+ for _ in 1..d.read_varint().expect("groups") {
+ let mut string = || {
+ d.read_compact_nullable_string()
+ .expect("a string")
+ .expect("not null")
+ };
+ groups.push((string(), string(), string(), string()));
+ d.read_tagged_fields().expect("group tagged fields");
+ }
+ d.read_tagged_fields().expect("tagged fields");
+ assert_eq!(d.remaining(), 0, "ListGroups trailing bytes");
+ (error, groups)
+}
+
+async fn describe(state: &GatewayState, version: i16, groups: &[&str]) -> Vec<Described> {
+ let body = build_describe_groups_request(version, groups, false);
+ let response = handle_request_bounded(state, API_KEY_DESCRIBE_GROUPS, version, body)
+ .await
+ .expect_response("DescribeGroups answers");
+ let mut d = Decoder::new(response);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let mut described = Vec::new();
+ for _ in 1..d.read_varint().expect("groups") {
+ let error = d.read_i16().expect("error_code");
+ let error_message = if version >= 6 {
+ d.read_compact_nullable_string().expect("error_message")
+ } else {
+ None
+ };
+ let mut string = || {
+ d.read_compact_nullable_string()
+ .expect("a string")
+ .expect("not null")
+ };
+ let (group_id, state, protocol_type, protocol_data) =
+ (string(), string(), string(), string());
+ let members = usize::try_from(d.read_varint().expect("members") - 1).expect("a count");
+ assert_eq!(members, 0, "only groups with no members are described here");
+ let _authorized_operations = d.read_i32().expect("authorized_operations");
+ d.read_tagged_fields().expect("group tagged fields");
+ described.push(Described {
+ error,
+ error_message,
+ group_id,
+ state,
+ protocol_type,
+ protocol_data,
+ members,
+ });
+ }
+ d.read_tagged_fields().expect("tagged fields");
+ assert_eq!(d.remaining(), 0, "DescribeGroups trailing bytes");
+ described
+}
+
+/// Joins a member to `group` and returns its id. The group is then `CompletingRebalance`.
+async fn join(state: &GatewayState, group: &str) -> String {
+ join_with(
+ state,
+ &JoinGroupParams {
+ group_id: group,
+ ..JoinGroupParams::default()
+ },
+ )
+ .await
+}
+
+/// [`join`] with `params`.
+async fn join_with(state: &GatewayState, params: &JoinGroupParams<'_>) -> String {
+ let body = build_join_group_request(JOIN_VERSION, params);
+ let response = handle_request_bounded(state, API_KEY_JOIN_GROUP, JOIN_VERSION, body)
+ .await
+ .expect_response("JoinGroup answers");
+ let mut d = Decoder::new(response);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ assert_eq!(d.read_i16().expect("error_code"), ERROR_NONE, "the join");
+ let _generation_id = d.read_i32().expect("generation_id");
+ let _protocol_name = d.read_nullable_string().expect("protocol_name");
+ let _leader = d.read_nullable_string().expect("leader");
+ d.read_nullable_string()
+ .expect("member_id")
+ .expect("a member id")
+}
+
+async fn leave(state: &GatewayState, group: &str, member_id: &str) {
+ let body = build_leave_group_request(LEAVE_VERSION, group, &[(member_id, None, None)]);
+ let response = handle_request_bounded(state, API_KEY_LEAVE_GROUP, LEAVE_VERSION, body)
+ .await
+ .expect_response("LeaveGroup answers");
+ let mut d = Decoder::new(response);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ assert_eq!(d.read_i16().expect("error_code"), ERROR_NONE, "the leave");
+}
+
+/// The committed offset of partition 0 of `topic`, -1 when there is none.
+async fn committed(state: &GatewayState, group: &str, topic: &str) -> i64 {
+ let body = build_offset_fetch_request(FETCH_VERSION, group, Some(topic));
+ let response = handle_request_bounded(state, API_KEY_OFFSET_FETCH, FETCH_VERSION, body)
+ .await
+ .expect_response("OffsetFetch answers");
+ let mut d = Decoder::new(response);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ assert_eq!(d.read_varint().expect("groups"), 2, "one group");
+ let _group_id = d.read_compact_nullable_string().expect("group_id");
+ assert_eq!(d.read_varint().expect("topics"), 2, "one topic");
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ assert_eq!(d.read_varint().expect("partitions"), 2, "one partition");
+ let _index = d.read_i32().expect("partition_index");
+ let offset = d.read_i64().expect("committed_offset");
+ let _leader_epoch = d.read_i32().expect("committed_leader_epoch");
+ let _metadata = d.read_compact_nullable_string().expect("metadata");
+ assert_eq!(
+ d.read_i16().expect("error_code"),
+ ERROR_NONE,
+ "the partition"
+ );
+ d.read_tagged_fields().expect("partition tagged fields");
+ d.read_tagged_fields().expect("topic tagged fields");
+ assert_eq!(
+ d.read_i16().expect("group error_code"),
+ ERROR_NONE,
+ "the group"
+ );
+ offset
+}
+
+/// `(group_id, error_code)` per group of a v2 `DeleteGroups` answer.
+async fn delete_groups(state: &GatewayState, groups: &[&str]) -> Vec<(String, i16)> {
+ let body = build_delete_groups_request(DELETE_GROUPS_VERSION, groups);
+ let response =
+ handle_request_bounded(state, API_KEY_DELETE_GROUPS, DELETE_GROUPS_VERSION, body)
+ .await
+ .expect_response("DeleteGroups answers");
+ let mut d = Decoder::new(response);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let mut results = Vec::new();
+ for _ in 1..d.read_varint().expect("results") {
+ let group_id = d
+ .read_compact_nullable_string()
+ .expect("group_id")
+ .expect("not null");
+ results.push((group_id, d.read_i16().expect("error_code")));
+ d.read_tagged_fields().expect("result tagged fields");
+ }
+ d.read_tagged_fields().expect("tagged fields");
+ assert_eq!(d.remaining(), 0, "DeleteGroups trailing bytes");
+ results
+}
+
+/// The top-level error and `(topic, partition, error_code)` per partition of an `OffsetDelete`.
+async fn delete_offsets(
+ state: &GatewayState,
+ group: &str,
+ topics: &[(&str, &[i32])],
+) -> (i16, Vec<(String, i32, i16)>) {
+ let body = build_offset_delete_request(group, topics);
+ let response = handle_request_bounded(state, API_KEY_OFFSET_DELETE, 0, body)
+ .await
+ .expect_response("OffsetDelete answers");
+ let mut d = Decoder::new(response);
+ let error = d.read_i16().expect("error_code");
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let mut partitions = Vec::new();
+ for _ in 0..d.read_i32().expect("topics") {
+ let name = d
+ .read_nullable_string()
+ .expect("topic name")
+ .expect("not null");
+ for _ in 0..d.read_i32().expect("partitions") {
+ let index = d.read_i32().expect("partition_index");
+ partitions.push((name.clone(), index, d.read_i16().expect("error_code")));
+ }
+ }
+ assert_eq!(d.remaining(), 0, "OffsetDelete trailing bytes");
+ (error, partitions)
+}
+
+/// A consumer's subscription to `topics`, as the Java consumer sends it in `JoinGroup`.
+fn subscription(topics: &[&str]) -> Vec<u8> {
+ let mut bytes = 0_i16.to_be_bytes().to_vec();
+ let count = i32::try_from(topics.len()).expect("a few topics");
+ bytes.extend_from_slice(&count.to_be_bytes());
+ for topic in topics {
+ let len = i16::try_from(topic.len()).expect("a short name");
+ bytes.extend_from_slice(&len.to_be_bytes());
+ bytes.extend_from_slice(topic.as_bytes());
+ }
+ bytes.extend_from_slice(&(-1_i32).to_be_bytes()); // user_data: null
+ bytes
+}
+
+fn listed(group: &str, state: &str) -> Listed {
+ let group_type = "classic".to_string();
+ (
+ group.to_string(),
+ "consumer".to_string(),
+ state.to_string(),
+ group_type,
+ )
+}
+
+fn offsets_only(group: &str) -> Described {
+ Described {
+ error: ERROR_NONE,
+ error_message: None,
+ group_id: group.to_string(),
+ state: "Empty".to_string(),
+ protocol_type: "consumer".to_string(),
+ protocol_data: String::new(),
+ members: 0,
+ }
+}
+
+/// A group whose consumers committed and left has no members here. Kafka still lists and
+/// describes it, as `Empty`, while it holds offsets.
+#[tokio::test]
+#[serial]
+async fn given_offsets_and_no_members_list_and_describe_should_report_an_empty_group() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+ commit(&state, GROUP, "orders", 7).await;
+
+ assert_eq!(
+ list(&state, &[], &[]).await,
+ (ERROR_NONE, vec![listed(GROUP, "Empty")])
+ );
+ assert_eq!(
+ describe(&state, DESCRIBE_VERSION, &[GROUP, "never-committed"]).await,
+ vec![
+ offsets_only(GROUP),
+ Described {
+ state: "Dead".to_string(),
+ protocol_type: String::new(),
+ ..offsets_only("never-committed")
+ },
+ ]
+ );
+ assert_eq!(
+ describe(&state, 6, &[GROUP, "never-committed"]).await,
+ vec![
+ offsets_only(GROUP),
+ Described {
+ error: ERROR_GROUP_ID_NOT_FOUND,
+ error_message: Some("Group never-committed not found.".to_string()),
+ state: "Dead".to_string(),
+ protocol_type: String::new(),
+ ..offsets_only("never-committed")
+ },
+ ]
+ );
+}
+
+/// A live group is listed once, with its live state. After its last member leaves, the group is
+/// `Empty`, because its offsets are still in Iggy.
+#[tokio::test]
+#[serial]
+async fn given_a_live_group_with_offsets_should_list_it_once_and_empty_after_it_leaves() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+ commit(&state, GROUP, "orders", 7).await;
+
+ let member_id = join(&state, GROUP).await;
+ assert_eq!(
+ list(&state, &[], &[]).await,
+ (ERROR_NONE, vec![listed(GROUP, "CompletingRebalance")])
+ );
+
+ leave(&state, GROUP, &member_id).await;
+ assert_eq!(
+ list(&state, &[], &[]).await,
+ (ERROR_NONE, vec![listed(GROUP, "Empty")])
+ );
+ assert_eq!(
+ describe(&state, DESCRIBE_VERSION, &[GROUP]).await,
+ vec![offsets_only(GROUP)]
+ );
+}
+
+/// The filters apply to a group that only holds offsets as to any other: its state is `Empty` and
+/// its type is `classic`.
+#[tokio::test]
+#[serial]
+async fn given_filters_should_keep_an_offsets_only_group_only_when_they_match() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+ commit(&state, GROUP, "orders", 7).await;
+ let empty = (ERROR_NONE, vec![listed(GROUP, "Empty")]);
+
+ assert_eq!(list(&state, &[" empty "], &[]).await, empty);
+ assert_eq!(list(&state, &[], &["Classic"]).await, empty);
+ assert_eq!(list(&state, &["Stable"], &[]).await, (ERROR_NONE, vec![]));
+ assert_eq!(list(&state, &[], &["consumer"]).await, (ERROR_NONE, vec![]));
+}
+
+/// Groups come from a scan that answers for a while. A commit through this gateway adds its group
+/// at once, so a group does not wait for the next scan.
+#[tokio::test]
+#[serial]
+async fn given_a_commit_after_the_last_scan_should_list_its_group_at_once() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+ commit(&state, "first-app", "orders", 1).await;
+ assert_eq!(
+ list(&state, &[], &[]).await,
+ (ERROR_NONE, vec![listed("first-app", "Empty")])
+ );
+
+ commit(&state, "second-app", "orders", 2).await;
+
+ assert_eq!(
+ list(&state, &[], &[]).await,
+ (
+ ERROR_NONE,
+ vec![listed("first-app", "Empty"), listed("second-app", "Empty")]
+ )
+ );
+}
+
+/// A gateway that just started has no memory of the groups. Its scan finds the groups that another
+/// gateway, or an earlier run, committed for, on every topic.
+#[tokio::test]
+#[serial]
+async fn given_a_new_gateway_should_find_the_groups_of_every_topic() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ for topic in ["orders", "payments"] {
+ seed_topic(&server, topic).await;
+ }
+ let first = connected_state(&server).await;
+ commit(&first, "first-app", "orders", 1).await;
+ commit(&first, "second-app", "payments", 2).await;
+ commit(&first, "first-app", "payments", 3).await;
+ drop(first);
+
+ let restarted = connected_state(&server).await;
+
+ assert_eq!(
+ list(&restarted, &[], &[]).await,
+ (
+ ERROR_NONE,
+ vec![listed("first-app", "Empty"), listed("second-app", "Empty")]
+ )
+ );
+}
+
+/// Only an Iggy group named `kafka.cg.<group>` holds Kafka offsets. A native Iggy group on the
+/// same topic is not a Kafka group.
+#[tokio::test]
+#[serial]
+async fn given_a_native_iggy_group_should_not_list_it() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let raw = iggy_server::raw_client(&server).await;
+ let iggy_id = |name: &str| Identifier::named(name).expect("valid Iggy name");
+ raw.create_consumer_group(&iggy_id("kafka"), &iggy_id("orders"), "native-app")
+ .await
+ .expect("create a native Iggy group");
+ let state = connected_state(&server).await;
+
+ assert_eq!(list(&state, &[], &[]).await, (ERROR_NONE, vec![]));
+}
+
+/// Without a scan that found the groups, the gateway cannot tell `Empty` from `Dead`. It answers
+/// `COORDINATOR_LOAD_IN_PROGRESS`, which clients retry, rather than a list that leaves groups out.
+#[tokio::test]
+#[serial]
+async fn given_iggy_unreachable_before_any_scan_should_answer_coordinator_load_in_progress() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let state = connected_state(&server).await;
+ drop(server);
+
+ assert_eq!(
+ list(&state, &[], &[]).await,
+ (ERROR_COORDINATOR_LOAD_IN_PROGRESS, vec![])
+ );
+ assert_eq!(
+ list(&state, &["Stable"], &[]).await,
+ (ERROR_NONE, vec![]),
+ "a filter that leaves out Empty needs no scan"
+ );
+ let unknown = describe(&state, DESCRIBE_VERSION, &[GROUP]).await;
+ assert_eq!(
+ unknown,
+ vec![Described {
+ error: ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ state: String::new(),
+ protocol_type: String::new(),
+ ..offsets_only(GROUP)
+ }]
+ );
+}
+
+/// Kafka refuses to delete a group while it has members. After they leave, the group goes with
+/// its offsets, and the admin APIs no longer find it.
+#[tokio::test]
+#[serial]
+async fn given_a_group_with_members_delete_should_refuse_it_and_delete_it_after_they_leave() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+ commit(&state, GROUP, "orders", 7).await;
+ let member_id = join(&state, GROUP).await;
+
+ assert_eq!(
+ delete_groups(&state, &[GROUP]).await,
+ vec![(GROUP.to_string(), ERROR_NON_EMPTY_GROUP)]
+ );
+ assert_eq!(committed(&state, GROUP, "orders").await, 7);
+
+ leave(&state, GROUP, &member_id).await;
+ assert_eq!(
+ delete_groups(&state, &[GROUP]).await,
+ vec![(GROUP.to_string(), ERROR_NONE)]
+ );
+ assert_eq!(committed(&state, GROUP, "orders").await, -1);
+ assert_eq!(list(&state, &[], &[]).await, (ERROR_NONE, vec![]));
+ assert_eq!(
+ describe(&state, DESCRIBE_VERSION, &[GROUP]).await,
+ vec![Described {
+ state: "Dead".to_string(),
+ protocol_type: String::new(),
+ ..offsets_only(GROUP)
+ }]
+ );
+ assert_eq!(
+ delete_groups(&state, &[GROUP]).await,
+ vec![(GROUP.to_string(), ERROR_GROUP_ID_NOT_FOUND)]
+ );
+}
+
+/// A group goes from every topic that holds it. Other groups on those topics stay.
+#[tokio::test]
+#[serial]
+async fn given_a_group_on_several_topics_delete_should_remove_it_from_each_and_keep_the_others() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ for topic in ["orders", "payments"] {
+ seed_topic(&server, topic).await;
+ }
+ let state = connected_state(&server).await;
+ commit(&state, GROUP, "orders", 1).await;
+ commit(&state, GROUP, "payments", 2).await;
+ commit(&state, "other-app", "orders", 3).await;
+
+ assert_eq!(
+ delete_groups(&state, &[GROUP, "never-committed"]).await,
+ vec![
+ (GROUP.to_string(), ERROR_NONE),
+ ("never-committed".to_string(), ERROR_GROUP_ID_NOT_FOUND),
+ ]
+ );
+
+ assert_eq!(committed(&state, GROUP, "orders").await, -1);
+ assert_eq!(committed(&state, GROUP, "payments").await, -1);
+ assert_eq!(committed(&state, "other-app", "orders").await, 3);
+ assert_eq!(
+ list(&state, &[], &[]).await,
+ (ERROR_NONE, vec![listed("other-app", "Empty")])
+ );
+ let raw = iggy_server::raw_client(&server).await;
+ let iggy_id = |name: &str| Identifier::named(name).expect("valid Iggy name");
+ for topic in ["orders", "payments"] {
+ let found = raw
+ .get_consumer_group(
+ &iggy_id("kafka"),
+ &iggy_id(topic),
+ &iggy_id("kafka.cg.orders-app"),
+ )
+ .await
+ .expect("get the Iggy group");
+ assert!(found.is_none(), "the Iggy group is gone from {topic}");
+ }
+}
+
+/// Kafka keeps the offsets of a topic that a member subscribes to, and deletes the others, also
+/// for a partition that held none.
+#[tokio::test]
+#[serial]
+async fn given_a_member_subscribed_to_one_topic_delete_offsets_should_delete_only_the_others() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ for topic in ["orders", "payments"] {
+ seed_topic(&server, topic).await;
+ }
+ let state = connected_state(&server).await;
+ commit(&state, GROUP, "orders", 5).await;
+ commit(&state, GROUP, "payments", 6).await;
+ let orders = subscription(&["orders"]);
+ let protocols: &[(&str, &[u8])] = &[("range", &orders)];
+ let member_id = join_with(
+ &state,
+ &JoinGroupParams {
+ group_id: GROUP,
+ protocols,
+ ..JoinGroupParams::default()
+ },
+ )
+ .await;
+
+ assert_eq!(
+ delete_offsets(
+ &state,
+ GROUP,
+ &[("orders", &[0]), ("payments", &[0, 1]), ("missing", &[0])],
+ )
+ .await,
+ (
+ ERROR_NONE,
+ vec![
+ ("orders".to_string(), 0, ERROR_GROUP_SUBSCRIBED_TO_TOPIC),
+ ("payments".to_string(), 0, ERROR_NONE),
+ ("payments".to_string(), 1, ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ ("missing".to_string(), 0, ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ ]
+ )
+ );
+ assert_eq!(committed(&state, GROUP, "orders").await, 5);
+ assert_eq!(committed(&state, GROUP, "payments").await, -1);
+
+ leave(&state, GROUP, &member_id).await;
+ assert_eq!(
+ delete_offsets(&state, GROUP, &[("orders", &[0])]).await,
+ (ERROR_NONE, vec![("orders".to_string(), 0, ERROR_NONE)])
+ );
+ assert_eq!(committed(&state, GROUP, "orders").await, -1);
+ assert_eq!(
+ delete_offsets(&state, "never-committed", &[("orders", &[0])]).await,
+ (ERROR_GROUP_ID_NOT_FOUND, vec![])
+ );
+}
+
+/// A subscription that does not decode keeps every topic, as in Kafka. A group of another
+/// protocol keeps every offset while it has members.
+#[tokio::test]
+#[serial]
+async fn given_members_kafka_cannot_read_delete_offsets_should_keep_every_offset() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+ commit(&state, GROUP, "orders", 5).await;
+ let _member_id = join(&state, GROUP).await;
+ let _worker_id = join_with(
+ &state,
+ &JoinGroupParams {
+ group_id: "connect-cluster",
+ protocol_type: "connect",
+ ..JoinGroupParams::default()
+ },
+ )
+ .await;
+
+ assert_eq!(
+ delete_offsets(&state, GROUP, &[("orders", &[0])]).await,
+ (
+ ERROR_NONE,
+ vec![("orders".to_string(), 0, ERROR_GROUP_SUBSCRIBED_TO_TOPIC)]
+ )
+ );
+ assert_eq!(committed(&state, GROUP, "orders").await, 5);
+ assert_eq!(
+ delete_offsets(&state, "connect-cluster", &[("orders", &[0])]).await,
+ (ERROR_NON_EMPTY_GROUP, vec![])
+ );
+}
diff --git a/gateways/kafka/tests/version_firewall_tests.rs b/gateways/kafka/tests/version_firewall_tests.rs
index e0dce7872..b28a0a2a6 100644
--- a/gateways/kafka/tests/version_firewall_tests.rs
+++ b/gateways/kafka/tests/version_firewall_tests.rs
@@ -37,13 +37,13 @@ use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use iggy_gateway_kafka::protocol::api::{
- API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_DELETE_TOPICS,
- API_KEY_DESCRIBE_CONFIGS, API_KEY_DESCRIBE_GROUPS, API_KEY_FETCH, API_KEY_FIND_COORDINATOR,
- API_KEY_HEARTBEAT, API_KEY_INIT_PRODUCER_ID, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
- API_KEY_LIST_GROUPS, API_KEY_LIST_OFFSETS, API_KEY_METADATA, API_KEY_OFFSET_COMMIT,
- API_KEY_OFFSET_FETCH, API_KEY_PRODUCE, API_KEY_SYNC_GROUP, ERROR_INVALID_REQUEST, ERROR_NONE,
- ERROR_UNSUPPORTED_VERSION, advertised_min_version, handle_request, is_supported_version,
- supported_api_ranges,
+ API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_DELETE_GROUPS,
+ API_KEY_DELETE_TOPICS, API_KEY_DESCRIBE_CONFIGS, API_KEY_DESCRIBE_GROUPS, API_KEY_FETCH,
+ API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, API_KEY_INIT_PRODUCER_ID, API_KEY_JOIN_GROUP,
+ API_KEY_LEAVE_GROUP, API_KEY_LIST_GROUPS, API_KEY_LIST_OFFSETS, API_KEY_METADATA,
+ API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_DELETE, API_KEY_OFFSET_FETCH, API_KEY_PRODUCE,
+ API_KEY_SYNC_GROUP, ERROR_INVALID_REQUEST, ERROR_NONE, ERROR_UNSUPPORTED_VERSION,
+ advertised_min_version, handle_request, is_supported_version, supported_api_ranges,
};
use codec::Decoder;
@@ -58,18 +58,19 @@ use tcp::{
use wire::{
JoinGroupParams, OUT_OF_SCOPE_API_KEYS, SyncGroupParams, build_alter_configs_empty_request,
build_api_versions_flexible_request, build_create_topics_empty_request,
- build_delete_topics_empty_request, build_describe_configs_empty_request,
- build_describe_groups_request, build_fetch_empty_topics_request,
- build_find_coordinator_request, build_heartbeat_request, build_init_producer_id_request,
- build_join_group_request, build_leave_group_request, build_list_groups_request,
- build_list_offsets_request, build_metadata_all_topics_flexible,
+ build_delete_groups_request, build_delete_topics_empty_request,
+ build_describe_configs_empty_request, build_describe_groups_request,
+ build_fetch_empty_topics_request, build_find_coordinator_request, build_heartbeat_request,
+ build_init_producer_id_request, build_join_group_request, build_leave_group_request,
+ build_list_groups_request, build_list_offsets_request, build_metadata_all_topics_flexible,
build_metadata_all_topics_legacy, build_metadata_flexible_request_v10,
- build_offset_commit_request, build_offset_fetch_request, build_sync_group_request,
+ build_offset_commit_request, build_offset_delete_request, build_offset_fetch_request,
+ build_sync_group_request,
};
#[test]
-fn supported_ranges_table_has_nineteen_entries() {
- assert_eq!(supported_api_ranges().len(), 19);
+fn supported_ranges_table_has_twenty_one_entries() {
+ assert_eq!(supported_api_ranges().len(), 21);
}
#[test]
@@ -109,7 +110,8 @@ fn is_supported_version_matches_scope_table() {
/// in that order (Produce, Fetch, `ListOffsets`, Metadata, `OffsetCommit`, `OffsetFetch`,
/// `FindCoordinator`, `JoinGroup`, Heartbeat, `LeaveGroup`, `SyncGroup`, `DescribeGroups`,
/// `ListGroups`, `ApiVersions`, `CreateTopics`, `DeleteTopics`, `InitProducerId`,
-/// `DescribeConfigs`, `AlterConfigs`). `supported_ranges_table_has_nineteen_entries` and
+/// `DescribeConfigs`, `AlterConfigs`, `DeleteGroups`, `OffsetDelete`).
+/// `supported_ranges_table_has_twenty_one_entries` and
/// `is_supported_version_matches_scope_table` pin that both tables cover the same keys.
#[tokio::test]
async fn apiversions_advertises_exact_supported_ranges_v1() {
@@ -533,6 +535,8 @@ fn request_body_for_scoped_api(api_key: i16, name: &str, version: i16) -> Bytes
API_KEY_ALTER_CONFIGS => build_alter_configs_empty_request(version),
API_KEY_OFFSET_COMMIT => build_offset_commit_request(version, "scope-group", "orders", 0),
API_KEY_OFFSET_FETCH => build_offset_fetch_request(version, "scope-group", Some("orders")),
+ API_KEY_DELETE_GROUPS => build_delete_groups_request(version, &["scope-group"]),
+ API_KEY_OFFSET_DELETE => build_offset_delete_request("scope-group", &[("orders", &[0])]),
_ => Bytes::new(),
}
}
diff --git a/gateways/kafka/tools/kafka-tool/src/main.rs b/gateways/kafka/tools/kafka-tool/src/main.rs
index 40efe746c..9934caa06 100644
--- a/gateways/kafka/tools/kafka-tool/src/main.rs
+++ b/gateways/kafka/tools/kafka-tool/src/main.rs
@@ -30,6 +30,9 @@ use kafka_protocol::messages::join_group_request::*;
use kafka_protocol::messages::leave_group_request::MemberIdentity;
use kafka_protocol::messages::list_offsets_request::*;
use kafka_protocol::messages::offset_commit_request::*;
+use kafka_protocol::messages::offset_delete_request::{
+ OffsetDeleteRequestPartition, OffsetDeleteRequestTopic,
+};
use kafka_protocol::messages::offset_fetch_request::OffsetFetchRequestGroup;
use kafka_protocol::messages::produce_request::*;
use kafka_protocol::messages::txn_offset_commit_request::*;
@@ -573,6 +576,24 @@ fn build_payload(api_key: i16, version: i16) -> Result<Bytes> {
.encode(&mut buf, version)
.context("SaslAuthenticate")?;
}
+ 42 => {
+ DeleteGroupsRequest::default()
+ .with_groups_names(vec![GroupId::from(StrBytes::from_static_str("test-group"))])
+ .encode(&mut buf, version)
+ .context("DeleteGroups")?;
+ }
+ 47 => {
+ let topic = OffsetDeleteRequestTopic::default()
+ .with_name(TopicName::from(StrBytes::from_static_str("test-topic")))
+ .with_partitions(vec![
+ OffsetDeleteRequestPartition::default().with_partition_index(0),
+ ]);
+ OffsetDeleteRequest::default()
+ .with_group_id(GroupId::from(StrBytes::from_static_str("test-group")))
+ .with_topics(vec![topic])
+ .encode(&mut buf, version)
+ .context("OffsetDelete")?;
+ }
other => {
warn!("api_key={other}: no explicit builder — empty payload (framing test)");
}
diff --git a/gateways/kafka/tools/kafka-tool/src/response.rs b/gateways/kafka/tools/kafka-tool/src/response.rs
index d779da525..b81f6ad43 100644
--- a/gateways/kafka/tools/kafka-tool/src/response.rs
+++ b/gateways/kafka/tools/kafka-tool/src/response.rs
@@ -108,6 +108,7 @@ fn is_acceptable_verify_error(api_key: i16, error_code: i16) -> bool {
0..=2 => error_code == 6, // Produce/Fetch/ListOffsets stub: NOT_LEADER_OR_FOLLOWER
3 => error_code == 3, // Metadata stub: UNKNOWN_TOPIC_OR_PARTITION
19 | 32 | 33 => error_code == 41, // CreateTopics / DescribeConfigs / AlterConfigs stub
+ 47 => error_code == 69, // OffsetDelete: `test-group` holds no offsets
_ => false,
}
}