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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
