squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4240427544
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4366,46 +4416,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());
+ group.groupId(), groupEpoch, preferredServerAssignor,
groupSpecTimeMs + assignorTimeMs, assignmentResult.targetAssignment());
} else {
log.info("[GroupId {}] Computed a new target assignment for
epoch {} with '{}' assignor in {}ms.",
- group.groupId(), groupEpoch, preferredServerAssignor,
assignorTimeMs);
+ group.groupId(), groupEpoch, preferredServerAssignor,
groupSpecTimeMs + assignorTimeMs);
Review Comment:
Preserving parity with the timings before this change. We can split out the
two.
--
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]
