squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4482730971
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4365,46 +4431,86 @@ private UpdateTargetAssignmentResult<Assignment>
maybeUpdateTargetAssignment(
return
UpdateTargetAssignmentResult.fromLastTargetAssignment(group, updatedMember);
}
+ boolean offloadAssignor =
consumerGroupAssignorOffloadEnable(group.groupId());
+
String preferredServerAssignor = group.computePreferredServerAssignor(
member,
updatedMember
).orElse(defaultConsumerGroupAssignor.name());
- try {
- UpdatedMembersAndTargetAssignmentView<ConsumerGroupMember,
Assignment> updatedMembersAndTargetAssignment =
- new UpdatedMembersAndTargetAssignmentView<>(
- group.members(),
- group.staticMembers(),
- group.targetAssignment(),
- ConsumerGroupMember::instanceId
- );
-
updatedMembersAndTargetAssignment.addOrUpdateMember(updatedMember.memberId(),
updatedMember);
- long startTimeMs = time.milliseconds();
- GroupSpec groupSpec = new
GroupSpecBuilder.ConsumerGroupSpecBuilder()
- .withMembers(updatedMembersAndTargetAssignment.members())
- .withSubscriptionType(subscriptionType)
-
.withTargetAssignment(updatedMembersAndTargetAssignment.targetAssignment())
- .withInvertedTargetAssignment(group.invertedTargetAssignment())
- .withMetadataImage(metadataImage)
-
.withResolvedRegularExpressions(group.resolvedRegularExpressions())
- .build();
+ UpdatedMembersAndTargetAssignmentView<ConsumerGroupMember, Assignment>
updatedMembersAndTargetAssignment =
+ new UpdatedMembersAndTargetAssignmentView<>(
+ group.members(),
+ group.staticMembers(),
+ group.targetAssignment(),
+ ConsumerGroupMember::instanceId
+ );
+
updatedMembersAndTargetAssignment.addOrUpdateMember(updatedMember.memberId(),
updatedMember);
- TargetAssignmentBuilder assignmentResultBuilder =
+ // Use the same metadata image throughout.
+ CoordinatorMetadataImage metadataImage = this.metadataImage;
+
+ long groupSpecStartTimeMs = time.milliseconds();
+ GroupSpec groupSpec = new GroupSpecBuilder.ConsumerGroupSpecBuilder()
+ .withMembers(updatedMembersAndTargetAssignment.members())
+ .withSubscriptionType(subscriptionType)
+
.withTargetAssignment(updatedMembersAndTargetAssignment.targetAssignment())
+ .withInvertedTargetAssignment(group.invertedTargetAssignment())
+ .withMetadataImage(metadataImage)
+ .withResolvedRegularExpressions(group.resolvedRegularExpressions())
+ .withAssignorOffload(offloadAssignor)
+ .build();
+ long groupSpecTimeMs = time.milliseconds() - groupSpecStartTimeMs;
+
+ Supplier<TargetAssignmentBuilder.TargetAssignmentResult>
buildTargetAssignment = () -> {
+ long assignorStartTimeMs = time.milliseconds();
+ TargetAssignmentBuilder.TargetAssignmentResult assignmentResult =
new TargetAssignmentBuilder(groupEpoch,
consumerGroupAssignors.get(preferredServerAssignor))
.withTime(time)
.withMetadataImage(metadataImage)
- .withGroupSpec(groupSpec);
-
- TargetAssignmentBuilder.TargetAssignmentResult assignmentResult =
- assignmentResultBuilder.build();
- long assignorTimeMs = time.milliseconds() - startTimeMs;
+ .withGroupSpec(groupSpec)
+ .build();
+ long assignorTimeMs = time.milliseconds() - assignorStartTimeMs;
if (log.isDebugEnabled()) {
- log.debug("[GroupId {}] Computed a new target assignment for
epoch {} with '{}' assignor in {}ms: {}.",
- group.groupId(), groupEpoch, preferredServerAssignor,
assignorTimeMs, assignmentResult.targetAssignment());
+ log.debug("[GroupId {}] Computed a new target assignment for
epoch {} with '{}' assignor in {}ms after building the group spec in {}ms: {}.",
+ group.groupId(), groupEpoch, preferredServerAssignor,
assignorTimeMs, groupSpecTimeMs, assignmentResult.targetAssignment());
} else {
- log.info("[GroupId {}] Computed a new target assignment for
epoch {} with '{}' assignor in {}ms.",
- group.groupId(), groupEpoch, preferredServerAssignor,
assignorTimeMs);
+ log.info("[GroupId {}] Computed a new target assignment for
epoch {} with '{}' assignor in {}ms after building the group spec in {}ms.",
+ group.groupId(), groupEpoch, preferredServerAssignor,
assignorTimeMs, groupSpecTimeMs);
+ }
+
+ return assignmentResult;
+ };
+
+ if (offloadAssignor) {
+ Map<String, String> previousStaticMembers =
Map.copyOf(updatedMembersAndTargetAssignment.staticMembers());
+
+ inflightOffloadedAssignorEpochs.put(group.groupId(), groupEpoch);
+ executor.schedule(
+ targetAssignmentUpdateKey,
+ buildTargetAssignment::get,
+ (result, exception) ->
handleOffloadedConsumerTargetAssignmentResult(
+ group.groupId(),
+ groupEpoch,
+ previousStaticMembers,
+ result,
+ exception
+ )
+ );
+
+ // fromLastTargetAssignment looks up the assignment by instance
id, so it's fine not to
+ // use updatedMembersAndTargetAssignment.
+ return
UpdateTargetAssignmentResult.fromLastTargetAssignment(group, updatedMember);
Review Comment:
Filed as https://issues.apache.org/jira/browse/KAFKA-21431 and opened #24108
to address it
--
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]
