lucasbru commented on code in PR #21110:
URL: https://github.com/apache/kafka/pull/21110#discussion_r2605876778
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -699,17 +709,40 @@ public void
onHeartbeatSuccess(StreamsGroupHeartbeatResponse response) {
processAssignmentReceived(
toTasksAssignment(activeTasks),
toTasksAssignment(standbyTasks),
- toTasksAssignment(warmupTasks)
+ toTasksAssignment(warmupTasks),
+ isGroupReady
);
- } else {
- if (responseData.activeTasks() != null ||
- responseData.standbyTasks() != null ||
- responseData.warmupTasks() != null) {
+ } else if (responseData.activeTasks() != null ||
responseData.standbyTasks() != null || responseData.warmupTasks() != null) {
+ throw new IllegalStateException("Invalid response data, task
collections must be all null or all non-null: "
+ + responseData);
+ } else if (isGroupReady != targetAssignment.isGroupReady) {
+ // If the client did not provide a new assignment, but the group
is now ready or not ready anymore, so
+ // update the target assignment and reconcile it.
+ processAssignmentReceived(
+ targetAssignment.activeTasks,
+ targetAssignment.standbyTasks,
+ targetAssignment.warmupTasks,
+ isGroupReady
+ );
+ }
+ }
- throw new IllegalStateException("Invalid response data, task
collections must be all null or all non-null: "
- + responseData);
+ private boolean
isGroupReady(List<StreamsGroupHeartbeatResponseData.Status> statuses) {
+ if (statuses != null) {
+ for (final StreamsGroupHeartbeatResponseData.Status status :
statuses) {
+ switch
(StreamsGroupHeartbeatResponse.Status.fromCode(status.statusCode())) {
Review Comment:
Hmm, that may work. We can do it. But note that this will not be handled by
the versioning of the RPC code, so you have to check the versions and have
custom logic to avoid sending unknown statuses. So we have to be very careful
when adding new statuses
--
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]
