You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2026/08/13 18:40:52 UTC
(kafka) branch trunk updated: MINOR: few code cleanups for KIP-1071 (#23080)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 308095cf221 MINOR: few code cleanups for KIP-1071 (#23080)
308095cf221 is described below
commit 308095cf221c3695ad4bc5214ebc507a0b91705f
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Thu Aug 13 11:40:45 2026 -0700
MINOR: few code cleanups for KIP-1071 (#23080)
Reviewers: Lianet Magrans <lm...@confluent.io>
---
.../StreamsGroupHeartbeatRequestManager.java | 3 +-
.../consumer/internals/StreamsRebalanceData.java | 28 +++++++--
.../message/StreamsGroupHeartbeatResponse.json | 6 +-
.../internals/StreamsRebalanceDataTest.java | 73 ++++++++++++++++++++++
4 files changed, 101 insertions(+), 9 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 398e5e8929d..f9236be57f7 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
@@ -899,7 +899,8 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
return topicPartitions.stream()
.flatMap(partition ->
partition.partitions().stream().map(partitionId -> new TopicPartition(partition.topic(), partitionId)))
- .collect(Collectors.toList());
+ // toUnmodifiableList rather than toList, so that List.copyOf in EndpointPartitions is a no-op
+ .collect(Collectors.toUnmodifiableList());
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
index 183df95110c..258776f12c0 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
@@ -20,7 +20,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
@@ -131,17 +130,36 @@ public class StreamsRebalanceData {
public EndpointPartitions(final List<TopicPartition> activePartitions,
final List<TopicPartition> standbyPartitions) {
- this.activePartitions = activePartitions;
- this.standbyPartitions = standbyPartitions;
+ this.activePartitions = List.copyOf(Objects.requireNonNull(activePartitions, "Active partitions cannot be null"));
+ this.standbyPartitions = List.copyOf(Objects.requireNonNull(standbyPartitions, "Standby partitions cannot be null"));
}
public List<TopicPartition> activePartitions() {
- return new ArrayList<>(activePartitions);
+ return activePartitions;
}
public List<TopicPartition> standbyPartitions() {
- return new ArrayList<>(standbyPartitions);
+ return standbyPartitions;
}
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final EndpointPartitions that = (EndpointPartitions) o;
+ return Objects.equals(activePartitions, that.activePartitions)
+ && Objects.equals(standbyPartitions, that.standbyPartitions);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(activePartitions, standbyPartitions);
+ }
+
@Override
public String toString() {
return "EndpointPartitions {"
diff --git a/clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json b/clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json
index 2e6bb8a7706..1bdfef8f11d 100644
--- a/clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json
+++ b/clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json
@@ -76,14 +76,14 @@
{ "name": "EndpointInformationEpoch", "type": "int32", "versions": "0+",
"about": "The endpoint epoch set in the response"},
{ "name": "PartitionsByUserEndpoint", "type": "[]EndpointToPartitions", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "Global assignment information used for IQ. Null if unchanged since last heartbeat." ,
+ "about": "Global assignment information used for IQ. There is one entry per member, so several entries carry the same UserEndpoint when a client runs more than one stream thread; clients must union the entries of an endpoint. Null if unchanged since last heartbeat.",
"fields": [
{ "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
"about": "User-defined endpoint to connect to the node" },
{ "name": "ActivePartitions", "type": "[]TopicPartition", "versions": "0+",
- "about": "All topic partitions materialized by active tasks on the node" },
+ "about": "The source topic partitions, including repartition source topics, of the active tasks assigned to this member" },
{ "name": "StandbyPartitions", "type": "[]TopicPartition", "versions": "0+",
- "about": "All topic partitions materialized by standby and warm-up tasks on the node" }
+ "about": "The source topic partitions, including repartition source topics, of the standby and warm-up tasks assigned to this member" }
]
}
],
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java
index a7051afa55b..5ffb672648c 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
import org.junit.jupiter.api.Test;
@@ -207,6 +208,78 @@ public class StreamsRebalanceDataTest {
assertNotSame(emptyAssignment, copy);
}
+ @Test
+ public void endpointPartitionsShouldNotAcceptNulls() {
+ final Exception exception1 = assertThrows(
+ NullPointerException.class,
+ () -> new StreamsRebalanceData.EndpointPartitions(null, List.of())
+ );
+ assertEquals("Active partitions cannot be null", exception1.getMessage());
+ final Exception exception2 = assertThrows(
+ NullPointerException.class,
+ () -> new StreamsRebalanceData.EndpointPartitions(List.of(), null)
+ );
+ assertEquals("Standby partitions cannot be null", exception2.getMessage());
+ }
+
+ @Test
+ public void endpointPartitionsShouldNotBeModifiable() {
+ final StreamsRebalanceData.EndpointPartitions endpointPartitions = new StreamsRebalanceData.EndpointPartitions(
+ List.of(new TopicPartition("topic", 0)),
+ List.of(new TopicPartition("topic", 1))
+ );
+
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> endpointPartitions.activePartitions().add(new TopicPartition("topic", 2))
+ );
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> endpointPartitions.standbyPartitions().add(new TopicPartition("topic", 3))
+ );
+ }
+
+ @Test
+ public void endpointPartitionsShouldCopyPartitionsPassedToConstructor() {
+ final List<TopicPartition> activePartitions = new ArrayList<>(List.of(new TopicPartition("topic", 0)));
+ final List<TopicPartition> standbyPartitions = new ArrayList<>(List.of(new TopicPartition("topic", 1)));
+ final StreamsRebalanceData.EndpointPartitions endpointPartitions =
+ new StreamsRebalanceData.EndpointPartitions(activePartitions, standbyPartitions);
+
+ activePartitions.add(new TopicPartition("topic", 2));
+ standbyPartitions.add(new TopicPartition("topic", 3));
+
+ assertEquals(List.of(new TopicPartition("topic", 0)), endpointPartitions.activePartitions());
+ assertEquals(List.of(new TopicPartition("topic", 1)), endpointPartitions.standbyPartitions());
+ }
+
+ @Test
+ public void testEndpointPartitionsEqualsAndHashCode() {
+ final StreamsRebalanceData.EndpointPartitions endpointPartitions = new StreamsRebalanceData.EndpointPartitions(
+ List.of(new TopicPartition("topic", 0)),
+ List.of(new TopicPartition("topic", 1))
+ );
+ final StreamsRebalanceData.EndpointPartitions endpointPartitionsEqual = new StreamsRebalanceData.EndpointPartitions(
+ endpointPartitions.activePartitions(),
+ endpointPartitions.standbyPartitions()
+ );
+ final StreamsRebalanceData.EndpointPartitions unequalActivePartitions = new StreamsRebalanceData.EndpointPartitions(
+ List.of(new TopicPartition("topic", 2)),
+ endpointPartitions.standbyPartitions()
+ );
+ final StreamsRebalanceData.EndpointPartitions unequalStandbyPartitions = new StreamsRebalanceData.EndpointPartitions(
+ endpointPartitions.activePartitions(),
+ List.of(new TopicPartition("topic", 3))
+ );
+
+ assertEquals(endpointPartitions, endpointPartitionsEqual);
+ assertNotEquals(endpointPartitions, unequalActivePartitions);
+ assertNotEquals(endpointPartitions, unequalStandbyPartitions);
+ assertEquals(endpointPartitions.hashCode(), endpointPartitionsEqual.hashCode());
+ assertNotEquals(endpointPartitions.hashCode(), unequalActivePartitions.hashCode());
+ assertNotEquals(endpointPartitions.hashCode(), unequalStandbyPartitions.hashCode());
+ }
+
@Test
public void subtopologyShouldNotAcceptNulls() {
final Exception exception1 = assertThrows(