numinnex commented on code in PR #4231:
URL: https://github.com/apache/iggy/pull/4231#discussion_r4060685364
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -550,26 +831,36 @@ fn encode_api_versions_response(api_version: i16,
error_code: i16) -> Result<Byt
encode_message(&resp, api_version, 128)
}
+/// `topics` is `(kafka_topic, error_code, partitions_count)` -
`partitions_count` is `None`
+/// exactly when `error_code != ERROR_NONE` (a failed lookup has no partitions
to report), `Some`
+/// otherwise. Single-broker gateway: every partition of every topic reports
broker id 1 as its
+/// leader, sole replica and sole in-sync replica - there is no second broker
to be anything else.
fn encode_metadata_response(
response_version: i16,
- topics: &[StrBytes],
+ topics: &[(String, i16, Option<u32>)],
broker: &BrokerAdvertise,
- topic_error_override: i16,
) -> Result<Bytes> {
- // Stub has no topic catalog: echo requested names with
UNKNOWN_TOPIC_OR_PARTITION,
- // or a forced override (unused today; kept for symmetry with other
encoders).
- let topic_error = if topic_error_override == ERROR_NONE {
- ERROR_UNKNOWN_TOPIC_OR_PARTITION
- } else {
- topic_error_override
- };
-
let response_topics = topics
.iter()
- .map(|name| {
+ .map(|(name, error_code, partitions_count)| {
+ let partitions = partitions_count.map_or_else(Vec::new, |count| {
Review Comment:
**Response size is unbounded, and `bounds_guard`'s projection cannot see
it.**
The guard charges `RESPONSE_BYTES_PER_ELEMENT = 64` per *request* element,
justified in its own doc as "the largest fixed per-partition response entry is
~33 bytes" - a model that assumes 1 request element yields 1 response element.
That holds for Produce. Here one requested name expands to N server-side
partitions (Iggy caps at 1000), so 4096 duplicate names project as 256 KiB and
build ~100 MB of `MetadataResponsePartition` in one synchronous non-yielding
block. `send_response` checks only `i32::MAX`, never `max_frame_size`.
Suggest charging the encoded response against `max_frame_size` before
responding.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -440,9 +475,60 @@ fn handle_metadata(
api_version,
"Failed to decode Metadata request; closing connection"
);
- HandleOutcome::Close
+ return HandleOutcome::Close;
}
- }
+ };
+
+ // `None` (null array) means "all topics"; `Some(&[])` (explicit empty
array) means "no
+ // topics - brokers/cluster metadata only" (KIP-4's `describeCluster()`
shape); `Some(names)`
+ // means "look up exactly these" - the three must stay distinguishable (see
+ // `decode_metadata_topics`'s own doc). Listing is one bridge call per
stream this gateway's
+ // topic mapping can resolve to; looking up N specific topics is N bridge
calls, one per
+ // name, since Metadata's per-topic error_code needs to distinguish
"doesn't exist" from a
+ // real bridge failure for each name independently.
+ let resolved: Vec<(String, i16, Option<u32>)> = match requested {
+ None => match bridge.list_kafka_topics().await {
+ Ok(topics) => topics
+ .into_iter()
+ .map(|topic| (topic.kafka_topic, ERROR_NONE,
Some(topic.partitions_count)))
+ .collect(),
+ Err(error) => {
+ // Metadata has no top-level error field (same reason the
decode-failure arm
+ // above closes rather than encoding one): silently answering
"zero topics
+ // exist" here would be a lie, not a graceful degradation -
the bridge failed to
+ // answer, it did not confirm an empty catalog.
+ tracing::warn!(%error, "failed to list Kafka topics from Iggy
bridge; closing connection");
+ return HandleOutcome::Close;
+ }
+ },
+ Some(names) if names.is_empty() => Vec::new(),
+ Some(names) => {
+ let mut out = Vec::with_capacity(names.len());
+ for name in &names {
Review Comment:
**Unbounded bridge fan-out on the shared connection.**
`bounds_guard` allows 4096 elements, and this loop spends 2 Iggy round trips
per name (`iggy_bridge.rs:559,567`) with no dedup. `CreateTopics` costs up to 5
per topic, `ListOffsets` 1 per topic. All of it runs on the single lockstep
`IggyClient` shared by every connection (`iggy_bridge.rs:109-116`), so a ~12 KB
frame serializes ~8k round trips and head-of-line-blocks every other client's
control plane. Each call carries its own 15s timeout; nothing bounds the
request as a whole, and the wire `timeout_ms` is decoded but unused.
Suggest deduping names and capping bridge-backed topics per request well
below `MAX_REQUEST_ELEMENTS`, plus one wall-clock deadline for the handler.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -470,6 +556,201 @@ fn handle_versioned_request<T>(
}
}
+/// `CreateTopics` (`#3538`): decodes the request, validates and (unless
`validate_only`)
+/// provisions each requested topic through `bridge.ensure_stream_and_topic`,
and reports one
+/// result per topic - never fails the whole response for one bad topic among
several.
+async fn handle_create_topics(
+ api_version: i16,
+ body: Bytes,
+ max_frame_size: usize,
+ bridge: &dyn TopicCatalog,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_CREATE_TOPICS, api_version) {
+ return unsupported_version_response(API_KEY_CREATE_TOPICS,
api_version, |v| {
+ encode_create_topics_error_response(v, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let req = match decode_guarded::<CreateTopicsRequest>(api_version, body,
|v, b| {
+ validate_create_topics_shape(v, b, max_frame_size)
+ }) {
+ Ok(req) => req,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode CreateTopics request");
+ return respond_or_close(
+ encode_create_topics_error_response(api_version,
ERROR_INVALID_REQUEST),
+ "CreateTopics",
+ );
+ }
+ };
+
+ let mut results: Vec<CreateTopicResult> =
Vec::with_capacity(req.topics.len());
+ for topic in &req.topics {
+ results.push(create_one_topic(topic, api_version, req.validate_only,
bridge).await);
+ }
+ respond_or_close(
+ encode_create_topics_response(&req.topics, &results, api_version),
+ "CreateTopics",
+ )
+}
+
+/// Validates and (unless `validate_only`) creates one requested topic.
+///
+/// Sequential across a multi-topic request, not concurrent: `IggyBridge`
holds one lockstep
+/// `IggyClient` (`bridge::iggy_bridge`'s own doc - "the connection is
lockstep, one request in
+/// flight at a time"), so concurrent calls here would only queue behind each
other on the same
+/// mutex anyway; sequential keeps each topic's failure independent and easy
to reason about,
+/// with no ordering surprise from a scheduler interleaving them differently
across runs.
+async fn create_one_topic(
+ topic: &CreatableTopic,
+ api_version: i16,
+ validate_only: bool,
+ bridge: &dyn TopicCatalog,
+) -> CreateTopicResult {
+ if !topic.configs.is_empty() {
+ return Err(ERROR_INVALID_CONFIG);
+ }
+ // Checked here, not left to `ensure_stream_and_topic`'s own call: that
call is skipped
+ // entirely below for `validate_only`, and name validation is part of "can
this topic be
+ // created as specified," the exact thing `validate_only` promises to
check - skipping it
+ // would report success for a name (empty, padded, over Kafka's 249-byte
cap) that the real
+ // call always rejects.
+ validate_kafka_topic_name("kafka_topic", topic.name.0.as_str())
+ .map_err(|error| error.to_kafka_error_code())?;
+ let partition_count = validate_create_topic_shape(topic, api_version)?;
+
+ let kafka_topic = topic.name.0.as_str();
+ // Checked before the `validate_only` branch, not after: "can this topic
be created as
+ // specified" includes "does it already exist" - real Kafka's own
`validateOnly` still
+ // flags an existing topic, it doesn't only check format.
`get_kafka_topic` is read-only
+ // (the same lookup `handle_metadata` uses), so running it here doesn't
violate KIP-4's
+ // "don't create anything" promise even when `validate_only` is set.
+ //
+ // `ensure_stream_and_topic`'s own contract is intentionally idempotent
(get-or-create,
+ // matching-spec re-call is Ok) - correct for an internal "make sure this
exists" helper, but
+ // `CreateTopics` is not an upsert: the real `AdminClient.createTopics`
contract is
+ // `TOPIC_ALREADY_EXISTS` (36) for a topic that's already there even when
the requested spec
+ // matches exactly. The common "genuinely new topic" path still goes
straight through
+ // `ensure_stream_and_topic`'s own create-race handling unchanged below;
only "it's already
+ // there" short-circuits before ever reaching it.
+ match bridge.get_kafka_topic(kafka_topic).await {
Review Comment:
**TOCTOU: concurrent creates both report success.**
`get_kafka_topic` -> `.await` -> `ensure_stream_and_topic` is not atomic.
Two concurrent `CreateTopics` for the same new name both observe `Ok(None)`,
and the second is absorbed by the idempotent arm in `ensure_stream_and_topic`,
so both clients receive `ERROR_NONE`. Kafka guarantees exactly one `NONE` and
`TOPIC_ALREADY_EXISTS` for the loser. The lockstep client serializes each call
but not the pair.
Suggest deriving already-exists from the create result instead of a prior
read.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -470,6 +556,201 @@ fn handle_versioned_request<T>(
}
}
+/// `CreateTopics` (`#3538`): decodes the request, validates and (unless
`validate_only`)
+/// provisions each requested topic through `bridge.ensure_stream_and_topic`,
and reports one
+/// result per topic - never fails the whole response for one bad topic among
several.
+async fn handle_create_topics(
+ api_version: i16,
+ body: Bytes,
+ max_frame_size: usize,
+ bridge: &dyn TopicCatalog,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_CREATE_TOPICS, api_version) {
+ return unsupported_version_response(API_KEY_CREATE_TOPICS,
api_version, |v| {
+ encode_create_topics_error_response(v, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let req = match decode_guarded::<CreateTopicsRequest>(api_version, body,
|v, b| {
+ validate_create_topics_shape(v, b, max_frame_size)
+ }) {
+ Ok(req) => req,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode CreateTopics request");
+ return respond_or_close(
+ encode_create_topics_error_response(api_version,
ERROR_INVALID_REQUEST),
+ "CreateTopics",
+ );
+ }
+ };
+
+ let mut results: Vec<CreateTopicResult> =
Vec::with_capacity(req.topics.len());
Review Comment:
**Duplicate topic names in one request create a topic Kafka would not.**
Processing entries independently means the first occurrence is created and
returns 0, later ones return `TOPIC_ALREADY_EXISTS`. Kafka refuses the whole
name: `ControllerApis.createTopics` returns per-topic `INVALID_REQUEST` (42)
"Duplicate topic name." for *every* occurrence and creates nothing. Java
`AdminClient` keys futures by name, so one of the two results is silently
dropped.
Suggest pre-scanning for duplicates, marking all occurrences 42, and
skipping the bridge.
##########
gateways/kafka/tests/common/fake_bridge.rs:
##########
@@ -0,0 +1,184 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! In-memory [`TopicCatalog`] for protocol-level tests (decode edge cases,
version firewall,
+//! wire-shape assertions) that don't need a real `iggy-server` - only the
dispatch layer is
+//! under test in those files, not bridge behavior itself.
`bridge_iggy_integration_tests.rs` and
+//! `gateway_bridge_e2e_tests.rs` exercise real `IggyBridge` behavior against
a spawned server.
+
+#![allow(dead_code)] // Not every test file that includes this module uses
every method.
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+use async_trait::async_trait;
+use iggy::prelude::IggyError;
+use iggy_gateway_kafka::bridge::{BridgeError, KafkaTopicMetadata,
TopicCatalog};
+
+/// One topic's fake state: partition count, plus a single watermark applied
to every partition
+/// (real `IggyBridge` tracks watermarks per-partition, but no test in this
suite needs more than
+/// one distinguishable value per topic to tell LATEST from EARLIEST).
+#[derive(Clone, Copy)]
+struct TopicState {
+ partitions_count: u32,
+ /// What `high_watermarks` reports for every partition of this topic.
Distinct from `0` by
+ /// default specifically so a test seeding a nonzero watermark can tell
"the handler read the
+ /// real watermark" apart from "the handler (or this fake) always answers
0 regardless" - a
+ /// fake that only ever returns `Ok(0)` (this type's original shape) makes
every EARLIEST/
+ /// LATEST assertion pass identically whether the underlying logic is
correct or not.
+ watermark: i64,
+}
+
+/// Topics this fake already knows about, keyed by Kafka topic name. Seeded via
+/// [`FakeBridge::with_topic`]/[`FakeBridge::with_topic_and_watermark`];
`ensure_stream_and_topic`
+/// also inserts into it (watermark `0`), mirroring `IggyBridge`'s own
create-if-missing contract
+/// closely enough for wire-level assertions.
+#[derive(Default)]
+pub struct FakeBridge {
+ topics: Mutex<HashMap<String, TopicState>>,
+}
+
+impl FakeBridge {
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Seeds a topic with watermark `0` for every partition - equivalent to
+ /// `with_topic_and_watermark(kafka_topic, partitions_count, 0)`. Kept as
its own method: most
+ /// callers only care about partition count/existence, not a specific
watermark value.
+ #[must_use]
+ pub fn with_topic(self, kafka_topic: &str, partitions_count: u32) -> Self {
+ self.with_topic_and_watermark(kafka_topic, partitions_count, 0)
+ }
+
+ #[must_use]
+ pub fn with_topic_and_watermark(
+ self,
+ kafka_topic: &str,
+ partitions_count: u32,
+ watermark: i64,
+ ) -> Self {
+ self.topics
+ .lock()
+ .expect("fake bridge mutex poisoned")
+ .insert(
+ kafka_topic.to_string(),
+ TopicState {
+ partitions_count,
+ watermark,
+ },
+ );
+ self
+ }
+}
+
+#[async_trait]
Review Comment:
**The fake cannot produce any connection-shaped failure.**
Only `PartitionCountMismatch`, `PartitionOutOfRange` and `TopicNameNotFound`
are reachable here. Every transport error - `Timeout`, `Disconnected`,
`StaleClient`, `NotConnected`, `EmptyResponse` - is impossible, so
`to_kafka_error_code` is never exercised through a handler, and
`handle_metadata`'s list-failure `Close` branch has no coverage and cannot get
any with this fake. `list_kafka_topics` is infallible by construction.
Suggest an injectable `VecDeque<Result<..>>` per method.
##########
gateways/kafka/tests/common/fake_bridge.rs:
##########
@@ -0,0 +1,184 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! In-memory [`TopicCatalog`] for protocol-level tests (decode edge cases,
version firewall,
+//! wire-shape assertions) that don't need a real `iggy-server` - only the
dispatch layer is
+//! under test in those files, not bridge behavior itself.
`bridge_iggy_integration_tests.rs` and
+//! `gateway_bridge_e2e_tests.rs` exercise real `IggyBridge` behavior against
a spawned server.
+
+#![allow(dead_code)] // Not every test file that includes this module uses
every method.
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+use async_trait::async_trait;
+use iggy::prelude::IggyError;
+use iggy_gateway_kafka::bridge::{BridgeError, KafkaTopicMetadata,
TopicCatalog};
+
+/// One topic's fake state: partition count, plus a single watermark applied
to every partition
+/// (real `IggyBridge` tracks watermarks per-partition, but no test in this
suite needs more than
+/// one distinguishable value per topic to tell LATEST from EARLIEST).
+#[derive(Clone, Copy)]
+struct TopicState {
+ partitions_count: u32,
+ /// What `high_watermarks` reports for every partition of this topic.
Distinct from `0` by
+ /// default specifically so a test seeding a nonzero watermark can tell
"the handler read the
+ /// real watermark" apart from "the handler (or this fake) always answers
0 regardless" - a
+ /// fake that only ever returns `Ok(0)` (this type's original shape) makes
every EARLIEST/
+ /// LATEST assertion pass identically whether the underlying logic is
correct or not.
+ watermark: i64,
+}
+
+/// Topics this fake already knows about, keyed by Kafka topic name. Seeded via
+/// [`FakeBridge::with_topic`]/[`FakeBridge::with_topic_and_watermark`];
`ensure_stream_and_topic`
+/// also inserts into it (watermark `0`), mirroring `IggyBridge`'s own
create-if-missing contract
+/// closely enough for wire-level assertions.
+#[derive(Default)]
+pub struct FakeBridge {
+ topics: Mutex<HashMap<String, TopicState>>,
+}
+
+impl FakeBridge {
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Seeds a topic with watermark `0` for every partition - equivalent to
+ /// `with_topic_and_watermark(kafka_topic, partitions_count, 0)`. Kept as
its own method: most
+ /// callers only care about partition count/existence, not a specific
watermark value.
+ #[must_use]
+ pub fn with_topic(self, kafka_topic: &str, partitions_count: u32) -> Self {
+ self.with_topic_and_watermark(kafka_topic, partitions_count, 0)
+ }
+
+ #[must_use]
+ pub fn with_topic_and_watermark(
+ self,
+ kafka_topic: &str,
+ partitions_count: u32,
+ watermark: i64,
+ ) -> Self {
+ self.topics
+ .lock()
+ .expect("fake bridge mutex poisoned")
+ .insert(
+ kafka_topic.to_string(),
+ TopicState {
+ partitions_count,
+ watermark,
+ },
+ );
+ self
+ }
+}
+
+#[async_trait]
+impl TopicCatalog for FakeBridge {
+ async fn ensure_stream_and_topic(
+ &self,
+ kafka_topic: &str,
+ partition_count: u32,
+ ) -> Result<(), BridgeError> {
+ let existing = {
+ let mut topics = self.topics.lock().expect("fake bridge mutex
poisoned");
+ let existing = topics.get(kafka_topic).map(|state|
state.partitions_count);
+ if existing.is_none() {
+ topics.insert(
+ kafka_topic.to_string(),
+ TopicState {
+ partitions_count: partition_count,
+ watermark: 0,
+ },
+ );
+ }
+ existing
+ };
+ match existing {
+ Some(existing) if existing != partition_count => {
+ Err(BridgeError::PartitionCountMismatch {
+ topic: kafka_topic.to_string(),
+ existing,
+ requested: partition_count,
+ })
+ }
+ _ => Ok(()),
+ }
+ }
+
+ async fn high_watermarks(
+ &self,
+ kafka_topic: &str,
+ partitions: &[u32],
+ ) -> Result<Vec<(u32, Result<i64, BridgeError>)>, BridgeError> {
+ let state = {
+ let topics = self.topics.lock().expect("fake bridge mutex
poisoned");
+ topics.get(kafka_topic).copied()
+ };
+ let Some(state) = state else {
+ return Err(BridgeError::Iggy(IggyError::TopicNameNotFound(
+ kafka_topic.to_string(),
+ "kafka".to_string(),
+ )));
+ };
+ Ok(partitions
+ .iter()
+ .map(|&partition| {
+ let result = if partition < state.partitions_count {
+ Ok(state.watermark)
+ } else {
+ Err(BridgeError::PartitionOutOfRange {
+ topic: kafka_topic.to_string(),
+ partition,
+ partitions_count: state.partitions_count,
+ })
+ };
+ (partition, result)
+ })
+ .collect())
+ }
+
+ async fn get_kafka_topic(
Review Comment:
**Fake skips name validation, so tests pin the wrong error code.**
The real `IggyBridge` runs `validate_kafka_topic_name` first in
`get_kafka_topic`, `high_watermarks` and `ensure_stream_and_topic`. Metadata
and ListOffsets pass the raw wire name straight through, so for an illegal name
the fake answers `UNKNOWN_TOPIC_OR_PARTITION` (3) where production answers
`INVALID_TOPIC_EXCEPTION` (17).
Also missing the `partition_count == 0` guard. Suggest mirroring both.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -440,9 +475,60 @@ fn handle_metadata(
api_version,
"Failed to decode Metadata request; closing connection"
);
- HandleOutcome::Close
+ return HandleOutcome::Close;
}
- }
+ };
+
+ // `None` (null array) means "all topics"; `Some(&[])` (explicit empty
array) means "no
+ // topics - brokers/cluster metadata only" (KIP-4's `describeCluster()`
shape); `Some(names)`
+ // means "look up exactly these" - the three must stay distinguishable (see
+ // `decode_metadata_topics`'s own doc). Listing is one bridge call per
stream this gateway's
+ // topic mapping can resolve to; looking up N specific topics is N bridge
calls, one per
+ // name, since Metadata's per-topic error_code needs to distinguish
"doesn't exist" from a
+ // real bridge failure for each name independently.
+ let resolved: Vec<(String, i16, Option<u32>)> = match requested {
+ None => match bridge.list_kafka_topics().await {
+ Ok(topics) => topics
+ .into_iter()
+ .map(|topic| (topic.kafka_topic, ERROR_NONE,
Some(topic.partitions_count)))
+ .collect(),
+ Err(error) => {
+ // Metadata has no top-level error field (same reason the
decode-failure arm
+ // above closes rather than encoding one): silently answering
"zero topics
+ // exist" here would be a lie, not a graceful degradation -
the bridge failed to
+ // answer, it did not confirm an empty catalog.
+ tracing::warn!(%error, "failed to list Kafka topics from Iggy
bridge; closing connection");
+ return HandleOutcome::Close;
+ }
+ },
+ Some(names) if names.is_empty() => Vec::new(),
Review Comment:
**Empty `topics` array means "all topics" at v0.**
Kafka's own rule is `isAllTopics() = topics() == null || (topics().isEmpty()
&& version() == 0)`. This arm applies "empty = none" at every version, and the
gateway advertises Metadata from v0, so a v0 client asking for the whole
catalog is told the cluster is empty.
Exposure is narrow (modern Java refuses v0, librdkafka negotiates up), so
not a blocker - but the decode side is already correct, it is only this branch
that diverges.
Suggest gating on `api_version >= 1`.
##########
gateways/kafka/src/main.rs:
##########
@@ -36,24 +36,64 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let config = load_config()?;
+ // Connected before the listener binds: an unreachable Iggy backend at
startup should fail
+ // fast and loud (a non-zero exit an orchestrator's readiness probe sees
immediately), not
+ // accept Kafka connections that can then only ever answer bridge-backed
requests with a
+ // connectivity error. `IggyBridge::connect` itself bounds this in
wall-clock time (see its
+ // own `REQUEST_TIMEOUT` doc) - it cannot hang here even against a
silently-dropping address.
+ let bridge_config = IggyBridgeConfig::from_env()?;
+ let bridge = IggyBridge::connect(bridge_config)
+ .await
+ .map_err(|e| format!("failed to connect to Iggy: {e}"))?;
+ // Kept as a concrete `Arc<IggyBridge>`, not only the `Arc<dyn
TopicCatalog>` the server
+ // needs: `close()` takes `IggyBridge` by value, so tearing it down at
shutdown needs this
+ // binding's own `Arc::try_unwrap` once every clone the server handed out
is done with it -
+ // a bare `Arc<dyn TopicCatalog>` has no path back to the concrete type to
call it on.
+ let bridge = Arc::new(bridge);
+ let server_bridge: Arc<dyn iggy_gateway_kafka::bridge::TopicCatalog> =
bridge.clone();
+
let listener = bind_listener(&config.bind_addr)
.map_err(|e| format!("failed to bind {}: {e}", config.bind_addr))?;
- let server = KafkaGateway::new(config);
+ let server = KafkaGateway::new(config, server_bridge);
let (tx, rx) = broadcast::channel(1);
let mut server_task = tokio::spawn(async move { server.run(listener,
rx).await });
- tokio::select! {
- result = &mut server_task => {
- return Ok(result??);
- }
+ let server_result = tokio::select! {
+ result = &mut server_task => result,
() = shutdown_signal() => {
let _ = tx.send(());
+ server_task.await
}
- }
+ };
- server_task.await??;
- Ok(())
+ close_bridge(bridge).await;
Review Comment:
**`close_bridge` runs after the drain and can overrun the shutdown budget.**
`GatewayConfig`'s doc picks `shutdown_drain_timeout = 25s` specifically to
stay inside Kubernetes' 30s default grace. Running `close_bridge` after
`server_task.await` can add up to `REQUEST_TIMEOUT` (15s), since
`TcpClient::shutdown` takes the stream mutex a detached send task may still
hold. A drain finishing at 24.9s then a 15s close reaches ~40s and gets
SIGKILLed.
Suggest passing the remaining grace budget into `close_bridge`.
##########
gateways/kafka/docs/MANUAL_TESTING.md:
##########
@@ -75,11 +79,11 @@ connection-refuses before any assertion runs.
| A1 | Gateway starts | Run `iggy-gateway-kafka` | Binds to `:9093`, no panic
| Log shows bind address |
| A2 | ApiVersions v1 | `cargo run -p kafka-message-gen -- send --host
127.0.0.1:9093 --api-key 18 --version 1` | Response received | `ec=0`, non-zero
byte count |
| A3 | ApiVersions v3 (flexible) | Same with `--version 3` | Response received
| `ec=0` |
-| A4 | Metadata v0 | `send --host 127.0.0.1:9093 --api-key 3 --version 0` |
Stub broker in response | Topic entries show `ec=3`
(UNKNOWN_TOPIC_OR_PARTITION, stub) |
+| A4 | Metadata v0 | `send --host 127.0.0.1:9093 --api-key 3 --version 0` |
Real lookup via `IggyBridge` | Topic entries show `ec=3`
(UNKNOWN_TOPIC_OR_PARTITION) for a topic that doesn't exist yet on the
connected Iggy backend |
Review Comment:
**A4's pass criterion is now unreachable, so correct code reads as failing.**
`kafka-message-gen` builds Metadata v0 from `MetadataRequest::default()`,
whose `topics` is `Some(vec![])`, and it only sets `topics = None` from v1.
Under this PR an explicit empty array means "no topics", so the response
carries zero topic entries and the tool prints `ec=0` - never the documented
"Topic entries show `ec=3`".
Suggest running A4 at `--version 1`, or restating the criterion as "1
broker, 0 topic entries".
##########
gateways/kafka/src/protocol/responses.rs:
##########
@@ -230,89 +252,154 @@ fn encode_list_offsets_response_inner(
encode_message(&resp, version, 256)
}
+/// `timestamp`/`offset` default to `-1`, matching a real broker's own
error-path convention
+/// (`kafka_protocol`'s `#[derive(Default)]` would otherwise zero them, which
reads as "offset 0",
+/// a genuinely valid answer, rather than "no answer").
fn list_offsets_partition_response(
partition: i32,
error_code: i16,
) -> ListOffsetsPartitionResponse {
ListOffsetsPartitionResponse::default()
.with_partition_index(partition)
.with_error_code(error_code)
+ .with_timestamp(-1)
+ .with_offset(-1)
}
// ── CreateTopics
─────────────────────────────────────────────────────────────
-/// Well-formed `CreateTopics` response with a single placeholder topic.
+/// Whole-request `CreateTopics` failure (decode error, unsupported version).
+///
+/// One placeholder topic result carrying `error_code`, since no real
per-topic breakdown is
+/// possible when the request itself couldn't be read.
///
/// # Errors
///
/// Returns an error when `kafka_protocol` cannot encode the response at
`version`.
pub fn encode_create_topics_error_response(version: i16, error_code: i16) ->
Result<Bytes> {
let topics = vec![
- CreatableTopic::default()
- .with_num_partitions(1)
- .with_replication_factor(1),
+ CreatableTopicResult::default()
+ .with_error_code(error_code)
+ .with_error_message(None)
+ .with_num_partitions(-1)
+ .with_replication_factor(-1),
];
- encode_create_topics_response_inner(version, &topics, error_code)
+ let resp = CreateTopicsResponse::default().with_topics(topics);
+ encode_message(&resp, version, 256)
}
+/// One resolved `CreateTopics` outcome.
+///
+/// `Ok(partitions_created)` on success, `Err(error_code)` otherwise - a
per-topic failure (bad
+/// config, already exists with a different count) must not fail sibling
topics in the same
+/// request.
+pub type CreateTopicResult = std::result::Result<u32, i16>;
+
+/// Bridge-backed `CreateTopics` response.
+///
+/// One [`CreateTopicResult`] per topic in `topics`, in the same order
(`handle_create_topics`
+/// builds this by resolving each topic against
+/// [`crate::bridge::TopicCatalog::ensure_stream_and_topic`]).
+///
/// # Errors
///
/// Returns an error when `kafka_protocol` cannot encode the response at
`version`.
-pub fn encode_create_topics_response(version: i16, req: &CreateTopicsRequest)
-> Result<Bytes> {
- encode_create_topics_response_inner(version, &req.topics, ERROR_NONE)
+pub fn encode_create_topics_response(
+ topics: &[CreatableTopic],
+ results: &[CreateTopicResult],
+ version: i16,
+) -> Result<Bytes> {
+ let response_results = topics
+ .iter()
+ .zip(results)
+ .map(|(topic, result)| match result {
+ Ok(partitions_count) => CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(ERROR_NONE)
+ .with_error_message(None)
+
.with_num_partitions(i32::try_from(*partitions_count).unwrap_or(i32::MAX))
+ // Iggy's replication is cluster-wide (Raft over the whole
stream), not a
+ // per-topic knob this bridge can set - echoed back as 1
regardless of what was
+ // requested, matching this bridge's documented "RF is
accepted, not applied"
+ // policy (see README's CreateTopics section).
+ .with_replication_factor(1),
+ Err(error_code) => CreatableTopicResult::default()
+ .with_name(topic.name.clone())
+ .with_error_code(*error_code)
+ .with_error_message(None)
+ .with_num_partitions(-1)
+ .with_replication_factor(-1),
+ })
+ .collect();
+ let resp = CreateTopicsResponse::default().with_topics(response_results);
+ encode_message(&resp, version, 256)
}
-/// Resolve per-topic `CreateTopics` error.
+/// Broker default when `num_partitions == -1` (KIP-464). Real Kafka's own
broker default
+/// (`num.partitions`) is 1 out of the box; this bridge has no equivalent
per-deployment config
+/// yet, so 1 is hardcoded rather than invented.
+const DEFAULT_PARTITION_COUNT: u32 = 1;
+
+/// Validates one requested topic's `num_partitions`/`replication_factor`
shape.
+///
+/// Not whether the bridge call itself would succeed - existence conflicts
surface later, from
+/// `ensure_stream_and_topic`'s own result - and resolves the KIP-464
broker-default sentinel
+/// (`-1`) to a concrete partition count.
///
/// KIP-464: `num_partitions = -1` / `replication_factor = -1` mean broker
default when either
/// (a) the version is v4+, or (b) the topic carries a manual partition
assignment (valid on
-/// v2/v3 as well). Otherwise non-positive values are
[`ERROR_INVALID_PARTITIONS`] /
-/// [`ERROR_INVALID_REPLICATION_FACTOR`]. When validation passes, the stub
returns
-/// [`ERROR_NOT_CONTROLLER`] so clients do not believe the topic was created.
-const fn create_topics_topic_error(version: i16, topic: &CreatableTopic,
forced_error: i16) -> i16 {
- if forced_error != ERROR_NONE {
- return forced_error;
- }
-
+/// v2/v3 as well).
+///
+/// # Errors
+///
+/// Returns [`ERROR_INVALID_PARTITIONS`] or
[`ERROR_INVALID_REPLICATION_FACTOR`] if the shape is
+/// invalid for `version`.
+pub fn validate_create_topic_shape(
+ topic: &CreatableTopic,
+ version: i16,
+) -> std::result::Result<u32, i16> {
let broker_default_ok = version >= 4 || !topic.assignments.is_empty();
- let partitions_ok = if broker_default_ok {
- topic.num_partitions == -1 || topic.num_partitions > 0
+ // `num_partitions == -1` covers two distinct cases the wire can't tell
apart by that field
+ // alone (`CreatableTopic`'s own doc: "-1 if we are either specifying a
manual partition
+ // assignment or using the default partitions"): a client that left
partitioning to the
+ // broker, or one that supplied an explicit `assignments` list - one entry
per intended
+ // partition - and expects that count honored even though this bridge
can't honor the
+ // per-partition broker placement inside it. Falling back to
`DEFAULT_PARTITION_COUNT`
+ // unconditionally here would silently create a 1-partition topic for a
client that asked
+ // for N.
+ let partition_count = if topic.num_partitions == -1 && broker_default_ok {
+ if topic.assignments.is_empty() {
+ DEFAULT_PARTITION_COUNT
+ } else {
+ u32::try_from(topic.assignments.len()).map_err(|_|
ERROR_INVALID_PARTITIONS)?
+ }
+ } else if topic.num_partitions > 0 {
+ let count = u32::try_from(topic.num_partitions).map_err(|_|
ERROR_INVALID_PARTITIONS)?;
+ // An explicit count alongside a non-empty manual assignment is not
the normal shape a
+ // conforming client sends (the Java `NewTopic` constructors are
mutually exclusive on
+ // this), but nothing on the wire forbids it, and the two can disagree
- the assignment
+ // implies a different partition count than `num_partitions` states
outright. Neither is
+ // silently preferred: real Kafka's own analogue for a malformed
assignment is
+ // `INVALID_REPLICA_ASSIGNMENT` (39).
+ let assignment_count_matches = topic.assignments.is_empty()
+ || u64::try_from(topic.assignments.len()).is_ok_and(|len| len ==
u64::from(count));
+ if !assignment_count_matches {
+ return Err(ERROR_INVALID_REPLICA_ASSIGNMENT);
Review Comment:
**Manual assignments plus an explicit count should be rejected outright, and
error 39 is untested.**
Kafka rejects both shapes with `INVALID_REQUEST` (42): "A manual partition
assignment was specified, but numPartitions was not set to -1." (same for
replication factor). Here the combination is accepted when the counts happen to
agree and yields 39 when they disagree; `replication_ok` never consults
`assignments` at all. The comment above calling 39 "real Kafka's own analogue"
is contradicted by `ReplicationControlManager`.
Separately, 39 has zero test coverage - deleting this check leaves the suite
green.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -470,6 +556,201 @@ fn handle_versioned_request<T>(
}
}
+/// `CreateTopics` (`#3538`): decodes the request, validates and (unless
`validate_only`)
+/// provisions each requested topic through `bridge.ensure_stream_and_topic`,
and reports one
+/// result per topic - never fails the whole response for one bad topic among
several.
+async fn handle_create_topics(
+ api_version: i16,
+ body: Bytes,
+ max_frame_size: usize,
+ bridge: &dyn TopicCatalog,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_CREATE_TOPICS, api_version) {
+ return unsupported_version_response(API_KEY_CREATE_TOPICS,
api_version, |v| {
+ encode_create_topics_error_response(v, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let req = match decode_guarded::<CreateTopicsRequest>(api_version, body,
|v, b| {
+ validate_create_topics_shape(v, b, max_frame_size)
+ }) {
+ Ok(req) => req,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode CreateTopics request");
+ return respond_or_close(
+ encode_create_topics_error_response(api_version,
ERROR_INVALID_REQUEST),
+ "CreateTopics",
+ );
+ }
+ };
+
+ let mut results: Vec<CreateTopicResult> =
Vec::with_capacity(req.topics.len());
+ for topic in &req.topics {
+ results.push(create_one_topic(topic, api_version, req.validate_only,
bridge).await);
+ }
+ respond_or_close(
+ encode_create_topics_response(&req.topics, &results, api_version),
+ "CreateTopics",
+ )
+}
+
+/// Validates and (unless `validate_only`) creates one requested topic.
+///
+/// Sequential across a multi-topic request, not concurrent: `IggyBridge`
holds one lockstep
+/// `IggyClient` (`bridge::iggy_bridge`'s own doc - "the connection is
lockstep, one request in
+/// flight at a time"), so concurrent calls here would only queue behind each
other on the same
+/// mutex anyway; sequential keeps each topic's failure independent and easy
to reason about,
+/// with no ordering surprise from a scheduler interleaving them differently
across runs.
+async fn create_one_topic(
+ topic: &CreatableTopic,
+ api_version: i16,
+ validate_only: bool,
+ bridge: &dyn TopicCatalog,
+) -> CreateTopicResult {
+ if !topic.configs.is_empty() {
+ return Err(ERROR_INVALID_CONFIG);
+ }
+ // Checked here, not left to `ensure_stream_and_topic`'s own call: that
call is skipped
+ // entirely below for `validate_only`, and name validation is part of "can
this topic be
+ // created as specified," the exact thing `validate_only` promises to
check - skipping it
+ // would report success for a name (empty, padded, over Kafka's 249-byte
cap) that the real
+ // call always rejects.
+ validate_kafka_topic_name("kafka_topic", topic.name.0.as_str())
Review Comment:
**This validation is mutation-survivable.**
The comment is right that `validate_only` needs it, but no test exercises
it: `FakeBridge` skips `validate_kafka_topic_name` entirely, and no e2e case
sends an illegal name. Delete this line and `validate_only` happily reports
success for a whitespace-padded or 250-byte name, with the suite still green.
Suggest a handler test asserting `INVALID_TOPIC_EXCEPTION` (17) for a bad
name with `validate_only=true`.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -470,6 +556,201 @@ fn handle_versioned_request<T>(
}
}
+/// `CreateTopics` (`#3538`): decodes the request, validates and (unless
`validate_only`)
+/// provisions each requested topic through `bridge.ensure_stream_and_topic`,
and reports one
+/// result per topic - never fails the whole response for one bad topic among
several.
+async fn handle_create_topics(
+ api_version: i16,
+ body: Bytes,
+ max_frame_size: usize,
+ bridge: &dyn TopicCatalog,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_CREATE_TOPICS, api_version) {
+ return unsupported_version_response(API_KEY_CREATE_TOPICS,
api_version, |v| {
+ encode_create_topics_error_response(v, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let req = match decode_guarded::<CreateTopicsRequest>(api_version, body,
|v, b| {
+ validate_create_topics_shape(v, b, max_frame_size)
+ }) {
+ Ok(req) => req,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode CreateTopics request");
+ return respond_or_close(
+ encode_create_topics_error_response(api_version,
ERROR_INVALID_REQUEST),
+ "CreateTopics",
+ );
+ }
+ };
+
+ let mut results: Vec<CreateTopicResult> =
Vec::with_capacity(req.topics.len());
+ for topic in &req.topics {
+ results.push(create_one_topic(topic, api_version, req.validate_only,
bridge).await);
+ }
+ respond_or_close(
+ encode_create_topics_response(&req.topics, &results, api_version),
+ "CreateTopics",
+ )
+}
+
+/// Validates and (unless `validate_only`) creates one requested topic.
+///
+/// Sequential across a multi-topic request, not concurrent: `IggyBridge`
holds one lockstep
+/// `IggyClient` (`bridge::iggy_bridge`'s own doc - "the connection is
lockstep, one request in
+/// flight at a time"), so concurrent calls here would only queue behind each
other on the same
+/// mutex anyway; sequential keeps each topic's failure independent and easy
to reason about,
+/// with no ordering surprise from a scheduler interleaving them differently
across runs.
+async fn create_one_topic(
+ topic: &CreatableTopic,
+ api_version: i16,
+ validate_only: bool,
+ bridge: &dyn TopicCatalog,
+) -> CreateTopicResult {
+ if !topic.configs.is_empty() {
+ return Err(ERROR_INVALID_CONFIG);
+ }
+ // Checked here, not left to `ensure_stream_and_topic`'s own call: that
call is skipped
+ // entirely below for `validate_only`, and name validation is part of "can
this topic be
+ // created as specified," the exact thing `validate_only` promises to
check - skipping it
+ // would report success for a name (empty, padded, over Kafka's 249-byte
cap) that the real
+ // call always rejects.
+ validate_kafka_topic_name("kafka_topic", topic.name.0.as_str())
+ .map_err(|error| error.to_kafka_error_code())?;
+ let partition_count = validate_create_topic_shape(topic, api_version)?;
+
+ let kafka_topic = topic.name.0.as_str();
+ // Checked before the `validate_only` branch, not after: "can this topic
be created as
+ // specified" includes "does it already exist" - real Kafka's own
`validateOnly` still
+ // flags an existing topic, it doesn't only check format.
`get_kafka_topic` is read-only
+ // (the same lookup `handle_metadata` uses), so running it here doesn't
violate KIP-4's
+ // "don't create anything" promise even when `validate_only` is set.
+ //
+ // `ensure_stream_and_topic`'s own contract is intentionally idempotent
(get-or-create,
+ // matching-spec re-call is Ok) - correct for an internal "make sure this
exists" helper, but
+ // `CreateTopics` is not an upsert: the real `AdminClient.createTopics`
contract is
+ // `TOPIC_ALREADY_EXISTS` (36) for a topic that's already there even when
the requested spec
+ // matches exactly. The common "genuinely new topic" path still goes
straight through
+ // `ensure_stream_and_topic`'s own create-race handling unchanged below;
only "it's already
+ // there" short-circuits before ever reaching it.
+ match bridge.get_kafka_topic(kafka_topic).await {
+ Ok(Some(_existing)) => return Err(ERROR_TOPIC_ALREADY_EXISTS),
+ Ok(None) => {}
+ Err(error) => return Err(error.to_kafka_error_code()),
+ }
+
+ if validate_only {
+ // KIP-4: "check that the topics can be created as specified, but
don't create
+ // anything." Format and existence are already checked above; going
further (does
+ // ensure_topic's eventual PartitionCountMismatch also apply here)
would mean calling
+ // the bridge with create-on-miss semantics anyway, defeating the
"don't create
+ // anything" contract - so this reports success on format+existence
alone rather than
+ // fully simulating the real call.
+ return Ok(partition_count);
+ }
+
+ bridge
+ .ensure_stream_and_topic(kafka_topic, partition_count)
+ .await
+ .map(|()| partition_count)
+ .map_err(|error| error.to_kafka_error_code())
+}
+
+/// `ListOffsets` (`#3537`): resolves `earliest`/`latest` timestamp sentinels
(`-2`/`-1`) per
+/// requested partition through `bridge.high_watermarks`, one bridge call per
topic covering all
+/// its requested partitions at once.
+async fn handle_list_offsets(
+ api_version: i16,
+ body: Bytes,
+ max_frame_size: usize,
+ bridge: &dyn TopicCatalog,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_LIST_OFFSETS, api_version) {
+ return unsupported_version_response(API_KEY_LIST_OFFSETS, api_version,
|v| {
+ encode_list_offsets_error_response(v, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let req = match decode_guarded::<ListOffsetsRequest>(api_version, body,
|v, b| {
+ validate_list_offsets_shape(v, b, max_frame_size)
+ }) {
+ Ok(req) => req,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, "Failed to decode ListOffsets request");
+ return respond_or_close(
+ encode_list_offsets_error_response(api_version,
ERROR_INVALID_REQUEST),
+ "ListOffsets",
+ );
+ }
+ };
+
+ let mut results: Vec<Vec<ListOffsetsPartitionResult>> =
Vec::with_capacity(req.topics.len());
+ for topic in &req.topics {
+ results.push(resolve_list_offsets_topic(topic, bridge).await);
+ }
+ respond_or_close(
+ encode_list_offsets_response(api_version, &req, &results),
+ "ListOffsets",
+ )
+}
+
+/// Resolves every partition of one `ListOffsetsTopic`, in request order.
+///
+/// One `high_watermarks` call for the whole topic (covering every partition
this topic's
+/// request carries, valid or not), not one call per partition - matches
+/// `IggyBridge::high_watermarks`'s own one-round-trip batching contract. A
partition index that
+/// cannot even be a real Iggy partition (negative) is filtered out before the
bridge call rather
+/// than sent - `IggyBridge` takes `partitions: &[u32]`, so a negative index
has no wire
+/// representation to send anyway - and reported as
`UNKNOWN_TOPIC_OR_PARTITION` directly.
+async fn resolve_list_offsets_topic(
+ topic: &kafka_protocol::messages::list_offsets_request::ListOffsetsTopic,
+ bridge: &dyn TopicCatalog,
+) -> Vec<ListOffsetsPartitionResult> {
+ let kafka_topic = topic.name.0.as_str();
+ let valid_indices: Vec<u32> = topic
+ .partitions
+ .iter()
+ .filter_map(|p| u32::try_from(p.partition_index).ok())
+ .collect();
+
+ let watermarks = match bridge.high_watermarks(kafka_topic,
&valid_indices).await {
+ Ok(watermarks) => watermarks,
+ // Call-level failure (bad topic name, mapped stream doesn't exist,
bridge timeout) -
+ // every partition of this topic fails the same way, matching how a
real broker answers
+ // every partition of a topic it cannot see identically.
+ Err(error) => {
+ let error_code = error.to_kafka_error_code();
+ return topic.partitions.iter().map(|_| Err(error_code)).collect();
+ }
+ };
+ let watermarks: std::collections::HashMap<u32, std::result::Result<i64,
BridgeError>> =
+ watermarks.into_iter().collect();
+
+ topic
+ .partitions
+ .iter()
+ .map(|partition| {
+ let Ok(index) = u32::try_from(partition.partition_index) else {
+ return Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION);
+ };
+ match watermarks.get(&index) {
+ // -2 (earliest): every topic this bridge creates has
message_expiry left at
+ // ServerDefault (never-expire - see
IggyBridge::ensure_topic's own doc), and
+ // nothing in this gateway ever trims a partition yet, so the
earliest available
+ // offset is always 0 for a topic this bridge actually
manages. Real Kafka's
+ // own `timestamp` field for an earliest/latest sentinel query
is -1 regardless
+ // (it only carries a real value for an actual timestamp-based
lookup, which
+ // this bridge does not support - see the `_ =>` arm below).
+ Some(Ok(_watermark)) if partition.timestamp == -2 => Ok((-1,
0)),
Review Comment:
**EARLIEST returns a hardcoded 0 for topics this gateway did not create.**
The justifying comment is about topics *the bridge creates* having
`ServerDefault` expiry, but this path serves everything `list_kafka_topics`
returns, including topics created outside the gateway with real retention. Once
the segment cleaner trims those, 0 names a log-start offset that no longer
exists and a consumer with `auto.offset.reset=earliest` seeks into a hole.
Bounded today only because Fetch is still a stub.
Suggest exposing the real start offset on the bridge, or stating the
constraint as a precondition.
##########
gateways/kafka/tests/gateway_bridge_e2e_tests.rs:
##########
@@ -0,0 +1,292 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Full-stack `KafkaGateway` + real `IggyBridge` + real `iggy-server` tests,
over the actual
+//! Kafka wire (raw TCP frames via `tests/common/tcp.rs`) - the one
combination no other suite in
+//! this crate exercises. `bridge_iggy_integration_tests.rs` calls
`IggyBridge` methods directly
+//! (never through `KafkaGateway`/the wire protocol);
`api_handler_tests.rs`/`server_e2e_tests.rs`
+//! run `KafkaGateway` over the wire, but only against `FakeBridge` (never a
real Iggy backend).
+//! This file is the only place a real Kafka client's exact bytes, decoded by
the real protocol
+//! layer, reach real `IggyBridge` provisioning/lookup logic, and the real
response bytes that
+//! come back are decoded again on the way out.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/server.rs"]
+mod server;
+#[path = "common/tcp.rs"]
+mod tcp;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::sync::Arc;
+
+use iggy::prelude::{Identifier, IggyMessage, MessageClient, Partitioning,
TopicClient};
+use serial_test::serial;
+
+use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::protocol::api::{
+ API_KEY_CREATE_TOPICS, API_KEY_LIST_OFFSETS, API_KEY_METADATA, ERROR_NONE,
+ ERROR_TOPIC_ALREADY_EXISTS,
+};
+
+use codec::Decoder;
+use iggy_server::{TestServer, raw_client};
+use server::spawn_test_server;
+use tcp::round_trip;
+
+/// Connects a real `IggyBridge` to `server` and spawns a `KafkaGateway` in
front of it -
+/// the one combination this whole file exists to exercise.
+async fn spawn_gateway_over_real_bridge(
+ server: &TestServer,
+) -> (std::net::SocketAddr, tokio::sync::broadcast::Sender<()>) {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ spawn_test_server(Arc::new(bridge)).await
+}
+
+/// `CreateTopics` then `Metadata` for the same topic, both over the real wire
against a real
+/// Iggy backend: the topic `CreateTopics` provisions must be the exact one
`Metadata` reports
+/// back, partition count included - the two handlers going through the same
real bridge state
+/// is the thing no `FakeBridge`-backed test can prove.
+#[tokio::test]
+#[serial]
+async fn e2e_create_topics_then_metadata_round_trip() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (addr, _shutdown) = spawn_gateway_over_real_bridge(&server).await;
+
+ let create_body = wire::build_create_topics_simple_request(5, "orders", 3,
1, false);
+ let (_corr, create_resp) = round_trip(addr, API_KEY_CREATE_TOPICS, 5, 1,
&create_body).await;
+ let mut d = Decoder::new(create_resp);
+ let _throttle = d.read_i32().unwrap();
+ let _topics = d.read_varint().unwrap();
+ let _topic = d.read_compact_nullable_string().unwrap();
+ assert_eq!(
+ d.read_i16().unwrap(),
+ ERROR_NONE,
+ "CreateTopics must succeed"
+ );
+
+ let metadata_body = wire::build_metadata_flexible_request(&["orders"]);
+ let (_corr, metadata_resp) = round_trip(addr, API_KEY_METADATA, 9, 2,
&metadata_body).await;
+ let mut d = Decoder::new(metadata_resp);
+ let _throttle = d.read_i32().unwrap();
+ let brokers = d.read_varint().unwrap();
+ assert_eq!(brokers, 2, "one broker"); // N+1
+ let _node_id = d.read_i32().unwrap();
+ let _host = d.read_compact_nullable_string().unwrap();
+ let _port = d.read_i32().unwrap();
+ let _rack = d.read_compact_nullable_string().unwrap();
+ d.read_tagged_fields().unwrap(); // broker tagged fields
+ let _cluster_id = d.read_compact_nullable_string().unwrap();
+ let _controller_id = d.read_i32().unwrap();
+ let topics = d.read_varint().unwrap();
+ assert_eq!(topics, 2, "one topic"); // N+1
+ let error_code = d.read_i16().unwrap();
+ assert_eq!(
+ error_code, ERROR_NONE,
+ "Metadata must find the topic CreateTopics just provisioned"
+ );
+ let name = d.read_compact_nullable_string().unwrap();
+ assert_eq!(name, Some("orders".to_string()));
+ let _internal = d.read_bool().unwrap();
+ let partitions = d.read_varint().unwrap();
+ assert_eq!(
+ partitions, 4,
+ "must report the 3 partitions CreateTopics actually provisioned" // N+1
+ );
+}
+
+/// Wire-level proof of the `CreateTopics` idempotent-success fix: re-sending
the exact request
+/// an earlier call already satisfied must answer `TOPIC_ALREADY_EXISTS` (36),
not silently
+/// succeed a second time - through the real protocol decode/bridge/encode
path, not `FakeBridge`.
+#[tokio::test]
+#[serial]
+async fn e2e_create_topics_recreate_returns_topic_already_exists() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (addr, _shutdown) = spawn_gateway_over_real_bridge(&server).await;
+
+ let body = wire::build_create_topics_simple_request(5, "orders", 2, 1,
false);
+
+ let (_corr, first) = round_trip(addr, API_KEY_CREATE_TOPICS, 5, 1,
&body).await;
+ let mut d = Decoder::new(first);
+ let _throttle = d.read_i32().unwrap();
+ let _topics = d.read_varint().unwrap();
+ let _topic = d.read_compact_nullable_string().unwrap();
+ assert_eq!(
+ d.read_i16().unwrap(),
+ ERROR_NONE,
+ "first create must succeed"
+ );
+
+ let (_corr, second) = round_trip(addr, API_KEY_CREATE_TOPICS, 5, 2,
&body).await;
+ let mut d = Decoder::new(second);
+ let _throttle = d.read_i32().unwrap();
+ let _topics = d.read_varint().unwrap();
+ let _topic = d.read_compact_nullable_string().unwrap();
+ assert_eq!(
+ d.read_i16().unwrap(),
+ ERROR_TOPIC_ALREADY_EXISTS,
+ "recreating an existing topic over the real wire must not silently
succeed again"
+ );
+}
+
+/// Wire-level proof that `num_partitions = -1` with a manual `assignments`
list resolves the
+/// real partition count from the assignment length, not
`DEFAULT_PARTITION_COUNT` - verified
+/// against the actual Iggy topic `CreateTopics` provisioned, not just the
response bytes.
+#[tokio::test]
+#[serial]
+async fn
e2e_create_topics_with_manual_assignment_creates_the_real_iggy_partition_count()
{
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (addr, _shutdown) = spawn_gateway_over_real_bridge(&server).await;
+
+ let body = wire::build_create_topics_with_assignments_request(5, "orders",
4);
+ let (_corr, resp) = round_trip(addr, API_KEY_CREATE_TOPICS, 5, 1,
&body).await;
+ let mut d = Decoder::new(resp);
+ let _throttle = d.read_i32().unwrap();
+ let _topics = d.read_varint().unwrap();
+ let _topic = d.read_compact_nullable_string().unwrap();
+ assert_eq!(
+ d.read_i16().unwrap(),
+ ERROR_NONE,
+ "create with assignments must succeed"
+ );
+ let _error_message = d.read_compact_nullable_string().unwrap();
+ // `topic_config_error_code` is not a plain sequential field -
`kafka_protocol`'s own encoder
+ // (create_topics_response.rs) only ever writes it inside the
tagged-fields section, and only
+ // when non-zero; this bridge's response never sets it, so it never
appears on the wire here.
+ let num_partitions = d.read_i32().unwrap();
+ assert_eq!(
+ num_partitions, 4,
+ "response must echo the 4 partitions implied by the assignment list"
+ );
+
+ let raw = raw_client(&server).await;
+ let topic = raw
+ .get_topic(
+ &Identifier::named("kafka").expect("valid stream name"),
+ &Identifier::named("orders").expect("valid topic name"),
+ )
+ .await
+ .expect("get_topic call")
+ .expect("topic must exist on the real Iggy backend");
+ assert_eq!(
+ topic.partitions_count, 4,
+ "the real Iggy topic must actually have 4 partitions, not the
1-partition broker default"
+ );
+}
+
+/// Wire-level proof of the Metadata null-vs-empty-array fix: an explicit
empty `topics` array
+/// (KIP-4's `describeCluster()` shape) must list no topics even when real
topics exist on the
+/// backend, while a null array still lists all of them - distinguishable only
against a backend
+/// that actually has a topic to (not) list.
+#[tokio::test]
+#[serial]
+async fn e2e_metadata_empty_array_lists_no_topics_while_null_lists_all() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ bridge
+ .ensure_stream_and_topic("orders", 1)
+ .await
+ .expect("seed a real topic so empty-vs-null is actually
distinguishable");
+ let (addr, _shutdown) = spawn_test_server(Arc::new(bridge)).await;
+
+ let empty_body = wire::build_metadata_legacy_request_for_version(0, &[]);
Review Comment:
**This test pins the v0 divergence rather than catching it.**
The seeded topic above is deliberate ("so empty-vs-null is actually
distinguishable"), but the request is built at **version 0**, where Kafka
treats an empty array as *all topics*. A real broker returns `orders` here; the
assertion locks in returning nothing.
Suggest moving this case to v1+ and adding a v0 case that asserts the
all-topics behaviour.
##########
gateways/kafka/src/bridge/iggy_bridge.rs:
##########
@@ -526,4 +539,97 @@ impl IggyBridge {
);
watermark
}
+
+ /// Looks up `kafka_topic` without creating it. Unlike
[`Self::ensure_stream_and_topic`],
+ /// Metadata is read-only discovery - a client asking "does this topic
exist" must not have
+ /// the asking itself create the topic.
+ ///
+ /// # 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
failures.
+ pub async fn get_kafka_topic(
+ &self,
+ kafka_topic: &str,
+ ) -> Result<Option<KafkaTopicMetadata>, 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)?;
+ if with_request_timeout(self.client.get_stream(&stream_id))
Review Comment:
**The `get_stream` probe is redundant and doubles round trips.**
`get_topic` already returns `Ok(None)` when the stream is missing, and
`get_kafka_topic` collapses both misses to `Ok(None)` anyway - the comment on
`high_watermarks` states this same fact. Dropping it halves the cost of the two
hottest control-plane paths: every Metadata name lookup and every CreateTopics
existence check.
Suggest deleting the probe and keeping `get_topic` alone.
##########
gateways/kafka/README.md:
##########
@@ -1,8 +1,15 @@
# Kafka gateway (`iggy-gateway-kafka`)
-Foundation layer for
[apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener
on the Kafka wire port that decodes requests, validates scoped API keys and
versions, and returns stub responses.
+Foundation layer for
[apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener
on the Kafka wire port that decodes requests, validates scoped API keys and
versions, and dispatches to real or stub handlers depending on the API.
-> **Stub warning:** no API persists or reads real data yet. Produce, Fetch,
and ListOffsets return retriable `NOT_LEADER_OR_FOLLOWER` (6) so clients keep
data locally / retry elsewhere instead of trusting a fake success. CreateTopics
does **not** create topics; valid requests return `NOT_CONTROLLER` (41).
Metadata still reports requested topics as unknown. Persistence lands with the
Iggy bridge (see [docs/SCOPE.md](docs/SCOPE.md)).
+> **Partial stub warning:** Produce and Fetch still discard/never read data -
both return retriable
+> `NOT_LEADER_OR_FOLLOWER` (6) so clients keep data locally / retry elsewhere
instead of trusting a
+> fake success, until [#3535](https://github.com/apache/iggy/issues/3535)/
+> [#3536](https://github.com/apache/iggy/issues/3536) land. Metadata
+> ([#3534](https://github.com/apache/iggy/issues/3534)), CreateTopics
+> ([#3538](https://github.com/apache/iggy/issues/3538)) and ListOffsets
+> ([#3537](https://github.com/apache/iggy/issues/3537)) are real: they call
through to a real Iggy
+> backend via `IggyBridge` (see [docs/SCOPE.md](docs/SCOPE.md)).
## Run
Review Comment:
**Run section is now incorrect.**
`main.rs` connects to Iggy before binding, so a bare `cargo run -p
iggy-gateway-kafka` exits immediately with `IGGY_KAFKA_IGGY_PASSWORD must be
set`. The env table below also omits every `IGGY_KAFKA_IGGY_*` variable.
`MANUAL_TESTING.md` received this fix in the same PR; the README did not.
Suggest adding the running-`iggy-server` prerequisite and cross-linking the
connection config.
##########
gateways/kafka/src/bridge/error.rs:
##########
@@ -163,7 +174,13 @@ const fn iggy_error_to_kafka_code(err: &IggyError) -> i16 {
| IggyError::TcpError
| IggyError::TransientNotAccepted => ERROR_NOT_LEADER_OR_FOLLOWER,
IggyError::TransientNotCommitted => ERROR_REQUEST_TIMED_OUT,
- IggyError::TooManyPartitions => ERROR_INVALID_PARTITIONS,
+ // Not `ERROR_INVALID_PARTITIONS` (37): that code's own text, per
`kafka-protocol`'s
+ // table, is "Number of partitions is below 1" - the opposite
condition from "too many"
+ // (Iggy's server-side cap, above 1000). Reusing 37 for both
directions would return a
+ // client-visible error message that contradicts the actual request it
sent.
+ // `ERROR_INVALID_REQUEST` (42) has no such text mismatch and matches
this bridge's own
+ // convention elsewhere for a request-shape problem with no exact
Kafka analogue.
+ IggyError::TooManyPartitions => ERROR_INVALID_REQUEST,
_ => ERROR_UNKNOWN_SERVER_ERROR,
Review Comment:
**`RequestAlreadyApplied` reports a committed operation as a permanent
fault.**
Code 59 means the operation *did* commit. It falls into the catch-all here
and becomes `UNKNOWN_SERVER_ERROR` (-1), which Java clients treat as
non-retriable, so a `CreateTopics` replayed by the SDK's reconnect path reports
-1 for a topic that now exists. Wrong direction: success surfaced as a server
fault.
Suggest mapping it to `ERROR_NONE` for CreateTopics, or at minimum to a
retriable code.
##########
gateways/kafka/tests/gateway_bridge_e2e_tests.rs:
##########
@@ -0,0 +1,292 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Full-stack `KafkaGateway` + real `IggyBridge` + real `iggy-server` tests,
over the actual
+//! Kafka wire (raw TCP frames via `tests/common/tcp.rs`) - the one
combination no other suite in
+//! this crate exercises. `bridge_iggy_integration_tests.rs` calls
`IggyBridge` methods directly
+//! (never through `KafkaGateway`/the wire protocol);
`api_handler_tests.rs`/`server_e2e_tests.rs`
+//! run `KafkaGateway` over the wire, but only against `FakeBridge` (never a
real Iggy backend).
+//! This file is the only place a real Kafka client's exact bytes, decoded by
the real protocol
+//! layer, reach real `IggyBridge` provisioning/lookup logic, and the real
response bytes that
+//! come back are decoded again on the way out.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/server.rs"]
+mod server;
+#[path = "common/tcp.rs"]
+mod tcp;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::sync::Arc;
+
+use iggy::prelude::{Identifier, IggyMessage, MessageClient, Partitioning,
TopicClient};
+use serial_test::serial;
+
+use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::protocol::api::{
+ API_KEY_CREATE_TOPICS, API_KEY_LIST_OFFSETS, API_KEY_METADATA, ERROR_NONE,
+ ERROR_TOPIC_ALREADY_EXISTS,
+};
+
+use codec::Decoder;
+use iggy_server::{TestServer, raw_client};
+use server::spawn_test_server;
+use tcp::round_trip;
+
+/// Connects a real `IggyBridge` to `server` and spawns a `KafkaGateway` in
front of it -
+/// the one combination this whole file exists to exercise.
+async fn spawn_gateway_over_real_bridge(
+ server: &TestServer,
+) -> (std::net::SocketAddr, tokio::sync::broadcast::Sender<()>) {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ spawn_test_server(Arc::new(bridge)).await
+}
+
+/// `CreateTopics` then `Metadata` for the same topic, both over the real wire
against a real
+/// Iggy backend: the topic `CreateTopics` provisions must be the exact one
`Metadata` reports
+/// back, partition count included - the two handlers going through the same
real bridge state
+/// is the thing no `FakeBridge`-backed test can prove.
+#[tokio::test]
+#[serial]
+async fn e2e_create_topics_then_metadata_round_trip() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ let (addr, _shutdown) = spawn_gateway_over_real_bridge(&server).await;
+
+ let create_body = wire::build_create_topics_simple_request(5, "orders", 3,
1, false);
+ let (_corr, create_resp) = round_trip(addr, API_KEY_CREATE_TOPICS, 5, 1,
&create_body).await;
+ let mut d = Decoder::new(create_resp);
+ let _throttle = d.read_i32().unwrap();
+ let _topics = d.read_varint().unwrap();
+ let _topic = d.read_compact_nullable_string().unwrap();
+ assert_eq!(
+ d.read_i16().unwrap(),
+ ERROR_NONE,
+ "CreateTopics must succeed"
+ );
+
+ let metadata_body = wire::build_metadata_flexible_request(&["orders"]);
+ let (_corr, metadata_resp) = round_trip(addr, API_KEY_METADATA, 9, 2,
&metadata_body).await;
+ let mut d = Decoder::new(metadata_resp);
+ let _throttle = d.read_i32().unwrap();
+ let brokers = d.read_varint().unwrap();
+ assert_eq!(brokers, 2, "one broker"); // N+1
+ let _node_id = d.read_i32().unwrap();
+ let _host = d.read_compact_nullable_string().unwrap();
+ let _port = d.read_i32().unwrap();
+ let _rack = d.read_compact_nullable_string().unwrap();
+ d.read_tagged_fields().unwrap(); // broker tagged fields
+ let _cluster_id = d.read_compact_nullable_string().unwrap();
+ let _controller_id = d.read_i32().unwrap();
+ let topics = d.read_varint().unwrap();
+ assert_eq!(topics, 2, "one topic"); // N+1
+ let error_code = d.read_i16().unwrap();
+ assert_eq!(
+ error_code, ERROR_NONE,
+ "Metadata must find the topic CreateTopics just provisioned"
+ );
+ let name = d.read_compact_nullable_string().unwrap();
+ assert_eq!(name, Some("orders".to_string()));
+ let _internal = d.read_bool().unwrap();
+ let partitions = d.read_varint().unwrap();
Review Comment:
**The per-partition record has no assertion coverage.**
This is the only test that gets a populated partitions array, and it asserts
the count then returns without reading a single partition field. Every other
Metadata test decodes an empty array. librdkafka and the Java client pick the
produce/fetch broker from `leader_id` and the replica/ISR arrays, so a
regression emitting `leader_id = -1` or empty `replica_nodes` passes the whole
suite while making the topic look leaderless to real clients.
Suggest decoding one full partition record here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
