squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4472098283
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4421,11 +4527,81 @@ 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
+ ) {
+ Integer expectedTargetAssignmentEpoch =
inflightOffloadedAssignorEpochs.remove(groupId);
+ if (expectedTargetAssignmentEpoch == null || targetAssignmentEpoch !=
expectedTargetAssignmentEpoch) {
+ // The assignment epoch doesn't match the expected in-flight
assignment epoch.
+ // This should never happen since we only allow a single in-flight
assignor run per
+ // group and cancel assignor tasks before removing their epoch
from the in-flight map.
+ log.error("[GroupId {}] Discarding unexpected offloaded target
assignment for epoch {} (expected in-flight epoch was {}).",
+ groupId, targetAssignmentEpoch, expectedTargetAssignmentEpoch);
+ return new CoordinatorResult<>(List.of());
+ }
+
+ if (exception != null) {
+ log.error("[GroupId {}] Failed to compute a new target assignment
for epoch {}: {}.",
+ groupId, targetAssignmentEpoch, exception.getMessage(),
exception);
+ return new CoordinatorResult<>(List.of());
Review Comment:
Filed as https://issues.apache.org/jira/browse/KAFKA-21428 and
https://issues.apache.org/jira/browse/KAFKA-21429.
--
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]
