squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4239535349
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4422,11 +4512,73 @@ private UpdateTargetAssignmentResult<Assignment>
maybeUpdateTargetAssignment(
} else {
return new UpdateTargetAssignmentResult<>(groupEpoch,
Assignment.EMPTY);
}
- } catch (PartitionAssignorException ex) {
- String msg = String.format("Failed to compute a new target
assignment for epoch %d: %s",
- groupEpoch, ex.getMessage());
- log.error("[GroupId {}] {}.", group.groupId(), msg, ex);
- throw new UnknownServerException(msg, ex);
+ }
+ }
+
+ /**
+ * Handle the result of the asynchronous task that computes a consumer
group's
+ * target assignment when assignor offloading is enabled.
+ *
+ * @param groupId The group id.
+ * @param targetAssignmentEpoch The assignment epoch.
+ * @param previousStaticMembers The static members at schedule time,
keyed by instance id.
+ * @param result The computed target assignment.
+ * @param exception The exception if the computation failed.
+ * @return A CoordinatorResult containing the records to mutate the group
state.
+ */
+ private CoordinatorResult<Void, CoordinatorRecord>
handleOffloadedConsumerTargetAssignmentResult(
+ String groupId,
+ int targetAssignmentEpoch,
+ Map<String, String> previousStaticMembers,
+ TargetAssignmentBuilder.TargetAssignmentResult result,
+ Throwable exception
+ ) {
+ inflightOffloadedAssignorEpochs.remove(groupId, targetAssignmentEpoch);
Review Comment:
We can do it defensively. Right now it should be impossible to not be the
correct one since we cancel on every epoch bump after a rollback.
--
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]
