ryerraguntla commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3812643353
########## gateways/kafka/tests/listener_robustness_tests.rs: ########## @@ -0,0 +1,585 @@ +// 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 listener robustness - framing, pipelining, concurrency, edge cases. + +#[path = "common/codec.rs"] +mod codec; +#[path = "common/server.rs"] +mod server; +#[path = "common/tcp.rs"] +mod tcp; +#[path = "common/wire.rs"] +mod wire; + +use std::time::Duration; + +use bytes::{BufMut, BytesMut}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time; + +use iggy_gateway_kafka::GatewayConfig; +use iggy_gateway_kafka::protocol::api::{ + API_KEY_API_VERSIONS, API_KEY_FETCH, API_KEY_METADATA, API_KEY_PRODUCE, ERROR_INVALID_REQUEST, +}; + +use codec::Decoder; +use server::{spawn_test_server, spawn_test_server_with_config}; +use tcp::{ + ByteRead, build_request_frame, concat_frames, parse_response_payload, read_byte_with_timeout, + read_response_frame, read_response_frame_with_timeout, scan_for_error_code, +}; + +#[tokio::test] +async fn e2e_pipelined_requests_receive_responses_in_order() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let frame1 = build_request_frame(API_KEY_API_VERSIONS, 1, 1, Some("pipe-test"), &[]); + let frame2 = build_request_frame(API_KEY_API_VERSIONS, 1, 2, Some("pipe-test"), &[]); + let frame3 = build_request_frame(API_KEY_API_VERSIONS, 1, 3, Some("pipe-test"), &[]); + stream + .write_all(&concat_frames(&[frame1, frame2, frame3])) + .await + .expect("pipelined write"); + + for expected_corr in 1..=3 { + let payload = read_response_frame(&mut stream, 8 * 1024 * 1024).await; + let (corr, body) = parse_response_payload(API_KEY_API_VERSIONS, 1, payload); + assert_eq!(corr, expected_corr); + assert_eq!(Decoder::new(body).read_i16().unwrap(), 0); + } +} + +#[tokio::test] +async fn e2e_partial_length_prefix_then_remainder_accepted() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let frame = build_request_frame(API_KEY_API_VERSIONS, 1, 42, Some("partial-test"), &[]); + assert!(frame.len() > 6, "test frame long enough to split"); + + stream.write_all(&frame[..2]).await.expect("partial prefix"); + time::sleep(Duration::from_millis(50)).await; + stream.write_all(&frame[2..]).await.expect("remainder"); + + let payload = read_response_frame(&mut stream, 8 * 1024 * 1024).await; + let (corr, _) = parse_response_payload(API_KEY_API_VERSIONS, 1, payload); + assert_eq!(corr, 42); +} + +#[tokio::test] +async fn e2e_frame_within_custom_max_frame_size_accepted() { + let max_frame = 512; + let (addr, _shutdown) = spawn_test_server_with_config(GatewayConfig { + bind_addr: String::new(), + advertised_host: None, + advertised_port: None, + max_frame_size: max_frame, + max_connections: 1024, + idle_timeout: Duration::from_secs(5), + read_timeout: Duration::from_secs(5), + write_timeout: Duration::from_secs(5), + shutdown_drain_timeout: Duration::from_secs(5), + }) + .await; + + let mut stream = TcpStream::connect(addr).await.expect("connect"); + let frame = build_request_frame(API_KEY_API_VERSIONS, 1, 55, Some("max-frame-test"), &[]); + assert!( + frame.len() <= max_frame, + "ApiVersions frame must fit test max ({max_frame})" + ); + + stream.write_all(&frame).await.expect("write"); + let payload = read_response_frame(&mut stream, max_frame).await; + assert_eq!( + parse_response_payload(API_KEY_API_VERSIONS, 1, payload).0, + 55 + ); +} + +#[tokio::test] +async fn e2e_frame_exceeding_max_frame_size_closes_connection() { + let max_frame = 64; + let (addr, _shutdown) = spawn_test_server_with_config(GatewayConfig { + bind_addr: String::new(), + advertised_host: None, + advertised_port: None, + max_frame_size: max_frame, + max_connections: 1024, + idle_timeout: Duration::from_secs(5), + read_timeout: Duration::from_secs(5), + write_timeout: Duration::from_secs(5), + shutdown_drain_timeout: Duration::from_secs(5), + }) + .await; + + let mut stream = TcpStream::connect(addr).await.expect("connect"); + let mut frame = BytesMut::new(); + frame.put_i32(200); + frame.resize(4 + 200, 0); + stream.write_all(&frame).await.expect("oversized frame"); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "oversized frame should close connection (EOF)" + ); +} + +#[tokio::test] +async fn e2e_truncated_frame_body_closes_connection() { + // A truncated in-flight body closes only once the server's read_timeout elapses, so use a + // short read_timeout and wait longer than it to observe a genuine close, not a mere stall. + let (addr, _shutdown) = spawn_test_server_with_config(GatewayConfig { + bind_addr: String::new(), + advertised_host: None, + advertised_port: None, + max_frame_size: 8 * 1024 * 1024, + max_connections: 1024, + idle_timeout: Duration::from_secs(5), + read_timeout: Duration::from_secs(1), + write_timeout: Duration::from_secs(5), + shutdown_drain_timeout: Duration::from_secs(5), + }) + .await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let full = build_request_frame(API_KEY_API_VERSIONS, 1, 66, Some("trunc-test"), &[]); + let payload_len = u32::from_be_bytes([full[0], full[1], full[2], full[3]]) as usize; + assert!(full.len() >= 4 + payload_len); + + stream + .write_all(&full[..4 + payload_len / 2]) + .await + .expect("half body"); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(3)).await, + ByteRead::Closed, + "truncated body should close connection after read_timeout" + ); +} + +#[tokio::test] +async fn e2e_multiple_concurrent_connections_are_independent() { + let (addr, _shutdown) = spawn_test_server().await; + + let (r1, r2, r3) = tokio::join!( + tcp::round_trip(addr, API_KEY_API_VERSIONS, 1, 101, &[]), + tcp::round_trip(addr, API_KEY_API_VERSIONS, 1, 102, &[]), + tcp::round_trip(addr, API_KEY_API_VERSIONS, 1, 103, &[]), + ); + + assert_eq!(r1.0, 101); + assert_eq!(r2.0, 102); + assert_eq!(r3.0, 103); +} + +#[tokio::test] +async fn e2e_client_disconnect_mid_frame_allows_new_connection() { + let (addr, _shutdown) = spawn_test_server().await; + + { + let mut stream = TcpStream::connect(addr).await.expect("connect"); + let full = build_request_frame(API_KEY_API_VERSIONS, 1, 77, Some("abort-test"), &[]); + stream.write_all(&full[..8]).await.expect("partial write"); + drop(stream); + } + + time::sleep(Duration::from_millis(100)).await; + + let (corr, body) = tcp::round_trip(addr, API_KEY_API_VERSIONS, 1, 78, &[]).await; + assert_eq!(corr, 78); + assert_eq!(Decoder::new(body).read_i16().unwrap(), 0); +} + +#[tokio::test] +async fn e2e_response_frames_have_positive_big_endian_length_prefix() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let request = wire::build_api_versions_flexible_request("iggy-test", "0.1.0"); + let frame = build_request_frame(API_KEY_API_VERSIONS, 3, 200, Some("len-test"), &request); + stream.write_all(&frame).await.expect("write"); + + let mut len_buf = [0u8; 4]; + stream + .read_exact(&mut len_buf) + .await + .expect("length prefix"); Review Comment: Rewritten to read the raw prefix and assert it equals an independently-computed (in-process handle_request) expected length, byte-exact — no longer just "non-empty." The unguarded stream.read(...).await.expect(...) in e2e_client_eof_after_valid_frame_closes_connection_cleanly replaced with read_byte_with_timeout(...) == ByteRead::Closed. -- 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]
