squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4317378424
##########
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);
Review Comment:
The above numbers are wrong, since they use `HashMaps` instead of
`TimelineHashMaps`. See
https://github.com/apache/kafka/pull/23868#issuecomment-6272545664 for updated
numbers.
--
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]
