feat(gateways): iggy gateway for kafka listener

[ryerraguntla]

Which issue does this PR address?

Closes #3421

Rationale

Existing Kafka clients cannot talk to Iggy without a protocol translation layer. Issue #3421 calls for a foundation-layer TCP listener on the Kafka wire port (9093) so that any Kafka-compatible client can eventually route messages into Iggy streams without any client-side changes. This PR delivers that foundation: decode, version-firewall, and stub responses with no Iggy backend integration yet.

What changed?

Before this PR, Iggy had no surface that spoke the Kafka binary protocol. Kafka clients connecting on port 9093 would receive no response or a connection reset.

This PR adds a new gateways/kafka/ workspace crate (iggy-gateway-kafka) with a tokio-based TCP listener that accepts Kafka wire connections, auto-detects request header v1/v2, enforces a version-firewall (SUPPORTED_RANGES), decodes and stub-encodes six API keys (ApiVersions 18, Metadata 3, Produce 0, Fetch 1, ListOffsets 2, CreateTopics 19), and rejects unsupported keys or versions with UNSUPPORTED_VERSION (error 35) without dropping the connection. A kafka-message-gen helper tool generates golden wire fixtures used by the regression suite.

Key implementation notes:

• Produce hot path keeps RecordBatch as opaque Bytes — no RecordBatch decode in this layer.
• BrokerAdvertise is Arc-cloned per connection; no per-connection string allocation.
• Frame body read: bounded read() chunks, not read_buf — avoids stealing pipelined frame bytes; chunks zero-filled via resize.
• SUPPORTED_RANGES is the single source of truth; ApiVersions advertises exactly what the firewall allows.

103 regression tests across 12 suites cover: primitive codec round-trips, adversarial wire inputs, header v1/v2 parsing, version boundary matrix, per-API golden byte fixtures, stub error codes, and full TCP round-trips via a real KafkaServer listener.

Local Execution

• Passed
• Pre-commit hooks ran

# Fixture generation (required for decode_validation_tests)
cargo run -p kafka-message-gen -- generate \
  --output gateways/kafka/tools/kafka-tool/kafka_messages \
  --api-key 0    --api-key 1 --api-key 2 --api-key 19

# Regression suite
cargo test -p iggy-gateway-kafka

# Lint
cargo clippy -p iggy-gateway-kafka --all-targets -- -D warnings
cargo fmt --all --check

All 103 tests pass. clippy -D warnings clean. Manual smoke-test procedure followed per gateways/kafka/docs/MANUAL_TESTING.md (ApiVersions, Metadata, Produce, Fetch round-trips; version firewall; oversized frame rejection).

AI Usage

1. Tools: Claude Sonnet 4.6 (Cursor Cloud Agent)
2. Scope: Protocol codec implementation (requests.rs, responses.rs, codec.rs, header.rs), server framing (server.rs), error type (error.rs), regression test suites, documentation (SCOPE.md, TEST_SUITE.md, MANUAL_TESTING.md, kafka_api_keys_reference.md), and the kafka-tool fixture generator.
3. Verification: Every generated file was reviewed line by line. The 103-test regression suite was run locally against the implementation. Manual testing followed the procedure in MANUAL_TESTING.md using kcat and the kafka-message-gen tool against a live listener. Golden byte fixtures pin exact wire output against known-correct Kafka responses.
4. Explainability: Yes — can explain the code. The codec follows the Kafka protocol specification field-by-field; non-obvious decisions (opaque RecordBatch, SUPPORTED_RANGES governance model, Arc<BrokerAdvertise> per-connection, varint=0 null compact array) are documented inline or in SCOPE.md.
5. Some files(SCOPE.md and TEST_SUITE.md) in the docs folder are temporary. After the implementation of the complete feature, these files will be removed or merged with other documents.

[github-actions]

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

