From b15a91e383f75a609645ef6f58b8bdb9e9541518 Mon Sep 17 00:00:00 2001 From: Uladzislau Blok Date: Tue, 7 Jul 2026 16:28:17 +0200 Subject: [PATCH 1/2] failed test --- .../integration/IQv2EndpointToPartitionsIntegrationTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2EndpointToPartitionsIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2EndpointToPartitionsIntegrationTest.java index 5b3b789ab10de..fb9304053dd61 100644 --- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2EndpointToPartitionsIntegrationTest.java +++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2EndpointToPartitionsIntegrationTest.java @@ -110,6 +110,7 @@ public void shouldGetCorrectHostPartitionInformation(final String groupProtocolC streamOneProperties.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId).getPath() + "-ks1"); streamOneProperties.put(StreamsConfig.CLIENT_ID_CONFIG, appId + "-ks1"); streamOneProperties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:2020"); + streamOneProperties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3); streamOneProperties.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, groupProtocolConfig); if (usingStandbyReplicas) { streamOneProperties.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, numStandbyReplicas); From 447523d3ef3bd83a0e5f9ed5c68bc79c6df1839d Mon Sep 17 00:00:00 2001 From: Uladzislau Blok Date: Tue, 7 Jul 2026 21:35:03 +0200 Subject: [PATCH 2/2] merged view for IQ --- .../StreamsGroupHeartbeatRequestManager.java | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java index bd554051f30b4..0f2f327826626 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java @@ -834,8 +834,21 @@ private static Map activeTopicPartitions = getTopicPartitionList(endpoint.activePartitions()); List standbyTopicPartitions = getTopicPartitionList(endpoint.standbyPartitions()); StreamsGroupHeartbeatResponseData.Endpoint userEndpoint = endpoint.userEndpoint(); - StreamsRebalanceData.EndpointPartitions endpointPartitions = new StreamsRebalanceData.EndpointPartitions(activeTopicPartitions, standbyTopicPartitions); - partitionsByHost.put(new StreamsRebalanceData.HostInfo(userEndpoint.host(), userEndpoint.port()), endpointPartitions); + StreamsRebalanceData.HostInfo hostInfo = new StreamsRebalanceData.HostInfo(userEndpoint.host(), userEndpoint.port()); + + partitionsByHost.merge( + hostInfo, + new StreamsRebalanceData.EndpointPartitions(activeTopicPartitions, standbyTopicPartitions), + (existing, newPartitions) -> { + List mergedActive = existing.activePartitions(); + mergedActive.addAll(newPartitions.activePartitions()); + + List mergedStandby = existing.standbyPartitions(); + mergedStandby.addAll(newPartitions.standbyPartitions()); + + return new StreamsRebalanceData.EndpointPartitions(mergedActive, mergedStandby); + } + ); }); return partitionsByHost; }