You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2026/09/29 15:40:10 UTC
(kafka) branch trunk updated: MINOR: Fix CoordinatorResult javadoc and strengthen streams delay-timer tests (#23619)
This is an automated email from the ASF dual-hosted git repository.
lucasbru 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 3c7b0661c13 MINOR: Fix CoordinatorResult javadoc and strengthen streams delay-timer tests (#23619)
3c7b0661c13 is described below
commit 3c7b0661c132fc07b6e8da9daf991e31b0363b3e
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Tue Sep 29 17:40:01 2026 +0200
MINOR: Fix CoordinatorResult javadoc and strengthen streams delay-timer tests (#23619)
The replayRecords javadoc described it as "the records to replay", even
though it is a boolean flag controlling whether the coordinator applies
records to its in-memory state, not a list of records.
The two streams initial-rebalance-delay tests also could not tell
whether the target assignment they observed came from the delay timer
actually replaying its records, or from a later heartbeat recomputing
the assignment on its own, since both tests only checked the group state
after an additional heartbeat had already run past the timer firing.
They now assert the in-memory target assignment right after the timer
fires and before any further heartbeat, and the first test performs an
extra heartbeat inside the delay window so the topology is configured on
the persisted group rather than a throwaway instance, which is required
for the timer to compute and replay the assignment in the first place.
Reviewers: Sean Quah <sq...@confluent.io>
---
.../common/runtime/CoordinatorResult.java | 6 ++---
.../group/GroupMetadataManagerTest.java | 26 ++++++++++++++++++++++
2 files changed, 29 insertions(+), 3 deletions(-)
diff --git a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorResult.java b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorResult.java
index 16fa565953a..6f5f19e1f4e 100644
--- a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorResult.java
+++ b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorResult.java
@@ -97,7 +97,7 @@ public class CoordinatorResult<T, U> {
*
* @param records A non-null list of records.
* @param appendFuture The future to complete once the records are committed.
- * @param replayRecords The records to replay.
+ * @param replayRecords Whether to replay the records.
*/
public CoordinatorResult(
List<U> records,
@@ -113,7 +113,7 @@ public class CoordinatorResult<T, U> {
* @param records A non-null list of records.
* @param response A response.
* @param appendFuture The future to complete once the records are committed.
- * @param replayRecords The records to replay.
+ * @param replayRecords Whether to replay the records.
*/
public CoordinatorResult(
List<U> records,
@@ -130,7 +130,7 @@ public class CoordinatorResult<T, U> {
* @param records A non-null list of records.
* @param response A response.
* @param appendFuture The future to complete once the records are committed.
- * @param replayRecords The records to replay.
+ * @param replayRecords Whether to replay the records.
* @param isAtomic A boolean indicating whether the result is atomic or not.
*/
public CoordinatorResult(
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 94a26ea6f6b..2fd03c04c09 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -22350,11 +22350,30 @@ public class GroupMetadataManagerTest {
result.response().data()
);
+ // A second heartbeat, still within the delay window, is needed for the topology to be
+ // configured on the group (the first join computes it on a throwaway group instance).
+ context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(1)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of()));
+
assignor.prepareGroupAssignment(
Map.of(memberId, TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE, TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1))));
context.sleep(10000);
+ // The delay timer firing must update the target assignment without requiring a subsequent
+ // heartbeat.
+ StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
+ assertEquals(
+ TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE, TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1)),
+ group.targetAssignment(memberId, Optional.empty())
+ );
+
result = context.streamsGroupHeartbeat(
new StreamsGroupHeartbeatRequestData()
.setGroupId(groupId)
@@ -22428,6 +22447,13 @@ public class GroupMetadataManagerTest {
context.sleep(2000);
+ // The delay timer firing must update the target assignment without requiring a subsequent
+ // heartbeat.
+ assertEquals(
+ TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE, TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1)),
+ group.targetAssignment(memberId, Optional.empty())
+ );
+
result = context.streamsGroupHeartbeat(
new StreamsGroupHeartbeatRequestData()
.setGroupId(groupId)