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 <[email protected]> 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(), } }
