squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4306322828
##########
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:
> Does building the group spec take time? I wonder if the timing is useful
at all...
It does when offloading is enabled and it needs to deep copy the current
assignment. Here are some timings that Claude recorded with a homogeneous setup:
| Members | Spec build, offload off | Spec build, offload on | Assignment
build | Records build | Total, offload off | Total, offload on | Spec % of
total, offload off | Spec % of total, offload on |
|---|---|---|---|---|---|---|---|---|
| 100 | 0.002 | 0.058 | 0.075 | 0.022 | 0.107 | 0.153 | 1.9% | 37.9% |
| 500 | 0.018 | 0.285 | 0.194 | 0.023 | 0.238 | 0.468 | 7.6% | 60.9% |
| 1,000 | 0.026 | 0.495 | 0.311 | 0.030 | 0.461 | 0.916 | 5.6% | 54.0% |
| 5,000 | 0.122 | 2.539 | 1.513 | 0.106 | 1.685 | 4.404 | 7.2% | 57.7% |
| 10,000 | 0.279 | 5.186 | 3.140 | 0.202 | 3.754 | 8.547 | 7.4% | 60.7% |
Building the group spec takes half of the total assignment time.
--
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]
