ryerraguntla commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3608227213
########## 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: fixed ########## 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: fixed -- 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]
