You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "squah-confluent (via GitHub)" <gi...@apache.org> on 2026/10/03 17:28:47 UTC
[PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
squah-confluent opened a new pull request, #23688:
URL: https://github.com/apache/kafka/pull/23688
(no comment)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4239531500
##########
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:
It comes from https://github.com/apache/kafka/pull/23702. The original code allowed the earlier consumers to bump the latch more than once while the newer consumers wait behind the assignment batching or offload delay.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4256840760
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4421,11 +4527,81 @@ 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
+ ) {
+ Integer expectedTargetAssignmentEpoch = inflightOffloadedAssignorEpochs.remove(groupId);
+ if (expectedTargetAssignmentEpoch == null || targetAssignmentEpoch != expectedTargetAssignmentEpoch) {
+ // The assignment epoch doesn't match the expected in-flight assignment epoch.
+ // This should never happen since we only allow a single in-flight assignor run per
+ // group and cancel assignor tasks before removing their epoch from the in-flight map.
+ log.error("[GroupId {}] Discarding unexpected offloaded target assignment for epoch {} (expected in-flight epoch was {}).",
+ groupId, targetAssignmentEpoch, expectedTargetAssignmentEpoch);
+ return new CoordinatorResult<>(List.of());
+ }
+
+ if (exception != null) {
+ log.error("[GroupId {}] Failed to compute a new target assignment for epoch {}: {}.",
+ groupId, targetAssignmentEpoch, exception.getMessage(), exception);
+ return new CoordinatorResult<>(List.of());
Review Comment:
One concern here is that the exception may not be noticed. When the assignment is computed synchronously, the client gets an error at least. Here, it only shows up in the log. Should we have a metric so an admin could alert on it? It could be considered as a follow-up.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -31222,6 +31222,908 @@ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
);
}
+ @Test
+ public void testConsumerGroupAsyncTargetAssignment() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ // Member 1 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result1 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result1.response()
+ );
+
+ ConsumerGroupMember expectedMember1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember1),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember1)
+ ),
+ result1.records()
+ );
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 2 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result2 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId2)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result2.response()
+ );
+
+ ConsumerGroupMember expectedMember2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember2),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 3, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember2)
+ ),
+ result2.records()
+ );
+
+ // Assignor completes on executor.
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId1, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ ))
+ )));
+
+ // The target assignment is written.
+ List<MockCoordinatorExecutor.ExecutorResult<CoordinatorRecord>> tasks = context.processTasks();
+ assertEquals(1, tasks.size());
+ assertEquals(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId), tasks.get(0).key());
+ assertUnorderedRecordsEquals(
+ List.of(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ )),
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId2, Map.of())
+ ),
+ List.of(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentMetadataRecord(groupId, 2, context.time.milliseconds()))
+ ),
+ tasks.get(0).result().records()
+ );
+
+ // The next assignment is not started automatically.
+ assertFalse(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 1 heartbeats again and receives the first assignment.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result3 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(1));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(2)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of(
+ new ConsumerGroupHeartbeatResponseData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(List.of(0, 1, 2, 3, 4, 5))
+ ))),
+ result3.response()
+ );
+
+ ConsumerGroupMember expectedMember3 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(2)
+ .setPreviousMemberEpoch(1)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 2))
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember3)
+ ),
+ result3.records()
+ );
+
+ // The next assignor run is scheduled.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentDiscardedOnRollback() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ long groupMetadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)));
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 10))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withAssignment(memberId1, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(groupMetadataHash))
+ .build();
+
+ // Member 2 joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Simulate a failed write to the log.
+ context.rollback();
+
+ // Assignor completes on executor.
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId1, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2)
+ )),
+ memberId2, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5)
+ ))
+ )));
+
+ // No records are written.
+ List<MockCoordinatorExecutor.ExecutorResult<CoordinatorRecord>> tasks = context.processTasks();
+ assertEquals(
+ List.of(
+ new MockCoordinatorExecutor.ExecutorResult<>(
+ GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId),
+ new CoordinatorResult<>(List.of())
+ )
+ ),
+ tasks
+ );
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentCancelledOnRollbackAfterHeartbeatEpochBump() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ // Member joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ assertEquals(1, context.executor.size());
+
+ // Simulate a failed write to the log.
+ context.rollback();
+
+ // The member rejoins, re-bumping the group epoch to the same value.
+ // The stale assignor run is cancelled and replaced.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ assertEquals(2, context.executor.size());
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentCancelledOnRollbackAfterFence() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+ String memberId3 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ long groupMetadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)));
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2)), 10))
+ .build();
+ ConsumerGroupMember member2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5)), 10))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withMember(member2)
+ .withAssignment(memberId1, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2)))
+ .withAssignment(memberId2, mkAssignment(mkTopicAssignment(fooTopicId, 3, 4, 5)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(groupMetadataHash))
+ .build();
+
+ // Member 3 joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId3)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Simulate a failed write to the log.
+ context.rollback();
+
+ // Member 2 leaves the group, re-bumping the group epoch to the same value.
+ // The stale assignor run is cancelled.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(LEAVE_GROUP_MEMBER_EPOCH));
+
+ assertFalse(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentCancelledOnRollbackAfterRegexResolution() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicRegex("foo*")
+ .setServerAssignorName("range")
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withAssignmentEpoch(10))
+ .build();
+
+ // Member 2 joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor and regex resolution are scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ assertTrue(context.executor.isScheduled(groupId + "-regex"));
+
+ // Simulate a failed write to the log.
+ context.rollback();
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ assertTrue(context.executor.isScheduled(groupId + "-regex"));
+
+ // The regex resolution runs and bumps the group epoch to the same value.
+ // The stale assignor run is cancelled.
+ context.processTask(groupId + "-regex");
Review Comment:
How do we know that the stale assignment run was a no-op here?
##########
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:
We need to think through how we want to log when offloading is used. For instance, it would be nice to see the following steps in the logs:
1/ Scheduled computing of new target assignment....
2/ Computed new target...
3/ Installed new target... or Failed to install new target assignment...
Does building the group spec take time? I wonder if the timing is useful at all...
I also wonder if we want to have more timing details about the offloading. For instance, the time spent in the queue, the time it took the compute the assignment, the time spent in the queue before installing it, etc. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -607,6 +630,7 @@ private GroupMetadataManager(
this.streamsGroupAssignmentRefiner = streamsGroupAssignmentRefiner;
this.topicRegexResolver = new TopicRegexResolver(() -> authorizerPlugin, this.time);
this.topicHashCache = new HashMap<>();
+ this.inflightOffloadedAssignorEpochs = new HashMap<>();
Review Comment:
Something to explore as a follow-up: Could we store this in the group itself?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -31222,6 +31222,908 @@ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
);
}
+ @Test
+ public void testConsumerGroupAsyncTargetAssignment() {
Review Comment:
Should we also add tests for the following cases?
1/ The assignor throws an exception. We should also verify that the next heartbeat schedules a new run.
2/ The write of the new target assignment fails.
3/ Members using the classic protocol join while a run is in flight.
##########
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);
Review Comment:
nit: `schedule` returns false if a task is already scheduled for the key. It cannot happen today because of the `isScheduled` check above but should we only record the epoch when it returns true?
##########
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:
I think that there is an issue when all the members of the consumer group use the classic protocol. If the offloaded run is lost (e.g. the assignor throws an exception, the write of the new target assignment fails or the coordinator fails over before it is written), nothing triggers a new run. The classic heartbeat and sync paths never call `maybeUpdateTargetAssignment` and they only ask the members to rejoin when `memberEpoch < assignmentEpoch`. So the group stays in `Assigning` until a member rejoins. When the assignment is computed synchronously, the join request fails and the client retries it.
The following test reproduces it. It fails on the last assertion.
<details>
<summary>Test</summary>
```java
@Test
public void testClassicGroupHeartbeatToConsumerGroupAfterFailedAsyncTargetAssignment() throws Exception {
String groupId = "fooup";
String memberId1 = Uuid.randomUuid().toString();
String memberId2 = Uuid.randomUuid().toString();
Uuid fooTopicId = Uuid.randomUuid();
String fooTopicName = "foo";
ConsumerGroupPartitionAssignor assignor = mock(ConsumerGroupPartitionAssignor.class);
when(assignor.name()).thenReturn("range");
when(assignor.assign(any(), any())).thenThrow(new PartitionAssignorException("Assignment failed."));
CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
.addTopic(fooTopicId, fooTopicName, 6)
.addRacks()
.buildCoordinatorMetadataImage();
JoinGroupRequestData.JoinGroupRequestProtocolCollection protocols =
GroupMetadataManagerTestContext.toConsumerProtocol(List.of(fooTopicName), List.of());
// Consumer group with a single member using the classic protocol. The migration
// policy does not allow downgrading it to a classic group.
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, ConsumerGroupMigrationPolicy.UPGRADE.toString())
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
.withMetadataImage(metadataImage)
.withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
.withMember(new ConsumerGroupMember.Builder(memberId1)
.setState(MemberState.STABLE)
.setMemberEpoch(10)
.setPreviousMemberEpoch(10)
.setClientId(DEFAULT_CLIENT_ID)
.setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
.setSubscribedTopicNames(List.of(fooTopicName))
.setRebalanceTimeoutMs(5000)
.setClassicMemberMetadata(
new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
.setSessionTimeoutMs(5000)
.setSupportedProtocols(ConsumerGroupMember.classicProtocolListFromJoinRequestProtocolCollection(protocols))
)
.setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 10))
.build())
.withAssignment(memberId1, mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)))
.withAssignmentEpoch(10)
.withMetadataHash(computeGroupHash(Map.of(
fooTopicName, computeTopicHash(fooTopicName, metadataImage)
))))
.build();
// Member 2 joins with the classic protocol. The group epoch is bumped to 11 and an
// assignor run is scheduled. The join completes with the current generation.
GroupMetadataManagerTestContext.JoinResult joinResult = context.sendClassicGroupJoin(
new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
.withGroupId(groupId)
.withMemberId(memberId2)
.withSessionTimeoutMs(5000)
.withRebalanceTimeoutMs(5000)
.withProtocols(protocols)
.build()
);
joinResult.appendFuture.complete(null);
context.commit();
assertEquals(
new JoinGroupResponseData()
.setMemberId(memberId2)
.setGenerationId(10)
.setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
.setProtocolName("range"),
joinResult.joinFuture.get()
);
assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
// The assignor fails, so no target assignment is written.
assertEquals(
List.of(
new MockCoordinatorExecutor.ExecutorResult<>(
GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId),
new CoordinatorResult<>(List.of())
)
),
context.processTasks()
);
assertEquals(11, context.groupMetadataManager.consumerGroup(groupId).groupEpoch());
assertEquals(10, context.groupMetadataManager.consumerGroup(groupId).assignmentEpoch());
// Member 2 syncs and gets an empty assignment.
context.verifyClassicGroupSyncToConsumerGroup(
groupId,
memberId2,
10,
"range",
ConsumerProtocol.PROTOCOL_TYPE,
List.of()
);
// Both members heartbeat and are not asked to rejoin.
for (String memberId : List.of(memberId1, memberId2)) {
assertEquals(
new HeartbeatResponseData(),
context.sendClassicGroupHeartbeat(
new HeartbeatRequestData()
.setGroupId(groupId)
.setMemberId(memberId)
.setGenerationId(10)
).response()
);
}
// The target assignment is still behind the group epoch, so another assignor run
// must be scheduled.
assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
}
```
</details>
The same can already happen with the assignment interval but it is much more likely with offloading. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4356,6 +4415,13 @@ private UpdateTargetAssignmentResult<Assignment> maybeUpdateTargetAssignment(
return UpdateTargetAssignmentResult.fromLastTargetAssignment(group, updatedMember);
}
+ String targetAssignmentUpdateKey = groupTargetAssignmentUpdateKey(group.groupId());
+ if (executor.isScheduled(targetAssignmentUpdateKey)) {
+ // There is already an async assignor run in progress. We must not start another run
Review Comment:
This is not entirely true after a cancellation. `cancel` does not interrupt the computation so the cancelled run may still be running when we schedule the next one. Should we update the comment?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4421,11 +4527,81 @@ 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
+ ) {
+ Integer expectedTargetAssignmentEpoch = inflightOffloadedAssignorEpochs.remove(groupId);
Review Comment:
If we ever reach this branch with a stale run, we would also remove the epoch of the in-flight run. Should we only remove it when it matches `targetAssignmentEpoch`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4421,11 +4527,81 @@ 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
+ ) {
+ Integer expectedTargetAssignmentEpoch = inflightOffloadedAssignorEpochs.remove(groupId);
+ if (expectedTargetAssignmentEpoch == null || targetAssignmentEpoch != expectedTargetAssignmentEpoch) {
+ // The assignment epoch doesn't match the expected in-flight assignment epoch.
+ // This should never happen since we only allow a single in-flight assignor run per
+ // group and cancel assignor tasks before removing their epoch from the in-flight map.
+ log.error("[GroupId {}] Discarding unexpected offloaded target assignment for epoch {} (expected in-flight epoch was {}).",
+ groupId, targetAssignmentEpoch, expectedTargetAssignmentEpoch);
+ return new CoordinatorResult<>(List.of());
+ }
+
+ if (exception != null) {
+ log.error("[GroupId {}] Failed to compute a new target assignment for epoch {}: {}.",
+ groupId, targetAssignmentEpoch, exception.getMessage(), exception);
+ return new CoordinatorResult<>(List.of());
+ }
+
+ try {
+ ConsumerGroup consumerGroup = consumerGroup(groupId);
+ if (consumerGroup.groupEpoch() < targetAssignmentEpoch) {
+ // The assignment epoch is greater than the group epoch. This means that the
+ // assignment was built off a group state that was not successfully written to the
+ // log and was reverted. Discard the assignment.
+ log.debug("[GroupId {}] Discarding stale offloaded target assignment for epoch {} (current group epoch is {}).",
+ groupId, targetAssignmentEpoch, consumerGroup.groupEpoch());
+ return new CoordinatorResult<>(List.of());
+ }
+
+ if (consumerGroup.assignmentEpoch() >= targetAssignmentEpoch) {
Review Comment:
It seems that no test covers this case. According to Claude, it can be reached when the upgrade from a classic group is reverted and the group is upgraded again, because the upgrade sets the assignment epoch to the generation of the classic group. Should we add a test for it?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4332,6 +4359,38 @@ static boolean canComputeNextTargetAssignment(
return currentTimeMs >= assignmentTimestampMs + assignmentIntervalMs;
}
+ public static String groupTargetAssignmentUpdateKey(String groupId) {
Review Comment:
nit: Should we make it package private and move it next to the other key helpers (e.g. `consumerGroupJoinKey`)?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -549,6 +550,28 @@ 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.
+ *
+ * <p>Entries are added when an offloaded assignor run is queued on {@link #executor} and removed
+ * when the run completes or is cancelled.
+ *
+ * <p>This is used to detect when an in-flight offloaded assignor run is no longer valid due to
+ * a group coordinator rollback. After a rollback that resets the group epoch, an in-flight
+ * assignor run may:
+ * <ul>
+ * <li>finish before another epoch bump, in which case the current epoch of the group will be
+ * less than the assignment epoch of the run. We detect this case when the assignor run
+ * completes and discard the assignment.</li>
+ * <li>finish after another epoch bump, which we cannot reliably detect on completion of the
+ * assignor run. Instead, when bumping the group epoch, we check the assignment epoch of
+ * the in-flight assignor run if it exists and cancel the run if its assignment epoch is
+ * greater than the unbumped group epoch.</li>
Review Comment:
nit: The code compares against the new group epoch (`assignmentEpoch >= newGroupEpoch`). Should we phrase it the same way here?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -31222,6 +31222,908 @@ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
);
}
+ @Test
+ public void testConsumerGroupAsyncTargetAssignment() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ // Member 1 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result1 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result1.response()
+ );
+
+ ConsumerGroupMember expectedMember1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember1),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember1)
+ ),
+ result1.records()
+ );
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 2 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result2 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId2)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result2.response()
+ );
+
+ ConsumerGroupMember expectedMember2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember2),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 3, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember2)
+ ),
+ result2.records()
+ );
+
+ // Assignor completes on executor.
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId1, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ ))
+ )));
+
+ // The target assignment is written.
+ List<MockCoordinatorExecutor.ExecutorResult<CoordinatorRecord>> tasks = context.processTasks();
+ assertEquals(1, tasks.size());
+ assertEquals(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId), tasks.get(0).key());
+ assertUnorderedRecordsEquals(
+ List.of(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ )),
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId2, Map.of())
+ ),
+ List.of(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentMetadataRecord(groupId, 2, context.time.milliseconds()))
+ ),
+ tasks.get(0).result().records()
+ );
+
+ // The next assignment is not started automatically.
+ assertFalse(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 1 heartbeats again and receives the first assignment.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result3 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(1));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(2)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of(
+ new ConsumerGroupHeartbeatResponseData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(List.of(0, 1, 2, 3, 4, 5))
+ ))),
+ result3.response()
+ );
+
+ ConsumerGroupMember expectedMember3 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(2)
+ .setPreviousMemberEpoch(1)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 2))
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember3)
+ ),
+ result3.records()
+ );
+
+ // The next assignor run is scheduled.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentDiscardedOnRollback() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ long groupMetadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)));
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 10))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withAssignment(memberId1, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(groupMetadataHash))
+ .build();
+
+ // Member 2 joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Simulate a failed write to the log.
+ context.rollback();
+
+ // Assignor completes on executor.
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId1, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2)
+ )),
+ memberId2, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 3, 4, 5)
+ ))
+ )));
+
+ // No records are written.
+ List<MockCoordinatorExecutor.ExecutorResult<CoordinatorRecord>> tasks = context.processTasks();
+ assertEquals(
+ List.of(
+ new MockCoordinatorExecutor.ExecutorResult<>(
+ GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId),
+ new CoordinatorResult<>(List.of())
+ )
+ ),
+ tasks
+ );
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentCancelledOnRollbackAfterHeartbeatEpochBump() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ // Member joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ assertEquals(1, context.executor.size());
+
+ // Simulate a failed write to the log.
+ context.rollback();
+
+ // The member rejoins, re-bumping the group epoch to the same value.
+ // The stale assignor run is cancelled and replaced.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ assertEquals(2, context.executor.size());
Review Comment:
Should we execute them in order to ensure that the first one is really a no-op?
##########
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);
Review Comment:
`CoordinatorExecutorImpl` does not run a task which was cancelled or replaced in the meantime. This one and `poll()` run it anyway because the result of `tasks.remove` is ignored. This means that the tests verifying the cancellation do not really verify it. For instance, `testConsumerGroupAsyncTargetAssignmentDiscardedWhenLastMemberLeaves` only passes because the cancelled run is executed. I think that we should skip them:
```suggestion
if (!tasks.remove(task.key, task.task)) continue;
```
We need the same in `poll()`. With this, we could also verify that only the replacement run is executed in `testConsumerGroupAsyncTargetAssignmentCancelledOnRollbackAfterHeartbeatEpochBump` by replacing the last assertion with the following. It fails with the current mock because the cancelled run is applied and the replacement run is discarded.
```java
// Only the replacement run is executed and its target assignment is written.
assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
memberId, new MemberAssignmentImpl(mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
))
)));
assertEquals(
List.of(
new MockCoordinatorExecutor.ExecutorResult<>(
GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId),
new CoordinatorResult<>(List.of(
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId, mkAssignment(
mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
)),
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentMetadataRecord(groupId, 2, context.time.milliseconds())
))
)
),
context.processTasks()
);
```
##########
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 = () -> {
Review Comment:
nit: The supplier captures `group` and resolves the assignor on the background thread. Should we pass the group id and the assignor instead so it only captures immutable state?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -31222,6 +31222,908 @@ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
);
}
+ @Test
+ public void testConsumerGroupAsyncTargetAssignment() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ // Member 1 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result1 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result1.response()
+ );
+
+ ConsumerGroupMember expectedMember1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember1),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember1)
+ ),
+ result1.records()
+ );
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 2 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result2 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId2)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result2.response()
+ );
+
+ ConsumerGroupMember expectedMember2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember2),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 3, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember2)
+ ),
+ result2.records()
+ );
+
+ // Assignor completes on executor.
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId1, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ ))
+ )));
+
+ // The target assignment is written.
+ List<MockCoordinatorExecutor.ExecutorResult<CoordinatorRecord>> tasks = context.processTasks();
+ assertEquals(1, tasks.size());
+ assertEquals(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId), tasks.get(0).key());
+ assertUnorderedRecordsEquals(
+ List.of(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ )),
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId2, Map.of())
+ ),
+ List.of(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentMetadataRecord(groupId, 2, context.time.milliseconds()))
+ ),
+ tasks.get(0).result().records()
+ );
+
+ // The next assignment is not started automatically.
+ assertFalse(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 1 heartbeats again and receives the first assignment.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result3 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(1));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(2)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of(
+ new ConsumerGroupHeartbeatResponseData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(List.of(0, 1, 2, 3, 4, 5))
+ ))),
+ result3.response()
+ );
+
+ ConsumerGroupMember expectedMember3 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(2)
+ .setPreviousMemberEpoch(1)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 2))
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember3)
+ ),
+ result3.records()
+ );
+
+ // The next assignor run is scheduled.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentDiscardedOnRollback() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ long groupMetadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)));
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 10))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withAssignment(memberId1, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(groupMetadataHash))
+ .build();
+
+ // Member 2 joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Simulate a failed write to the log.
+ context.rollback();
Review Comment:
This is just an idea for the future. We have discussed having a snapshottable timer to simplify the management of timers from within the shard. I think that we could consider the same for the executor. Overall, a task would only be scheduled if committed and discarded otherwise.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4239531500
##########
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:
Comes from https://github.com/apache/kafka/pull/23702. The original code allowed the earlier consumers to bump the latch multiple times while the newer consumers wait behind the assignment batching or offload delay.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4240352671
##########
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:
Most of them are indeed not that useful. We can be more selective and ditch the tests whose stated purpose is testing validation for example.
The remaining tests kept would be
* `testConsumerGroupHeartbeatIsAccessibleWhenNewGroupCoordinatorIsEnabled` - It's the happy path test
* `testConsumerGroupHeartbeatWithRegularExpression` - Regexes interact with assignments
* `testRejoiningStaticMemberGetsAssignmentsBackWhenNewGroupCoordinatorIsEnabled` - Borderline choice
* `testConsumerGroupHeartbeatRebalance` - Assignments with multiple members
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub) via jira" <ji...@kafka.apache.org>.
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:
This also impacts the much more normal case where the last consumer member of a mixed group leaves. The group epoch is bumped but the member's partitions are not redistributed to the remaining classic members.
I filed this 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4306202599
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -607,6 +630,7 @@ private GroupMetadataManager(
this.streamsGroupAssignmentRefiner = streamsGroupAssignmentRefiner;
this.topicRegexResolver = new TopicRegexResolver(() -> authorizerPlugin, this.time);
this.topicHashCache = new HashMap<>();
+ this.inflightOffloadedAssignorEpochs = new HashMap<>();
Review Comment:
This is challenging, since on the first heartbeat that creates the group, the `Group` instance is not the one that gets persisted into the in-memory state. Only after the first replay do we get a stable `Group` instance.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub) via jira" <ji...@kafka.apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4472098283
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4421,11 +4527,81 @@ 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
+ ) {
+ Integer expectedTargetAssignmentEpoch = inflightOffloadedAssignorEpochs.remove(groupId);
+ if (expectedTargetAssignmentEpoch == null || targetAssignmentEpoch != expectedTargetAssignmentEpoch) {
+ // The assignment epoch doesn't match the expected in-flight assignment epoch.
+ // This should never happen since we only allow a single in-flight assignor run per
+ // group and cancel assignor tasks before removing their epoch from the in-flight map.
+ log.error("[GroupId {}] Discarding unexpected offloaded target assignment for epoch {} (expected in-flight epoch was {}).",
+ groupId, targetAssignmentEpoch, expectedTargetAssignmentEpoch);
+ return new CoordinatorResult<>(List.of());
+ }
+
+ if (exception != null) {
+ log.error("[GroupId {}] Failed to compute a new target assignment for epoch {}: {}.",
+ groupId, targetAssignmentEpoch, exception.getMessage(), exception);
+ return new CoordinatorResult<>(List.of());
Review Comment:
Filed as https://issues.apache.org/jira/browse/KAFKA-21428 and https://issues.apache.org/jira/browse/KAFKA-21429.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4239527734
##########
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:
That is what it does. I don't follow what you mean. The real coordinator executor does not have ordering guarantees when there are multiple executor threads.
The test that uses it is `testConsumerGroupAsyncTargetAssignmentCancelledOnRollbackAfterRegexResolution` and it wants a specific order where regex resolution completes and bumps the epoch before the async assignor.
##########
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:
It comes from https://github.com/apache/kafka/pull/23702. The original code allowed the earlier consumers to bump the latch multiple times while the newer consumers wait behind the assignment batching or offload delay.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub) via jira" <ji...@kafka.apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4472110626
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -31222,6 +31222,908 @@ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
);
}
+ @Test
+ public void testConsumerGroupAsyncTargetAssignment() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ // Member 1 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result1 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result1.response()
+ );
+
+ ConsumerGroupMember expectedMember1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember1),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember1)
+ ),
+ result1.records()
+ );
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 2 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result2 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId2)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result2.response()
+ );
+
+ ConsumerGroupMember expectedMember2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember2),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 3, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember2)
+ ),
+ result2.records()
+ );
+
+ // Assignor completes on executor.
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId1, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ ))
+ )));
+
+ // The target assignment is written.
+ List<MockCoordinatorExecutor.ExecutorResult<CoordinatorRecord>> tasks = context.processTasks();
+ assertEquals(1, tasks.size());
+ assertEquals(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId), tasks.get(0).key());
+ assertUnorderedRecordsEquals(
+ List.of(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ )),
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId2, Map.of())
+ ),
+ List.of(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentMetadataRecord(groupId, 2, context.time.milliseconds()))
+ ),
+ tasks.get(0).result().records()
+ );
+
+ // The next assignment is not started automatically.
+ assertFalse(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 1 heartbeats again and receives the first assignment.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result3 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(1));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(2)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of(
+ new ConsumerGroupHeartbeatResponseData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(List.of(0, 1, 2, 3, 4, 5))
+ ))),
+ result3.response()
+ );
+
+ ConsumerGroupMember expectedMember3 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(2)
+ .setPreviousMemberEpoch(1)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 2))
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember3)
+ ),
+ result3.records()
+ );
+
+ // The next assignor run is scheduled.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentDiscardedOnRollback() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ long groupMetadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)));
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 10))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withAssignment(memberId1, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(groupMetadataHash))
+ .build();
+
+ // Member 2 joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Simulate a failed write to the log.
+ context.rollback();
Review Comment:
Filed as https://issues.apache.org/jira/browse/KAFKA-21312. Exploring the implementation at https://github.com/apache/kafka/pull/23923.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4239535349
##########
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:
We can do it defensively. Right now it should be impossible to not be the correct one since we cancel on every epoch bump after a rollback.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4239527734
##########
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:
That is what it does. I don't follow the concern. The real coordinator executor does not have ordering guarantees when there are multiple executor threads.
The test that uses it is `testConsumerGroupAsyncTargetAssignmentCancelledOnRollbackAfterRegexResolution` and it wants a specific order where regex resolution completes and bumps the epoch before the async assignor.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub) via jira" <ji...@kafka.apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4472110626
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -31222,6 +31222,908 @@ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
);
}
+ @Test
+ public void testConsumerGroupAsyncTargetAssignment() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .build();
+
+ // Member 1 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result1 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result1.response()
+ );
+
+ ConsumerGroupMember expectedMember1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember1),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 2, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember1)
+ ),
+ result1.records()
+ );
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 2 joins the group and gets no assignment yet.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result2 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId2)
+ .setMemberEpoch(1)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of())),
+ result2.response()
+ );
+
+ ConsumerGroupMember expectedMember2 = new ConsumerGroupMember.Builder(memberId2)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(1)
+ .setPreviousMemberEpoch(0)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(Map.of())
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember2),
+ GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 3, computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ))),
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember2)
+ ),
+ result2.records()
+ );
+
+ // Assignor completes on executor.
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId1, new MemberAssignmentImpl(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ ))
+ )));
+
+ // The target assignment is written.
+ List<MockCoordinatorExecutor.ExecutorResult<CoordinatorRecord>> tasks = context.processTasks();
+ assertEquals(1, tasks.size());
+ assertEquals(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId), tasks.get(0).key());
+ assertUnorderedRecordsEquals(
+ List.of(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId1, mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)
+ )),
+ GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId2, Map.of())
+ ),
+ List.of(GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentMetadataRecord(groupId, 2, context.time.milliseconds()))
+ ),
+ tasks.get(0).result().records()
+ );
+
+ // The next assignment is not started automatically.
+ assertFalse(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Member 1 heartbeats again and receives the first assignment.
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord> result3 = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId1)
+ .setMemberEpoch(1));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(memberId1)
+ .setMemberEpoch(2)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of(
+ new ConsumerGroupHeartbeatResponseData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(List.of(0, 1, 2, 3, 4, 5))
+ ))),
+ result3.response()
+ );
+
+ ConsumerGroupMember expectedMember3 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(2)
+ .setPreviousMemberEpoch(1)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 2))
+ .build();
+
+ assertRecordsEquals(
+ List.of(
+ GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember3)
+ ),
+ result3.records()
+ );
+
+ // The next assignor run is scheduled.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+ }
+
+ @Test
+ public void testConsumerGroupAsyncTargetAssignmentDiscardedOnRollback() {
+ String groupId = "fooup";
+ String memberId1 = Uuid.randomUuid().toString();
+ String memberId2 = Uuid.randomUuid().toString();
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 6)
+ .buildCoordinatorMetadataImage();
+
+ long groupMetadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)));
+
+ ConsumerGroupMember member1 = new ConsumerGroupMember.Builder(memberId1)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setClientId(DEFAULT_CLIENT_ID)
+ .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)), 10))
+ .build();
+
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNOR_OFFLOAD_ENABLE_CONFIG, true)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG, List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(member1)
+ .withAssignment(memberId1, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1, 2, 3, 4, 5)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(groupMetadataHash))
+ .build();
+
+ // Member 2 joins the group and gets no assignment yet.
+ context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId2)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setServerAssignor("range")
+ .setSubscribedTopicNames(List.of(fooTopicName))
+ .setTopicPartitions(List.of()));
+
+ // Assignor is scheduled on executor.
+ assertTrue(context.executor.isScheduled(GroupMetadataManager.groupTargetAssignmentUpdateKey(groupId)));
+
+ // Simulate a failed write to the log.
+ context.rollback();
Review Comment:
I like this idea since it would eliminate most of the tests
Filed as https://issues.apache.org/jira/browse/KAFKA-21312. Exploring the implementation at https://github.com/apache/kafka/pull/23923.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4239531500
##########
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:
Comes from https://github.com/apache/kafka/pull/23702. The original code allowed the earlier consumers to bump the latch multiple times while the newer consumers wait behind the assignment batching or offload delay.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4240161341
##########
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:
Hm, it's very tempting to move it there, but it runs counter to the current design? We don't start or cancel timers or tasks in the replay path and `removeGroup` is part of the replay path.
##########
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:
Hm, it's very tempting to move it there, but I think it runs counter to the current design? We don't start or cancel timers or tasks in the replay path and `removeGroup` is part of the replay path.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
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. 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on PR #23688:
URL: https://github.com/apache/kafka/pull/23688#issuecomment-5984879752
Also includes a test fix from #23702. Previously the implementation of that test method didn't matter. With offloading, it is more likely to miscount the number of started consumers.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4240428055
##########
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:
Updated the javadoc
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub) via jira" <ji...@kafka.apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4471519246
##########
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:
> I also wonder if we want to have more timing details about the offloading. For instance, the time spent in the queue, the time it took the compute the assignment, the time spent in the queue before installing it, etc. What do you think?
This will need some refactoring since we already calculate those metrics in the `ExecutorService` layer.
Filed as https://issues.apache.org/jira/browse/KAFKA-21427
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4239527734
##########
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:
That is what it does. I don't follow what you mean. The real coordinator executor does not have ordering guarantees when there are multiple executor threads.
The test that uses it is `testConsumerGroupAsyncTargetAssignmentCancelledOnRollbackAfterRegexResolution` and it wants a specific order where regex resolution completes before the async assignor.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4240352671
##########
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:
Most of them are indeed not that useful. We can be more selective and ditch the tests whose stated purpose is testing validation for example.
The remaining tests kept would be
* `testConsumerGroupHeartbeatIsAccessibleWhenNewGroupCoordinatorIsEnabled` - It's the happy path test
* `testConsumerGroupHeartbeatWithRegularExpression` - Regexes interact with assignments
* `testRejoiningStaticMemberGetsAssignmentsBackWhenNewGroupCoordinatorIsEnabled` - Borderline choice
* `testConsumerGroupHeartbeatRebalance` - Assignments with multiple members
Tests not kept would be:
* `testConsumerGroupHeartbeatIsInaccessibleWhenDisabledByStaticConfig`
* `testConsumerGroupHeartbeatIsInaccessibleWhenFeatureFlagNotEnabled`
* `testConsumerGroupHeartbeatWithInvalidRegularExpression`
* `testEmptyConsumerGroupId`
* `testConsumerGroupHeartbeatWithEmptySubscription` - Borderline choice
* `testStaticMemberRemovedAfterSessionTimeoutExpiryWhenNewGroupCoordinatorIsEnabled`
* `testUpdateConsumerGroupHeartbeatConfigSuccessful`
* `testConsumerGroupHeartbeatFailureIfMemberIdMissingForVersionsAbove0`
* `testMemberIdGeneratedOnServerWhenApiVersionIs0`
* `testFencedMemberCanRejoinWithEpochZero`
* `testDuplicateFullHeartbeatInStableState`
* `testDuplicateFullHeartbeatWhileWaitingForPartitions`
* `testDuplicateFullHeartbeatDuringRevocation`
* `testDuplicateFullHeartbeatWithRevocationAck`
* `testStaticMembersRejoinWithDifferentServerAssignor`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4240428745
##########
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:
Updated to be more selective about which heartbeat tests run with batching and offloading.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23688:
URL: https://github.com/apache/kafka/pull/23688#discussion_r4240161341
##########
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:
Hm, it's very tempting to move it there, but I think it runs counter to the current coordinator design? We don't start or cancel timers or tasks in the replay path and `removeGroup` is part of the replay path.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20292 [14/N]: Implement assignor offloading for consumer groups [kafka]
Posted by "squah-confluent (via GitHub) via jira" <ji...@kafka.apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org