dajac commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4206670887
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -551,6 +552,12 @@ GroupMetadataManager build() {
*/
private long lastMetadataImageWithNewTopics = -1L;
+ /**
+ * A map of in-flight offloaded assignor runs. The keys are group ids and
values are the
+ * assignment epochs for each run.
+ */
Review Comment:
Could we explain the rules to maintain this data structure?
##########
coordinator-common/src/testFixtures/java/org/apache/kafka/coordinator/common/runtime/MockCoordinatorExecutor.java:
##########
@@ -103,4 +104,17 @@ public List<ExecutorResult<T>> poll() {
queue.clear();
return results;
}
+
+ public ExecutorResult<T> poll(String key) {
+ Iterator<ExecutorTask<?>> it = queue.iterator();
+ while (it.hasNext()) {
+ ExecutorTask<?> task = it.next();
+ if (task.key.equals(key)) {
+ it.remove();
+ tasks.remove(task.key, task.task);
+ return new ExecutorResult<>(task.key, task.execute());
+ }
+ }
+ return null;
+ }
Review Comment:
This feels a bit weird. If I understand it correctly, it skips all the tasks
before `key` and then executes the task associated to `key`. Isn't it error
prone?
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -3880,11 +3880,14 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
new Thread {
override def run : Unit = {
consumer.subscribe(util.Set.of(topic))
+ var started = false
try {
while (consumerThreadRunning.get()) {
consumer.poll(JDuration.ofSeconds(5))
- if (!consumer.assignment.isEmpty && startLatch.getCount > 0L)
+ if (!started && !consumer.assignment.isEmpty) {
Review Comment:
Why do we need to change this?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1389,6 +1397,7 @@ private void convertToClassicGroup(
// Directly update the states instead of replaying the records because
// the classicGroup reference is needed for triggering the rebalance.
+ cancelTargetAssignmentUpdate(consumerGroup.groupId());
Review Comment:
Should this be part of `removeGroup` to ensure that we clean it up when the
group is removed?
##########
core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala:
##########
@@ -399,9 +445,29 @@ class ConsumerGroupHeartbeatRequestTest(cluster:
ClusterInstance) extends GroupC
// Verify the response.
assertNotNull(consumerGroupHeartbeatResponse.data.memberId)
- assertEquals(2, consumerGroupHeartbeatResponse.data.memberEpoch)
+ assertEquals(if (isConsumerAssignorOffloadEnabled) 1 else 2,
consumerGroupHeartbeatResponse.data.memberEpoch)
assertEquals(new ConsumerGroupHeartbeatResponseData.Assignment(),
consumerGroupHeartbeatResponse.data.assignment)
+ // When assignor offload is enabled, the initial assignment is available
in a later heartbeat.
+ if (isConsumerAssignorOffloadEnabled) {
Review Comment:
I wonder whether adding this to all the tests bring any value? Having it
once makes sense though.
##########
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:
It seems a bit weird to combine both times. What's the reasoning?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4422,11 +4512,73 @@ 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
+ ) {
+ inflightOffloadedAssignorEpochs.remove(groupId, targetAssignmentEpoch);
Review Comment:
Do we need to ensure that `targetAssignmentEpoch` was the correct one?
--
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]
