You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2026/07/31 06:25:14 UTC
[PR] KAFKA-20870: Add test for the owned-task report for a single-role assignment change [kafka]
mjsax opened a new pull request, #23002:
URL: https://github.com/apache/kafka/pull/23002
The broker reads the three owned-task lists in a StreamsGroupHeartbeat as a report of what the member holds only when all three are non-null; with any of them null it cannot tell that a task was released, so the member fails tonever acknowledges the revocation and the group stays in rebalancing. A change confined to one role therefore has to resend the other two lists as well, even though they did not change.
While the client implements this already correctly, there is no explicit test for it. The problem is, that KIP-1071 does not specify this requirement atm. This PR adds an explicit test to avoid any accidental regression.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20870: Add test for the owned-task report for a single-role assignment change [kafka]
Posted by "zheguang (via GitHub)" <gi...@apache.org>.
zheguang commented on code in PR #23002:
URL: https://github.com/apache/kafka/pull/23002#discussion_r3772449463
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java:
##########
@@ -1831,6 +1832,93 @@ public void testBuildingHeartbeatAssignmentSentWhenChanged(final MemberState mem
assertEquals(List.of(), nonJoiningRequestDataWithChanges.warmupTasks());
}
+ private enum OwnedTaskRole { ACTIVE, STANDBY, WARMUP }
+
+ @ParameterizedTest
+ @EnumSource(OwnedTaskRole.class)
+ public void testBuildingHeartbeatAllOwnedTaskListsSentWhenOnlyOneRoleChanges(final OwnedTaskRole changingRole) {
+ // The broker reads the owned-task lists as a report of what the member holds only when all three of them are
+ // non-null; if any is null it cannot tell that a task was released, and the member effectively fails to
+ // acknowledges the revocation. So a change confined to a single role has to resend the other two lists as well,
+ // even though they did not change.
+ final StreamsGroupHeartbeatRequestManager.HeartbeatState heartbeatState =
+ new StreamsGroupHeartbeatRequestManager.HeartbeatState(
+ streamsRebalanceData,
+ membershipManager,
+ 1234,
+ time
+ );
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ heartbeatState.buildRequestData();
+ when(membershipManager.state()).thenReturn(MemberState.STABLE);
+
+ final Set<StreamsRebalanceData.TaskId> otherActiveTasks =
+ Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_NAME_1, 0));
+ final Set<StreamsRebalanceData.TaskId> otherStandbyTasks =
+ Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_NAME_1, 1));
+ final Set<StreamsRebalanceData.TaskId> otherWarmupTasks =
+ Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_NAME_2, 2));
+ final Set<StreamsRebalanceData.TaskId> changingTask =
+ Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_NAME_2, 3));
+
+ final Function<Set<StreamsRebalanceData.TaskId>, StreamsRebalanceData.Assignment> assignmentWhereRoleHolds =
+ tasksOfChangingRole -> {
+ switch (changingRole) {
+ case ACTIVE:
+ return new StreamsRebalanceData.Assignment(
+ tasksOfChangingRole, otherStandbyTasks, otherWarmupTasks, true);
+ case STANDBY:
+ return new StreamsRebalanceData.Assignment(
+ otherActiveTasks, tasksOfChangingRole, otherWarmupTasks, true);
+ default:
+ return new StreamsRebalanceData.Assignment(
+ otherActiveTasks, otherStandbyTasks, tasksOfChangingRole, true);
+ }
+ };
+ final StreamsRebalanceData.Assignment withoutTheTask = assignmentWhereRoleHolds.apply(Set.of());
+ final StreamsRebalanceData.Assignment withTheTask = assignmentWhereRoleHolds.apply(changingTask);
+
+ streamsRebalanceData.setReconciledAssignment(withoutTheTask);
+ heartbeatState.buildRequestData();
+ assertNull(heartbeatState.buildRequestData().activeTasks());
+
+ // The role gains a task; the other two roles are untouched.
+ streamsRebalanceData.setReconciledAssignment(withTheTask);
+ assertOwnedTasksFullyReported(withTheTask, heartbeatState.buildRequestData());
+ assertNull(heartbeatState.buildRequestData().activeTasks());
+
+ // The role loses it again; the other two roles are untouched. Its own list has to go out as an empty list
Review Comment:
For comment "Its own list has to go out as an empty list" -> how about stressing "all 3 lists have to be reported as non-null"?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java:
##########
@@ -1831,6 +1832,93 @@ public void testBuildingHeartbeatAssignmentSentWhenChanged(final MemberState mem
assertEquals(List.of(), nonJoiningRequestDataWithChanges.warmupTasks());
}
+ private enum OwnedTaskRole { ACTIVE, STANDBY, WARMUP }
+
+ @ParameterizedTest
+ @EnumSource(OwnedTaskRole.class)
+ public void testBuildingHeartbeatAllOwnedTaskListsSentWhenOnlyOneRoleChanges(final OwnedTaskRole changingRole) {
+ // The broker reads the owned-task lists as a report of what the member holds only when all three of them are
+ // non-null; if any is null it cannot tell that a task was released, and the member effectively fails to
+ // acknowledges the revocation. So a change confined to a single role has to resend the other two lists as well,
+ // even though they did not change.
+ final StreamsGroupHeartbeatRequestManager.HeartbeatState heartbeatState =
+ new StreamsGroupHeartbeatRequestManager.HeartbeatState(
+ streamsRebalanceData,
+ membershipManager,
+ 1234,
+ time
+ );
+ when(membershipManager.state()).thenReturn(MemberState.JOINING);
+ heartbeatState.buildRequestData();
+ when(membershipManager.state()).thenReturn(MemberState.STABLE);
+
+ final Set<StreamsRebalanceData.TaskId> otherActiveTasks =
+ Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_NAME_1, 0));
+ final Set<StreamsRebalanceData.TaskId> otherStandbyTasks =
+ Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_NAME_1, 1));
+ final Set<StreamsRebalanceData.TaskId> otherWarmupTasks =
+ Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_NAME_2, 2));
+ final Set<StreamsRebalanceData.TaskId> changingTask =
+ Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_NAME_2, 3));
+
+ final Function<Set<StreamsRebalanceData.TaskId>, StreamsRebalanceData.Assignment> assignmentWhereRoleHolds =
+ tasksOfChangingRole -> {
+ switch (changingRole) {
+ case ACTIVE:
+ return new StreamsRebalanceData.Assignment(
+ tasksOfChangingRole, otherStandbyTasks, otherWarmupTasks, true);
+ case STANDBY:
+ return new StreamsRebalanceData.Assignment(
+ otherActiveTasks, tasksOfChangingRole, otherWarmupTasks, true);
+ default:
+ return new StreamsRebalanceData.Assignment(
+ otherActiveTasks, otherStandbyTasks, tasksOfChangingRole, true);
+ }
+ };
+ final StreamsRebalanceData.Assignment withoutTheTask = assignmentWhereRoleHolds.apply(Set.of());
+ final StreamsRebalanceData.Assignment withTheTask = assignmentWhereRoleHolds.apply(changingTask);
+
+ streamsRebalanceData.setReconciledAssignment(withoutTheTask);
+ heartbeatState.buildRequestData();
+ assertNull(heartbeatState.buildRequestData().activeTasks());
+
+ // The role gains a task; the other two roles are untouched.
+ streamsRebalanceData.setReconciledAssignment(withTheTask);
+ assertOwnedTasksFullyReported(withTheTask, heartbeatState.buildRequestData());
+ assertNull(heartbeatState.buildRequestData().activeTasks());
Review Comment:
If we also assert null on the other 2 lists, that'd match the current implementation fully -- worth it if there's a potential for that part of the code to drift, I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20870: Add test for the owned-task report for a single-role assignment change [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #23002:
URL: https://github.com/apache/kafka/pull/23002#issuecomment-5284953220
@OmniaGM Similar -- would be nice to cherry-pick to `4.4` to close the test gap there, too. (Otherwise, I need to make a note for myself to cherry-pick after 4.4.0 was released....)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20870: Add test for the owned-task report for a single-role assignment change [kafka]
Posted by "OmniaGM (via GitHub)" <gi...@apache.org>.
OmniaGM commented on PR #23002:
URL: https://github.com/apache/kafka/pull/23002#issuecomment-5296330833
> @OmniaGM Similar -- would be nice to cherry-pick to `4.4` to close the test gap there, too. (Otherwise, I need to make a note for myself to cherry-pick after 4.4.0 was released....)
Seems small enough please @mjsax go ahead and cherry pick them to 4.4.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20870: Add test for the owned-task report for a single-role assignment change [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #23002:
URL: https://github.com/apache/kafka/pull/23002#issuecomment-5321870343
Cherry-picked to `4.4`, `4.3`, and `4.2` branches.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20870: Add test for the owned-task report for a single-role assignment change [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #23002:
URL: https://github.com/apache/kafka/pull/23002
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20870: Add test for the owned-task report for a single-role assignment change [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #23002:
URL: https://github.com/apache/kafka/pull/23002#discussion_r3771122278
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java:
##########
@@ -1831,6 +1832,93 @@ public void testBuildingHeartbeatAssignmentSentWhenChanged(final MemberState mem
assertEquals(List.of(), nonJoiningRequestDataWithChanges.warmupTasks());
}
+ private enum OwnedTaskRole { ACTIVE, STANDBY, WARMUP }
+
+ @ParameterizedTest
+ @EnumSource(OwnedTaskRole.class)
+ public void testBuildingHeartbeatAllOwnedTaskListsSentWhenOnlyOneRoleChanges(final OwnedTaskRole changingRole) {
+ // The broker reads the owned-task lists as a report of what the member holds only when all three of them are
+ // non-null; if any is null it cannot tell that a task was released, and the member effectively fails tp
Review Comment:
```suggestion
// non-null; if any is null it cannot tell that a task was released, and the member effectively fails to
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org