This is an automated email from the ASF dual-hosted git repository. krishvishal pushed a commit to branch kafka-4x-clients in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 3ed9857813d06480bfcfee0245f66590334f9d27 Author: Krishna Vishal <[email protected]> AuthorDate: Sat Nov 7 13:15:17 2026 +0530 feat(gateways): serve ApiVersions v4 and v5 to Kafka 4 clients Kafka 4.0 to 4.3 clients send ApiVersions v4 and 4.4 clients send v5, so each connection needed a second request at v3. --- .github/actions/rust/pre-merge/action.yml | 4 +- core/kafka_wire/src/api.rs | 3 + core/kafka_wire/src/api_versions.rs | 209 +++++++++++++++++++++ core/kafka_wire/src/bounds_guard.rs | 9 +- core/kafka_wire/src/lib.rs | 1 + core/kafka_wire/tests/version_sweep.rs | 59 ++++-- gateways/kafka/README.md | 4 + gateways/kafka/docs/MANUAL_TESTING.md | 12 +- gateways/kafka/docs/SCOPE.md | 9 +- gateways/kafka/docs/TEST_SUITE.md | 17 +- gateways/kafka/docs/kafka_api_keys_reference.md | 2 +- .../kafka/src/protocol/handlers/api_versions.rs | 71 ++++--- gateways/kafka/src/protocol/handlers/mod.rs | 2 +- gateways/kafka/tests/api_handler_tests.rs | 95 +++++++++- gateways/kafka/tests/cluster_failover_tests.rs | 2 +- gateways/kafka/tests/common/scope.rs | 2 +- gateways/kafka/tests/common/wire.rs | 13 ++ gateways/kafka/tests/configs_java_client_tests.rs | 22 +-- gateways/kafka/tests/golden_wire_fixtures_tests.rs | 4 +- gateways/kafka/tests/java/ConfigsProbe.java | 24 +-- .../kafka/tests/java/OffsetsForTimesProbe.java | 2 +- gateways/kafka/tests/kafka_client_e2e_tests.rs | 76 +++++++- gateways/kafka/tests/version_firewall_tests.rs | 44 ++--- 23 files changed, 573 insertions(+), 113 deletions(-) diff --git a/.github/actions/rust/pre-merge/action.yml b/.github/actions/rust/pre-merge/action.yml index b9266ee0c..0fa6f0bb6 100644 --- a/.github/actions/rust/pre-merge/action.yml +++ b/.github/actions/rust/pre-merge/action.yml @@ -400,8 +400,8 @@ runs: # Read the tags out of the suites themselves; hardcoding them here lets a bump in a test # desync silently, which pre-pulls the superseded image and puts the real one back # inside the budget this block exists to avoid. The failover suite runs the same - # clients against a cluster. - E2E_IMAGES=$(grep -ohP '^const (KCAT|KAFKA)_IMAGE: &str = "\K[^"]+' \ + # clients against a cluster. `KAFKA_OLDEST_IMAGE` matches too. + E2E_IMAGES=$(grep -ohP '^const (KCAT|KAFKA)\w*_IMAGE: &str = "\K[^"]+' \ gateways/kafka/tests/kafka_client_e2e_tests.rs \ gateways/kafka/tests/cluster_failover_tests.rs | sort -u) if [[ -z "$E2E_IMAGES" ]]; then diff --git a/core/kafka_wire/src/api.rs b/core/kafka_wire/src/api.rs index 75624f0fa..06c4a7118 100644 --- a/core/kafka_wire/src/api.rs +++ b/core/kafka_wire/src/api.rs @@ -198,6 +198,9 @@ pub const ERROR_INVALID_RECORD: i16 = ResponseError::InvalidRecord.code(); pub const ERROR_MISMATCHED_ENDPOINT_TYPE: i16 = ResponseError::MismatchedEndpointType.code(); /// `DescribeCluster` v1+ named an endpoint type Kafka does not define. pub const ERROR_UNSUPPORTED_ENDPOINT_TYPE: i16 = ResponseError::UnsupportedEndpointType.code(); +/// `ApiVersions` v5 named another cluster or another node (KIP-1242). A Java client then closes +/// its connections and starts again from its bootstrap servers. +pub const ERROR_REBOOTSTRAP_REQUIRED: i16 = ResponseError::RebootstrapRequired.code(); #[derive(Debug, Clone, Copy)] pub struct ApiVersionRange { diff --git a/core/kafka_wire/src/api_versions.rs b/core/kafka_wire/src/api_versions.rs new file mode 100644 index 000000000..739874ac2 --- /dev/null +++ b/core/kafka_wire/src/api_versions.rs @@ -0,0 +1,209 @@ +// 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. + +//! `ApiVersions` requests up to v5, one version past what `kafka-protocol` decodes. +//! +//! `kafka-protocol` 0.18 stops at v4. v5 (KIP-1242) adds `ClusterId` and `NodeId` to the request: +//! the broker that the client means to reach. The response gains no field, so the v4 encoder +//! writes a v5 response body. + +use bytes::Bytes; +use kafka_protocol::messages::ApiVersionsRequest; +use kafka_protocol::protocol::Decodable; +use kafka_protocol::protocol::buf::ByteBuf; + +use crate::bounds_guard::validate_api_versions_shape; +use crate::codec::decode_guarded; +use crate::error::Result; + +/// Highest `ApiVersions` version that `kafka-protocol` decodes and encodes. +pub const CODEC_MAX_VERSION: i16 = 4; + +/// The broker that an `ApiVersions` v5 request names (KIP-1242). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct IntendedBroker { + pub cluster_id: String, + pub node_id: i32, +} + +/// Decodes an `ApiVersions` request at `version`, and returns the broker that it names. +/// +/// Only v5 names one, and only with a `ClusterId`. A Java client from 4.4 sends both fields when +/// its metadata knows the cluster, and neither on a connection to a bootstrap server. +/// +/// # Errors +/// +/// [`crate::KafkaWireError::Malformed`] when the body does not decode at `version`. +pub fn decode_request(version: i16, body: Bytes) -> Result<Option<IntendedBroker>> { + if version <= CODEC_MAX_VERSION { + decode_guarded::<ApiVersionsRequest>(version, body, validate_api_versions_shape)?; + return Ok(None); + } + let request = decode_guarded::<RequestV5>(version, body, validate_api_versions_shape)?; + Ok(request.cluster_id.map(|cluster_id| IntendedBroker { + cluster_id, + node_id: request.node_id, + })) +} + +/// The v5 fields after the client software name and version. +struct RequestV5 { + cluster_id: Option<String>, + node_id: i32, +} + +impl Decodable for RequestV5 { + fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> anyhow::Result<Self> { + anyhow::ensure!(version == 5, "ApiVersions v{version} has no v5 fields"); + anyhow::ensure!(compact_string(buf)?.is_some(), "ClientSoftwareName is null"); + anyhow::ensure!( + compact_string(buf)?.is_some(), + "ClientSoftwareVersion is null" + ); + let cluster_id = compact_string(buf)?; + let node_id = buf.try_get_i32()?; + skip_tagged_fields(buf)?; + Ok(Self { + cluster_id, + node_id, + }) + } +} + +/// A compact nullable string: a varint of the byte count plus one, where 0 means null. +fn compact_string<B: ByteBuf>(buf: &mut B) -> anyhow::Result<Option<String>> { + let Some(len) = unsigned_varint(buf)?.checked_sub(1) else { + return Ok(None); + }; + let bytes = buf.try_get_bytes(usize::try_from(len)?)?; + Ok(Some(String::from_utf8(bytes.to_vec())?)) +} + +fn skip_tagged_fields<B: ByteBuf>(buf: &mut B) -> anyhow::Result<()> { + for _ in 0..unsigned_varint(buf)? { + let _tag = unsigned_varint(buf)?; + let size = unsigned_varint(buf)?; + buf.try_get_bytes(usize::try_from(size)?)?; + } + Ok(()) +} + +/// The varint reader of `kafka-protocol`: at most 5 bytes, in `u32`. The bounds guard reads a +/// varint the same way, so its walk and this one stop at the same byte. +fn unsigned_varint<B: ByteBuf>(buf: &mut B) -> anyhow::Result<u32> { + let mut value = 0_u32; + for shift in [0, 7, 14, 21, 28] { + let byte = buf.try_get_u8()?; + value |= u32::from(byte & 0x7F) << shift; + if byte < 0x80 { + break; + } + } + Ok(value) +} + +#[cfg(test)] +mod tests { + use bytes::{BufMut, BytesMut}; + + use super::*; + + fn put_compact_string(body: &mut BytesMut, value: Option<&str>) { + let Some(value) = value else { + body.put_u8(0); + return; + }; + body.put_u8(u8::try_from(value.len() + 1).expect("a short test string")); + body.put_slice(value.as_bytes()); + } + + fn v5_body(cluster_id: Option<&str>, node_id: i32, tagged_fields: &[u8]) -> Bytes { + let mut body = BytesMut::new(); + put_compact_string(&mut body, Some("client")); + put_compact_string(&mut body, Some("1")); + put_compact_string(&mut body, cluster_id); + body.put_i32(node_id); + body.put_slice(tagged_fields); + body.freeze() + } + + #[test] + fn given_v5_with_a_cluster_id_should_return_the_broker_it_names() { + let named = decode_request(5, v5_body(Some("cluster-a"), 2, &[0])).unwrap(); + assert_eq!( + named, + Some(IntendedBroker { + cluster_id: "cluster-a".to_owned(), + node_id: 2, + }) + ); + } + + #[test] + fn given_v5_without_a_cluster_id_should_name_no_broker() { + assert_eq!(decode_request(5, v5_body(None, -1, &[0])).unwrap(), None); + } + + #[test] + fn given_v5_with_an_unknown_tagged_field_should_skip_it() { + // One tagged field: tag 7, two bytes of data. + let body = v5_body(Some("cluster-a"), 2, &[1, 7, 2, 0xAB, 0xCD]); + assert!(decode_request(5, body).unwrap().is_some()); + } + + #[test] + fn given_a_version_below_v5_should_name_no_broker() { + let mut v3 = BytesMut::new(); + put_compact_string(&mut v3, Some("client")); + put_compact_string(&mut v3, Some("1")); + v3.put_u8(0); + let v3 = v3.freeze(); + assert_eq!(decode_request(3, v3.clone()).unwrap(), None); + assert_eq!(decode_request(4, v3).unwrap(), None); + assert_eq!(decode_request(0, Bytes::new()).unwrap(), None); + } + + #[test] + fn given_a_v5_body_with_trailing_bytes_should_be_malformed() { + assert!(decode_request(5, v5_body(Some("cluster-a"), 2, &[0, 0])).is_err()); + } + + #[test] + fn given_a_v4_body_at_v5_should_be_malformed() { + let mut v4 = BytesMut::new(); + put_compact_string(&mut v4, Some("client")); + put_compact_string(&mut v4, Some("1")); + v4.put_u8(0); + assert!(decode_request(5, v4.freeze()).is_err()); + } + + #[test] + fn given_a_cluster_id_that_is_not_utf8_should_be_malformed() { + let mut body = BytesMut::new(); + put_compact_string(&mut body, Some("client")); + put_compact_string(&mut body, Some("1")); + body.put_slice(&[3, 0xFF, 0xFE]); + body.put_i32(2); + body.put_u8(0); + assert!(decode_request(5, body.freeze()).is_err()); + } + + #[test] + fn given_v6_should_be_refused() { + assert!(decode_request(6, v5_body(Some("cluster-a"), 2, &[0])).is_err()); + } +} diff --git a/core/kafka_wire/src/bounds_guard.rs b/core/kafka_wire/src/bounds_guard.rs index 3f22b95e6..a99dfa08f 100644 --- a/core/kafka_wire/src/bounds_guard.rs +++ b/core/kafka_wire/src/bounds_guard.rs @@ -108,7 +108,7 @@ pub const GUARDED_RANGES: &[ApiVersionRange] = &[ range(API_KEY_DESCRIBE_GROUPS, 0, 6), range(API_KEY_LIST_GROUPS, 0, 5), range(API_KEY_SASL_HANDSHAKE, 0, 1), - range(API_KEY_API_VERSIONS, 0, 3), + range(API_KEY_API_VERSIONS, 0, 5), range(API_KEY_CREATE_TOPICS, 2, 5), range(API_KEY_DELETE_TOPICS, 1, 5), range(API_KEY_INIT_PRODUCER_ID, 0, 5), @@ -959,6 +959,8 @@ pub fn validate_init_producer_id_shape(version: i16, body: &Bytes) -> Result<()> /// Mirrors the field order `ApiVersionsRequest::decode` walks. v0-2 have an empty body (no /// length-prefixed fields to bound), so this is a no-op below v3. /// +/// For v5 it follows [`crate::api_versions::decode_request`], because the codec stops at v4. +/// /// # Errors /// /// Returns an error when a declared string length cannot fit in the bytes remaining in the @@ -975,6 +977,11 @@ pub fn validate_api_versions_shape(version: i16, body: &Bytes) -> Result<()> { let mut c = ShapeCursor::new(body.clone(), usize::MAX); c.compact_string(false)?; c.compact_string(false)?; + if version >= 5 { + // `ClusterId` and `NodeId` (KIP-1242). + c.compact_string(true)?; + let _node_id = c.read_i32()?; + } c.tagged_fields()?; Ok(()) } diff --git a/core/kafka_wire/src/lib.rs b/core/kafka_wire/src/lib.rs index 6bd23e5b2..0cf34e058 100644 --- a/core/kafka_wire/src/lib.rs +++ b/core/kafka_wire/src/lib.rs @@ -22,6 +22,7 @@ pub mod acl; pub mod api; +pub mod api_versions; pub mod bounds_guard; pub mod codec; pub mod dedup; diff --git a/core/kafka_wire/tests/version_sweep.rs b/core/kafka_wire/tests/version_sweep.rs index f71dc28f3..77af100ca 100644 --- a/core/kafka_wire/tests/version_sweep.rs +++ b/core/kafka_wire/tests/version_sweep.rs @@ -16,14 +16,15 @@ // under the License. //! Every validator, at every version it guards, against the bytes `kafka_protocol` itself -//! encodes: a well-formed request with one element in every array is accepted and decodes, every -//! strict prefix of it is refused, and the versions just outside the guarded range are refused. +//! encodes, or bytes built by hand where it cannot encode a version: a well-formed request with +//! one element in every array is accepted and decodes, every strict prefix of it is refused, and +//! the versions just outside the guarded range are refused. //! //! This is the crate's proof that each walk stays aligned with the decoder's field order. A walk //! that reads one field too few or too many fails the prefix check, because a cut inside that //! field goes unnoticed or a complete body reads as truncated. -use bytes::{Bytes, BytesMut}; +use bytes::{BufMut, Bytes, BytesMut}; use iggy_kafka_wire::Result; use iggy_kafka_wire::api::{ API_KEY_ALTER_CONFIGS, API_KEY_API_VERSIONS, API_KEY_CREATE_PARTITIONS, API_KEY_CREATE_TOPICS, @@ -34,6 +35,7 @@ use iggy_kafka_wire::api::{ API_KEY_METADATA, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_DELETE, API_KEY_OFFSET_FETCH, API_KEY_PRODUCE, API_KEY_SASL_AUTHENTICATE, API_KEY_SASL_HANDSHAKE, API_KEY_SYNC_GROUP, }; +use iggy_kafka_wire::api_versions::{CODEC_MAX_VERSION, decode_request}; use iggy_kafka_wire::bounds_guard::{ guarded_range, validate_alter_configs_shape, validate_api_versions_shape, validate_create_partitions_shape, validate_create_topics_shape, validate_delete_groups_shape, @@ -113,15 +115,31 @@ fn sweep<R: Encodable + Decodable>( api_key: i16, build: impl Fn(i16) -> R, validate: impl Fn(i16, &Bytes) -> Result<()>, +) { + sweep_bodies( + api_key, + |version| encode(&build(version), version), + validate, + |version, body| decode_exhaustive::<R>(version, body).map(drop), + ); +} + +/// [`sweep`] over bodies that `body` builds and `decode` reads, for versions that +/// `kafka_protocol` cannot encode. +fn sweep_bodies( + api_key: i16, + body: impl Fn(i16) -> Bytes, + validate: impl Fn(i16, &Bytes) -> Result<()>, + decode: impl Fn(i16, Bytes) -> Result<()>, ) { let range = guarded_range(api_key).unwrap_or_else(|| panic!("api key {api_key} has no guarded range")); for version in range.min_version..=range.max_version { - let body = encode(&build(version), version); + let body = body(version); validate(version, &body).unwrap_or_else(|error| { panic!("api {api_key} v{version}: a well-formed body was refused: {error}") }); - if let Err(error) = decode_exhaustive::<R>(version, body.clone()) { + if let Err(error) = decode(version, body.clone()) { panic!("api {api_key} v{version}: the decoder disagrees with the guard: {error}"); } for cut in 0..body.len() { @@ -132,15 +150,13 @@ fn sweep<R: Encodable + Decodable>( ); } } - let body = encode(&build(range.max_version), range.max_version); assert!( - validate(range.max_version + 1, &body).is_err(), + validate(range.max_version + 1, &body(range.max_version)).is_err(), "api {api_key}: v{} is above the guarded range and must be refused", range.max_version + 1 ); - let body = encode(&build(range.min_version), range.min_version); assert!( - validate(range.min_version - 1, &body).is_err(), + validate(range.min_version - 1, &body(range.min_version)).is_err(), "api {api_key}: v{} is below the guarded range and must be refused", range.min_version - 1 ); @@ -439,20 +455,35 @@ fn sasl_handshake() { ); } +/// `kafka_protocol` encodes `ApiVersions` only up to v4, so the v5 body is built by hand, and every +/// version decodes through `decode_request`, as the gateway decodes it. #[test] fn api_versions() { - sweep( + sweep_bodies( API_KEY_API_VERSIONS, |version| { - if version >= 3 { - ApiVersionsRequest::default() + if version > CODEC_MAX_VERSION { + let mut body = BytesMut::new(); + // `ClientSoftwareName`, `ClientSoftwareVersion` and `ClusterId`, as compact + // strings, then `NodeId` and no tagged fields. + for field in ["client", "1", "cluster"] { + body.put_u8(u8::try_from(field.len() + 1).expect("a short field")); + body.put_slice(field.as_bytes()); + } + body.put_i32(1); + body.put_u8(0); + body.freeze() + } else if version >= 3 { + let request = ApiVersionsRequest::default() .with_client_software_name(text("client")) - .with_client_software_version(text("1")) + .with_client_software_version(text("1")); + encode(&request, version) } else { - ApiVersionsRequest::default() + encode(&ApiVersionsRequest::default(), version) } }, validate_api_versions_shape, + |version, body| decode_request(version, body).map(drop), ); } diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index 54299b15e..9474c8783 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -359,6 +359,10 @@ cargo run -p iggy-gateway-kafka as the leader of every partition. Any gateway can serve any partition, because the Iggy SDK sends each partition request to its primary. A client of a stopped or crashed gateway goes on through another at its next Metadata. +- A Java client from Kafka 4.4 names the node that it wants to reach in ApiVersions (KIP-1242). A + gateway with another node id answers `REBOOTSTRAP_REQUIRED` (129), and the client starts again + from its bootstrap servers. So each address in `IGGY_KAFKA_BROKERS` must reach the gateway with + that node id, and not a load balancer in front of several gateways. - The owner of log partition 0 is the controller. Only the controller creates and deletes topics and adds partitions. The others answer `NOT_CONTROLLER` (41), and Kafka admin clients retry at the controller. diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md index ae4788512..14c66b3ce 100644 --- a/gateways/kafka/docs/MANUAL_TESTING.md +++ b/gateways/kafka/docs/MANUAL_TESTING.md @@ -110,7 +110,7 @@ sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="iggy" password="iggy"; EOF docker run --rm --network host -v /tmp/client.properties:/tmp/client.properties:ro \ - apache/kafka:3.9.0 /opt/kafka/bin/kafka-broker-api-versions.sh \ + apache/kafka:4.4.0 /opt/kafka/bin/kafka-broker-api-versions.sh \ --bootstrap-server 127.0.0.1:9095 --command-config /tmp/client.properties ``` @@ -141,7 +141,7 @@ Then list each principal's ACLs with the real admin client, swapping the usernam ```bash docker run --rm --network host -v /tmp/client.properties:/tmp/client.properties:ro \ - apache/kafka:3.9.0 /opt/kafka/bin/kafka-acls.sh \ + apache/kafka:4.4.0 /opt/kafka/bin/kafka-acls.sh \ --bootstrap-server 127.0.0.1:9095 --command-config /tmp/client.properties --list ``` @@ -217,7 +217,7 @@ IGGY_KAFKA_LISTENERS=PLAINTEXT://127.0.0.1:9093,SSL://127.0.0.1:9094,SASL_SSL:// ``` `KC` below is `docker run --rm --network host -v /tmp/certs:/certs:ro edenhill/kcat:1.7.1`, and -`KAFKA` is `docker run --rm --network host -v /tmp/certs:/certs:ro -v /tmp/client.properties:/tmp/client.properties:ro apache/kafka:3.9.0`. +`KAFKA` is `docker run --rm --network host -v /tmp/certs:/certs:ro -v /tmp/client.properties:/tmp/client.properties:ro apache/kafka:4.4.0`. | ID | Test | Command | Pass criteria | Last run | | ---- | ------ | --------- | --------------- | ---------- | @@ -376,9 +376,13 @@ unset). With a bridge, A5 writes one record to `test-topic`: `ec=0` if it exists For each API key, test **min−1**, **min**, **max**, **max+1** using `kafka-message-gen send --host 127.0.0.1:9093` with `--version N`. +`kafka-message-gen` encodes ApiVersions only up to v4, as `kafka-protocol` 0.18 does. For v5, run +`kafka-broker-api-versions.sh` from `apache/kafka:4.4.0`. It prints +`ApiVersions(18): 0 to 5 [usable: 5]`. + | API key | Name | Min | Max | Test versions | | --------- | ------ | ----- | ----- | --------------- | -| 18 | ApiVersions | 0 | 3 | −1, 0, 3, 4 | +| 18 | ApiVersions | 0 | 5 | −1, 0, 5, 6 | | 3 | Metadata | 0 | 9 | −1, 0, 9, 10 | | 0 | Produce | 3 | 9 | 2, 3, 9, 10 | | 1 | Fetch | 4 | 12 | 3, 4, 12, 13 | diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md index c70b528a8..55ad39a9a 100644 --- a/gateways/kafka/docs/SCOPE.md +++ b/gateways/kafka/docs/SCOPE.md @@ -98,7 +98,7 @@ Use this table when configuring clients or generating wire fixtures with `kafka- | 14 | SyncGroup | 0–5 | v4 | | 15 | DescribeGroups | 0–6 | v5 | | 16 | ListGroups | 0–5 | v3 | -| 18 | ApiVersions | 0–3 | v3 | +| 18 | ApiVersions | 0–5 | v3 | | 19 | CreateTopics | 2–5 | v5 | | 20 | DeleteTopics | 1–5 | v4 | | 22 | InitProducerId | 0–5 | v2 | @@ -110,6 +110,13 @@ Use this table when configuring clients or generating wire fixtures with `kafka- | 47 | OffsetDelete | 0 | never | | 60 | DescribeCluster | 0–2 | v0 | +The gateway supports Java clients from Kafka 4.0 on. Clients 4.0 to 4.3 open ApiVersions at v4, +and 4.4 clients open it at v5. v5 names the cluster and the node that the client wants to reach +(KIP-1242). If either one is not this gateway's, the gateway answers `REBOOTSTRAP_REQUIRED` (129), +and the client starts again from its bootstrap servers. `kafka-protocol` 0.18 stops at +ApiVersions v4, so `iggy_kafka_wire::api_versions` decodes the two v5 fields. The v5 response has +no new field, so the v4 encoder writes it. + ### Cluster id Metadata v2+ and DescribeCluster report one id, so `kafka-cluster.sh cluster-id`, Connect and diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index ed5744678..73a4c7866 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -110,13 +110,16 @@ already-built `iggy-server` in the same target directory, and a host JDK 17+ (`j consumer `offsetsForTimes` probe. Missing any of them makes it skip with a printed reason rather than fail, which is what lets `cargo test -p iggy-gateway-kafka` stay usable without them. -Client containers run with `--pull never`, so pull the two images first. Without them the suite -skips and names the missing one. +Client containers run with `--pull never`, so pull the images first. Without them the suite +skips and names the missing one. The Java tools come from Kafka 4.4.0, the newest Kafka 4 release. +One test also runs the tools of 4.0.2, the oldest Kafka 4 release. Clients 4.0 to 4.3 open +ApiVersions at v4, and 4.4 clients open it at v5. ```bash cargo build --bin iggy-server docker pull edenhill/kcat:1.7.1 -docker pull apache/kafka:3.9.0 +docker pull apache/kafka:4.4.0 +docker pull apache/kafka:4.0.2 KAFKA_E2E_REQUIRED=1 cargo test -p iggy-gateway-kafka --test kafka_client_e2e_tests ``` @@ -133,14 +136,14 @@ because they cover cases a test does not assert, but the load-bearing ones now r ## Failover suite -`cluster_failover_tests.rs` uses the same two images and `KAFKA_E2E_REQUIRED` as the real-client -suite, and skips in the same way. It needs no JDK. Cargo builds the gateway binary for the test, -but `iggy-server` comes from another package, so build it first: +`cluster_failover_tests.rs` uses the kcat image, the Kafka 4.4.0 image and `KAFKA_E2E_REQUIRED`, as +the real-client suite does. It skips in the same way. It needs no JDK. Cargo builds the gateway +binary for the test, but `iggy-server` comes from another package, so build it first: ```bash cargo build --bin iggy-server docker pull edenhill/kcat:1.7.1 -docker pull apache/kafka:3.9.0 +docker pull apache/kafka:4.4.0 KAFKA_E2E_REQUIRED=1 cargo test -p iggy-gateway-kafka --test cluster_failover_tests ``` diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md index 654711651..07bf6e1eb 100644 --- a/gateways/kafka/docs/kafka_api_keys_reference.md +++ b/gateways/kafka/docs/kafka_api_keys_reference.md @@ -291,7 +291,7 @@ Key new minimums: | Fetch | v4-v12 | v17 | 5 versions behind | | ListOffsets | v1-v7 | v9 | 2 versions behind | | Metadata | v0-v9 | v12 | 3 versions behind | -| ApiVersions | v0-v3 | v4 | 1 version behind | +| ApiVersions | v0-v5 | v4 | current, with v5 of Kafka 4.4 (KIP-1242) | | CreateTopics | v2-v5 | v7 | 2 versions behind | | FindCoordinator | v0-v4 | v6 | 2 versions behind | | JoinGroup | v0-v9 | v9 | current | diff --git a/gateways/kafka/src/protocol/handlers/api_versions.rs b/gateways/kafka/src/protocol/handlers/api_versions.rs index cdcc1acbe..72f4a65e3 100644 --- a/gateways/kafka/src/protocol/handlers/api_versions.rs +++ b/gateways/kafka/src/protocol/handlers/api_versions.rs @@ -18,22 +18,28 @@ //! `ApiVersions` (API key 18). use bytes::Bytes; +use iggy_kafka_wire::api_versions::{CODEC_MAX_VERSION, decode_request}; +use kafka_protocol::messages::ApiVersionsResponse; use kafka_protocol::messages::api_versions_response::ApiVersion; -use kafka_protocol::messages::{ApiVersionsRequest, ApiVersionsResponse}; use crate::error::Result; use crate::protocol::api::{ API_KEY_API_VERSIONS, ApiVersionRange, ConnectionState, ERROR_INVALID_REQUEST, ERROR_NONE, - ERROR_UNSUPPORTED_VERSION, HandleOutcome, acl_advertised_ranges, advertised_min_version, - is_supported_version, sasl_advertised_ranges, supported_api_ranges, + ERROR_REBOOTSTRAP_REQUIRED, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome, + acl_advertised_ranges, advertised_min_version, is_supported_version, sasl_advertised_ranges, + supported_api_ranges, }; -use crate::protocol::bounds_guard::validate_api_versions_shape; -use crate::protocol::handlers::{decode_guarded, encode_message, respond_or_close}; +use crate::protocol::handlers::{encode_message, respond_or_close}; +/// Java clients 4.0 to 4.3 open at v4, and 4.4 clients at v5, so both get an answer without a +/// KIP-511 retry. +/// +/// v4 changes only how a broker reports a feature at version 0, and this gateway reports no +/// features. v5 names the broker that the client means to reach (KIP-1242). pub const RANGE: ApiVersionRange = ApiVersionRange { api_key: API_KEY_API_VERSIONS, min_version: 0, - max_version: 3, + max_version: 5, }; /// Which of the listener-dependent keys one connection is told about. @@ -63,7 +69,12 @@ impl Advertisement { clippy::unused_async, reason = "the shared handler signature, kept until a handler awaits the bridge" )] -pub async fn handle(connection: &ConnectionState, api_version: i16, body: Bytes) -> HandleOutcome { +pub async fn handle( + state: &GatewayState, + connection: &ConnectionState, + api_version: i16, + body: Bytes, +) -> HandleOutcome { if !is_supported_version(API_KEY_API_VERSIONS, api_version) { // KIP-511: reply with v0 when the requested version is not understood. return respond_or_close( @@ -71,25 +82,35 @@ pub async fn handle(connection: &ConnectionState, api_version: i16, body: Bytes) "ApiVersions", ); } - match decode_guarded::<ApiVersionsRequest>(api_version, body, validate_api_versions_shape) { - Ok(_) => respond_or_close( - encode_response(api_version, ERROR_NONE, connection.advertisement()), - "ApiVersions", - ), + let error_code = match decode_request(api_version, body) { + // KIP-1242, as a Kafka broker applies it: the client's metadata names another cluster or + // another node for this address. That follows an address that moved to another gateway, + // or a gateway without the bridge that restarted with a new random cluster id. + Ok(Some(intended)) + if intended.cluster_id != state.cluster_id.as_str() + || intended.node_id != state.node_id => + { + tracing::debug!( + cluster_id = intended.cluster_id, + node_id = intended.node_id, + gateway_cluster_id = %state.cluster_id, + gateway_node_id = state.node_id, + "ApiVersions names another broker, so the client must rebootstrap" + ); + ERROR_REBOOTSTRAP_REQUIRED + } + Ok(_) => ERROR_NONE, Err(error) => { // debug!, not warn!: attacker-controlled, not operator-actionable (see the same // note on the Produce decode-failure arm). tracing::debug!(%error, "failed to decode ApiVersions request"); - respond_or_close( - encode_response( - api_version, - ERROR_INVALID_REQUEST, - connection.advertisement(), - ), - "ApiVersions", - ) + ERROR_INVALID_REQUEST } - } + }; + respond_or_close( + encode_response(api_version, error_code, connection.advertisement()), + "ApiVersions", + ) } /// Advertise exactly what the firewall allows, so a client never negotiates a version this @@ -137,17 +158,17 @@ pub fn encode_response( let resp = ApiVersionsResponse::default() .with_error_code(error_code) .with_api_keys(api_keys); - encode_message(&resp, api_version, 128) + // The v5 response has the fields of v4, and `kafka-protocol` encodes only up to v4. + encode_message(&resp, api_version.min(CODEC_MAX_VERSION), 128) } #[cfg(test)] mod tests { use super::*; - use iggy_kafka_wire::codec::decode_exhaustive; #[test] fn decode_v3_requires_software_fields() { - assert!(decode_exhaustive::<ApiVersionsRequest>(3, Bytes::new()).is_err()); + assert!(decode_request(3, Bytes::new()).is_err()); } #[test] @@ -161,6 +182,6 @@ mod tests { 0x06, b'0', b'.', b'1', b'.', b'0', // compact string (len 5) 0x00, // empty tagged fields ]); - decode_exhaustive::<ApiVersionsRequest>(3, body).unwrap(); + assert_eq!(decode_request(3, body).unwrap(), None); } } diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs index dacf2979c..cb3ec4a43 100644 --- a/gateways/kafka/src/protocol/handlers/mod.rs +++ b/gateways/kafka/src/protocol/handlers/mod.rs @@ -126,7 +126,7 @@ pub async fn dispatch( API_KEY_METADATA => metadata::handle(state, connection, principal, api_version, body).await, API_KEY_OFFSET_COMMIT => offset_commit::handle(state, connection, api_version, body).await, API_KEY_OFFSET_FETCH => offset_fetch::handle(state, connection, api_version, body).await, - API_KEY_API_VERSIONS => api_versions::handle(connection, api_version, body).await, + API_KEY_API_VERSIONS => api_versions::handle(state, connection, api_version, body).await, API_KEY_CREATE_TOPICS => create_topics::handle(state, connection, api_version, body).await, API_KEY_DELETE_TOPICS => { delete_topics::handle(state, connection, principal, api_version, body).await diff --git a/gateways/kafka/tests/api_handler_tests.rs b/gateways/kafka/tests/api_handler_tests.rs index 4491b8d8f..1832e7c94 100644 --- a/gateways/kafka/tests/api_handler_tests.rs +++ b/gateways/kafka/tests/api_handler_tests.rs @@ -34,7 +34,7 @@ use iggy_gateway_kafka::cluster_id::ClusterId; use iggy_gateway_kafka::protocol::api::{ API_KEY_API_VERSIONS, API_KEY_CREATE_PARTITIONS, API_KEY_CREATE_TOPICS, API_KEY_DELETE_TOPICS, API_KEY_FETCH, API_KEY_LIST_OFFSETS, API_KEY_METADATA, API_KEY_PRODUCE, ERROR_INVALID_REQUEST, - ERROR_NONE, ERROR_NOT_CONTROLLER, ERROR_NOT_LEADER_OR_FOLLOWER, + ERROR_NONE, ERROR_NOT_CONTROLLER, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_REBOOTSTRAP_REQUIRED, ERROR_UNKNOWN_TOPIC_OR_PARTITION, ERROR_UNSUPPORTED_VERSION, GatewayState, handle_request, handle_request_bounded, is_supported_version, supported_api_ranges, }; @@ -44,10 +44,11 @@ use fixtures::load_fixture_body_or_skip; use scope::default_broker; use tcp::{build_metadata_legacy_request, build_produce_v3_body}; use wire::{ - build_api_versions_flexible_request, build_create_partitions_request, - build_delete_topics_request_with_name, build_metadata_all_topics_legacy, - build_metadata_flexible_request, build_metadata_flexible_request_v10, - build_metadata_legacy_request_for_version, build_produce_legacy_request, + build_api_versions_flexible_request, build_api_versions_v5_request, + build_create_partitions_request, build_delete_topics_request_with_name, + build_metadata_all_topics_legacy, build_metadata_flexible_request, + build_metadata_flexible_request_v10, build_metadata_legacy_request_for_version, + build_produce_legacy_request, }; // ── ApiVersions ───────────────────────────────────────────────────────────── @@ -113,6 +114,82 @@ async fn api_versions_v3_response_flexible_format() { } } +/// `kafka-protocol` encodes `ApiVersions` only up to v4, and v5 adds no response field, so a v5 +/// request gets the v4 body. +#[tokio::test] +async fn api_versions_v5_answers_with_the_v4_body() { + let v4 = handle_request( + API_KEY_API_VERSIONS, + 4, + build_api_versions_flexible_request("iggy-test", "0.1.0"), + &default_broker(), + ) + .await + .expect_response("ApiVersions v4 answers"); + let v5 = handle_request( + API_KEY_API_VERSIONS, + 5, + build_api_versions_v5_request(None, -1), + &default_broker(), + ) + .await + .expect_response("ApiVersions v5 answers"); + assert_eq!(v4, v5); + assert_eq!(Decoder::new(v5).read_i16().unwrap(), ERROR_NONE); +} + +/// KIP-1242: a Java client from 4.4 names the cluster and the node that its metadata gives for +/// this address. The gateway serves the request when both are its own, or when the client names +/// none, as it does on a connection to a bootstrap server. +#[tokio::test] +async fn api_versions_v5_naming_this_gateway_answers_the_versions() { + let state = GatewayState::stub(8 * 1024 * 1024).with_node_id(2); + let cluster_id = state.cluster_id.to_string(); + for body in [ + build_api_versions_v5_request(Some(&cluster_id), 2), + build_api_versions_v5_request(None, -1), + ] { + let response = handle_request_bounded(&state, API_KEY_API_VERSIONS, 5, body) + .await + .expect_response("ApiVersions answers"); + let mut d = Decoder::new(response); + assert_eq!(d.read_i16().unwrap(), ERROR_NONE); + assert_eq!( + usize::try_from(d.read_varint().unwrap() - 1).expect("api count fits usize"), + supported_api_ranges().len() + ); + } +} + +/// KIP-1242: a client whose metadata names another node or another cluster for this address +/// gets `REBOOTSTRAP_REQUIRED`, as from a Kafka broker, and starts again from its bootstrap +/// servers. +#[tokio::test] +async fn api_versions_v5_naming_another_broker_asks_for_a_rebootstrap() { + let state = GatewayState::stub(8 * 1024 * 1024).with_node_id(2); + let cluster_id = state.cluster_id.to_string(); + let another_cluster = ClusterId::random().to_string(); + for (body, named) in [ + ( + build_api_versions_v5_request(Some(&cluster_id), 3), + "another node", + ), + ( + build_api_versions_v5_request(Some(&another_cluster), 2), + "another cluster", + ), + ] { + let response = handle_request_bounded(&state, API_KEY_API_VERSIONS, 5, body) + .await + .expect_response("REBOOTSTRAP_REQUIRED is an answer, not a close"); + assert_eq!( + Decoder::new(response).read_i16().unwrap(), + ERROR_REBOOTSTRAP_REQUIRED, + "{named}" + ); + } +} + // ── Metadata ───────────────────────────────────────────────────────────────── #[tokio::test] @@ -188,8 +265,8 @@ async fn unknown_api_key_closes_connection() { #[test] fn version_support_table_is_applied() { - assert!(is_supported_version(API_KEY_API_VERSIONS, 3)); - assert!(!is_supported_version(API_KEY_API_VERSIONS, 10)); + assert!(is_supported_version(API_KEY_API_VERSIONS, 5)); + assert!(!is_supported_version(API_KEY_API_VERSIONS, 6)); assert!(is_supported_version(API_KEY_METADATA, 1)); assert!(!is_supported_version(API_KEY_METADATA, -1)); } @@ -340,7 +417,9 @@ async fn handle_request_succeeds_for_every_supported_version_with_fixture() { } if api_key == API_KEY_API_VERSIONS { for version in min_ver..=max_ver { - let body = if version >= 3 { + let body = if version >= 5 { + build_api_versions_v5_request(None, -1) + } else if version >= 3 { build_api_versions_flexible_request("iggy-test", "0.1.0") } else { Bytes::new() diff --git a/gateways/kafka/tests/cluster_failover_tests.rs b/gateways/kafka/tests/cluster_failover_tests.rs index b2653ba6c..ca3766b9e 100644 --- a/gateways/kafka/tests/cluster_failover_tests.rs +++ b/gateways/kafka/tests/cluster_failover_tests.rs @@ -100,7 +100,7 @@ use iggy_gateway_kafka::protocol::api::{ use deployment::{Deployment, KAFKA_STREAM, NODES, TAIL_LINES, tail}; use docker_clients::{Container, missing_prerequisite, run_to_end, skip}; -const KAFKA_IMAGE: &str = "apache/kafka:3.9.0"; +const KAFKA_IMAGE: &str = "apache/kafka:4.4.0"; const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1"; /// The topic of the load. Two partitions per node, so a stop of any node moves primaries. diff --git a/gateways/kafka/tests/common/scope.rs b/gateways/kafka/tests/common/scope.rs index 93bbbe169..c0a0d5651 100644 --- a/gateways/kafka/tests/common/scope.rs +++ b/gateways/kafka/tests/common/scope.rs @@ -39,7 +39,7 @@ pub const SCOPED_API_KEYS: &[(i16, &str, i16, i16)] = &[ (14, "SyncGroup", 0, 5), (15, "DescribeGroups", 0, 6), (16, "ListGroups", 0, 5), - (18, "ApiVersions", 0, 3), + (18, "ApiVersions", 0, 5), (19, "CreateTopics", 2, 5), (20, "DeleteTopics", 1, 5), (22, "InitProducerId", 0, 5), diff --git a/gateways/kafka/tests/common/wire.rs b/gateways/kafka/tests/common/wire.rs index 9f0fce73a..e09b95a2a 100644 --- a/gateways/kafka/tests/common/wire.rs +++ b/gateways/kafka/tests/common/wire.rs @@ -102,6 +102,19 @@ pub fn build_api_versions_flexible_request(software_name: &str, software_version enc.freeze() } +/// `ApiVersions` v5 body: the v3 fields, then the `ClusterId` and `NodeId` of the broker that the +/// client means to reach (KIP-1242). A client sends a null `ClusterId` and `NodeId` -1 to a +/// bootstrap server. +pub fn build_api_versions_v5_request(cluster_id: Option<&str>, node_id: i32) -> Bytes { + let mut enc = Encoder::with_capacity(64); + enc.write_compact_nullable_string(Some("iggy-test")); + enc.write_compact_nullable_string(Some("0.1.0")); + enc.write_compact_nullable_string(cluster_id); + enc.write_i32(node_id); + enc.write_empty_tagged_fields(); + enc.freeze() +} + /// Legacy Metadata request body for a specific version (v0–v8). pub fn build_metadata_legacy_request_for_version(version: i16, topic_names: &[&str]) -> Bytes { let mut enc = Encoder::with_capacity(64); diff --git a/gateways/kafka/tests/configs_java_client_tests.rs b/gateways/kafka/tests/configs_java_client_tests.rs index bdefa6ce4..17ad8d0b3 100644 --- a/gateways/kafka/tests/configs_java_client_tests.rs +++ b/gateways/kafka/tests/configs_java_client_tests.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -//! Kafka 3.9 `AdminClient` describe/alter against the gateway, on the host JDK. +//! Kafka 4.4 `AdminClient` describe/alter against the gateway, on the host JDK. //! //! The docker `--network host` suite cannot reach a loopback gateway from macOS. //! This test publishes nothing into a container: it binds `127.0.0.1` and runs @@ -131,8 +131,8 @@ fn try_output_bounded(command: &mut Command, what: &str) -> Result<Output, Strin const JARS: &[(&str, &str)] = &[ ( - "kafka-clients-3.9.0.jar", - "https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.9.0/kafka-clients-3.9.0.jar";, + "kafka-clients-4.4.0.jar", + "https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/4.4.0/kafka-clients-4.4.0.jar";, ), ( "slf4j-api-1.7.36.jar", @@ -143,21 +143,21 @@ const JARS: &[(&str, &str)] = &[ "https://repo1.maven.org/maven2/org/slf4j/slf4j-simple/1.7.36/slf4j-simple-1.7.36.jar";, ), ( - "lz4-java-1.8.0.jar", - "https://repo1.maven.org/maven2/org/lz4/lz4-java/1.8.0/lz4-java-1.8.0.jar";, + "lz4-java-1.11.2.jar", + "https://repo1.maven.org/maven2/at/yawk/lz4/lz4-java/1.11.2/lz4-java-1.11.2.jar";, ), ( - "snappy-java-1.1.10.5.jar", - "https://repo1.maven.org/maven2/org/xerial/snappy/snappy-java/1.1.10.5/snappy-java-1.1.10.5.jar";, + "snappy-java-1.1.10.7.jar", + "https://repo1.maven.org/maven2/org/xerial/snappy/snappy-java/1.1.10.7/snappy-java-1.1.10.7.jar";, ), ( - "zstd-jni-1.5.6-4.jar", - "https://repo1.maven.org/maven2/com/github/luben/zstd-jni/1.5.6-4/zstd-jni-1.5.6-4.jar";, + "zstd-jni-1.5.7-17.jar", + "https://repo1.maven.org/maven2/com/github/luben/zstd-jni/1.5.7-17/zstd-jni-1.5.7-17.jar";, ), ]; fn jar_dir() -> PathBuf { - PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../target/kafka-3.9-admin-jars") + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../target/kafka-4.4-admin-jars") } fn ensure_jars(dir: &Path) -> Result<(), String> { @@ -232,7 +232,7 @@ async fn spawn_gateway(server: &TestServer) -> SocketAddr { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn kafka_39_admin_client_describes_and_alters_topic_configs() { +async fn kafka_4_admin_client_describes_and_alters_topic_configs() { if java_missing() { return; } diff --git a/gateways/kafka/tests/golden_wire_fixtures_tests.rs b/gateways/kafka/tests/golden_wire_fixtures_tests.rs index 7a25ce96b..0152f082a 100644 --- a/gateways/kafka/tests/golden_wire_fixtures_tests.rs +++ b/gateways/kafka/tests/golden_wire_fixtures_tests.rs @@ -59,7 +59,7 @@ async fn golden_apiversions_v3_flexible_response_fixture() { 0x00, 0x0E, 0x00, 0x00, 0x00, 0x05, 0x00, // key 14: SyncGroup 0-5 0x00, 0x0F, 0x00, 0x00, 0x00, 0x06, 0x00, // key 15: DescribeGroups 0-6 0x00, 0x10, 0x00, 0x00, 0x00, 0x05, 0x00, // key 16: ListGroups 0-5 - 0x00, 0x12, 0x00, 0x00, 0x00, 0x03, 0x00, // key 18: ApiVersions 0-3 + 0x00, 0x12, 0x00, 0x00, 0x00, 0x05, 0x00, // key 18: ApiVersions 0-5 0x00, 0x13, 0x00, 0x02, 0x00, 0x05, 0x00, // key 19: CreateTopics 2-5 0x00, 0x14, 0x00, 0x01, 0x00, 0x05, 0x00, // key 20: DeleteTopics 1-5 0x00, 0x16, 0x00, 0x00, 0x00, 0x05, 0x00, // key 22: InitProducerId 0-5 @@ -100,7 +100,7 @@ async fn golden_apiversions_v1_response_fixture() { 0x00, 0x0E, 0x00, 0x00, 0x00, 0x05, // key 14: SyncGroup 0-5 0x00, 0x0F, 0x00, 0x00, 0x00, 0x06, // key 15: DescribeGroups 0-6 0x00, 0x10, 0x00, 0x00, 0x00, 0x05, // key 16: ListGroups 0-5 - 0x00, 0x12, 0x00, 0x00, 0x00, 0x03, // key 18: ApiVersions 0-3 + 0x00, 0x12, 0x00, 0x00, 0x00, 0x05, // key 18: ApiVersions 0-5 0x00, 0x13, 0x00, 0x02, 0x00, 0x05, // key 19: CreateTopics 2-5 0x00, 0x14, 0x00, 0x01, 0x00, 0x05, // key 20: DeleteTopics 1-5 0x00, 0x16, 0x00, 0x00, 0x00, 0x05, // key 22: InitProducerId 0-5 diff --git a/gateways/kafka/tests/java/ConfigsProbe.java b/gateways/kafka/tests/java/ConfigsProbe.java index b01671269..167bed7e5 100644 --- a/gateways/kafka/tests/java/ConfigsProbe.java +++ b/gateways/kafka/tests/java/ConfigsProbe.java @@ -31,14 +31,14 @@ import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.errors.InvalidConfigurationException; /** - * Kafka 3.9 AdminClient probe for DescribeConfigs, the deprecated AlterConfigs API and - * IncrementalAlterConfigs, the only alter API a 4.x client has. - * Prints elapsed time for 50 describes and does not fail on a latency threshold. + * Kafka 4.4 AdminClient probe for DescribeConfigs and IncrementalAlterConfigs, the only alter API + * that a Kafka 4 client has. Kafka 4.0 removed {@code Admin.alterConfigs}, so the AlterConfigs API + * is tested on the wire instead. Prints elapsed time for 50 describes and does not fail on a + * latency threshold. */ public final class ConfigsProbe { private ConfigsProbe() {} - @SuppressWarnings("deprecation") public static void main(String[] args) throws Exception { if (args.length != 2) { throw new IllegalArgumentException("usage: ConfigsProbe <bootstrap> <topic>"); @@ -68,7 +68,9 @@ public final class ConfigsProbe { System.out.println("cleanup.policy=absent"); try { - admin.alterConfigs(Map.of(resource, config("no.such", "1"))).all().get(); + admin.incrementalAlterConfigs(Map.of(resource, List.of(op("no.such", "1", AlterConfigOp.OpType.SET)))) + .all() + .get(); throw new AssertionError("unknown key was accepted"); } catch (ExecutionException exception) { if (!(exception.getCause() instanceof InvalidConfigurationException)) { @@ -80,14 +82,18 @@ public final class ConfigsProbe { Config stillDefault = describe(admin, resource); require("-1".equals(stillDefault.get("retention.ms").value()), "unknown key wrote retention"); - admin.alterConfigs(Map.of(resource, config("retention.ms", "8000")), new AlterConfigsOptions().validateOnly(true)) + admin.incrementalAlterConfigs( + Map.of(resource, List.of(op("retention.ms", "8000", AlterConfigOp.OpType.SET))), + new AlterConfigsOptions().validateOnly(true)) .all() .get(); Config validated = describe(admin, resource); require("-1".equals(validated.get("retention.ms").value()), "validate_only wrote retention"); System.out.println("validate_only_retention.ms=" + validated.get("retention.ms").value()); - admin.alterConfigs(Map.of(resource, config("retention.ms", "8000"))).all().get(); + admin.incrementalAlterConfigs(Map.of(resource, List.of(op("retention.ms", "8000", AlterConfigOp.OpType.SET)))) + .all() + .get(); Config altered = describe(admin, resource); ConfigEntry changed = altered.get("retention.ms"); require("8000".equals(changed.value()), "altered retention.ms"); @@ -136,10 +142,6 @@ public final class ConfigsProbe { return admin.describeConfigs(List.of(resource)).all().get().get(resource); } - private static Config config(String name, String value) { - return new Config(List.of(new ConfigEntry(name, value))); - } - private static AlterConfigOp op(String name, String value, AlterConfigOp.OpType type) { return new AlterConfigOp(new ConfigEntry(name, value), type); } diff --git a/gateways/kafka/tests/java/OffsetsForTimesProbe.java b/gateways/kafka/tests/java/OffsetsForTimesProbe.java index 7c8f39524..602c7486f 100644 --- a/gateways/kafka/tests/java/OffsetsForTimesProbe.java +++ b/gateways/kafka/tests/java/OffsetsForTimesProbe.java @@ -27,7 +27,7 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.ByteArrayDeserializer; /** - * Kafka 3.9 consumer probe for {@code KafkaConsumer.offsetsForTimes}, which plain consumers and + * Kafka 4 consumer probe for {@code KafkaConsumer.offsetsForTimes}, which plain consumers and * Kafka Streams seek by time with. Flink, Spark and {@code --by-duration} resets go through * {@code AdminClient.listOffsets} instead. Prints one {@code time=<ms> offset=<offset|null>} line * per time. diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs index 3fdcaf3ef..73a9ea7b6 100644 --- a/gateways/kafka/tests/kafka_client_e2e_tests.rs +++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs @@ -82,7 +82,10 @@ use iggy_gateway_kafka::tls::{ClientAuth, TlsConfig}; use server::{spawn_gateway as spawn_listeners, spawn_test_server_with_authenticator}; const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1"; -const KAFKA_IMAGE: &str = "apache/kafka:3.9.0"; +/// The newest Kafka 4 release. The gateway supports Java clients from Kafka 4.0 on. +const KAFKA_IMAGE: &str = "apache/kafka:4.4.0"; +/// The oldest Kafka 4 release, for the one test that runs both ends of the range. +const KAFKA_OLDEST_IMAGE: &str = "apache/kafka:4.0.2"; const ROOT_USER: &str = "iggy"; const ROOT_PASSWORD: &str = "iggy"; /// Password for every non-root principal these tests create. @@ -1038,6 +1041,77 @@ async fn given_the_java_client_should_authenticate_and_list_the_sasl_apis() { ); } +/// Java clients 4.0 to 4.3 open `ApiVersions` at v4, and 4.4 clients at v5 (KIP-1242). Each finds +/// its version in the advertisement, so neither asks again at v3 (KIP-511). The oldest Kafka 4 +/// client then produces and reads back in a group, as the newest one does in the other tests. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_kafka_4_0_and_4_4_clients_should_use_their_own_api_versions_and_serve_records() { + let Some((_server, gateway)) = bridged_stack().await else { + return; + }; + if !image_present(KAFKA_OLDEST_IMAGE) + && skip(&format!( + "client image {KAFKA_OLDEST_IMAGE} is not pulled; run `docker pull {KAFKA_OLDEST_IMAGE}`" + )) + { + return; + } + let bootstrap = gateway.to_string(); + for (image, usable) in [(KAFKA_OLDEST_IMAGE, 4), (KAFKA_IMAGE, 5)] { + let versions = run_client( + image, + &[ + "/opt/kafka/bin/kafka-broker-api-versions.sh", + "--bootstrap-server", + &bootstrap, + ], + &[], + ) + .expect_ran("java client api-versions"); + assert!( + versions.contains(&format!("ApiVersions(18): 0 to 5 [usable: {usable}]")), + "{image} must use ApiVersions v{usable}, got: {versions}" + ); + } + + let produce = format!( + "printf 'k1:alpha\\nk2:beta\\n' | /opt/kafka/bin/kafka-console-producer.sh \ + --bootstrap-server {bootstrap} --topic {TOPIC} --property parse.key=true \ + --property key.separator=: --producer-property max.block.ms=20000" + ); + let produced = run_client(KAFKA_OLDEST_IMAGE, &["sh", "-c", &produce], &[]) + .expect_ran("java 4.0 console producer"); + assert!( + !produced.contains("ERROR"), + "the 4.0 producer must send, got: {produced}" + ); + let consumed = run_client( + KAFKA_OLDEST_IMAGE, + &[ + "/opt/kafka/bin/kafka-console-consumer.sh", + "--bootstrap-server", + &bootstrap, + "--topic", + TOPIC, + "--group", + "kafka-4-0", + "--from-beginning", + "--max-messages", + "2", + "--timeout-ms", + "20000", + "--property", + "print.key=true", + ], + &[], + ) + .expect_ran("java 4.0 console consumer in a group"); + assert!( + consumed.contains("k1\talpha") && consumed.contains("k2\tbeta"), + "the 4.0 consumer must read both records, got: {consumed}" + ); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn given_distinct_principals_when_listing_acls_should_describe_each_differently() { // The spec's category T procedure. Root alone proves nothing about the mapping, because it diff --git a/gateways/kafka/tests/version_firewall_tests.rs b/gateways/kafka/tests/version_firewall_tests.rs index b51e8e546..8a753f7a7 100644 --- a/gateways/kafka/tests/version_firewall_tests.rs +++ b/gateways/kafka/tests/version_firewall_tests.rs @@ -58,11 +58,11 @@ use tcp::{ }; use wire::{ JoinGroupParams, OUT_OF_SCOPE_API_KEYS, SyncGroupParams, build_alter_configs_empty_request, - build_api_versions_flexible_request, build_create_partitions_empty_request, - build_create_topics_empty_request, build_delete_groups_request, - build_delete_topics_empty_request, build_describe_configs_empty_request, - build_describe_groups_request, build_fetch_empty_topics_request, - build_find_coordinator_request, build_heartbeat_request, + build_api_versions_flexible_request, build_api_versions_v5_request, + build_create_partitions_empty_request, build_create_topics_empty_request, + build_delete_groups_request, build_delete_topics_empty_request, + build_describe_configs_empty_request, build_describe_groups_request, + build_fetch_empty_topics_request, build_find_coordinator_request, build_heartbeat_request, build_incremental_alter_configs_empty_request, build_init_producer_id_request, build_join_group_request, build_leave_group_request, build_list_groups_request, build_list_offsets_request, build_metadata_all_topics_flexible, @@ -187,12 +187,8 @@ async fn apiversions_advertises_exact_supported_ranges_v3_flexible() { #[tokio::test] async fn apiversions_all_versions_return_success() { - for version in 0i16..=3 { - let request = if version >= 3 { - build_api_versions_flexible_request("iggy-test", "0.1.0") - } else { - Bytes::new() - }; + for version in 0i16..=5 { + let request = api_versions_body(version); let body = handle_request(API_KEY_API_VERSIONS, version, request, &default_broker()) .await .expect_response("test request has acks != 0 and expects a response"); @@ -201,6 +197,18 @@ async fn apiversions_all_versions_return_success() { } } +/// The body a client sends at `version`. From v5, it names no broker, as on a connection to a +/// bootstrap server. +fn api_versions_body(version: i16) -> Bytes { + if version >= 5 { + build_api_versions_v5_request(None, -1) + } else if version >= 3 { + build_api_versions_flexible_request("iggy-test", "0.1.0") + } else { + Bytes::new() + } +} + #[tokio::test] async fn apiversions_out_of_range_returns_unsupported_in_body() { let body = handle_request(API_KEY_API_VERSIONS, 99, Bytes::new(), &default_broker()) @@ -470,13 +478,7 @@ fn request_body_for_scoped_api(api_key: i16, name: &str, version: i16) -> Bytes build_metadata_all_topics_legacy(version) } } - API_KEY_API_VERSIONS => { - if version >= 3 { - build_api_versions_flexible_request("iggy-test", "0.1.0") - } else { - Bytes::new() - } - } + API_KEY_API_VERSIONS => api_versions_body(version), API_KEY_PRODUCE => { // `_or_skip` (not bare `load_fixture_body`) so `KAFKA_FIXTURES_REQUIRED=1` panics on // a missing in-range fixture instead of this falling through to a synthetic body and @@ -631,15 +633,15 @@ async fn apiversions_v0_and_v2_e2e_return_success() { } #[tokio::test] -async fn apiversions_v4_out_of_range_e2e_returns_unsupported() { +async fn apiversions_v6_out_of_range_e2e_returns_unsupported() { let (addr, _shutdown) = spawn_test_server().await; - let (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 4, 350, &[]).await; + let (corr, body) = round_trip(addr, API_KEY_API_VERSIONS, 6, 350, &[]).await; assert_eq!(corr, 350); let mut d = Decoder::new(body); assert_eq!( d.read_i16().unwrap(), ERROR_UNSUPPORTED_VERSION, - "ApiVersions v4 must return UNSUPPORTED_VERSION per KIP-511" + "ApiVersions v6 must return UNSUPPORTED_VERSION per KIP-511" ); }
