feat(gateways): report Kafka group members' client id and host

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

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) |

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<StrBytes>,
+    /// 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<StrBytes>,
+    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,

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<StrBytes>,
+    /// 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<JoinResult> {
-    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 {