• /ready - back to S-waiting-on-review after addressing feedback
• /author - flip to S-waiting-on-author while you finish changes
• /request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 82.76%. Comparing base (4eb879e) to head (e803716).

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3519      +/-   ##
============================================
- Coverage     82.78%   82.76%   -0.03%     
  Complexity     1299     1299              
============================================
  Files          1199     1199              
  Lines        161885   161887       +2     
  Branches     131360   131464     +104     
============================================
- Hits         134019   133985      -34     
+ Misses        24322    24243      -79     
- Partials       3544     3659     +115     

Components	Coverage Δ
Rust Core	83.32% <ø> (+0.04%)	⬆️
Java SDK	66.06% <ø> (ø)
C# SDK	74.77% <ø> (-1.21%)	⬇️
Python SDK	89.98% <ø> (ø)
PHP SDK	84.26% <ø> (ø)
Node SDK	96.21% <ø> (ø)
Go SDK	68.53% <ø> (ø)
see 70 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs.

If you need a review, please ensure CI is green and the PR is rebased on the latest master. Don't hesitate to ping the maintainers - either @core on Discord or by mentioning them directly here on the PR.

Thank you for your contribution!

[ryerraguntla]

/ready

[atharvalade]

I ran this locally to verify the behavior before commenting, sharing the setup so it's easy to reproduce.

Started the gateway:

cargo build -p iggy-gateway-kafka
KAFKA_BIND_ADDR=127.0.0.1:19093 ./target/debug/iggy-gateway-kafka

Then sent raw frames with this script:

import socket, struct, time

def frame(p): return struct.pack('>i', len(p)) + p

def send(payload, timeout=3.0):
    s = socket.create_connection(('127.0.0.1', 19093), timeout=timeout)
    s.sendall(frame(payload)); s.settimeout(timeout)
    try: return s.recv(65536)
    except socket.timeout: return None

# 1. Produce v3 with acks=0, empty topics. Spec says the broker must not respond.
hdr = struct.pack('>hhih', 0, 3, 42, -1)
body = struct.pack('>hhii', -1, 0, 1000, 0)
resp = send(hdr + body)
print('acks=0 Produce:', resp.hex() if resp else 'no response')

# 2. ListOffsets v0, below the firewall min, so the error path encodes at v0.
hdr = struct.pack('>hhih', 2, 0, 7, -1)
body = struct.pack('>i', -1) + struct.pack('>i', 1) + struct.pack('>h', 1) + b't' \
     + struct.pack('>i', 1) + struct.pack('>iqi', 0, -1, 1)
resp = send(hdr + body)
print('ListOffsets v0:', resp.hex())

# 3. Metadata v1 asking for topic "orders".
hdr = struct.pack('>hhih', 3, 1, 9, -1)
body = struct.pack('>i', 1) + struct.pack('>h', 6) + b'orders'
resp = send(hdr + body)
print('Metadata has "orders":', b'orders' in resp, '| has "unknown-topic":', b'unknown-topic' in resp)

# 4. Open a connection and send nothing, measure when the server closes it.
s = socket.create_connection(('127.0.0.1', 19093), timeout=20)
start = time.time()
s.recv(1)
print(f'idle connection closed after {time.time()-start:.1f}s')

Output I got:

acks=0 Produce: 0000000c0000002a0000000000000000
ListOffsets v0: 0000001c00000007000000010000000000010000000000230000000000000000
Metadata has "orders": False | has "unknown-topic": True
idle connection closed after 15.0s

So: acks=0 gets a 16 byte response when the spec says silence, the ListOffsets v0 error response leaves 4 unparsed bytes for a v0 client (bare i64 where the old_style_offsets array should be), Metadata never echoes the requested topic name, and idle connections are dropped after 15s. Details in the inline comments.

[slbotbm]

@ryerraguntla seems like you pushed opensearch connector work to this PR as well?

[ryerraguntla]

/ready

[ryerraguntla]

/author