squah-confluent commented on code in PR #21696:
URL: https://github.com/apache/kafka/pull/21696#discussion_r2916799306
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3947,17 +3944,39 @@ private UpdateTargetAssignmentResult<Assignment>
maybeUpdateTargetAssignment(
* @param updatedMember The updated member (optional).
* @param metadataImage The metadata image.
* @param records The list to accumulate any new records.
+ * @param returnedStatus A mutable collection of status to be
returned in the response.
* @return The new target assignment for the updated member, or EMPTY if
no member specified.
*/
- private TasksTuple updateStreamsTargetAssignment(
+ private UpdateTargetAssignmentResult<TasksTuple>
maybeUpdateStreamsTargetAssignment(
StreamsGroup group,
int groupEpoch,
Optional<StreamsGroupMember> updatedMember,
ConfiguredTopology configuredTopology,
CoordinatorMetadataImage metadataImage,
List<CoordinatorRecord> records,
+ Optional<List<Status>> returnedStatus,
Map<String, String> assignmentConfigs
) {
+ boolean initialDelayActive =
timer.isScheduled(streamsInitialRebalanceKey(group.groupId()));
+ if (initialDelayActive) {
+ returnedStatus.ifPresent(statusList -> statusList.add(
+ new Status()
+
.setStatusCode(StreamsGroupHeartbeatResponse.Status.ASSIGNMENT_DELAYED.code())
+ .setStatusDetail("Assignment delayed due to the configured
initial rebalance delay.")
+ ));
+
+ return new UpdateTargetAssignmentResult<>(
+ Math.max(1, group.assignmentEpoch()),
+ updatedMember.map(member ->
group.targetAssignment(member.memberId()))
Review Comment:
Fixed it, thanks!
--
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]
