numinnex commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3572071392
########## gateways/kafka/tests/review_regression_tests.rs: ########## @@ -0,0 +1,450 @@ +// 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. + +//! Regression tests for PR #3519 review findings (atharvalade, Jul 2026). +//! +//! These encode Kafka-client-correct behavior. Several fail until the +//! corresponding protocol/server fixes land. Review Comment: This says several tests fail until the fixes land, but the fixes are part of this same PR, so the comment reads as if the suite merges red. If it is stale, please drop it; if any test here still fails, that needs resolving before merge. ########## gateways/kafka/tests/common/fixtures.rs: ########## @@ -0,0 +1,54 @@ +// 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. + +//! Fixture loaders — compiled into each integration test binary via `#[path]`. +#![allow(dead_code)] + +use std::path::PathBuf; + +use bytes::Bytes; + +use iggy_gateway_kafka::protocol::codec::Decoder; +use iggy_gateway_kafka::protocol::header::{RequestHeader, request_header_version}; + +pub fn fixtures_dir() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tools/kafka-tool/kafka_messages") +} + +pub fn fixture_exists(api_key: i16, api_name: &str, version: i16) -> bool { + let filename = format!("{api_key:03}_{api_name}_v{version}.bin"); + fixtures_dir().join(filename).is_file() +} + +/// Load request body bytes from a kafka-tool `.bin` fixture (skips frame header). +pub fn load_fixture_body(api_key: i16, api_name: &str, version: i16) -> Bytes { + let filename = format!("{api_key:03}_{api_name}_v{version}.bin"); + let path = fixtures_dir().join(&filename); + let data = std::fs::read(&path).unwrap_or_else(|e| panic!("failed to read {filename}: {e}")); Review Comment: On a fresh clone, `cargo test -p iggy-gateway-kafka` panics here for the whole `decode_validation_tests` suite, two firewall tests, and one e2e test, because `kafka_messages/` is gitignored and only CI runs `ci-wire-fixtures.sh generate`. Meanwhile `handler_regression_tests` guards with `fixture_exists()` and silently passes with zero assertions, and `scope_coverage_tests` falls back to hand-built bodies. Three different policies for the same missing-fixture condition. Suggest unifying on one: either skip with an explicit message pointing at the generation script, or generate fixtures from a build step so plain `cargo test` works for contributors. ########## .github/actions/rust/pre-merge/action.yml: ########## @@ -269,6 +269,20 @@ runs: compile_duration=$((compile_end - compile_start)) echo "::notice::Tests compiled in ${compile_duration}s ($(date -ud @${compile_duration} +'%M:%S'))" + # decode_validation_tests need gitignored wire fixtures. Generate when + # iggy-gateway-kafka is in the DAG test scope (rust-gateway job or parent + # rust job — both run gateway tests when gateways/** changes). + NEEDS_KAFKA_FIXTURES=false + if [[ -z "$NEXTEST_FILTER" ]]; then Review Comment: An empty `NEXTEST_FILTER` means every full-workspace test run generates the fixtures, building `kafka-message-gen` plus the whole `kafka-protocol` dependency tree even when `gateways/**` is untouched. Worth also keying on the changed-paths signal the component DAG already computes, if it is available at this point. ########## gateways/kafka/src/protocol/api.rs: ########## @@ -0,0 +1,459 @@ +// 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. + +use bytes::Bytes; + +use crate::error::{KafkaProtocolError, Result}; +use crate::protocol::codec::{Decoder, Encoder}; +use crate::protocol::requests::{ + ProduceDecodeResult, decode_create_topics_request, decode_fetch_request, + decode_list_offsets_request, decode_produce_request, +}; +use crate::protocol::responses::{ + encode_create_topics_error_response, encode_create_topics_response, + encode_fetch_error_response, encode_fetch_response, encode_list_offsets_error_response, + encode_list_offsets_response, encode_produce_error_response, encode_produce_response, +}; + +pub const API_KEY_PRODUCE: i16 = 0; +pub const API_KEY_FETCH: i16 = 1; +pub const API_KEY_LIST_OFFSETS: i16 = 2; +pub const API_KEY_METADATA: i16 = 3; +pub const API_KEY_API_VERSIONS: i16 = 18; +pub const API_KEY_CREATE_TOPICS: i16 = 19; + +pub const DEFAULT_KAFKA_PORT: u16 = 9093; + +pub const ERROR_NONE: i16 = 0; +pub const ERROR_UNKNOWN_TOPIC_OR_PARTITION: i16 = 3; +pub const ERROR_UNSUPPORTED_VERSION: i16 = 35; +pub const ERROR_INVALID_PARTITIONS: i16 = 37; +pub const ERROR_INVALID_REQUEST: i16 = 42; + +const MAX_SUPPORTED_METADATA_VERSION: i16 = 9; + +/// Sentinel for `topic_authorized_operations` / `cluster_authorized_operations` when ACLs are not supported. +const AUTHORIZED_OPS_UNKNOWN: i32 = i32::MIN; + +#[derive(Debug, Clone)] +pub struct BrokerAdvertise { + pub host: String, + pub port: i32, +} + +impl Default for BrokerAdvertise { + fn default() -> Self { + Self { + host: "127.0.0.1".to_string(), + port: i32::from(DEFAULT_KAFKA_PORT), + } + } +} + +#[derive(Debug, Clone, Copy)] +pub struct ApiVersionRange { + pub api_key: i16, + pub min_version: i16, + pub max_version: i16, +} + +static SUPPORTED_RANGES: &[ApiVersionRange] = &[ + ApiVersionRange { + api_key: API_KEY_PRODUCE, + min_version: 3, + max_version: 9, + }, + ApiVersionRange { + api_key: API_KEY_FETCH, + min_version: 4, + max_version: 12, + }, + ApiVersionRange { + api_key: API_KEY_LIST_OFFSETS, + min_version: 1, + max_version: 6, + }, + ApiVersionRange { + api_key: API_KEY_METADATA, + min_version: 0, + max_version: 9, + }, + ApiVersionRange { + api_key: API_KEY_API_VERSIONS, + min_version: 0, + max_version: 3, + }, + ApiVersionRange { + api_key: API_KEY_CREATE_TOPICS, + min_version: 2, + max_version: 5, + }, +]; + +#[must_use] +pub fn supported_api_ranges() -> &'static [ApiVersionRange] { + SUPPORTED_RANGES +} + +/// Handles one decoded request frame and returns the response to write back, +/// or `None` when the wire protocol forbids a response (Produce with `acks=0`). +pub fn handle_request( + api_key: i16, + api_version: i16, + body: Bytes, + broker: &BrokerAdvertise, +) -> Option<Bytes> { + if api_key == API_KEY_PRODUCE { + return handle_produce_request(api_version, body); + } + Some(handle_other_request(api_key, api_version, body, broker)) +} + +/// Produce is the only request the wire protocol allows to go unanswered +/// (`acks=0`), so it gets its own `Option`-returning path. +fn handle_produce_request(api_version: i16, body: Bytes) -> Option<Bytes> { + if !is_supported_version(API_KEY_PRODUCE, api_version) { + return Some(encode_produce_error_response( + api_version, + ERROR_UNSUPPORTED_VERSION, + )); + } + match decode_produce_request(api_version, body) { + // acks=0 is fire-and-forget: the client isn't reading a response, so + // sending one desyncs the next correlation id it expects. + ProduceDecodeResult::Ok(req) if req.acks == 0 => None, + ProduceDecodeResult::Ok(req) => Some(encode_produce_response(api_version, &req)), + ProduceDecodeResult::Err { + acks: Some(0), + error, + } => { + tracing::warn!( + "Failed to decode Produce request with acks=0 (no response): {:?}", + error + ); + None + } + ProduceDecodeResult::Err { error, .. } => { + tracing::warn!("Failed to decode Produce request: {:?}", error); + Some(encode_produce_error_response( + api_version, + ERROR_INVALID_REQUEST, + )) + } + } +} + +fn handle_other_request( + api_key: i16, + api_version: i16, + body: Bytes, + broker: &BrokerAdvertise, +) -> Bytes { + match api_key { + API_KEY_API_VERSIONS => { + if is_supported_version(api_key, api_version) { + encode_api_versions_response(api_version, ERROR_NONE) + } else { + // KIP-511: reply with v0 when the requested version is not understood. + encode_api_versions_response(0, ERROR_UNSUPPORTED_VERSION) + } + } + API_KEY_METADATA => { + if is_supported_version(api_key, api_version) { + encode_metadata_response(api_version, api_version, body, broker, ERROR_NONE) + } else { + let response_version = api_version.clamp(0, MAX_SUPPORTED_METADATA_VERSION); Review Comment: nit: for a Metadata version above the supported max (say v13), the body is encoded at the clamped v9 while the client parses at v13, so the `UNSUPPORTED_VERSION` signal never survives the parse. Version-negotiating clients cap at the advertised v9 and never hit this, so it only affects clients that skip ApiVersions; closing the connection may be the more honest contract, consistent with the unknown-api-key discussion. ########## gateways/kafka/src/protocol/responses.rs: ########## @@ -0,0 +1,370 @@ +// 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. + +//! Kafka response encoders (stub implementations). + +#![allow(clippy::pedantic)] + +use crate::protocol::api::{ERROR_INVALID_PARTITIONS, ERROR_NONE}; +use crate::protocol::codec::Encoder; +use crate::protocol::requests::{ + CreateTopicsRequest, FetchRequest, ListOffsetsRequest, ProducePartitionData, ProduceRequest, + ProduceTopicData, +}; +use bytes::Bytes; + +/// Well-formed Produce response with a single placeholder topic/partition. +pub fn encode_produce_error_response(version: i16, error_code: i16) -> Bytes { + let topics = vec![ProduceTopicData { + topic: String::new(), // TODO topic name will be populated in the end to end functional completion + partitions: vec![ProducePartitionData { + partition: 0, + records: None, + }], + }]; + encode_produce_response_inner(version, &topics, error_code) +} + +pub fn encode_produce_response(version: i16, req: &ProduceRequest) -> Bytes { Review Comment: Produce acks with `error_code=0, base_offset=0`, so any real producer pointed at the gateway treats its data as durably written while the gateway discards it. Understood this is the stub phase, but from the client's perspective this failure mode is silent data loss. Two suggestions: a prominent warning in the README ("returns success without persisting"), and consider returning a retriable error instead of fake success until the bridge lands, so default client retry behavior keeps the data on the client side. Same concern for CreateTopics answering success while a subsequent Metadata reports the topic as unknown. ########## gateways/kafka/tests/common/tcp.rs: ########## @@ -0,0 +1,165 @@ +// 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. + +//! TCP round-trip helpers — compiled into each integration test binary via `#[path]`. +#![allow(dead_code)] + +use std::net::SocketAddr; +use std::time::Duration; + +use bytes::{BufMut, Bytes, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time; + +use iggy_gateway_kafka::protocol::codec::Decoder; +use iggy_gateway_kafka::protocol::header::{request_header_version, response_header_version}; + +/// Build a complete length-prefixed Kafka request frame (header + body). +pub fn build_request_frame( + api_key: i16, + api_version: i16, + correlation_id: i32, + client_id: Option<&str>, + body: &[u8], +) -> Bytes { + let hdr_ver = request_header_version(api_key, api_version); + let mut enc = iggy_gateway_kafka::protocol::codec::Encoder::with_capacity(64 + body.len()); + enc.write_i16(api_key); + enc.write_i16(api_version); + enc.write_i32(correlation_id); + if hdr_ver >= 2 { + enc.write_compact_nullable_string(client_id); + enc.write_empty_tagged_fields(); + } else { + enc.write_nullable_string(client_id) + .expect("test client_id fits i16"); + } + enc.write_bytes(body); + + let payload = enc.freeze(); + let payload_len = i32::try_from(payload.len()).expect("test payload fits i32"); + let mut frame = BytesMut::with_capacity(4 + payload.len()); + frame.put_i32(payload_len); + frame.extend_from_slice(&payload); + frame.freeze() +} + +/// Parse correlation id and response body from a raw response payload (no length prefix). +pub fn parse_response_payload(api_key: i16, api_version: i16, payload: Bytes) -> (i32, Bytes) { + let resp_hdr_ver = response_header_version(api_key, api_version); + let mut d = Decoder::new(payload); + let correlation_id = d.read_i32().expect("correlation_id"); + if resp_hdr_ver >= 1 { + d.read_tagged_fields().expect("response tagged fields"); + } + let body = d.read_bytes(d.remaining()).expect("response body"); + (correlation_id, body) +} + +/// Read one length-prefixed response frame from the stream. +pub async fn read_response_frame(stream: &mut TcpStream, max_size: usize) -> Bytes { + let mut len_buf = [0u8; 4]; + stream + .read_exact(&mut len_buf) + .await + .expect("response length prefix"); + let frame_len_i32 = i32::from_be_bytes(len_buf); + assert!(frame_len_i32 > 0, "response frame length must be positive"); + let frame_len = usize::try_from(frame_len_i32).expect("positive i32 frame length fits usize"); + assert!( + frame_len <= max_size, + "response frame too large: {frame_len}" + ); + let mut buf = vec![0u8; frame_len]; + stream.read_exact(&mut buf).await.expect("response body"); + Bytes::from(buf) +} + +/// Minimal Produce v3 body: nullable `transactional_id`, acks, timeout, empty topics array. +pub fn build_produce_v3_body(acks: i16, topics_count: i32) -> Bytes { + let mut body = BytesMut::new(); + body.put_i16(-1); // null transactional_id + body.put_i16(acks); + body.put_i32(1_000); // timeout_ms + body.put_i32(topics_count); + body.freeze() +} + +/// Legacy Metadata request body listing topic names (non-flexible, v0–v8). +pub fn build_metadata_legacy_request(topic_names: &[&str]) -> Bytes { + let mut body = BytesMut::new(); + body.put_i32(i32::try_from(topic_names.len()).expect("topic name count fits i32")); + for name in topic_names { + let name_bytes = name.as_bytes(); + let len = i16::try_from(name_bytes.len()).expect("topic name fits i16"); + body.put_i16(len); + body.extend_from_slice(name_bytes); + } + body.freeze() +} + +/// Read one length-prefixed response frame, returning `None` on timeout. +pub async fn read_response_frame_with_timeout( + stream: &mut TcpStream, + max_size: usize, + timeout: Duration, +) -> Option<Bytes> { + time::timeout(timeout, read_response_frame(stream, max_size)) + .await + .ok() +} + +/// Concatenate multiple length-prefixed frames (for pipelining tests). +pub fn concat_frames(frames: &[Bytes]) -> Bytes { + let total: usize = frames.iter().map(Bytes::len).sum(); + let mut out = BytesMut::with_capacity(total); + for frame in frames { + out.extend_from_slice(frame); + } + out.freeze() +} + +/// Read one byte or return `None` on EOF / timeout. +pub async fn read_byte_with_timeout(stream: &mut TcpStream, timeout: Duration) -> Option<u8> { Review Comment: `Ok(Ok(_)) => Some(..), _ => None` conflates EOF with timeout. Every "server must close the connection" test built on this helper also passes when the server merely stalls for the timeout window without closing, so the close semantics are never actually verified. Suggest a three-state return (byte / closed / timeout) with the close tests asserting closed explicitly. ########## gateways/kafka/Cargo.toml: ########## @@ -0,0 +1,59 @@ +# 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. + +[package] +name = "iggy-gateway-kafka" +version = "0.1.0" +description = "Kafka wire protocol gateway foundation for Apache Iggy" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "kafka", "gateway", "streaming"] +homepage = "https://iggy.apache.org"; +documentation = "https://iggy.apache.org/docs"; +repository = "https://github.com/apache/iggy"; +readme = "README.md" +publish = false + +[[bin]] +name = "iggy-gateway-kafka" +path = "src/main.rs" + +[dependencies] +bytes = { workspace = true } +socket2 = "0.6.4" Review Comment: nit: `socket2` is already a workspace dependency at the root; use `{ workspace = true }` to avoid version drift. ########## gateways/kafka/src/protocol/responses.rs: ########## @@ -0,0 +1,370 @@ +// 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. + +//! Kafka response encoders (stub implementations). + +#![allow(clippy::pedantic)] + +use crate::protocol::api::{ERROR_INVALID_PARTITIONS, ERROR_NONE}; +use crate::protocol::codec::Encoder; +use crate::protocol::requests::{ + CreateTopicsRequest, FetchRequest, ListOffsetsRequest, ProducePartitionData, ProduceRequest, + ProduceTopicData, +}; +use bytes::Bytes; + +/// Well-formed Produce response with a single placeholder topic/partition. +pub fn encode_produce_error_response(version: i16, error_code: i16) -> Bytes { + let topics = vec![ProduceTopicData { + topic: String::new(), // TODO topic name will be populated in the end to end functional completion + partitions: vec![ProducePartitionData { + partition: 0, + records: None, + }], + }]; + encode_produce_response_inner(version, &topics, error_code) +} + +pub fn encode_produce_response(version: i16, req: &ProduceRequest) -> Bytes { + encode_produce_response_inner(version, &req.topics, ERROR_NONE) +} + +fn encode_produce_response_inner( + version: i16, + topics: &[ProduceTopicData], + partition_error: i16, +) -> Bytes { + let flexible = version >= 9; + let mut e = Encoder::with_capacity(512); + + if flexible { + e.write_varint((topics.len() + 1) as u64); + } else { + e.write_i32(i32::try_from(topics.len()).expect("topic count bounded")); + } + + for topic in topics { + if flexible { + e.write_compact_nullable_string(Some(&topic.topic)); + } else { + e.write_nullable_string_unchecked(Some(&topic.topic)); + } + + if flexible { + e.write_varint((topic.partitions.len() + 1) as u64); + } else { + e.write_i32(i32::try_from(topic.partitions.len()).expect("partition count bounded")); + } + + for p in &topic.partitions { + e.write_i32(p.partition); + e.write_i16(partition_error); + e.write_i64(0); + if version >= 2 { + e.write_i64(-1); + } + if version >= 5 { + e.write_i64(0); + } + if version >= 8 { + if flexible { + e.write_varint(1); + e.write_compact_nullable_string(None); + } else { + e.write_i32(0); + e.write_nullable_string_unchecked(None); + } + } + if flexible { + e.write_empty_tagged_fields(); + } + } + + if flexible { + e.write_empty_tagged_fields(); + } + } + + if version >= 1 { + e.write_i32(0); + } + if flexible { + e.write_empty_tagged_fields(); + } + + e.freeze() +} + +/// Well-formed Fetch response. Uses top-level `error_code` at v7+, or a single +/// placeholder topic/partition with per-partition `error_code` below v7. +pub fn encode_fetch_error_response(version: i16, error_code: i16) -> Bytes { + use crate::protocol::requests::{FetchPartition, FetchTopic}; + + if version >= 7 { + return encode_fetch_response_inner(version, &[], Some(error_code), error_code); + } + + let topics = vec![FetchTopic { + topic: String::new(), + partitions: vec![FetchPartition { + partition: 0, + fetch_offset: 0, + partition_max_bytes: 1, + }], + }]; + encode_fetch_response_inner(version, &topics, Some(ERROR_NONE), error_code) +} + +pub fn encode_fetch_response(version: i16, req: &FetchRequest) -> Bytes { + encode_fetch_response_inner(version, &req.topics, Some(ERROR_NONE), ERROR_NONE) +} + +fn encode_fetch_response_inner( + version: i16, + topics: &[crate::protocol::requests::FetchTopic], + top_level_error: Option<i16>, + partition_error: i16, +) -> Bytes { + let flexible = version >= 12; + let mut e = Encoder::with_capacity(512); + + if version >= 1 { + e.write_i32(0); + } + if version >= 7 { + e.write_i16(top_level_error.unwrap_or(ERROR_NONE)); + e.write_i32(0); + } + + if flexible { + e.write_varint((topics.len() + 1) as u64); + } else { + e.write_i32(i32::try_from(topics.len()).expect("topic count bounded")); + } + + for topic in topics { + if flexible { + e.write_compact_nullable_string(Some(&topic.topic)); + } else { + e.write_nullable_string_unchecked(Some(&topic.topic)); + } + + if flexible { + e.write_varint((topic.partitions.len() + 1) as u64); + } else { + e.write_i32(i32::try_from(topic.partitions.len()).expect("partition count bounded")); + } + + for partition in &topic.partitions { + e.write_i32(partition.partition); + e.write_i16(partition_error); + e.write_i64(0); // high_watermark + if version >= 4 { + e.write_i64(0); // last_stable_offset + } + if version >= 5 { + e.write_i64(0); // log_start_offset + } + if version >= 4 { + if flexible { + e.write_varint(1); // empty aborted_transactions + } else { + e.write_i32(0); // empty aborted_transactions + } + } + if version >= 11 { + e.write_i32(-1); // preferred_read_replica + } + if flexible { + e.write_compact_nullable_bytes(None); + } else { + e.write_null_bytes(); + } + if flexible { + e.write_empty_tagged_fields(); + } + } + + if flexible { + e.write_empty_tagged_fields(); + } + } + + if flexible { + e.write_empty_tagged_fields(); + } + + e.freeze() +} + +/// Well-formed ListOffsets response with a single placeholder topic/partition. +pub fn encode_list_offsets_error_response(version: i16, error_code: i16) -> Bytes { + use crate::protocol::requests::{ListOffsetsPartition, ListOffsetsTopic}; + + let topics = vec![ListOffsetsTopic { + topic: String::new(), + partitions: vec![ListOffsetsPartition { + partition: 0, + timestamp: -1, + }], + }]; + encode_list_offsets_response_inner(version, &topics, error_code) +} + +pub fn encode_list_offsets_response(version: i16, req: &ListOffsetsRequest) -> Bytes { + encode_list_offsets_response_inner(version, &req.topics, ERROR_NONE) +} + +fn encode_list_offsets_response_inner( + version: i16, + topics: &[crate::protocol::requests::ListOffsetsTopic], + partition_error: i16, +) -> Bytes { + let flexible = version >= 6; + let mut e = Encoder::with_capacity(256); + + if version >= 2 { + e.write_i32(0); + } + + if flexible { + e.write_varint((topics.len() + 1) as u64); + } else { + e.write_i32(i32::try_from(topics.len()).expect("topic count bounded")); + } + + for topic in topics { + if flexible { + e.write_compact_nullable_string(Some(&topic.topic)); + } else { + e.write_nullable_string_unchecked(Some(&topic.topic)); + } + + if flexible { + e.write_varint((topic.partitions.len() + 1) as u64); + } else { + e.write_i32(i32::try_from(topic.partitions.len()).expect("partition count bounded")); + } + + for partition in &topic.partitions { + e.write_i32(partition.partition); + e.write_i16(partition_error); + + if version == 0 { + // v0 has no `timestamp`/`offset` fields; it returns the legacy + // `old_style_offsets` ARRAY (i32 count + i64 entries) instead. + // Empty since this stub never resolves a real offset. + e.write_i32(0); + } else { + e.write_i64(-1); // timestamp: -1 = not available (Kafka sentinel) + e.write_i64(0); // offset + if version >= 4 { + e.write_i32(-1); // leader_epoch + } + } + if flexible { + e.write_empty_tagged_fields(); + } + } + + if flexible { + e.write_empty_tagged_fields(); + } + } + + if flexible { + e.write_empty_tagged_fields(); + } + + e.freeze() +} + +/// Well-formed CreateTopics response with a single placeholder topic. +pub fn encode_create_topics_error_response(version: i16, error_code: i16) -> Bytes { + use crate::protocol::requests::CreatableTopic; + + let topics = vec![CreatableTopic { + name: String::new(), + num_partitions: 1, + replication_factor: 1, + }]; + encode_create_topics_response_inner(version, &topics, error_code) +} + +pub fn encode_create_topics_response(version: i16, req: &CreateTopicsRequest) -> Bytes { + encode_create_topics_response_inner(version, &req.topics, ERROR_NONE) +} + +fn encode_create_topics_response_inner( + version: i16, + topics: &[crate::protocol::requests::CreatableTopic], + topic_error: i16, +) -> Bytes { + let flexible = version >= 5; + let mut e = Encoder::with_capacity(256); + + if version >= 2 { + e.write_i32(0); + } + + if flexible { + e.write_varint((topics.len() + 1) as u64); + } else { + e.write_i32(i32::try_from(topics.len()).expect("topic count bounded")); + } + + for topic in topics { + if flexible { + e.write_compact_nullable_string(Some(&topic.name)); + } else { + e.write_nullable_string_unchecked(Some(&topic.name)); + } + + let error_code = if topic_error != ERROR_NONE { + topic_error + } else if topic.num_partitions <= 0 { Review Comment: `num_partitions = -1` is valid in CreateTopics v4+ (KIP-464: use the broker default), and the Java AdminClient sends -1 when the partition count is unspecified. This branch returns `INVALID_PARTITIONS` for it, where a real broker accepts it. Suggest treating -1 as valid and reserving error 37 for `0` and `< -1`. Same consideration applies to `replication_factor = -1`. ########## gateways/kafka/tests/common/tcp.rs: ########## @@ -0,0 +1,165 @@ +// 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. + +//! TCP round-trip helpers — compiled into each integration test binary via `#[path]`. +#![allow(dead_code)] + +use std::net::SocketAddr; +use std::time::Duration; + +use bytes::{BufMut, Bytes, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time; + +use iggy_gateway_kafka::protocol::codec::Decoder; +use iggy_gateway_kafka::protocol::header::{request_header_version, response_header_version}; + +/// Build a complete length-prefixed Kafka request frame (header + body). +pub fn build_request_frame( + api_key: i16, + api_version: i16, + correlation_id: i32, + client_id: Option<&str>, + body: &[u8], +) -> Bytes { + let hdr_ver = request_header_version(api_key, api_version); + let mut enc = iggy_gateway_kafka::protocol::codec::Encoder::with_capacity(64 + body.len()); + enc.write_i16(api_key); + enc.write_i16(api_version); + enc.write_i32(correlation_id); + if hdr_ver >= 2 { + enc.write_compact_nullable_string(client_id); + enc.write_empty_tagged_fields(); + } else { + enc.write_nullable_string(client_id) + .expect("test client_id fits i16"); + } + enc.write_bytes(body); + + let payload = enc.freeze(); + let payload_len = i32::try_from(payload.len()).expect("test payload fits i32"); + let mut frame = BytesMut::with_capacity(4 + payload.len()); + frame.put_i32(payload_len); + frame.extend_from_slice(&payload); + frame.freeze() +} + +/// Parse correlation id and response body from a raw response payload (no length prefix). +pub fn parse_response_payload(api_key: i16, api_version: i16, payload: Bytes) -> (i32, Bytes) { + let resp_hdr_ver = response_header_version(api_key, api_version); + let mut d = Decoder::new(payload); + let correlation_id = d.read_i32().expect("correlation_id"); + if resp_hdr_ver >= 1 { + d.read_tagged_fields().expect("response tagged fields"); + } + let body = d.read_bytes(d.remaining()).expect("response body"); + (correlation_id, body) +} + +/// Read one length-prefixed response frame from the stream. +pub async fn read_response_frame(stream: &mut TcpStream, max_size: usize) -> Bytes { Review Comment: No timeout here, and most suites call this directly. A regression that makes the server drop a response turns into an indefinite test hang instead of a failure. The `_with_timeout` variant already exists; wrapping the default path in a generous timeout would keep CI failures diagnosable. ########## gateways/kafka/tools/kafka-tool/src/response.rs: ########## @@ -0,0 +1,420 @@ +// 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. + +//! Kafka response frame parsing and human-readable summaries for `send` / `verify`. + +use bytes::Bytes; +use kafka_protocol::messages::{ + ApiVersionsResponse, CreateTopicsResponse, FetchResponse, ListOffsetsResponse, + MetadataResponse, ProduceResponse, +}; +use kafka_protocol::protocol::Decodable; + +/// Parsed view of one length-prefixed Kafka response payload (excluding the 4-byte frame length). +pub struct ResponseSummary { + pub frame_bytes: usize, + pub correlation_id: i32, + pub response_header_version: i16, + pub correlation_match: bool, + /// Highest-severity non-zero error found, or `0` when all decoded codes are zero. + pub primary_error_code: i16, + pub details: Vec<String>, + pub decode_note: Option<String>, +} + +impl ResponseSummary { + #[must_use] + pub fn has_nonzero_error(&self) -> bool { + self.primary_error_code != 0 + } + + pub fn print(&self, api_name: &str, version: i16, quiet: bool) { + let sym = if self.has_nonzero_error() { + "⚠" + } else { + "✓" + }; + let ec_label = format_error_code(self.primary_error_code); + let corr = if self.correlation_match { + format!("{}", self.correlation_id) + } else { + format!("{} (expected correlation mismatch)", self.correlation_id) + }; + + if quiet { + println!( + "{sym} {api_name} v{version} → {}B ec={} ({ec_label})", + self.frame_bytes, self.primary_error_code + ); + return; + } + + println!( + "{sym} {api_name} v{version} frame={}B correlation={corr} resp_hdr=v{} primary_ec={} ({ec_label})", + self.frame_bytes, self.response_header_version, self.primary_error_code + ); + for line in &self.details { + println!(" {line}"); + } + if let Some(note) = &self.decode_note { + println!(" note: {note}"); + } + } +} + +/// Analyze a response payload for the given request `(api_key, api_version)`. +pub fn analyze_response( + api_key: i16, + api_version: i16, + request_correlation_id: i32, + payload: &[u8], +) -> ResponseSummary { + let frame_bytes = payload.len(); + if payload.len() < 4 { + return ResponseSummary { + frame_bytes, + correlation_id: 0, + response_header_version: 0, + correlation_match: false, + primary_error_code: -1, + details: vec!["payload shorter than correlation_id".into()], + decode_note: Some("truncated response".into()), + }; + } + + let correlation_id = i32::from_be_bytes(payload[0..4].try_into().expect("4 bytes")); + let resp_hdr_ver = response_header_version(api_key, api_version); + let body_start = if resp_hdr_ver >= 1 { + 5 // correlation_id + empty tagged fields (0x00) + } else { + 4 + }; + + if payload.len() < body_start { + return ResponseSummary { + frame_bytes, + correlation_id, + response_header_version: resp_hdr_ver, + correlation_match: correlation_id == request_correlation_id, + primary_error_code: -1, + details: vec![format!( + "truncated after correlation (need {body_start} bytes)" + )], + decode_note: None, + }; + } + + let body = &payload[body_start..]; + let mut details = Vec::new(); + let mut codes = Vec::new(); + let mut decode_note = None; + + if body.len() == 2 { + let ec = i16::from_be_bytes(body.try_into().expect("2 bytes")); + codes.push(ec); + details.push(format!( + "error-only body: error_code={ec} ({})", + format_error_code(ec) + )); + } else { + match decode_body(api_key, api_version, body, &mut details, &mut codes) { + Ok(()) => {} + Err(e) => { + decode_note = Some(format!("schema decode failed: {e:#}")); + details.push(format!("raw_body_hex={}", hex::encode(body))); + } + } + } + + let primary_error_code = codes.iter().copied().filter(|&c| c != 0).max().unwrap_or(0); + + ResponseSummary { + frame_bytes, + correlation_id, + response_header_version: resp_hdr_ver, + correlation_match: correlation_id == request_correlation_id, + primary_error_code, + details, + decode_note, + } +} + +fn optional_topic_name(name: &Option<kafka_protocol::messages::TopicName>) -> String { + name.as_ref() + .map(|n| n.0.as_str().to_string()) + .unwrap_or_else(|| "<null>".into()) +} + +fn topic_name(name: &kafka_protocol::messages::TopicName) -> String { + name.0.as_str().to_string() +} + +fn decode_body( + api_key: i16, + api_version: i16, + body: &[u8], + details: &mut Vec<String>, + codes: &mut Vec<i16>, +) -> anyhow::Result<()> { + let mut buf = Bytes::copy_from_slice(body); + match api_key { + 18 => { + let resp = ApiVersionsResponse::decode(&mut buf, api_version)?; + codes.push(resp.error_code); + details.push(format!( + "top_level.error_code={} ({})", + resp.error_code, + format_error_code(resp.error_code) + )); + details.push(format!("api_keys={}", resp.api_keys.len())); + if api_version >= 1 { + details.push(format!("throttle_time_ms={}", resp.throttle_time_ms)); + } + for (i, k) in resp.api_keys.iter().enumerate().take(8) { + details.push(format!( + "api_keys[{i}]: key={} min={} max={}", + k.api_key, k.min_version, k.max_version + )); + } + if resp.api_keys.len() > 8 { + details.push(format!("… {} more api_keys", resp.api_keys.len() - 8)); + } + } + 3 => { + let resp = MetadataResponse::decode(&mut buf, api_version)?; + if api_version >= 3 { + details.push(format!("throttle_time_ms={}", resp.throttle_time_ms)); + } + details.push(format!("brokers={}", resp.brokers.len())); + if let Some(b) = resp.brokers.first() { + details.push(format!( + "brokers[0]: id={} host={} port={}", + b.node_id.0, b.host, b.port + )); + } + details.push(format!("topics={}", resp.topics.len())); + for (i, t) in resp.topics.iter().enumerate().take(4) { + codes.push(t.error_code); + let name = optional_topic_name(&t.name); + details.push(format!( + "topics[{i}]: name={name} ec={} ({}) partitions={}", + t.error_code, + format_error_code(t.error_code), + t.partitions.len() + )); + } + if resp.topics.len() > 4 { + details.push(format!("… {} more topics", resp.topics.len() - 4)); + } + } + 0 => { + let resp = ProduceResponse::decode(&mut buf, api_version)?; + if api_version >= 1 { + details.push(format!("throttle_time_ms={}", resp.throttle_time_ms)); + } + details.push(format!("topics={}", resp.responses.len())); + for (ti, topic) in resp.responses.iter().enumerate().take(4) { + let name = topic_name(&topic.name); + details.push(format!( + "topics[{ti}]: name={name} partitions={}", + topic.partition_responses.len() + )); + for (pi, p) in topic.partition_responses.iter().enumerate().take(4) { + codes.push(p.error_code); + details.push(format!( + " partitions[{pi}]: index={} ec={} ({}) offset={}", + p.index, + p.error_code, + format_error_code(p.error_code), + p.base_offset + )); + } + } + } + 1 => { + let resp = FetchResponse::decode(&mut buf, api_version)?; + if api_version >= 1 { + details.push(format!("throttle_time_ms={}", resp.throttle_time_ms)); + } + if api_version >= 7 { + codes.push(resp.error_code); + details.push(format!( + "top_level.error_code={} ({}) session_id={}", + resp.error_code, + format_error_code(resp.error_code), + resp.session_id + )); + } + details.push(format!("topics={}", resp.responses.len())); + for (ti, topic) in resp.responses.iter().enumerate().take(4) { + let name = topic_name(&topic.topic); + details.push(format!( + "topics[{ti}]: name={name} partitions={}", + topic.partitions.len() + )); + for (pi, p) in topic.partitions.iter().enumerate().take(4) { + codes.push(p.error_code); + details.push(format!( + " partitions[{pi}]: index={} ec={} ({}) hw={}", + p.partition_index, + p.error_code, + format_error_code(p.error_code), + p.high_watermark + )); + } + } + } + 2 => { + let resp = ListOffsetsResponse::decode(&mut buf, api_version)?; + if api_version >= 2 { + details.push(format!("throttle_time_ms={}", resp.throttle_time_ms)); + } + details.push(format!("topics={}", resp.topics.len())); + for (ti, topic) in resp.topics.iter().enumerate().take(4) { + let name = topic_name(&topic.name); + details.push(format!( + "topics[{ti}]: name={name} partitions={}", + topic.partitions.len() + )); + for (pi, p) in topic.partitions.iter().enumerate().take(4) { + codes.push(p.error_code); + details.push(format!( + " partitions[{pi}]: index={} ec={} ({}) offset={}", + p.partition_index, + p.error_code, + format_error_code(p.error_code), + p.offset + )); + } + } + } + 19 => { + let resp = CreateTopicsResponse::decode(&mut buf, api_version)?; + details.push(format!("throttle_time_ms={}", resp.throttle_time_ms)); + details.push(format!("topics={}", resp.topics.len())); + for (i, t) in resp.topics.iter().enumerate().take(4) { + codes.push(t.error_code); + let name = topic_name(&t.name); + details.push(format!( + "topics[{i}]: name={name} ec={} ({})", + t.error_code, + format_error_code(t.error_code) + )); + } + } + other => { + details.push(format!("no schema decoder for api_key={other}")); + if body.len() >= 2 { + let ec = i16::from_be_bytes(body[0..2].try_into().expect("2 bytes")); + codes.push(ec); + details.push(format!( + "body[0..2] as i16={ec} ({}) — may not be top-level error_code", + format_error_code(ec) + )); + } + } + } + Ok(()) +} + +fn format_error_code(code: i16) -> &'static str { + match code { + 0 => "NONE", + 1 => "OFFSET_OUT_OF_RANGE", + 2 => "CORRUPT_MESSAGE", + 3 => "UNKNOWN_TOPIC_OR_PARTITION", + 35 => "UNSUPPORTED_VERSION", + 36 => "TOPIC_ALREADY_EXISTS", + 37 => "INVALID_PARTITIONS", + 42 => "INVALID_REQUEST", + -1 => "UNKNOWN", + _ => "OTHER", + } +} + +fn request_header_version(api_key: i16, api_version: i16) -> i16 { + let flex_from = first_flexible_version(api_key); + match flex_from { + Some(fv) if api_version >= fv => 2, + _ => 1, + } +} + +fn response_header_version(api_key: i16, api_version: i16) -> i16 { + if api_key == 18 { + return 0; + } + if request_header_version(api_key, api_version) >= 2 { + 1 + } else { + 0 + } +} + +/// First flexible protocol version per API key (matches gateway `header.rs` / kafka-tool framing). +fn first_flexible_version(api_key: i16) -> Option<i16> { Review Comment: This is the fourth copy of the flexible-version table in the PR (gateway `header.rs`, this crate's `main.rs`, here, and the mirror in `header_tests.rs`), and `GATEWAY_REGISTRY` in `main.rs` hand-mirrors `SUPPORTED_RANGES` with only a comment tying them together. Drift between the copies is silent. `kafka-protocol` already exposes per-message `header_version()`, which would remove both copies in this crate; alternatively the tool could depend on `iggy-gateway-kafka` and use `supported_api_ranges()` directly. ########## gateways/kafka/src/main.rs: ########## @@ -0,0 +1,62 @@ +// 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. + +use tokio::net::TcpListener; +use tokio::signal; +use tokio::sync::broadcast; + +use iggy_gateway_kafka::server::init_tracing; +use iggy_gateway_kafka::{KafkaServer, ServerConfig}; + +#[tokio::main] +async fn main() -> Result<(), Box<dyn std::error::Error>> { + init_tracing(); + + let mut config = ServerConfig::default(); + if let Ok(bind_addr) = std::env::var("KAFKA_BIND_ADDR") { + config.bind_addr = bind_addr; + } + if let Ok(advertised_host) = std::env::var("KAFKA_ADVERTISED_HOST") { + config.advertised_host = Some(advertised_host); + } + if let Ok(advertised_port) = std::env::var("KAFKA_ADVERTISED_PORT") { + config.advertised_port = Some( + advertised_port + .parse() + .map_err(|e| format!("invalid KAFKA_ADVERTISED_PORT `{advertised_port}`: {e}"))?, + ); + } + let listener = TcpListener::bind(&config.bind_addr) + .await + .map_err(|e| format!("failed to bind {}: {e}", config.bind_addr))?; + let server = KafkaServer::new(config); + + 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??); + } + _ = signal::ctrl_c() => { Review Comment: nit: only SIGINT is handled, so `docker stop` (SIGTERM) skips the graceful drain the TaskTracker machinery otherwise provides. A `tokio::signal::unix::signal(SignalKind::terminate())` branch in this select would cover it. ########## gateways/kafka/tools/kafka-tool/src/main.rs: ########## @@ -0,0 +1,894 @@ +// 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. + +use anyhow::{Context, Result}; +use bytes::{BufMut, Bytes, BytesMut}; +use clap::{Parser, Subcommand}; +use kafka_protocol::messages::*; +use kafka_protocol::protocol::{Encodable, StrBytes}; +use std::path::PathBuf; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tracing::{info, warn}; + +mod response; + +#[derive(Parser)] +#[command( + name = "kafka-message-gen", + about = "Generate Kafka wire protocol binary messages for all API keys and versions", + long_about = "Generates correctly-framed Kafka protocol requests from Kafka 4.1.0 schemas.\n\ +Each output .bin file is TCP-ready: [len:i32][api_key:i16][api_version:i16][correlation_id:i32][client_id][payload]" +)] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Subcommand)] +enum Command { + /// List all supported API keys with name and version range + List, + /// Generate binary .bin files for all API keys and versions + Generate { + #[arg(short, long, default_value = "kafka_messages")] + output: PathBuf, + /// Filter to API key(s), repeatable: --api-key 0 --api-key 1 + #[arg(long, action = clap::ArgAction::Append)] + api_key: Vec<i16>, + /// Filter to a single version + #[arg(long)] + version: Option<i16>, + /// Print hex dump to stdout + #[arg(long)] + hex: bool, + }, + /// Send messages to a live Kafka-compatible server and show responses + Send { + #[arg(long, default_value = "127.0.0.1:9092")] + host: String, + #[arg(long)] + api_key: Option<i16>, + #[arg(long)] + version: Option<i16>, + #[arg(long, default_value = "5000")] + timeout_ms: u64, + /// Compact one-line output (default is verbose decoded response) + #[arg(long)] + quiet: bool, + }, + Verify { + #[arg(long, default_value = "127.0.0.1:9092")] + host: String, + /// Limit to these API keys (repeatable). Defaults to all gateway-scoped keys. + #[arg(long, action = clap::ArgAction::Append)] + api_key: Vec<i16>, + /// Limit to a single protocol version + #[arg(long)] + version: Option<i16>, + #[arg(long, default_value = "5000")] + timeout_ms: u64, + /// Stop on the first failure + #[arg(long)] + fail_fast: bool, + /// Use the full Kafka 4.1 registry (for real brokers), not the Iggy gateway scope + #[arg(long)] + all_apis: bool, + /// Compact one-line output (default is verbose decoded response) + #[arg(long)] + quiet: bool, + }, +} + +// ── API Registry ───────────────────────────────────────────────────────────── +// Source: validVersions in apache/kafka trunk JSON schema files, Kafka 4.1.0 +// Format: (api_key, name, min_version, max_version) +const API_REGISTRY: &[(i16, &str, i16, i16)] = &[ + (0, "Produce", 3, 13), + (1, "Fetch", 4, 18), + (2, "ListOffsets", 1, 11), + (3, "Metadata", 0, 13), + (8, "OffsetCommit", 2, 10), + (9, "OffsetFetch", 1, 10), + (10, "FindCoordinator", 0, 6), + (11, "JoinGroup", 0, 9), + (12, "Heartbeat", 0, 4), + (13, "LeaveGroup", 0, 5), + (14, "SyncGroup", 0, 5), + (15, "DescribeGroups", 0, 6), + (16, "ListGroups", 0, 5), + (17, "SaslHandshake", 0, 1), + (18, "ApiVersions", 0, 5), + (19, "CreateTopics", 2, 7), + (20, "DeleteTopics", 1, 6), + (21, "DeleteRecords", 0, 2), + (22, "InitProducerId", 0, 6), + (23, "OffsetForLeaderEpoch", 2, 4), + (24, "AddPartitionsToTxn", 0, 5), + (25, "AddOffsetsToTxn", 0, 4), + (26, "EndTxn", 0, 5), + (27, "WriteTxnMarkers", 1, 2), + (28, "TxnOffsetCommit", 0, 5), + (29, "DescribeAcls", 1, 3), + (30, "CreateAcls", 1, 3), + (31, "DeleteAcls", 1, 3), + (32, "DescribeConfigs", 1, 4), + (33, "AlterConfigs", 0, 2), + (34, "AlterReplicaLogDirs", 1, 2), + (35, "DescribeLogDirs", 1, 5), + (36, "SaslAuthenticate", 0, 2), + (37, "CreatePartitions", 0, 3), + (38, "CreateDelegationToken", 1, 3), + (39, "RenewDelegationToken", 1, 2), + (40, "ExpireDelegationToken", 1, 2), + (41, "DescribeDelegationToken", 1, 3), + (42, "DeleteGroups", 0, 2), + (43, "ElectLeaders", 0, 2), + (44, "IncrementalAlterConfigs", 0, 1), + (45, "AlterPartitionReassignments", 0, 1), + (46, "ListPartitionReassignments", 0, 1), + (47, "OffsetDelete", 0, 0), + (48, "DescribeClientQuotas", 0, 1), + (49, "AlterClientQuotas", 0, 1), + (50, "DescribeUserScramCredentials", 0, 0), + (51, "AlterUserScramCredentials", 0, 0), + (55, "DescribeQuorum", 2, 3), + (56, "AlterPartition", 2, 3), + (57, "UpdateFeatures", 0, 2), + (60, "DescribeCluster", 0, 2), + (61, "DescribeProducers", 0, 0), + (64, "UnregisterBroker", 0, 0), + (65, "DescribeTransactions", 0, 0), + (66, "ListTransactions", 0, 1), + (67, "AllocateProducerIds", 0, 0), + (68, "ConsumerGroupHeartbeat", 0, 1), + (69, "ConsumerGroupDescribe", 0, 1), + (71, "GetTelemetrySubscriptions", 0, 0), + (72, "PushTelemetry", 0, 0), + (74, "AssignReplicasToDirs", 0, 0), + (75, "DescribeTopicPartitions", 0, 0), + (76, "ListClientMetricsResources", 0, 0), +]; + +/// Iggy Kafka gateway #3421 scope — mirrors `SUPPORTED_RANGES` in `iggy_gateway_kafka`. +const GATEWAY_REGISTRY: &[(i16, &str, i16, i16)] = &[ + (0, "Produce", 3, 9), + (1, "Fetch", 4, 12), + (2, "ListOffsets", 1, 6), + (3, "Metadata", 0, 9), + (18, "ApiVersions", 0, 3), + (19, "CreateTopics", 2, 5), +]; + +// ── Flexible version table ──────────────────────────────────────────────────── +// Source: flexibleVersions field in each Kafka JSON schema. +// Returns the first version using compact encoding, or None if never flexible. +fn first_flexible_version(api_key: i16) -> Option<i16> { + match api_key { + 0 => Some(9), + 1 => Some(12), + 2 => Some(6), + 3 => Some(9), + 8 => Some(8), + 9 => Some(6), + 10 => Some(3), + 11 => Some(6), + 12 => Some(4), + 13 => Some(4), + 14 => Some(4), + 15 => Some(5), + 16 => Some(3), + 17 => None, + 18 => Some(3), + 19 => Some(5), + 20 => Some(4), + 21 => Some(2), + 22 => Some(2), + 23 => Some(4), + 24 => Some(3), + 25 => Some(3), + 26 => Some(3), + 27 => Some(1), + 28 => Some(3), + 29 => Some(2), + 30 => Some(2), + 31 => Some(2), + 32 => Some(4), + 33 => Some(2), + 34 => Some(2), + 35 => Some(2), + 36 => Some(2), + 37 => Some(2), + 38 => Some(2), + 39 => Some(2), + 40 => Some(2), + 41 => Some(2), + 42 => Some(2), + 43 => Some(2), + 44 => Some(1), + 45 => Some(1), + 46 => Some(1), + 47 => Some(0), + 48 => Some(1), + 49 => Some(1), + 50 => Some(0), + 51 => Some(0), + 55 => Some(2), + 56 => Some(2), + 57 => Some(1), + 60 => Some(0), + 61 => Some(0), + 64 => Some(0), + 65 => Some(0), + 66 => Some(0), + 67 => Some(0), + 68 => Some(0), + 69 => Some(0), + 71 => Some(0), + 72 => Some(0), + 74 => Some(0), + 75 => Some(0), + 76 => Some(0), + _ => None, + } +} + +// ── Request framing ─────────────────────────────────────────────────────────── +// Wire format (Kafka protocol spec): +// [total_length: i32] big-endian, excludes self +// [api_key: i16] +// [api_version: i16] +// [correlation_id: i32] +// header v1: [client_id: NULLABLE_STRING] +// header v2: [client_id: COMPACT_NULLABLE_STRING] [request_header_tagged_fields] +// [payload: bytes] + +fn write_unsigned_varint(buf: &mut BytesMut, mut value: u64) { + loop { + let mut byte = (value & 0x7F) as u8; + value >>= 7; + if value != 0 { + byte |= 0x80; + } + buf.put_u8(byte); + if value == 0 { + break; + } + } +} + +fn write_compact_nullable_string(buf: &mut BytesMut, value: Option<&str>) { + match value { + None => write_unsigned_varint(buf, 0), + Some(s) => { + write_unsigned_varint(buf, (s.len() + 1) as u64); + buf.put_slice(s.as_bytes()); + } + } +} + +fn frame_request( + api_key: i16, + api_version: i16, + correlation_id: i32, + client_id: &str, + payload: &[u8], + flexible: bool, +) -> Bytes { + let mut header = BytesMut::new(); + header.put_i16(api_key); + header.put_i16(api_version); + header.put_i32(correlation_id); + if flexible { + write_compact_nullable_string(&mut header, Some(client_id)); + header.put_u8(0); // empty request-header tagged fields + } else { + header.put_i16(i16::try_from(client_id.len()).expect("client_id fits i16")); + header.put_slice(client_id.as_bytes()); + } + + let blen = header.len() + payload.len(); + let mut buf = BytesMut::with_capacity(4 + blen); + buf.put_i32(i32::try_from(blen).expect("frame fits i32")); + buf.put_slice(&header); + buf.put_slice(payload); + buf.freeze() +} + +// ── Payload builders ────────────────────────────────────────────────────────── +// Build the API-specific encoded body for a given api_key and version. +// All required fields contain realistic non-zero values. +// Returns raw bytes WITHOUT the framing header. +fn build_payload(api_key: i16, version: i16) -> Result<Bytes> { + let mut buf = BytesMut::new(); + match api_key { + 18 => { + let mut r = ApiVersionsRequest::default(); + if version >= 3 { + r.client_software_name = StrBytes::from_static_str("kafka-message-gen"); + r.client_software_version = StrBytes::from_static_str("0.1.0"); + } + r.encode(&mut buf, version).context("ApiVersions")?; + } + 3 => { + let mut r = MetadataRequest::default(); + if version >= 1 { + r.topics = None; + } + if version >= 4 { + r.allow_auto_topic_creation = true; + } + if version >= 8 { + r.include_cluster_authorized_operations = false; + r.include_topic_authorized_operations = false; + } + r.encode(&mut buf, version).context("Metadata")?; + } + 0 => { + use kafka_protocol::messages::produce_request::*; + use kafka_protocol::records::{ + Compression, Record, RecordBatchEncoder, RecordEncodeOptions, TimestampType, + }; + let rec = Record { + transactional: false, + control: false, + partition_leader_epoch: 0, + producer_id: -1, + producer_epoch: -1, + timestamp_type: TimestampType::Creation, + offset: 0, + sequence: 0, + timestamp: 1_700_000_000_000, + key: Some(Bytes::from_static(b"test-key")), + value: Some(Bytes::from_static(b"test-value")), + headers: indexmap::IndexMap::new(), + }; + let mut rb = BytesMut::new(); + RecordBatchEncoder::encode( + &mut rb, + [rec].iter(), + &RecordEncodeOptions { + version: 2, + compression: Compression::None, + }, + ) + .context("RecordBatch encode")?; + let pd = TopicProduceData::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partition_data(vec![ + PartitionProduceData::default() + .with_index(0) + .with_records(Some(rb.freeze())), + ]); + let mut r = ProduceRequest::default() + .with_acks(-1) + .with_timeout_ms(5000) + .with_topic_data(vec![pd]); + if version >= 3 { + r.transactional_id = None; + } + r.encode(&mut buf, version).context("Produce")?; + } + 1 => { + use kafka_protocol::messages::fetch_request::*; + let fp = FetchPartition::default() + .with_partition(0) + .with_fetch_offset(0) + .with_partition_max_bytes(1_048_576); + let ft = FetchTopic::default() + .with_topic(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![fp]); + let mut r = FetchRequest::default() + .with_replica_id(BrokerId(-1)) + .with_max_wait_ms(500) + .with_min_bytes(1) + .with_topics(vec![ft]); + if version >= 3 { + r.max_bytes = 52_428_800; + } + if version >= 4 { + r.isolation_level = 0; + } + if version >= 7 { + r.session_id = 0; + r.session_epoch = -1; + } + r.encode(&mut buf, version).context("Fetch")?; + } + 2 => { + use kafka_protocol::messages::list_offsets_request::*; + let p = ListOffsetsPartition::default() + .with_partition_index(0) + .with_timestamp(-1); + let t = ListOffsetsTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![p]); + ListOffsetsRequest::default() + .with_replica_id(BrokerId(-1)) + .with_isolation_level(0) + .with_topics(vec![t]) + .encode(&mut buf, version) + .context("ListOffsets")?; + } + 8 => { + use kafka_protocol::messages::offset_commit_request::*; + let p = OffsetCommitRequestPartition::default() + .with_partition_index(0) + .with_committed_offset(42) + .with_committed_metadata(Some(StrBytes::from_static_str(""))); + let t = OffsetCommitRequestTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![p]); + OffsetCommitRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_topics(vec![t]) + .encode(&mut buf, version) + .context("OffsetCommit")?; + } + 9 => { + OffsetFetchRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .encode(&mut buf, version) + .context("OffsetFetch")?; + } + 10 => { + FindCoordinatorRequest::default() + .with_key(StrBytes::from_static_str("test-group")) + .with_key_type(0) + .encode(&mut buf, version) + .context("FindCoordinator")?; + } + 11 => { + use kafka_protocol::messages::join_group_request::*; + let p = JoinGroupRequestProtocol::default() + .with_name(StrBytes::from_static_str("range")) + .with_metadata(Bytes::from_static(b"\x00\x00\x00\x01\x00\x0atest-topic")); + JoinGroupRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_session_timeout_ms(30_000) + .with_rebalance_timeout_ms(300_000) + .with_member_id(StrBytes::from_static_str("")) + .with_protocol_type(StrBytes::from_static_str("consumer")) + .with_protocols(vec![p]) + .encode(&mut buf, version) + .context("JoinGroup")?; + } + 12 => { + HeartbeatRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_generation_id(1) + .with_member_id(StrBytes::from_static_str("test-member-1")) + .encode(&mut buf, version) + .context("Heartbeat")?; + } + 13 => { + LeaveGroupRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_member_id(StrBytes::from_static_str("test-member-1")) + .encode(&mut buf, version) + .context("LeaveGroup")?; + } + 14 => { + SyncGroupRequest::default() + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_generation_id(1) + .with_member_id(StrBytes::from_static_str("test-member-1")) + .with_protocol_type(Some(StrBytes::from_static_str("consumer"))) + .with_protocol_name(Some(StrBytes::from_static_str("range"))) + .encode(&mut buf, version) + .context("SyncGroup")?; + } + 15 => { + DescribeGroupsRequest::default() + .with_groups(vec![GroupId::from(StrBytes::from_static_str("test-group"))]) + .with_include_authorized_operations(false) + .encode(&mut buf, version) + .context("DescribeGroups")?; + } + 16 => { + ListGroupsRequest::default() + .encode(&mut buf, version) + .context("ListGroups")?; + } + 17 => { + SaslHandshakeRequest::default() + .with_mechanism(StrBytes::from_static_str("PLAIN")) + .encode(&mut buf, version) + .context("SaslHandshake")?; + } + 19 => { + use kafka_protocol::messages::create_topics_request::*; + let t = CreatableTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str( + "iggy-test-topic", + ))) + .with_num_partitions(1) + .with_replication_factor(1); + CreateTopicsRequest::default() + .with_topics(vec![t]) + .with_timeout_ms(30_000) + .with_validate_only(false) + .encode(&mut buf, version) + .context("CreateTopics")?; + } + 20 => { + use kafka_protocol::messages::delete_topics_request::*; + let r = if version >= 6 { + DeleteTopicsRequest::default() + .with_topics(vec![DeleteTopicState::default().with_name(Some( + TopicName::from(StrBytes::from_static_str("iggy-test-topic")), + ))]) + .with_timeout_ms(30_000) + } else { + DeleteTopicsRequest::default() + .with_topic_names(vec![TopicName::from(StrBytes::from_static_str( + "iggy-test-topic", + ))]) + .with_timeout_ms(30_000) + }; + r.encode(&mut buf, version).context("DeleteTopics")?; + } + 21 => { + use kafka_protocol::messages::delete_records_request::*; + let p = DeleteRecordsPartition::default() + .with_partition_index(0) + .with_offset(0); + let t = DeleteRecordsTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![p]); + DeleteRecordsRequest::default() + .with_topics(vec![t]) + .with_timeout_ms(30_000) + .encode(&mut buf, version) + .context("DeleteRecords")?; + } + 22 => { + InitProducerIdRequest::default() + .with_transactional_id(None) + .with_transaction_timeout_ms(60_000) + .encode(&mut buf, version) + .context("InitProducerId")?; + } + 24 => { + use kafka_protocol::messages::add_partitions_to_txn_request::*; + let t = AddPartitionsToTxnTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![0i32]); + AddPartitionsToTxnRequest::default() + .with_v3_and_below_transactional_id(TransactionalId(StrBytes::from_static_str( + "test-txn", + ))) + .with_v3_and_below_producer_id(ProducerId(100)) + .with_v3_and_below_producer_epoch(1) + .with_v3_and_below_topics(vec![t]) + .encode(&mut buf, version) + .context("AddPartitionsToTxn")?; + } + 25 => { + AddOffsetsToTxnRequest::default() + .with_transactional_id(TransactionalId(StrBytes::from_static_str("test-txn"))) + .with_producer_id(ProducerId(100)) + .with_producer_epoch(1) + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .encode(&mut buf, version) + .context("AddOffsetsToTxn")?; + } + 26 => { + EndTxnRequest::default() + .with_transactional_id(TransactionalId(StrBytes::from_static_str("test-txn"))) + .with_producer_id(ProducerId(100)) + .with_producer_epoch(1) + .with_committed(true) + .encode(&mut buf, version) + .context("EndTxn")?; + } + 28 => { + use kafka_protocol::messages::txn_offset_commit_request::*; + let p = TxnOffsetCommitRequestPartition::default() + .with_partition_index(0) + .with_committed_offset(42) + .with_committed_metadata(Some(StrBytes::from_static_str(""))); + let t = TxnOffsetCommitRequestTopic::default() + .with_name(TopicName::from(StrBytes::from_static_str("test-topic"))) + .with_partitions(vec![p]); + TxnOffsetCommitRequest::default() + .with_transactional_id(TransactionalId(StrBytes::from_static_str("test-txn"))) + .with_group_id(GroupId::from(StrBytes::from_static_str("test-group"))) + .with_producer_id(ProducerId(100)) + .with_producer_epoch(1) + .with_topics(vec![t]) + .encode(&mut buf, version) + .context("TxnOffsetCommit")?; + } + 32 => { + use kafka_protocol::messages::describe_configs_request::*; + let r = DescribeConfigsResource::default() + .with_resource_type(2) + .with_resource_name(StrBytes::from_static_str("test-topic")); + DescribeConfigsRequest::default() + .with_resources(vec![r]) + .encode(&mut buf, version) + .context("DescribeConfigs")?; + } + 36 => { + SaslAuthenticateRequest::default() + .with_auth_bytes(Bytes::from_static(b"\x00iggy\x00secret")) + .encode(&mut buf, version) + .context("SaslAuthenticate")?; + } + other => { + warn!("api_key={other}: no explicit builder — empty payload (framing test)"); + } + } + Ok(buf.freeze()) +} + +// Build a complete framed Kafka request message ready for TCP transmission. +fn build_framed(api_key: i16, version: i16, corr: i32) -> Result<Bytes> { + let payload = build_payload(api_key, version)?; + let flexible = first_flexible_version(api_key) + .map(|fv| version >= fv) + .unwrap_or(false); + Ok(frame_request( + api_key, + version, + corr, + "kafka-message-gen", + &payload, + flexible, + )) +} + +// ── Commands ────────────────────────────────────────────────────────────────── + +fn cmd_list() { + println!( + "{:<6} {:<42} {:<10} {:<10} {:<8}", + "Key", "Name", "MinVer", "MaxVer", "Count" + ); + println!("{}", "─".repeat(78)); + for &(k, n, min, max) in API_REGISTRY { + println!( + "{:<6} {:<42} {:<10} {:<10} {:<8}", + k, + n, + min, + max, + max - min + 1 + ); + } + let total: i16 = API_REGISTRY + .iter() + .map(|&(_, _, min, max)| max - min + 1) + .sum(); + println!("{}", "─".repeat(78)); + println!( + "Total: {} API keys | {} versioned messages", + API_REGISTRY.len(), + total + ); +} + +async fn cmd_generate( + out: PathBuf, + filter_keys: Vec<i16>, + fv: Option<i16>, + hex_dump: bool, +) -> Result<()> { + tokio::fs::create_dir_all(&out).await?; + let (mut n, mut corr) = (0usize, 1i32); + for &(ak, name, min, max) in API_REGISTRY { + if !filter_keys.is_empty() && !filter_keys.contains(&ak) { + continue; + } + for v in min..=max { + if fv.is_some_and(|fv| fv != v) { + continue; + } + match build_framed(ak, v, corr) { + Ok(msg) => { + let fname = format!("{:03}_{}_v{}.bin", ak, name, v); + tokio::fs::write(out.join(&fname), &msg).await?; + if hex_dump { + println!("── {} v{} ({} bytes) ──", name, v, msg.len()); + println!("{}", hex::encode(&msg)); + println!(); + } else { + info!(" {} ({} bytes)", fname, msg.len()); + } + n += 1; + corr += 1; + } + Err(e) => warn!("SKIP {} v{}: {e}", name, v), + } + } + } + println!("\n✓ Generated {n} messages → {}/", out.display()); + println!( + " Quick test: cat {}/018_ApiVersions_v3.bin | nc 127.0.0.1 9092 | xxd", + out.display() + ); + Ok(()) +} + +async fn connect(host: &str) -> Result<TcpStream> { + TcpStream::connect(host) + .await + .with_context(|| format!("Cannot connect to {host}")) +} + +async fn read_kafka_response(stream: &mut TcpStream) -> std::io::Result<Vec<u8>> { + let mut lb = [0u8; 4]; + stream.read_exact(&mut lb).await?; + let frame_len = i32::from_be_bytes(lb); + if frame_len <= 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("invalid response frame length: {frame_len}"), + )); + } + let mut body = vec![ + 0u8; + usize::try_from(frame_len).map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + "response frame length does not fit usize", + ) + })? + ]; + stream.read_exact(&mut body).await?; + Ok(body) +} + +async fn run_send( + host: &str, + registry: &[(i16, &str, i16, i16)], + filter_keys: &[i16], + fv: Option<i16>, + toms: u64, + fail_fast: bool, + quiet: bool, +) -> Result<(usize, usize)> { + let mut stream = connect(host).await?; + info!("Connected to {host}"); + let (mut ok, mut fail, mut corr) = (0usize, 0usize, 1i32); + 'outer: for &(ak, name, min, max) in registry { + if !filter_keys.is_empty() && !filter_keys.contains(&ak) { + continue; + } + for v in min..=max { + if fv.is_some_and(|wanted| wanted != v) { + continue; + } + let msg = match build_framed(ak, v, corr) { + Ok(m) => m, + Err(e) => { + warn!("Build {} v{}: {e}", name, v); + fail += 1; + if fail_fast { + break 'outer; + } + continue; + } + }; + if let Err(e) = stream.write_all(&msg).await { + println!("✗ {name} v{v} → write error: {e}"); + fail += 1; + stream = connect(host).await?; + if fail_fast { + break 'outer; + } + corr += 1; + continue; + } + + let res = tokio::time::timeout( + std::time::Duration::from_millis(toms), + read_kafka_response(&mut stream), + ) + .await; + + match res { + Ok(Ok(r)) => { + let summary = response::analyze_response(ak, v, corr, &r); + summary.print(name, v, quiet); + ok += 1; + } Review Comment: `verify` counts a pass for any response frame that can be read off the wire: `has_nonzero_error()`, `correlation_match`, and schema-decode failures are computed in `analyze_response` but never affect `ok`/`fail`, so a gateway encoding regression still exits 0. For the gateway registry, suggest failing on correlation mismatch, on `decode_note` (schema decode failure), and on unexpected non-zero error codes, so `verify` can be trusted as a smoke gate. -- 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]
