From 8273e4e7b5f9ca027861552286e3c24d6e0d35e9 Mon Sep 17 00:00:00 2001 From: Krishna Vishal Date: Mon, 26 Oct 2026 12:46:12 +0530 Subject: [PATCH] feat(gateways): report Kafka group members' client id and host DescribeGroups left CLIENT-ID and HOST empty. A member now keeps the client id and peer address of the connection it last joined through, and its id starts with the client id. --- gateways/kafka/docs/CONSUMER_GROUPS.md | 9 +- gateways/kafka/docs/TEST_SUITE.md | 2 +- gateways/kafka/src/group/mod.rs | 12 +- gateways/kafka/src/group/state.rs | 122 +++++++++++++++--- gateways/kafka/src/protocol/api.rs | 91 ++++++++++++- .../src/protocol/handlers/delete_groups.rs | 2 + .../src/protocol/handlers/describe_groups.rs | 4 +- .../kafka/src/protocol/handlers/join_group.rs | 17 ++- .../src/protocol/handlers/list_groups.rs | 2 + gateways/kafka/src/protocol/handlers/mod.rs | 2 +- .../src/protocol/handlers/offset_delete.rs | 2 + gateways/kafka/src/server.rs | 6 +- gateways/kafka/tests/consumer_group_tests.rs | 30 +++-- .../kafka/tests/kafka_client_e2e_tests.rs | 74 +++++++++++ 14 files changed, 333 insertions(+), 42 deletions(-) diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md index e87fb0df16..f8375ed7e3 100644 --- a/gateways/kafka/docs/CONSUMER_GROUPS.md +++ b/gateways/kafka/docs/CONSUMER_GROUPS.md @@ -57,9 +57,12 @@ consumers stopped, because that group is `Empty` while it holds offsets. Without `DeleteGroups` and `OffsetDelete` ([below](#deleting-groups-and-offsets)). See [OFFSET_STORAGE.md](OFFSET_STORAGE.md) for offset behavior and limits. -Member `client_id` and `client_host` are empty. The coordinator does not keep the request -header's client id or the connection's peer address. `include_authorized_operations` is answered -with the omitted sentinel. There is no group ACL bitfield. `ListGroups` v5 sets `group_type` to +Each member reports the `client_id` and `client_host` of the connection that it last joined +through, as Kafka does. The client id is the first one that a request header on that connection +named, cut to 1,024 bytes. The host is a `/` and the peer IP address, for example `/10.0.0.7`. A +new member id starts with the client id, or with the `group_instance_id` of a static member, +then a UUID. A client that names neither gets `member-` and a UUID. `include_authorized_operations` +is answered with the omitted sentinel. There is no group ACL bitfield. `ListGroups` v5 sets `group_type` to `classic`, which is the only protocol this coordinator runs. ## Groups that only hold offsets diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index fe9c028885..3e23962cb7 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -79,7 +79,7 @@ file under `tests/` anymore. | [`listener_robustness_tests.rs`](../tests/listener_robustness_tests.rs) | TCP listener robustness — framing, pipelining, concurrency, connection limits | No | | [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No | | [`tls_listener_tests.rs`](../tests/tls_listener_tests.rs) | The `SSL` and `SASL_SSL` listeners over a real socket with certificates minted per test: each listener advertising its own port, a plaintext client refused on the TLS port, SASL inside TLS, client certificates mapped to users through a stub `PrincipalResolver` (`required`, `requested`, unknown user, a subject no rule maps, a failing or hung lookup, `SASL_SSL` asking for no certificate), a multi-MiB response to a slow reader, the handshake budget, certificate reload without a restart, and the startup refusals | No | -| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. `kafka-cluster cluster-id` prints the same id across a gateway restart. Lookups by time resolve through `kafka-get-offsets` (a time, 0, `-3` and a future time), `kafka-consumer-groups --reset-offsets --to-datetime`, and the consumer's `offsetsForTimes` in a probe compiled with the host `javac`. librdkafka and the Java client connect with `security.protocol=SSL` and `SASL_SSL`, and a client certificate admits each as the Iggy user it names. With auto topic creation on, the Java producer sends to a topic nobody created. The only suite that can catch a client-compatibility bug, since every other one hand-builds frames | No, but needs Docker, a built `iggy-server` and a host JDK (`javac`) | +| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. `kafka-consumer-groups --describe --members` shows the client id and host of a member. `kafka-cluster cluster-id` prints the same id across a gateway restart. Lookups by time resolve through `kafka-get-offsets` (a time, 0, `-3` and a future time), `kafka-consumer-groups --reset-offsets --to-datetime`, and the consumer's `offsetsForTimes` in a probe compiled with the host `javac`. librdkafka and the Java client connect with `security.protocol=SSL` and `SASL_SSL`, and a client certificate admits each as the Iggy user it names. With auto topic creation on, the Java producer sends to a topic nobody created. The only suite that can catch a client-compatibility bug, since every other one hand-builds frames | No, but needs Docker, a built `iggy-server` and a host JDK (`javac`) | | [`sasl_real_server_tests.rs`](../tests/sasl_real_server_tests.rs) | `IggyAuthenticator` against a real, spawned `iggy-server`: twice as many logins as a small session table holds all succeed, because each check signs out | No (needs the `iggy-server` binary - see Prerequisites) | | [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases, and the bridge dialing a TLS-only server over TLS | No (needs the `iggy-server` binary - see Prerequisites) | | [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | No (needs the `iggy-server` binary - see Prerequisites) | diff --git a/gateways/kafka/src/group/mod.rs b/gateways/kafka/src/group/mod.rs index 1a22a66841..e98fbaa3a2 100644 --- a/gateways/kafka/src/group/mod.rs +++ b/gateways/kafka/src/group/mod.rs @@ -30,7 +30,7 @@ mod offset_work; mod state; pub use offset_work::{RunningCommit, RunningDeletion}; -pub use state::{MAX_GROUP_ID_BYTES, is_valid_group_id}; +pub use state::{MAX_CLIENT_ID_BYTES, MAX_GROUP_ID_BYTES, is_valid_group_id}; use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -157,6 +157,10 @@ pub struct JoinRequest { pub rebalance_timeout: Duration, pub member_id: StrBytes, pub group_instance_id: Option, + /// The `client_id` of the connection's request headers, and its peer address in Kafka's + /// `client_host` form. Empty when the request did not come through a connection. + pub client_id: StrBytes, + pub client_host: StrBytes, pub protocol_type: StrBytes, /// `(name, metadata)` in request order; the order is this member's preference vote. pub protocols: Vec<(StrBytes, Bytes)>, @@ -185,6 +189,8 @@ impl From<(i16, &JoinGroupRequest)> for JoinRequest { }, member_id: owned_str(&request.member_id), group_instance_id: request.group_instance_id.as_ref().map(owned_str), + client_id: StrBytes::new(), + client_host: StrBytes::new(), protocol_type: owned_str(&request.protocol_type), protocols: request .protocols @@ -380,6 +386,8 @@ impl LeaveResult { pub struct MemberDescription { pub member_id: StrBytes, pub group_instance_id: Option, + pub client_id: StrBytes, + pub client_host: StrBytes, /// Subscription metadata for the group's selected protocol. Empty until a protocol is chosen. pub metadata: Bytes, pub assignment: Bytes, @@ -959,6 +967,8 @@ mod tests { rebalance_timeout: Duration::from_secs(10), member_id: StrBytes::new(), group_instance_id: None, + client_id: StrBytes::new(), + client_host: StrBytes::new(), protocol_type: StrBytes::from_static_str(CONSUMER_PROTOCOL_TYPE), protocols: vec![(StrBytes::from_static_str("range"), Bytes::new())], require_known_member_id: false, diff --git a/gateways/kafka/src/group/state.rs b/gateways/kafka/src/group/state.rs index d9c5efcf23..7c53a52dc2 100644 --- a/gateways/kafka/src/group/state.rs +++ b/gateways/kafka/src/group/state.rs @@ -86,8 +86,17 @@ const MAX_PROTOCOL_TYPE_BYTES: usize = 255; /// to leave this much room or the id itself will not fit a legacy string. const MEMBER_ID_SUFFIX_BYTES: usize = 1 + 36; -/// Prefix for a generated member id when the client sent no `group_instance_id`. Kafka uses the -/// header's `client_id`, which handlers do not receive. +/// Cap on the `client_id` that a member keeps. +/// +/// Kafka sets none, but a header can carry 32 KiB, and every member keeps the id twice: as its +/// client id, and as the start of its member id. Stock clients send well under 1 KiB, a Kafka +/// Streams thread included. +pub const MAX_CLIENT_ID_BYTES: usize = 1024; + +const _: () = assert!(MAX_CLIENT_ID_BYTES + MEMBER_ID_SUFFIX_BYTES <= MAX_LEGACY_STRING_BYTES); + +/// Prefix for a generated member id when the client sent no `group_instance_id` and no +/// `client_id`. const DEFAULT_MEMBER_PREFIX: &str = "member"; /// Floor on how long a parked waiter sleeps. Every deadline a tick leaves behind is strictly in @@ -132,6 +141,9 @@ struct Departures { pub struct Member { group_instance_id: Option, + /// The `client_id` and `client_host` of the connection that the member last joined through. + client_id: StrBytes, + client_host: StrBytes, session_timeout: Duration, rebalance_timeout: Duration, protocols: Vec<(StrBytes, Bytes)>, @@ -164,6 +176,8 @@ impl Member { fn new(request: &JoinRequest, now: Instant, max_rebalance_timeout: Duration) -> Self { Self { group_instance_id: request.group_instance_id.as_ref().map(owned_str), + client_id: request.client_id.clone(), + client_host: request.client_host.clone(), session_timeout: request.session_timeout, rebalance_timeout: request.rebalance_timeout.min(max_rebalance_timeout), protocols: retained_protocols(&request.protocols), @@ -178,6 +192,7 @@ impl Member { /// Keeps `group_instance_id`: a rejoin that names one was already checked against it, and one /// that names none does not make a static member dynamic, as in Kafka. fn rejoin(&mut self, request: &JoinRequest, now: Instant, max_rebalance_timeout: Duration) { + self.note_client(request); self.session_timeout = request.session_timeout; self.rebalance_timeout = request.rebalance_timeout.min(max_rebalance_timeout); self.protocols = retained_protocols(&request.protocols); @@ -189,6 +204,11 @@ impl Member { self.join_response = None; } + fn note_client(&mut self, request: &JoinRequest) { + self.client_id.clone_from(&request.client_id); + self.client_host.clone_from(&request.client_host); + } + fn roster_bytes(&self, member_id: &StrBytes) -> usize { roster_entry_bytes(member_id, self.group_instance_id.as_ref(), &self.protocols) } @@ -238,9 +258,8 @@ impl Member { self.group_instance_id.as_ref().map(|id| id.len()), ); } - // `client_id` and `client_host` stay empty: the coordinator does not retain them. - total += kafka_string_len(flexible, 0); - total += kafka_string_len(flexible, 0); + total += kafka_string_len(flexible, self.client_id.len()); + total += kafka_string_len(flexible, self.client_host.len()); let metadata_len = if include_blobs { protocol_name.map_or(0, |name| self.metadata_len(name)) } else { @@ -849,6 +868,8 @@ impl GroupState { .map(|(member_id, member)| MemberDescription { member_id: member_id.clone(), group_instance_id: member.group_instance_id.clone(), + client_id: member.client_id.clone(), + client_host: member.client_host.clone(), metadata: if summary { Bytes::new() } else { @@ -1205,7 +1226,7 @@ fn join_new_member( ); return reject(groups, ERROR_COORDINATOR_NOT_AVAILABLE); } - let member_id = generate_member_id(request.group_instance_id.as_ref()); + let member_id = generate_member_id(request); // Checked before an id is handed out too, so a member that could never be admitted is // not sent round a `MEMBER_ID_REQUIRED` trip first. if group.roster_overflows(config, &member_id, request) { @@ -1258,6 +1279,7 @@ fn rejoin_step( } Phase::Stable | Phase::CompletingRebalance => { if let Some(member) = group.members.get_mut(&request.member_id) { + member.note_client(request); member.session_timeout = request.session_timeout; member.session_deadline = now + request.session_timeout; // Answering without consuming the snapshot would leave an uncollected one on a @@ -1957,7 +1979,7 @@ fn replace_static( request: &JoinRequest, now: Instant, ) -> Step { - let new_member_id = generate_member_id(request.group_instance_id.as_ref()); + let new_member_id = generate_member_id(request); if group.roster_overflows_replacing(config, old_member_id, &new_member_id, request) { return Step::Respond(JoinResult::error( ERROR_GROUP_MAX_SIZE_REACHED, @@ -2032,12 +2054,15 @@ fn park_or_respond(group: &mut GroupState, member_id: &StrBytes, now: Instant) - } } -/// Kafka shapes a member id as `clientId-UUID`. Handlers never see the header's `client_id`, so -/// the static-membership id stands in for it when present. The UUID matters on its own: a -/// counter would repeat across a gateway restart and turn a stale client's `UNKNOWN_MEMBER_ID` -/// into a silent identity theft. -fn generate_member_id(group_instance_id: Option<&StrBytes>) -> StrBytes { - let prefix = group_instance_id.map_or(DEFAULT_MEMBER_PREFIX, StrBytes::as_str); +/// Kafka shapes a member id as `groupInstanceId-UUID` for a static member, and as `clientId-UUID` +/// for any other. The UUID matters on its own: a counter would repeat across a gateway restart and +/// turn a stale client's `UNKNOWN_MEMBER_ID` into a silent identity theft. +fn generate_member_id(request: &JoinRequest) -> StrBytes { + let prefix = match &request.group_instance_id { + Some(instance_id) => instance_id.as_str(), + None if !request.client_id.is_empty() => request.client_id.as_str(), + None => DEFAULT_MEMBER_PREFIX, + }; StrBytes::from_string(format!("{prefix}-{}", Uuid::new_v4())) } @@ -2196,6 +2221,8 @@ mod tests { rebalance_timeout: Duration::from_secs(rebalance_secs), member_id: StrBytes::from_string(member_id.to_owned()), group_instance_id: None, + client_id: StrBytes::new(), + client_host: StrBytes::new(), protocol_type: StrBytes::from_static_str("consumer"), protocols: protocols .iter() @@ -2564,8 +2591,8 @@ mod tests { } /// The frame check has to match what `DescribeGroups` actually encodes, header included. - /// Fields are past the compact-varint boundary and the member has an instance id, which v4+ - /// writes and v0-v3 does not. + /// Fields, the client id among them, are past the compact-varint boundary, and the member has + /// an instance id, which v4+ writes and v0-v3 does not. #[test] fn given_a_described_group_when_pricing_the_frame_should_match_the_encoder() { let config = config(); @@ -2671,6 +2698,69 @@ mod tests { assert!(groups.contains_key(&group_id())); } + /// `kafka-consumer-groups --describe --members` shows these two columns. A member that + /// rejoins through another connection is shown there. + #[test] + fn given_a_member_when_describing_should_report_the_client_of_its_last_join() { + let config = config(); + let mut groups = Groups::new(); + let now = Instant::now(); + let mut request = request("", &["range"]); + request.client_id = StrBytes::from_static_str("consumer-g-1"); + request.client_host = StrBytes::from_static_str("/192.0.2.10"); + let Step::Respond(joined) = join_step(&mut groups, &config, &request, now) else { + panic!("an immediate join must answer"); + }; + assert_eq!(joined.error, ERROR_NONE); + assert_eq!( + described_client(&mut groups, now), + ("consumer-g-1".to_owned(), "/192.0.2.10".to_owned()) + ); + + request.member_id = joined.member_id; + request.client_host = StrBytes::from_static_str("/192.0.2.11"); + join_step(&mut groups, &config, &request, now); + assert_eq!( + described_client(&mut groups, now), + ("consumer-g-1".to_owned(), "/192.0.2.11".to_owned()) + ); + } + + /// Kafka starts a member id with the instance id of a static member, and with the client id + /// of any other. + #[test] + fn given_a_client_id_when_joining_should_start_the_member_id_with_it() { + let config = config(); + let now = Instant::now(); + let member_id = |request: &JoinRequest| { + let Step::Respond(joined) = join_step(&mut Groups::new(), &config, request, now) else { + panic!("an immediate join must answer"); + }; + assert_eq!(joined.error, ERROR_NONE); + joined.member_id.as_str().to_owned() + }; + let mut request = request("", &["range"]); + assert!(member_id(&request).starts_with("member-")); + + request.client_id = StrBytes::from_static_str("consumer-g-1"); + assert!(member_id(&request).starts_with("consumer-g-1-")); + + request.group_instance_id = Some(StrBytes::from_static_str("static-1")); + assert!(member_id(&request).starts_with("static-1-")); + } + + /// The client id and host of the only member of the group, as `DescribeGroups` reports them. + fn described_client(groups: &mut Groups, now: Instant) -> (String, String) { + let described = describe_groups(groups, &[group_id()], 6, usize::MAX, now); + let group = described[0].as_ref().unwrap(); + assert_eq!(group.members.len(), 1); + let member = &group.members[0]; + ( + member.client_id.as_str().to_owned(), + member.client_host.as_str().to_owned(), + ) + } + /// A `ConsumerProtocolSubscription` as a consumer sends it: its version, then its fields. fn subscription(version: i16, topics: &[&str]) -> Bytes { let mut buf = BytesMut::new(); @@ -2954,6 +3044,8 @@ mod tests { let wide = "x".repeat(130); let mut request = request("", &["range"]); request.group_instance_id = Some(StrBytes::from_string(wide.clone())); + request.client_id = StrBytes::from_string(wide.clone()); + request.client_host = StrBytes::from_static_str("/192.0.2.10"); request.protocols = vec![(StrBytes::from_string(wide), Bytes::from(vec![b'm'; 130]))]; let step = join_step(groups, config, &request, now); let Step::Respond(result) = step else { diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs index c98c73e239..8d1f4cfdcf 100644 --- a/gateways/kafka/src/protocol/api.rs +++ b/gateways/kafka/src/protocol/api.rs @@ -15,14 +15,16 @@ // specific language governing permissions and limitations // under the License. +use std::net::SocketAddr; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Mutex, PoisonError}; +use std::sync::{Arc, Mutex, OnceLock, PoisonError}; use std::time::Duration; use bytes::Bytes; use kafka_protocol::messages::{ DescribeAclsRequest, SaslAuthenticateRequest, SaslHandshakeRequest, }; +use kafka_protocol::protocol::StrBytes; use tokio::sync::{OwnedMutexGuard, Semaphore}; use tokio_util::sync::CancellationToken; @@ -30,7 +32,7 @@ use crate::auth::AuthenticatedPrincipal; use crate::bridge::IggyBridge; use crate::cluster_id::ClusterId; use crate::error::Result; -use crate::group::{GroupCoordinator, GroupCoordinatorConfig, GroupMember}; +use crate::group::{GroupCoordinator, GroupCoordinatorConfig, GroupMember, MAX_CLIENT_ID_BYTES}; use crate::protocol::acl::{ self, AclBinding, AclFilter, PrincipalPermissions, encode_describe_acls_error_response, encode_describe_acls_response, @@ -329,6 +331,11 @@ pub struct ConnectionState { /// The member that last sent a Heartbeat here. An offset call caps its hold by that member's /// session, since its next heartbeat waits behind the call. heartbeat_member: Mutex>, + /// The first `client_id` that a request header on this connection named, cut to + /// [`MAX_CLIENT_ID_BYTES`]. A Kafka client sends one id in every header of a connection. + client_id: OnceLock, + /// The peer's IP address as Kafka reports a member's `client_host`: a `/`, then the address. + client_host: StrBytes, /// Keys the Produce session of a producer without a producer id that this gateway gave out. id: ConnectionId, } @@ -359,6 +366,33 @@ impl ConnectionState { } } + /// The same state, for a connection from `peer`. An IPv4 client of a dual-stack listener is + /// reported by its IPv4 address, as Kafka does. + #[must_use] + pub fn with_peer(mut self, peer: SocketAddr) -> Self { + self.client_host = StrBytes::from_string(format!("/{}", peer.ip().to_canonical())); + self + } + + /// Keeps `client_id` if no earlier header named one. The copy is owned: the header is a view + /// into its frame, and a group member keeps the id for as long as it stays in the group. + pub fn note_client_id(&self, client_id: Option<&str>) { + if let Some(client_id) = client_id.filter(|client_id| !client_id.is_empty()) { + self.client_id.get_or_init(|| { + let end = client_id.floor_char_boundary(MAX_CLIENT_ID_BYTES); + StrBytes::from_string(client_id[..end].to_owned()) + }); + } + } + + pub(crate) fn client_id(&self) -> StrBytes { + self.client_id.get().cloned().unwrap_or_default() + } + + pub(crate) fn client_host(&self) -> StrBytes { + self.client_host.clone() + } + /// What `ApiVersions` tells this connection about the listener-dependent keys. #[must_use] pub const fn advertisement(&self) -> Advertisement { @@ -744,4 +778,57 @@ mod tests { ); } } + + #[test] + fn given_request_headers_should_keep_the_first_client_id_that_one_names() { + let connection = ConnectionState::default(); + connection.note_client_id(None); + connection.note_client_id(Some("")); + assert_eq!(connection.client_id().as_str(), ""); + + connection.note_client_id(Some("consumer-g-1")); + connection.note_client_id(Some("consumer-g-2")); + assert_eq!(connection.client_id().as_str(), "consumer-g-1"); + } + + /// A 2-byte character that crosses the cap is left out whole, so the id stays UTF-8. + #[test] + fn given_a_client_id_past_the_cap_should_keep_its_start() { + let connection = ConnectionState::default(); + let client_id = format!("{}é", "c".repeat(MAX_CLIENT_ID_BYTES - 1)); + connection.note_client_id(Some(&client_id)); + assert_eq!( + connection.client_id().as_str(), + &client_id[..MAX_CLIENT_ID_BYTES - 1] + ); + } + + /// Kafka shows a member's host as `/` and the address, without the port. + #[test] + fn given_a_peer_should_report_its_address_as_kafka_does() { + let peer: SocketAddr = "192.0.2.10:51234".parse().unwrap(); + assert_eq!( + ConnectionState::default() + .with_peer(peer) + .client_host() + .as_str(), + "/192.0.2.10" + ); + let peer: SocketAddr = "[::ffff:192.0.2.10]:51234".parse().unwrap(); + assert_eq!( + ConnectionState::default() + .with_peer(peer) + .client_host() + .as_str(), + "/192.0.2.10" + ); + let peer: SocketAddr = "[2001:db8::1]:51234".parse().unwrap(); + assert_eq!( + ConnectionState::default() + .with_peer(peer) + .client_host() + .as_str(), + "/2001:db8::1" + ); + } } diff --git a/gateways/kafka/src/protocol/handlers/delete_groups.rs b/gateways/kafka/src/protocol/handlers/delete_groups.rs index 247cba8047..09b2adf8e0 100644 --- a/gateways/kafka/src/protocol/handlers/delete_groups.rs +++ b/gateways/kafka/src/protocol/handlers/delete_groups.rs @@ -335,6 +335,8 @@ mod tests { rebalance_timeout: Duration::from_secs(10), member_id: StrBytes::new(), group_instance_id: None, + client_id: StrBytes::new(), + client_host: StrBytes::new(), protocol_type: StrBytes::from_static_str("consumer"), protocols: vec![(StrBytes::from_static_str("range"), Bytes::new())], require_known_member_id, diff --git a/gateways/kafka/src/protocol/handlers/describe_groups.rs b/gateways/kafka/src/protocol/handlers/describe_groups.rs index 158d3b4dd0..9d35b71885 100644 --- a/gateways/kafka/src/protocol/handlers/describe_groups.rs +++ b/gateways/kafka/src/protocol/handlers/describe_groups.rs @@ -199,11 +199,11 @@ fn error_group( } fn described_member(member: MemberDescription) -> DescribedGroupMember { - // `client_id` and `client_host` stay empty: the coordinator does not retain the request - // header's client id, and handlers are not given the connection's peer address. DescribedGroupMember::default() .with_member_id(member.member_id) .with_group_instance_id(member.group_instance_id) + .with_client_id(member.client_id) + .with_client_host(member.client_host) .with_member_metadata(member.metadata) .with_member_assignment(member.assignment) } diff --git a/gateways/kafka/src/protocol/handlers/join_group.rs b/gateways/kafka/src/protocol/handlers/join_group.rs index 5323b491ec..f1ea1cfd16 100644 --- a/gateways/kafka/src/protocol/handlers/join_group.rs +++ b/gateways/kafka/src/protocol/handlers/join_group.rs @@ -29,8 +29,8 @@ use kafka_protocol::protocol::StrBytes; use crate::error::Result; use crate::group::{JoinRequest, JoinResult}; use crate::protocol::api::{ - API_KEY_JOIN_GROUP, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION, - GatewayState, HandleOutcome, is_supported_version, + API_KEY_JOIN_GROUP, ApiVersionRange, ConnectionState, ERROR_INVALID_REQUEST, + ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome, is_supported_version, }; use crate::protocol::bounds_guard::validate_join_group_shape; use crate::protocol::handlers::{ @@ -48,7 +48,12 @@ pub const RANGE: ApiVersionRange = ApiVersionRange { /// rejects, so an error response carries an empty name instead. const FIRST_NULLABLE_PROTOCOL_NAME_VERSION: i16 = 7; -pub async fn handle(state: &GatewayState, 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_JOIN_GROUP, api_version) { return unsupported_version_response(API_KEY_JOIN_GROUP, api_version, |version| { encode_error_response(version, ERROR_UNSUPPORTED_VERSION) @@ -71,7 +76,11 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand tracing::debug!(%reason, "JoinGroup rejoin reason"); } - let join_request = JoinRequest::from((api_version, &request)); + let join_request = JoinRequest { + client_id: connection.client_id(), + client_host: connection.client_host(), + ..JoinRequest::from((api_version, &request)) + }; // The join can park for a whole rebalance timeout, and the decoded request is a view into // the frame it arrived in. drop(request); diff --git a/gateways/kafka/src/protocol/handlers/list_groups.rs b/gateways/kafka/src/protocol/handlers/list_groups.rs index 9acce6c7fe..ec3d91af0f 100644 --- a/gateways/kafka/src/protocol/handlers/list_groups.rs +++ b/gateways/kafka/src/protocol/handlers/list_groups.rs @@ -287,6 +287,8 @@ mod tests { rebalance_timeout: Duration::from_secs(10), member_id: StrBytes::new(), group_instance_id: None, + client_id: StrBytes::new(), + client_host: StrBytes::new(), protocol_type: StrBytes::from_static_str(CONSUMER_PROTOCOL_TYPE), protocols: vec![(StrBytes::from_static_str("range"), Bytes::new())], require_known_member_id: false, diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs index 4dfa6590ed..ff65d71679 100644 --- a/gateways/kafka/src/protocol/handlers/mod.rs +++ b/gateways/kafka/src/protocol/handlers/mod.rs @@ -140,7 +140,7 @@ pub async fn dispatch( API_KEY_FIND_COORDINATOR => { find_coordinator::handle(state, connection, api_version, body).await } - API_KEY_JOIN_GROUP => join_group::handle(state, api_version, body).await, + API_KEY_JOIN_GROUP => join_group::handle(state, connection, api_version, body).await, API_KEY_HEARTBEAT => heartbeat::handle(state, connection, api_version, body).await, API_KEY_LEAVE_GROUP => leave_group::handle(state, api_version, body).await, API_KEY_SYNC_GROUP => sync_group::handle(state, api_version, body).await, diff --git a/gateways/kafka/src/protocol/handlers/offset_delete.rs b/gateways/kafka/src/protocol/handlers/offset_delete.rs index 41ca44b80a..0e80f69e38 100644 --- a/gateways/kafka/src/protocol/handlers/offset_delete.rs +++ b/gateways/kafka/src/protocol/handlers/offset_delete.rs @@ -408,6 +408,8 @@ mod tests { rebalance_timeout: Duration::from_secs(10), member_id: StrBytes::new(), group_instance_id: None, + client_id: StrBytes::new(), + client_host: StrBytes::new(), protocol_type: StrBytes::from_string(protocol_type.to_owned()), protocols: vec![(StrBytes::from_static_str("range"), Bytes::new())], require_known_member_id: false, diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs index ba0f429f5e..5b63db61cc 100644 --- a/gateways/kafka/src/server.rs +++ b/gateways/kafka/src/server.rs @@ -1039,7 +1039,10 @@ async fn handle_connection( where S: AsyncRead + AsyncWrite + Unpin + Send, { - let connection_state = connection.listener.connection_state(); + let connection_state = connection + .listener + .connection_state() + .with_peer(connection.peer); let bridge = connection.state.bridge.clone(); let served = serve_frames(stream, connection, certificate_name, &connection_state).await; // A producer without a producer id has its Produce session by connection, so it ends here. @@ -1145,6 +1148,7 @@ where client_id = req.client_id.as_deref().unwrap_or(""), "received request" ); + connection_state.note_client_id(req.client_id.as_deref()); // A group request can park for a whole rebalance timeout. `client_id` and the tagged // fields are views into the frame, so keeping them would pin the frame for that long // after the handler has let go of its own copy. diff --git a/gateways/kafka/tests/consumer_group_tests.rs b/gateways/kafka/tests/consumer_group_tests.rs index 11eb7d9736..58252faf34 100644 --- a/gateways/kafka/tests/consumer_group_tests.rs +++ b/gateways/kafka/tests/consumer_group_tests.rs @@ -74,6 +74,8 @@ const HEARTBEAT_VERSION: i16 = 4; const LEAVE_VERSION: i16 = 5; const SESSION_TIMEOUT_MS: i32 = 10_000; const REBALANCE_TIMEOUT_MS: i32 = 20_000; +/// The `client_id` of every request header that `write_request` sends. +const CLIENT_ID: &str = "consumer-group-test"; // ── Fixtures ──────────────────────────────────────────────────────────────── @@ -1498,6 +1500,9 @@ struct Described { } struct DescribedMember { + member_id: String, + client_id: String, + client_host: String, metadata: Bytes, assignment: Bytes, } @@ -1525,20 +1530,21 @@ impl DescribeResponse { let member_count = read_array_count(&mut decoder, flexible); let mut members = Vec::with_capacity(member_count); for _ in 0..member_count { - let _member_id = read_nullable(&mut decoder, flexible); + let member_id = read_nullable(&mut decoder, flexible).unwrap_or_default(); if version >= 4 { let _instance_id = read_nullable(&mut decoder, flexible); } - let client_id = read_nullable(&mut decoder, flexible); - let client_host = read_nullable(&mut decoder, flexible); - assert_eq!(client_id.as_deref(), Some("")); - assert_eq!(client_host.as_deref(), Some("")); + let client_id = read_nullable(&mut decoder, flexible).unwrap_or_default(); + let client_host = read_nullable(&mut decoder, flexible).unwrap_or_default(); let metadata = read_bytes_field(&mut decoder, flexible); let assignment = read_bytes_field(&mut decoder, flexible); if flexible { decoder.read_tagged_fields().unwrap(); } members.push(DescribedMember { + member_id, + client_id, + client_host, metadata, assignment, }); @@ -2255,6 +2261,12 @@ async fn given_a_live_group_over_tcp_should_describe_and_list_it_through_the_rea assert_eq!(described.groups[0].group_id, GROUP); assert_eq!(described.groups[0].state, "Stable"); assert_eq!(described.groups[0].members.len(), 1); + // The client id of the request headers and the address of the socket, as Kafka reports them. + let described_member = &described.groups[0].members[0]; + assert_eq!(described_member.member_id, member); + assert!(member.starts_with(&format!("{CLIENT_ID}-"))); + assert_eq!(described_member.client_id, CLIENT_ID); + assert_eq!(described_member.client_host, "/127.0.0.1"); write_request( &mut stream, @@ -2278,13 +2290,7 @@ async fn write_request( correlation_id: i32, body: &Bytes, ) { - let frame = build_request_frame( - api_key, - api_version, - correlation_id, - Some("consumer-group-test"), - body, - ); + let frame = build_request_frame(api_key, api_version, correlation_id, Some(CLIENT_ID), body); stream.write_all(&frame).await.expect("write request"); } diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs index ed3f86941a..c836faa7f6 100644 --- a/gateways/kafka/tests/kafka_client_e2e_tests.rs +++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs @@ -1553,6 +1553,80 @@ fn consume_as_static_member(gateway: SocketAddr, count: usize) -> String { .expect_ran("java console consumer as a static member") } +/// `kafka-consumer-groups --describe --members` shows the client id and host of each member, as +/// Kafka does (#4511). A static member stays in its group for its whole session after its +/// consumer exits, so the tool still finds it. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn given_a_java_consumer_when_describing_the_members_should_show_its_client_id_and_host() { + let Some((_server, gateway)) = bridged_stack().await else { + return; + }; + produce_with_kcat(gateway, &RECORDS); + let consumed = run_client( + KAFKA_IMAGE, + &[ + "/opt/kafka/bin/kafka-console-consumer.sh", + "--bootstrap-server", + &gateway.to_string(), + "--topic", + TOPIC, + "--group", + "e2e-members", + "--consumer-property", + "group.instance.id=e2e-members-0", + "--consumer-property", + "client.id=e2e-members-client", + "--consumer-property", + "session.timeout.ms=60000", + "--from-beginning", + "--max-messages", + &RECORDS.len().to_string(), + "--timeout-ms", + "20000", + "--property", + "print.offset=true", + "--property", + "print.key=true", + ], + &[], + ) + .expect_ran("java console consumer with a client id"); + assert_eq!( + records_read(&consumed).len(), + RECORDS.len(), + "the consumer must join the group and read every record, got: {consumed}" + ); + + let described = run_client( + KAFKA_IMAGE, + &[ + "/opt/kafka/bin/kafka-consumer-groups.sh", + "--bootstrap-server", + &gateway.to_string(), + "--describe", + "--group", + "e2e-members", + "--members", + ], + &[], + ) + .expect_ran("kafka-consumer-groups --describe --members"); + // GROUP, CONSUMER-ID, GROUP-INSTANCE-ID, HOST, CLIENT-ID, then #PARTITIONS. Matched as a + // run of fields, as the reset test above does. + let fields: Vec<&str> = described.split_whitespace().collect(); + let shown = fields.windows(5).any(|row| { + row[0] == "e2e-members" + && row[1].starts_with("e2e-members-0-") + && row[2] == "e2e-members-0" + && row[3] == "/127.0.0.1" + && row[4] == "e2e-members-client" + }); + assert!( + shown, + "the member must show its client id and host, got: {described}" + ); +} + /// KIP-345 with librdkafka, which sends `JoinGroup` below v9: the returning member is told the /// previous leader, acts as a follower, and gets its old assignment back through `SyncGroup`. /// Without replacement the second run would wait on a rebalance its old incarnation never rejoins.