You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ra...@apache.org on 2026/10/08 15:19:53 UTC
(iggy) 01/01: Merge branch 'master' into feat/gateways-3546-describe-alter-configs
This is an automated email from the ASF dual-hosted git repository.
ryerraguntla pushed a commit to branch feat/gateways-3546-describe-alter-configs
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit afa9b80a1025c4a7703966dad2730ada63a3312d
Merge: 6f8a92783 09420cf4d
Author: ryerraguntla <ry...@users.noreply.github.com>
AuthorDate: Thu Oct 8 11:19:47 2026 -0400
Merge branch 'master' into feat/gateways-3546-describe-alter-configs
.config/nextest.toml | 1 +
.config/rail.toml | 36 +-
.devcontainer/Dockerfile | 2 +-
.github/actions/cpp-bazel/pre-merge/action.yml | 16 +-
.github/actions/rust/pre-merge/action.yml | 258 +-
.github/config/edge-image-variants.json | 117 +
.github/config/publish.yml | 28 +-
.github/workflows/_common.yml | 39 +-
.github/workflows/_detect.yml | 27 +
.github/workflows/_test.yml | 18 +-
.github/workflows/coverage-baseline.yml | 32 +-
.github/workflows/post-merge.yml | 61 +-
.github/workflows/pre-merge.yml | 5 +
.github/workflows/publish.yml | 2 +-
.pr4372-comments.json | 0
Cargo.lock | 8 +-
Cargo.toml | 1 -
Dockerfile | 2 +-
bdd/java/Dockerfile | 2 +-
bdd/java/build.gradle.kts | 6 +-
bdd/java/gradle/wrapper/gradle-wrapper.properties | 2 +-
bdd/java/gradlew | 4 +-
bdd/php/Dockerfile | 2 +-
bdd/python/Dockerfile | 2 +-
bdd/python/uv.lock | 6 +-
bdd/rust/Dockerfile | 2 +-
codecov.yml | 7 +
core/ai/mcp/Dockerfile | 2 +-
core/bench/dashboard/frontend/src/api/mod.rs | 9 +-
.../frontend/src/components/app_content.rs | 28 +-
.../frontend/src/components/chart/trend_chart.rs | 4 +-
.../frontend/src/components/layout/sidebar.rs | 4 +-
.../dashboard/frontend/src/state/benchmark.rs | 22 +-
core/bench/dashboard/server/Dockerfile | 2 +-
core/bench/report/Cargo.toml | 1 -
core/bench/report/src/types/hardware.rs | 2 +-
core/bench/src/utils/client_factory.rs | 20 +-
core/binary_protocol/src/codes.rs | 7 +-
core/binary_protocol/src/consensus/command.rs | 8 +-
.../src/consensus/consumer_session.rs | 32 +-
core/binary_protocol/src/consensus/header.rs | 121 +-
core/binary_protocol/src/consensus/operation.rs | 24 +-
core/binary_protocol/src/dispatch.rs | 8 +-
core/binary_protocol/src/framing.rs | 6 +-
core/binary_protocol/src/lib.rs | 4 +-
core/binary_protocol/src/primitives/consumer.rs | 29 +-
.../src/requests/system/attach_consumer_session.rs | 87 -
.../src/requests/system/bind_session.rs | 155 +
core/binary_protocol/src/requests/system/mod.rs | 6 +-
.../src/requests/system/retire_sessions.rs | 117 +
.../src/requests/users/login_register.rs | 60 +-
.../src/requests/users/login_register_with_pat.rs | 17 +-
.../src/responses/messages/poll_routing.rs | 8 +-
.../src/responses/messages/send_messages.rs | 14 +-
core/binary_protocol/src/version.rs | 145 +-
core/cli/src/args/consumer_offset.rs | 6 +-
.../binary_consumer_offsets/get_consumer_offset.rs | 1 +
core/common/src/error/iggy_error.rs | 8 +
core/common/src/lib.rs | 1 +
core/common/src/traits/binary_impls/mod.rs | 2 +-
.../traits/binary_impls/personal_access_tokens.rs | 25 +-
core/common/src/traits/binary_impls/users.rs | 25 +-
core/common/src/traits/binary_transport.rs | 46 +
core/common/src/traits/consumer_offset_client.rs | 7 +
core/common/src/traits/message_client.rs | 1 +
core/common/src/traits/mod.rs | 1 +
core/common/src/types/consumer/consumer_kind.rs | 78 +-
core/common/src/types/streaming_stats.rs | 4 +-
core/configs/src/server_config/metadata.rs | 12 +-
core/configs/src/server_config/partition.rs | 11 +-
core/configs/src/server_config/sharding.rs | 79 +-
core/connectors/runtime/Dockerfile | 2 +-
core/consensus/Cargo.toml | 1 +
core/consensus/src/client_table.rs | 3844 ++-
core/consensus/src/impls.rs | 368 +-
core/consensus/src/lib.rs | 28 +-
core/consensus/src/metadata_helpers.rs | 421 +-
core/consensus/src/observability.rs | 2 +
core/consensus/src/plane_helpers.rs | 152 +-
core/integration/src/harness/disk.rs | 1 +
core/integration/tests/cli/consumer_offset/mod.rs | 1 +
.../test_consumer_offset_external_group_command.rs | 149 +
.../test_consumer_offset_get_command.rs | 6 +-
.../test_consumer_offset_set_command.rs | 6 +-
.../tests/cluster/client_table_adversarial.rs | 112 +-
.../tests/cluster/client_table_restart.rs | 308 +-
core/integration/tests/cluster/crash_durability.rs | 56 +
.../tests/cluster/crash_recovery_corruption.rs | 30 +-
.../tests/cluster/fast_primary_rejoin.rs | 1 +
core/integration/tests/cluster/partition_dedup.rs | 548 +-
.../tests/cluster/partition_primary_routing.rs | 121 +-
.../tests/cluster/partition_state_transfer.rs | 54 +-
.../tests/connectors/runtime/http_state.rs | 2 +-
.../verify_consumer_group_partition_assignment.rs | 136 +-
core/integration/tests/sdk/producer/mod.rs | 52 +
core/integration/tests/sdk/protocol_version.rs | 25 +
core/integration/tests/server/cg.rs | 10 +-
.../tests/server/consumer_offset_quota_vsr.rs | 4 +
.../tests/server/descriptor_exhaustion_vsr.rs | 71 +-
core/integration/tests/server/http_rbac.rs | 91 +
core/integration/tests/server/http_vsr.rs | 176 +-
core/integration/tests/server/raw_tcp.rs | 4 +
.../server/scenarios/authentication_scenario.rs | 38 +-
...umer_group_auto_commit_reconnection_scenario.rs | 2 +
...er_group_new_messages_after_restart_scenario.rs | 1 +
.../scenarios/external_group_offsets_scenario.rs | 142 +
core/integration/tests/server/scenarios/mod.rs | 1 +
.../tests/server/scenarios/permissions_scenario.rs | 11 +-
.../scenarios/reconnect_after_restart_scenario.rs | 29 +-
.../stale_client_consumer_group_scenario.rs | 4 +-
core/journal/src/durable_storage.rs | 57 +-
core/journal/src/file_storage.rs | 37 +-
core/journal/src/local_gate.rs | 75 +-
core/journal/src/partition_journal.rs | 90 +
core/journal/src/partition_journal/segments.rs | 127 +-
core/journal/src/prepare_journal.rs | 248 +-
core/journal/src/superblock.rs | 49 +-
core/message_bus/src/connection_cap.rs | 2 +-
core/message_bus/src/replica/auth.rs | 25 +-
core/message_bus/src/replica/handshake.rs | 230 +-
core/message_bus/tests/common/mod.rs | 2 +
core/message_bus/tests/head_of_line.rs | 47 +-
core/message_bus/tests/replica_auth_handshake.rs | 68 +
core/metadata/Cargo.toml | 4 +-
core/metadata/src/applied_frontier.rs | 37 +-
core/metadata/src/impls/metadata.rs | 2519 +-
core/metadata/src/impls/recovery.rs | 491 +-
core/metadata/src/stm/authz.rs | 2 +
core/metadata/src/stm/mux.rs | 2 +-
core/metadata/src/stm/result.rs | 15 +-
core/metadata/src/stm/snapshot.rs | 105 +-
core/metadata/src/stm/stream.rs | 104 +-
core/metadata/src/stm/user.rs | 33 +-
core/partitions/src/consumer_offset_capacity.rs | 31 +-
core/partitions/src/iggy_index_writer.rs | 31 +-
core/partitions/src/iggy_partition.rs | 25259 +++++++++++++------
core/partitions/src/iggy_partitions.rs | 279 +-
core/partitions/src/io.rs | 1634 ++
core/partitions/src/journal.rs | 355 +-
core/partitions/src/lib.rs | 19 +-
core/partitions/src/messages_writer.rs | 23 +-
core/partitions/src/offset_recovery.rs | 23 +-
core/partitions/src/offset_storage.rs | 279 +-
core/partitions/src/partition_storage.rs | 288 +-
core/partitions/src/persistence.rs | 323 +-
core/partitions/src/segment_recovery.rs | 376 +-
core/partitions/src/state_transfer.rs | 3305 ++-
core/partitions/src/types.rs | 20 +-
core/partitions/tests/consumer_offset_history.rs | 236 +-
core/sdk/Cargo.toml | 2 +
core/sdk/README.md | 13 +-
core/sdk/src/clients/client.rs | 9 +-
core/sdk/src/clients/producer.rs | 93 +-
core/sdk/src/clients/producer_builder.rs | 25 +-
core/sdk/src/clients/producer_dispatcher.rs | 1 +
core/sdk/src/http/consumer_offsets.rs | 72 +-
core/sdk/src/http/messages.rs | 1 +
core/sdk/src/lib.rs | 3 +-
core/sdk/src/poll_routing.rs | 166 +-
core/sdk/src/quic/quic_client.rs | 647 +-
core/sdk/src/session.rs | 248 +-
.../stream_builder/build/build_iggy_consumer.rs | 3 +
.../stream_builder/build/build_iggy_producer.rs | 1 +
.../stream_builder/config/config_iggy_consumer.rs | 41 +-
.../stream_builder/config/config_iggy_producer.rs | 20 +-
core/sdk/src/stream_builder/iggy_stream.rs | 35 +-
.../sdk/src/stream_builder/iggy_stream_consumer.rs | 28 +-
core/sdk/src/tcp/tcp_client.rs | 747 +-
core/sdk/src/vsr.rs | 240 +-
core/sdk/src/websocket/websocket_client.rs | 986 +-
core/server/Cargo.toml | 1 +
core/server/Dockerfile | 2 +-
core/server/README.md | 121 +-
core/server/config.toml | 62 +-
core/server/src/boot/listeners.rs | 2 +
core/server/src/boot/mod.rs | 173 +-
core/server/src/boot/recovery.rs | 1 +
core/server/src/boot/threads.rs | 16 +-
core/server/src/consumer_group.rs | 39 +-
core/server/src/consumer_group/lease.rs | 207 +-
core/server/src/consumer_group/liveness.rs | 969 +-
core/server/src/dispatch/authz.rs | 10 +-
core/server/src/dispatch/host.rs | 41 +-
core/server/src/dispatch/login_error.rs | 10 +-
core/server/src/dispatch/mod.rs | 391 +-
core/server/src/dispatch/partition.rs | 803 +-
core/server/src/dispatch/reads.rs | 82 +-
core/server/src/dispatch/session_ops.rs | 313 +-
core/server/src/dispatch/submit.rs | 29 +-
core/server/src/dispatch/test_support.rs | 2 +
core/server/src/http.rs | 2 +
core/server/src/http/admission.rs | 11 +-
core/server/src/http/error.rs | 23 +
core/server/src/http/extractor.rs | 1 +
core/server/src/http/handlers.rs | 118 +-
core/server/src/http/session.rs | 246 +-
core/server/src/http/state.rs | 155 +-
core/server/src/http/submit.rs | 186 +-
core/server/src/http/wire.rs | 2 +-
core/server/src/namespace.rs | 18 +-
core/server/src/partition_helpers.rs | 785 +-
core/server/src/partition_reconciler.rs | 203 +-
core/server/src/responses.rs | 7 +-
core/server/src/server_error.rs | 30 +
core/server/src/session_manager.rs | 659 +-
core/server/src/sysinfo_printer.rs | 21 +-
core/server/src/sysinfo_probe.rs | 159 +-
core/server/src/wire.rs | 11 +-
core/server_common/src/consensus_message.rs | 22 +-
core/server_common/src/fs_utils.rs | 126 +-
core/server_common/src/iobuf.rs | 37 +-
core/server_common/src/poll.rs | 4 +-
core/server_common/src/segment_storage/mod.rs | 2 +-
core/server_common/src/sharding/mod.rs | 3 +
core/shard/Cargo.toml | 3 +
core/shard/src/builder.rs | 12 +
core/shard/src/lib.rs | 873 +-
core/shard/src/metrics.rs | 159 +-
core/shard/src/partition_io.rs | 2925 +++
core/shard/src/poll.rs | 74 +-
core/shard/src/poll/completion.rs | 2 +-
core/shard/src/poll/completion_tests.rs | 468 +-
core/shard/src/poll/test_support.rs | 30 +-
core/shard/src/poll/timeout_tests.rs | 17 +-
core/shard/src/router.rs | 257 +-
core/simulator/README.md | 5 +-
core/simulator/src/bin/workload-fuzz.rs | 13 -
core/simulator/src/client.rs | 57 +-
.../simulator/src/consumer_offset_history_tests.rs | 29 +-
core/simulator/src/executor/time.rs | 4 +-
core/simulator/src/lib.rs | 341 +-
core/simulator/src/packet.rs | 3 +-
core/simulator/src/replica.rs | 9 +-
core/simulator/src/storage/purge.rs | 356 +-
core/simulator/src/storage/tests.rs | 261 +-
core/simulator/src/workload/invariants.rs | 2 +-
core/simulator/src/workload/ops/mod.rs | 25 +-
.../src/workload/ops/store_consumer_offset.rs | 2 +-
core/simulator/src/workload/state_checker.rs | 1 +
examples/java/build.gradle.kts | 2 +-
.../java/gradle/wrapper/gradle-wrapper.properties | 2 +-
examples/java/gradlew | 4 +-
.../iggy/examples/MultiTenantAccessTest.java | 4 +-
examples/rust/src/new-sdk/consumer/main.rs | 1 +
foreign/cpp/BUILD.bazel | 50 +-
foreign/cpp/MODULE.bazel | 19 +-
foreign/cpp/MODULE.bazel.lock | 2 +-
.../Fixtures/IggyServerFixture.cs | 4 +-
.../Iggy_SDK.Tests.Integration/HeartbeatTests.cs | 5 +-
.../TcpMessageStream.PollRouting.cs | 32 +-
.../Implementations/TcpMessageStream.Vsr.cs | 64 +-
.../IggyClient/Implementations/TcpMessageStream.cs | 18 +-
foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs | 2 +-
foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs | 50 +-
foreign/csharp/Iggy_SDK/Vsr/LoginRegister.cs | 52 +-
foreign/csharp/Iggy_SDK/Vsr/VsrConnection.cs | 22 +-
foreign/csharp/Iggy_SDK/Vsr/VsrError.cs | 3 +
foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs | 20 +-
.../VsrTests/ConsensusSessionTests.cs | 68 +-
.../VsrTests/EndpointFailoverTests.cs | 257 +-
.../Iggy_SDK_Tests/VsrTests/LoginRegisterTests.cs | 38 +-
foreign/csharp/Iggy_SDK_Tests/VsrTests/MockNode.cs | 7 +-
.../Iggy_SDK_Tests/VsrTests/PrimaryPollingTests.cs | 78 +-
.../Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs | 4 +-
.../VsrTests/VsrProtocolDriftTests.cs | 33 +-
foreign/csharp/README.md | 41 +-
foreign/go/README.md | 15 +-
foreign/go/client/tcp/tcp_connect_test.go | 47 +-
foreign/go/client/tcp/tcp_core.go | 30 +-
foreign/go/client/tcp/tcp_core_test.go | 34 +-
foreign/go/client/tcp/tcp_failover_test.go | 99 +-
foreign/go/client/tcp/tcp_poll_routing.go | 33 +-
foreign/go/client/tcp/tcp_poll_routing_test.go | 110 +-
foreign/go/client/tcp/tcp_session_management.go | 77 +-
foreign/go/client/tcp/tcp_testing_test.go | 5 +
foreign/go/errors/errors.yaml | 15 +
foreign/go/errors/errors_gen.go | 52 +
foreign/go/errors/errors_test.go | 6 +
foreign/go/internal/command/code.go | 2 +-
foreign/go/internal/vsr/operation.go | 8 +-
foreign/go/internal/vsr/protocol_parity_test.go | 19 +-
foreign/go/internal/vsr/register.go | 42 +-
foreign/go/internal/vsr/register_test.go | 40 +-
foreign/go/internal/vsr/session.go | 13 +-
foreign/java/README.md | 14 +
foreign/java/gradle/libs.versions.toml | 6 +-
.../java/gradle/wrapper/gradle-wrapper.properties | 2 +-
foreign/java/gradlew | 4 +-
.../iggy/client/async/tcp/AsyncTcpConnection.java | 111 +-
.../apache/iggy/client/async/tcp/PollRouter.java | 51 +-
.../client/async/tcp/vsr/ConsensusSession.java | 79 +-
.../iggy/client/async/tcp/vsr/VsrLoginCodec.java | 43 +-
.../iggy/client/async/tcp/vsr/VsrOperation.java | 4 +
.../client/async/tcp/vsr/VsrRequestEncoder.java | 25 +-
.../client/async/tcp/vsr/VsrResponseHandler.java | 2 +-
.../org/apache/iggy/exception/IggyErrorCode.java | 3 +
.../java/org/apache/iggy/serde/CommandCode.java | 2 +-
.../test/java/org/apache/iggy/TestConstants.java | 9 +
.../client/async/AsyncClientIntegrationTest.java | 25 +-
.../client/async/AsyncConnectionPoolAuthTest.java | 7 +-
.../AsyncIggyTcpClientEndpointFailoverTest.java | 2 +-
.../tcp/AsyncIggyTcpClientEventLoopGroupTest.java | 2 +-
.../AsyncIggyTcpClientTransientFailoverTest.java | 194 +-
.../tcp/AsyncTcpConnectionConcurrencyTest.java | 2 +-
.../client/async/tcp/ClientRoutingStateTest.java | 5 +-
.../client/async/tcp/vsr/VsrLoginCodecTest.java | 58 +-
.../async/tcp/vsr/VsrRequestEncoderTest.java | 25 +-
.../async/tcp/vsr/VsrResponseHandlerTest.java | 7 +-
.../iggy/client/blocking/IntegrationTest.java | 4 +-
.../client/blocking/tcp/MessagesTcpClientTest.java | 9 +-
.../apache/iggy/exception/IggyErrorCodeTest.java | 2 +
foreign/node/README.md | 19 +-
foreign/node/scripts/check-vsr-protocol.mjs | 30 +-
foreign/node/src/client/client.socket.test.ts | 241 +-
foreign/node/src/client/client.socket.ts | 78 +-
foreign/node/src/e2e/tcp.system.e2e.ts | 7 +
foreign/node/src/wire/command-set.ts | 2 +
foreign/node/src/wire/command.code.ts | 4 +-
foreign/node/src/wire/error.code.test.ts | 4 +
foreign/node/src/wire/error.code.ts | 3 +
foreign/node/src/wire/system/index.ts | 1 +
.../node/src/wire/system/snapshot.command.test.ts | 107 +
foreign/node/src/wire/system/snapshot.command.ts | 103 +
foreign/node/src/wire/vsr/index.ts | 33 +-
foreign/node/src/wire/vsr/operation.test.ts | 7 +-
foreign/node/src/wire/vsr/operation.ts | 11 +-
foreign/node/src/wire/vsr/register.test.ts | 41 +-
foreign/node/src/wire/vsr/register.ts | 40 +-
foreign/node/src/wire/vsr/session.test.ts | 15 +-
foreign/node/src/wire/vsr/session.ts | 17 +-
foreign/node/src/wire/vsr/vsr.test.ts | 10 +-
foreign/php/Dockerfile.test | 2 +-
foreign/python/README.md | 30 +-
foreign/python/apache_iggy.pyi | 18 +-
foreign/python/src/client.rs | 11 +-
foreign/python/src/send_message.rs | 13 +-
foreign/python/tests/test_producer.py | 76 +-
.../swift/Sources/Iggy/Errors/IggyErrorCode.swift | 6 +
foreign/swift/Tests/IggyTests/Fixtures/golden.json | 14 +-
foreign/swift/Tools/golden-vectors/src/main.rs | 16 +
gateways/kafka/Cargo.toml | 3 +
gateways/kafka/Dockerfile | 2 +-
gateways/kafka/README.md | 52 +-
gateways/kafka/docs/ACL_MAPPING.md | 5 +-
gateways/kafka/docs/CONSUMER_GROUPS.md | 24 +-
gateways/kafka/docs/MANUAL_TESTING.md | 18 +-
gateways/kafka/docs/OFFSET_STORAGE.md | 297 +-
gateways/kafka/docs/SCOPE.md | 21 +-
gateways/kafka/docs/TEST_SUITE.md | 3 +-
gateways/kafka/docs/kafka_api_keys_reference.md | 14 +-
gateways/kafka/scripts/ci-wire-fixtures.sh | 2 +-
gateways/kafka/src/bridge/error.rs | 32 +-
gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 33 +-
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 686 +
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 29 +-
gateways/kafka/src/bridge/iggy_bridge/produce.rs | 9 +-
gateways/kafka/src/bridge/iggy_bridge/topics.rs | 25 +-
gateways/kafka/src/bridge/mod.rs | 6 +-
gateways/kafka/src/group/mod.rs | 68 +-
gateways/kafka/src/group/state.rs | 298 +-
gateways/kafka/src/protocol/acl.rs | 4 +-
gateways/kafka/src/protocol/api.rs | 37 +-
gateways/kafka/src/protocol/bounds_guard.rs | 262 +-
gateways/kafka/src/protocol/handlers/heartbeat.rs | 13 +-
gateways/kafka/src/protocol/handlers/metadata.rs | 2 +-
gateways/kafka/src/protocol/handlers/mod.rs | 33 +-
.../kafka/src/protocol/handlers/offset_commit.rs | 321 +
.../kafka/src/protocol/handlers/offset_fetch.rs | 564 +
.../kafka/tests/bridge_iggy_integration_tests.rs | 2 +-
gateways/kafka/tests/common/scope.rs | 2 +
gateways/kafka/tests/common/wire.rs | 103 +-
gateways/kafka/tests/consumer_group_tests.rs | 10 +-
gateways/kafka/tests/fetch_real_bridge_tests.rs | 8 +-
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 4 +
gateways/kafka/tests/kafka_client_e2e_tests.rs | 450 +-
gateways/kafka/tests/metadata_real_bridge_tests.rs | 2 +-
gateways/kafka/tests/offsets_real_bridge_tests.rs | 364 +
gateways/kafka/tests/produce_real_bridge_tests.rs | 2 +-
gateways/kafka/tests/server_e2e_tests.rs | 8 +-
gateways/kafka/tests/version_firewall_tests.rs | 7 +-
gateways/kafka/tools/kafka-tool/src/main.rs | 15 +-
justfile | 6 +-
rust-toolchain.toml | 2 +-
scripts/ci/coverage-baseline-affected.sh | 6 +-
scripts/ci/cpp-coverage.sh | 84 +
scripts/ci/edge-affected-images.sh | 150 -
scripts/ci/edge-image-catalog.sh | 88 +
scripts/ci/sync-rustc-version.sh | 21 +
web/package-lock.json | 18 +-
389 files changed, 49157 insertions(+), 19452 deletions(-)
diff --cc gateways/kafka/README.md
index b473f9ef3,a1cd72c8b..bc5229f0a
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@@ -280,11 -303,12 +304,13 @@@ Full reasoning, including what was reje
`src/bridge/` is the SDK integration layer: connects to Iggy, maps Kafka topics to Iggy
streams/topics, provisions them on demand, looks up high watermarks (one or many partitions of
- a topic per call) for `ListOffsets`, and probes and polls partitions for Fetch.
+ a topic per call) for `ListOffsets`, probes and polls partitions for Fetch, and stores and reads
+ group offsets for OffsetCommit and OffsetFetch.
Produce ([#3535](https://github.com/apache/iggy/issues/3535)), Fetch ([#3536](https://github.com/apache/iggy/issues/3536)), ListOffsets
([#3537](https://github.com/apache/iggy/issues/3537)), Metadata ([#3534](https://github.com/apache/iggy/issues/3534)), CreateTopics
- ([#3538](https://github.com/apache/iggy/issues/3538)), DescribeConfigs and AlterConfigs call it.
-([#3538](https://github.com/apache/iggy/issues/3538)), OffsetCommit and OffsetFetch
++([#3538](https://github.com/apache/iggy/issues/3538)), DescribeConfigs and AlterConfigs , OffsetCommit and OffsetFetchcall it.
+Topic configuration behavior is in [docs/CONFIGS.md](docs/CONFIGS.md).
+ ([#3542](https://github.com/apache/iggy/issues/3542)) call it.
Tested by `bridge`'s own unit tests, `tests/bridge_iggy_integration_tests.rs` and the
`tests/*_real_bridge_tests.rs` suites. All but the unit tests start a real `iggy-server`.
@@@ -475,9 -499,11 +501,11 @@@ cut before that threshold is reached -
### Concurrency ceiling
-- One `IggyClient` serves Produce, Metadata and CreateTopics for every Kafka connection, one Iggy
- request at a time. Fetch polls use 4 more, one per read slot. Topic probes use 1 more. Offset
- calls use 8 more: a commit client and a read client per offset slot.
+- One `IggyClient` serves Produce, Metadata, CreateTopics, DescribeConfigs and AlterConfigs for
+ every Kafka connection, one Iggy request at a time. Fetch polls use 4 more, one per read slot.
- Topic probes use 1 more.
++ Topic probes use 1 more. Offset calls use 8 more: a commit client and a read client per offset slot.
+ - OffsetCommit and OffsetFetch cost one Iggy call per partition, spread over the offset slots. In a
+ cluster, an offset commit client can also open one connection per node for stores.
- `IGGY_KAFKA_MAX_CONNECTIONS` does not change that. A Produce client pool is a TODO in
[docs/SCOPE.md](docs/SCOPE.md).
- Order: set `max.in.flight.requests.per.connection=1`, or `retries=0`. Otherwise a retried batch
diff --cc gateways/kafka/docs/MANUAL_TESTING.md
index e1b4aa356,301fc3a9c..c323c0d34
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@@ -229,8 -229,8 +231,8 @@@ For each API key, test **min−1**, **m
| ---- | ------ | ---------------------- | --------------------------- |
| B1 | ApiVersions negotiation | `error_code=0`; body lists 14 API keys with correct min/max | KIP-511 exception: still answers, `error_code=35` (UNSUPPORTED_VERSION), v0 response header regardless of the request's own encoding |
| B2 | Metadata out-of-range | N/A | **Connection closes**, no response sent - Metadata has no top-level error field to carry a version-correct error in |
- | B3 | Produce/Fetch/ListOffsets/CreateTopics/InitProducerId out-of-range | N/A | **Connection closes** for both above-max and below-min - `kafka_protocol`'s schema floor for each of these five messages equals `SUPPORTED_RANGES`' own min, so there is no encodable error response below min either (see `SCOPE.md`'s Governance model) |
- | B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,10,11,12,13,14,18,19,22,32,33 only — no OffsetCommit/OffsetFetch, and no transaction keys (24, 25, 26, 28) |
-| B3 | Produce/Fetch/ListOffsets/CreateTopics/InitProducerId/OffsetCommit/OffsetFetch out-of-range | N/A | **Connection closes** for both above-max and below-min - `kafka_protocol`'s schema floor for each of these seven messages equals `SUPPORTED_RANGES`' own min, so there is no encodable error response below min either (see `SCOPE.md`'s Governance model) |
-| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,8,9,10,11,12,13,14,18,19,22 only, and no transaction keys (24, 25, 26, 28) |
++| B3 | Produce/Fetch/ListOffsets/CreateTopics/InitProducerId/OffsetCommit/OffsetFetch out-of-range | N/A | **Connection closes** for both above-max and below-min - `kafka_protocol`'s schema floor for each of these five messages equals `SUPPORTED_RANGES`' own min, so there is no encodable error response below min either (see `SCOPE.md`'s Governance model) |
++| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,8,9,10,11,12,13,14,18,19,22,32,33 only, and no transaction keys (24, 25, 26, 28) |
An out-of-range version only ever produces `error_code=35` on ApiVersions (B1); every other API
key's out-of-range case closes the connection - see B2/B3. InitProducerId and Produce also send
diff --cc gateways/kafka/docs/SCOPE.md
index 16c96352b,60b3a6b67..a5353f123
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@@ -2,9 -2,14 +2,14 @@@
## Issue #3421 — in scope (this iteration)
-A TCP listener on the Kafka wire port. It decodes requests, validates scoped API keys, versions and wire formats, and answers them. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics, OffsetCommit and OffsetFetch use Iggy.
+A TCP listener on the Kafka wire port. It decodes requests, validates scoped API keys, versions and wire formats, and answers them. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics, DescribeConfigs and AlterConfigs use Iggy.
- **Stub semantics (important):** without a bridge, every API answers with a stub. Produce discards the payload and answers with retriable `NOT_LEADER_OR_FOLLOWER` (6). Fetch and ListOffsets answer 6 too. CreateTopics, DescribeConfigs and AlterConfigs answer `NOT_CONTROLLER` (41), so clients do not believe topics were created or configs were read or changed. Do not read `ec=0` from a stub as durable storage. Produce is the one API that stores records once you configure a bridge ([#3535](h [...]
-**Stub semantics (important):** without a bridge, every API answers with a stub. Produce discards
-the payload and answers with retriable `NOT_LEADER_OR_FOLLOWER` (6). Fetch and ListOffsets answer 6
-too. CreateTopics validates the request but answers `NOT_CONTROLLER` (41), so clients do not believe
-topics were created. OffsetCommit and OffsetFetch answer retriable `COORDINATOR_LOAD_IN_PROGRESS`
-(14). Do not read `ec=0` from a stub as durable storage. Produce is the one API that stores records
-once you configure a bridge ([#3535](https://github.com/apache/iggy/issues/3535)).
++**Stub semantics (important):** without a bridge, every API answers with a stub. Produce discards
++the payload and answers with retriable `NOT_LEADER_OR_FOLLOWER` (6). Fetch and ListOffsets answer 6
++too. CreateTopics, DescribeConfigs and AlterConfigs answer `NOT_CONTROLLER` (41), so clients do not believe
++topics were created or configs were read or changed. OffsetCommit and OffsetFetch answer retriable `COORDINATOR_LOAD_IN_PROGRESS`
++(14). Do not read `ec=0` from a stub as durable storage. Produce is the one API that stores records once you configure
++a bridge ([#3535](https://github.com/apache/iggy/issues/3535)).
| Deliverable | Status | Location |
| ------------- | -------- | ---------- |
@@@ -29,7 -34,7 +34,7 @@@ Expand `SUPPORTED_RANGES` only after a
**Every unsupported-version case closes the connection, for every listed key** - not just above
the encoder max. `kafka_protocol`'s schema floor for each supported message happens to equal
`SUPPORTED_RANGES`' own min today (Produce 3, Fetch 4, ListOffsets 1, Metadata 0, ApiVersions 0,
- CreateTopics 2, InitProducerId 0, DescribeConfigs 1, AlterConfigs 0, and 0 for the four consumer-group keys), so there is no version below an API's
-CreateTopics 2, InitProducerId 0, OffsetCommit 2, OffsetFetch 1, and 0 for the five group coordination keys), so there is no version below an API's
++CreateTopics 2, InitProducerId 0, OffsetCommit 2, OffsetFetch 1,DescribeConfigs 1, AlterConfigs 0, and 0 for the five coordination keys), so there is no version below an API's
min that the crate can actually encode a response for either - `unsupported_version_response`
still tries, but the encode attempt fails and the connection closes rather than sending a
malformed body.
diff --cc gateways/kafka/scripts/ci-wire-fixtures.sh
index 42e35a4c5,93b24fe93..b6ca97836
--- a/gateways/kafka/scripts/ci-wire-fixtures.sh
+++ b/gateways/kafka/scripts/ci-wire-fixtures.sh
@@@ -24,7 -24,7 +24,7 @@@ set -euo pipefai
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 10 11 12 13 14 19 22 32 33)
-FIXTURE_API_KEYS=(0 1 2 8 9 10 11 12 13 14 19 22)
++FIXTURE_API_KEYS=(0 1 2 8 9 10 11 12 13 14 19 22 32 33)
usage() {
echo "Usage: $0 {generate|cleanup}" >&2
diff --cc gateways/kafka/src/bridge/iggy_bridge/mod.rs
index b1c7698af,cb09163a8..13003776a
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@@ -37,7 -38,9 +38,9 @@@ mod topics
use fetch::{FetchPool, LazyClient};
pub(crate) use fetch::{FetchSlot, PartitionProbe, TopicProbe};
+pub use topics::{KafkaTopicMetadata, StreamTopicCache, TopicCreationOutcome, TopicLoad};
+ use group_offsets::OffsetPool;
+ pub use group_offsets::{OFFSET_GROUP_PREFIX, OffsetCalls};
-pub use topics::{KafkaTopicMetadata, TopicCreationOutcome};
/// Passes attempted, after the first, before [`IggyBridge::connect`] gives up and returns `Err`.
///
diff --cc gateways/kafka/src/bridge/iggy_bridge/topics.rs
index 948b3512c,2e74f1b2f..8c3d5dac4
--- a/gateways/kafka/src/bridge/iggy_bridge/topics.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
@@@ -20,12 -20,10 +20,12 @@@
use std::collections::{HashMap, HashSet};
use iggy::prelude::{
- Identifier, IggyError, IggyExpiry, StreamClient, Topic, TopicClient, TopicCreateOptions,
- Identifier, IggyClient, IggyError, StreamClient, TopicClient, TopicCreateOptions, TopicDetails,
++ Identifier,IggyClient, IggyError, IggyExpiry, StreamClient, Topic, TopicClient, TopicCreateOptions,
+ TopicDetails, TopicUpdateOptions,
};
use kafka_protocol::protocol::StrBytes;
-use tracing::{debug, info};
+use tokio::time::{Instant, timeout_at};
+use tracing::{debug, info, warn};
use super::{IggyBridge, with_request_timeout};
use crate::bridge::error::BridgeError;
@@@ -283,47 -289,9 +292,47 @@@ impl IggyBridge
// No separate get_stream probe: get_topic already answers Ok(None) when the stream
// itself is missing (see high_watermarks' own doc on this same fact), so a probe first
// would just pay a second round trip to learn something this one call already tells us.
- with_request_timeout(self.client.get_topic(&stream_id, &topic_id)).await
+ with_request_timeout(client.get_topic(&stream_id, &topic_id)).await
}
+ /// Sets `message_expiry` on the Iggy topic backing `kafka_topic`.
+ ///
+ /// Does not create the topic and does not rename it: `name` is the mapped Iggy
+ /// topic name, which `update_topic` also takes as the topic's new name.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`BridgeError::InvalidKafkaTopicName`] if `kafka_topic` fails Kafka's own
+ /// topic-naming rules. Returns [`BridgeError::Timeout`] if a call takes longer than
+ /// `REQUEST_TIMEOUT`. Returns [`BridgeError::Iggy`] for connectivity, auth, or a
+ /// missing topic.
+ pub async fn update_kafka_topic_message_expiry(
+ &self,
+ kafka_topic: &str,
+ message_expiry: IggyExpiry,
+ ) -> Result<(), BridgeError> {
+ validate_kafka_topic_name("kafka_topic", kafka_topic)?;
+ let (stream_name, topic_name) = self.config.topic_mapping.resolve(kafka_topic);
+ let stream_id = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
+ let topic_id = Identifier::named(topic_name).map_err(BridgeError::Iggy)?;
+ let options = TopicUpdateOptions {
+ message_expiry: Some(message_expiry),
+ ..TopicUpdateOptions::default()
+ };
+ with_request_timeout(
+ self.client
+ .update_topic(&stream_id, &topic_id, topic_name, &options),
+ )
+ .await
+ }
+
+ async fn topics_in_stream(&self, stream_name: &str) -> Result<Vec<Topic>, BridgeError> {
+ let stream_id = Identifier::named(stream_name).map_err(BridgeError::Iggy)?;
+ // A missing stream lists as empty, the same fact `get_kafka_topics` relies on, so a
+ // name that resolves here and is absent from the list is "no such topic".
+ with_request_timeout(self.client.get_topics(&stream_id)).await
+ }
+
/// Resolves many Kafka-visible names at once, one entry per input in the same order.
///
/// Batches by the Iggy stream each name resolves to, rather than paying one round trip per
diff --cc gateways/kafka/src/bridge/mod.rs
index f6b4f4520,64e9f5fd4..8be006ae6
--- a/gateways/kafka/src/bridge/mod.rs
+++ b/gateways/kafka/src/bridge/mod.rs
@@@ -17,9 -17,10 +17,9 @@@
//! Iggy SDK integration layer.
//!
-//! Maps Kafka topics to Iggy streams/topics, exposes create-if-missing provisioning,
-//! high-watermark lookups and group offsets, and translates Iggy errors to Kafka wire error codes.
-//! `ListOffsets`, Produce, Fetch, Metadata, `CreateTopics`, `OffsetCommit` and `OffsetFetch` call
-//! it.
+//! Maps Kafka topics to Iggy streams/topics, exposes create-if-missing provisioning and
- //! high-watermark lookups, and translates Iggy errors to Kafka wire error codes. `ListOffsets`,
- //! Produce, Fetch, Metadata, `CreateTopics`, `DescribeConfigs` and `AlterConfigs` call it.
++//! high-watermark lookups, group and translates Iggy errors to Kafka wire error codes. `ListOffsets`,
++//! Produce, Fetch, Metadata, `CreateTopics`, `OffsetCommit`, `OffsetFetch`, `DescribeConfigs` and `AlterConfigs` call it.
pub mod config;
pub mod error;
@@@ -29,6 -30,7 +29,6 @@@ pub mod topic_map
pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig};
pub use error::BridgeError;
pub use iggy_bridge::{
- IggyBridge, KafkaTopicMetadata, StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget,
- IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, TopicCreationOutcome,
- TopicTarget,
++ IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget,
};
pub use topic_map::{TopicMapping, TopicOverride};
diff --cc gateways/kafka/src/protocol/api.rs
index 5af935308,2e8358134..87e335b99
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@@ -26,10 -26,9 +26,10 @@@ use kafka_protocol::messages::
use tokio::sync::Semaphore;
use tokio_util::sync::CancellationToken;
+use crate::auth::AuthenticatedPrincipal;
use crate::bridge::IggyBridge;
use crate::error::Result;
- use crate::group::{GroupCoordinator, GroupCoordinatorConfig};
+ use crate::group::{GroupCoordinator, GroupCoordinatorConfig, GroupMember};
use crate::protocol::acl::{
self, AclBinding, AclFilter, PrincipalPermissions, encode_describe_acls_error_response,
encode_describe_acls_response,
@@@ -39,8 -38,8 +39,8 @@@ use crate::protocol::bounds_guard::
};
use crate::protocol::handlers::init_producer_id::ProducerIdAllocator;
use crate::protocol::handlers::{
- api_versions, create_topics, decode_guarded, dispatch, fetch, find_coordinator, heartbeat,
- init_producer_id, join_group, leave_group, list_offsets, metadata, offset_commit, offset_fetch,
+ alter_configs, api_versions, create_topics, decode_guarded, describe_configs, dispatch, fetch,
- find_coordinator, heartbeat, init_producer_id, join_group, leave_group, list_offsets, metadata,
++ find_coordinator, heartbeat, init_producer_id, join_group, leave_group, list_offsets, metadata, offset_commit, offset_fetch,
produce, respond_or_close, sync_group,
};
use crate::protocol::probe_board::ProbeBoard;
@@@ -277,10 -274,10 +281,12 @@@ static SUPPORTED_RANGES: &[ApiVersionRa
fetch::RANGE,
list_offsets::RANGE,
metadata::RANGE,
+ offset_commit::RANGE,
+ offset_fetch::RANGE,
api_versions::RANGE,
create_topics::RANGE,
+ describe_configs::RANGE,
+ alter_configs::RANGE,
init_producer_id::RANGE,
find_coordinator::RANGE,
join_group::RANGE,
diff --cc gateways/kafka/src/protocol/bounds_guard.rs
index d348c37e7,9f1faa3bd..811500a30
--- a/gateways/kafka/src/protocol/bounds_guard.rs
+++ b/gateways/kafka/src/protocol/bounds_guard.rs
@@@ -1728,36 -1774,115 +1878,146 @@@ mod tests
assert!(validate_leave_group_shape(4, &body, 1_024).is_err());
}
+ /// `validate_describe_configs_shape` and `validate_alter_configs_shape` had no POC of their
+ /// own before this - every sibling guard above pins a rejection for both the legacy
+ /// (`i32`) and flexible (compact varint) count forms, and these two never did.
+ #[test]
+ fn describe_configs_v1_huge_legacy_count_rejected() {
+ let body = Bytes::from_static(&[0x7F, 0xFF, 0xFF, 0xFF]);
+ assert!(validate_describe_configs_shape(1, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ /// The flexible (v4+) sibling of the legacy test above: a huge compact-array-count varint
+ /// for `resources` on its own, exceeding `MAX_COLLECTION_LEN` regardless of what bytes
+ /// remain in the frame.
+ #[test]
+ fn describe_configs_v4_huge_compact_count_rejected() {
+ let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF, 0xFF, 0x0F]); // u32::MAX, 5-byte varint
+ assert!(validate_describe_configs_shape(4, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ #[test]
+ fn alter_configs_v0_huge_legacy_count_rejected() {
+ let body = Bytes::from_static(&[0x7F, 0xFF, 0xFF, 0xFF]);
+ assert!(validate_alter_configs_shape(0, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ /// The flexible (v2+) sibling of the legacy test above: a huge compact-array-count varint
+ /// for `resources` on its own, exceeding `MAX_COLLECTION_LEN` regardless of what bytes
+ /// remain in the frame.
+ #[test]
+ fn alter_configs_v2_huge_compact_count_rejected() {
+ let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF, 0xFF, 0x0F]); // u32::MAX, 5-byte varint
+ assert!(validate_alter_configs_shape(2, &body, TEST_MAX_FRAME_SIZE).is_err());
+ #[test]
+ fn offset_commit_v2_null_metadata_accepted() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x01, b'g', // group_id
+ 0xFF, 0xFF, 0xFF, 0xFF, // generation_id
+ 0x00, 0x00, // member_id
+ 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, // retention_time_ms
+ 0x00, 0x00, 0x00, 0x01, // topics: 1
+ 0x00, 0x01, b't', // name
+ 0x00, 0x00, 0x00, 0x01, // partitions: 1
+ 0x00, 0x00, 0x00, 0x00, // partition_index
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, // committed_offset
+ 0xFF, 0xFF, // committed_metadata null
+ ]);
+ assert!(validate_offset_commit_shape(2, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetCommitRequest>(2, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_commit_v8_flexible_body_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, b'g', // group_id
+ 0x00, 0x00, 0x00, 0x01, // generation_id
+ 0x02, b'm', // member_id
+ 0x00, // group_instance_id null
+ 0x02, // topics: 1
+ 0x02, b't', // name
+ 0x02, // partitions: 1
+ 0x00, 0x00, 0x00, 0x00, // partition_index
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, // committed_offset
+ 0xFF, 0xFF, 0xFF, 0xFF, // committed_leader_epoch
+ 0x01, // committed_metadata ""
+ 0x00, // partition tagged fields
+ 0x00, // topic tagged fields
+ 0x00, // request tagged fields
+ ]);
+ assert!(validate_offset_commit_shape(8, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetCommitRequest>(8, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_commit_v9_huge_partitions_count_rejected() {
+ let body = Bytes::from_static(&[
+ 0x02, b'g', // group_id
+ 0x00, 0x00, 0x00, 0x01, // generation_id
+ 0x02, b'm', // member_id
+ 0x00, // group_instance_id null
+ 0x02, // topics: 1
+ 0x02, b't', // name
+ 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // partitions count
+ ]);
+ assert!(validate_offset_commit_shape(9, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ #[test]
+ fn offset_fetch_v1_partition_indexes_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, // partition_indexes: 2
+ 0x00, 0x00, 0x00, 0x00, // 0
+ 0x00, 0x00, 0x00, 0x01, // 1
+ ]);
+ assert!(validate_offset_fetch_shape(1, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetFetchRequest>(1, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_fetch_v7_null_topics_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, b'g', // group_id
+ 0x00, // topics null: every committed offset
+ 0x01, // require_stable
+ 0x00, // tagged fields
+ ]);
+ assert!(validate_offset_fetch_shape(7, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetFetchRequest>(7, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_fetch_v9_group_with_member_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, // groups: 1
+ 0x02, b'g', // group_id
+ 0x00, // member_id null
+ 0xFF, 0xFF, 0xFF, 0xFF, // member_epoch
+ 0x02, // topics: 1
+ 0x02, b't', // name
+ 0x02, // partition_indexes: 1
+ 0x00, 0x00, 0x00, 0x00, // 0
+ 0x00, // topic tagged fields
+ 0x00, // group tagged fields
+ 0x00, // require_stable
+ 0x00, // tagged fields
+ ]);
+ assert!(validate_offset_fetch_shape(9, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetFetchRequest>(9, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_fetch_v6_huge_topics_count_rejected() {
+ let body = Bytes::from_static(&[0x02, b'g', 0xFF, 0xFF, 0xFF, 0xFF, 0x0F]);
+ assert!(validate_offset_fetch_shape(6, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ #[test]
+ fn offset_fetch_v1_null_topics_rejected() {
+ let body = Bytes::from_static(&[0x00, 0x01, b'g', 0xFF, 0xFF, 0xFF, 0xFF]);
+ assert!(validate_offset_fetch_shape(1, &body, TEST_MAX_FRAME_SIZE).is_err());
}
}
diff --cc gateways/kafka/src/protocol/handlers/mod.rs
index 394040acc,f8d58a80d..4a7b2b693
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@@ -36,21 -34,24 +36,26 @@@ pub mod join_group
pub mod leave_group;
pub mod list_offsets;
pub mod metadata;
+ pub mod offset_commit;
+ pub mod offset_fetch;
pub mod produce;
pub mod sync_group;
+pub(crate) mod topic_config;
use bytes::{Buf, Bytes, BytesMut};
+ use iggy::prelude::IggyError;
use kafka_protocol::messages::TransactionalId;
use kafka_protocol::protocol::{Decodable, Encodable};
use tokio::runtime::{Handle, RuntimeFlavor};
+ use tokio::time::Instant;
+use crate::auth::AuthenticatedPrincipal;
+ use crate::bridge::BridgeError;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
- API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, 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_OFFSETS, API_KEY_METADATA, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_FETCH,
+ API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_DESCRIBE_CONFIGS,
+ 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_OFFSETS, API_KEY_METADATA,
++ API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP, 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,
@@@ -90,13 -90,13 +95,15 @@@ pub async fn dispatch
API_KEY_FETCH => fetch::handle(state, connection, api_version, body).await,
API_KEY_LIST_OFFSETS => list_offsets::handle(state, api_version, body).await,
API_KEY_METADATA => metadata::handle(state, api_version, body).await,
+ API_KEY_OFFSET_COMMIT => offset_commit::handle(state, connection, api_version, body).await,
+ API_KEY_OFFSET_FETCH => offset_fetch::handle(state, connection, api_version, body).await,
API_KEY_API_VERSIONS => api_versions::handle(state, api_version, body).await,
API_KEY_CREATE_TOPICS => create_topics::handle(state, api_version, body).await,
+ API_KEY_DESCRIBE_CONFIGS => describe_configs::handle(state, api_version, body).await,
+ API_KEY_ALTER_CONFIGS => alter_configs::handle(state, principal, api_version, body).await,
API_KEY_FIND_COORDINATOR => find_coordinator::handle(state, api_version, body).await,
API_KEY_JOIN_GROUP => join_group::handle(state, api_version, body).await,
- API_KEY_HEARTBEAT => heartbeat::handle(state, api_version, body).await,
+ 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_INIT_PRODUCER_ID => init_producer_id::handle(state, api_version, body).await,
diff --cc gateways/kafka/tests/common/wire.rs
index b9dd3c6bb,e11cc82ff..dc3b838a5
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@@ -31,23 -30,8 +31,21 @@@ use kafka_protocol::protocol::{Decodabl
use super::codec::Encoder;
+/// Encodes any `kafka_protocol` request or response type at `version`.
+pub fn encode<M: Encodable>(message: &M, version: i16) -> Bytes {
+ let mut buf = BytesMut::new();
+ message.encode(&mut buf, version).expect("encode request");
+ buf.freeze()
+}
+
+/// Decodes any `kafka_protocol` request or response type at `version`.
+pub fn decode<M: Decodable>(body: Bytes, version: i16) -> M {
+ let mut buf = body;
+ 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)] = &[
- (8, "OffsetCommit"),
- (9, "OffsetFetch"),
(15, "DescribeGroups"),
(16, "ListGroups"),
(20, "DeleteTopics"),
diff --cc gateways/kafka/tests/version_firewall_tests.rs
index 665e4ef70,4ee8cee68..d0fc438dd
--- a/gateways/kafka/tests/version_firewall_tests.rs
+++ b/gateways/kafka/tests/version_firewall_tests.rs
@@@ -37,9 -37,9 +37,9 @@@ use tokio::io::AsyncWriteExt
use tokio::net::TcpStream;
use iggy_gateway_kafka::protocol::api::{
- API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, 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_OFFSETS, API_KEY_METADATA, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_FETCH,
+ API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_DESCRIBE_CONFIGS,
+ 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_OFFSETS, API_KEY_METADATA,
++ API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP, 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,
@@@ -520,8 -520,8 +521,10 @@@ fn request_body_for_scoped_api(api_key
},
),
API_KEY_INIT_PRODUCER_ID => build_init_producer_id_request(version, None),
+ API_KEY_DESCRIBE_CONFIGS => build_describe_configs_empty_request(version),
+ 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")),
_ => Bytes::new(),
}
}