You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2026/09/18 22:56:50 UTC
[PR] KAFKA-20665: Wire-up refiner [kafka]
mjsax opened a new pull request, #23519:
URL: https://github.com/apache/kafka/pull/23519
This PR completes the first version of the refiner. In addition to the main wiring, this PR also fixes a bug in load handling, and improves how warm-up tasks are planet.
Part of KIP-1071.
--
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-20665: Wire-up refiner [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #23519:
URL: https://github.com/apache/kafka/pull/23519
--
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-20665: Wire-up refiner [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23519:
URL: https://github.com/apache/kafka/pull/23519#discussion_r4363656546
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -871,6 +990,15 @@ static boolean isCaughtUp(
return endOffset - offset <= acceptableRecoveryLag;
}
+ /**
+ * Whether the member has restored any of the task: it reports a position for it, and that position is not the
+ * cap the client reports before a restore has begun.
+ */
+ private static boolean hasRestoredSomething(final MemberTaskOffsets memberTaskOffsets, final TaskId task) {
+ final Long offset = offsetOf(memberTaskOffsets.taskOffsets(), task);
+ return offset != null && offset != Long.MAX_VALUE;
Review Comment:
https://github.com/apache/kafka/pull/23942
--
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-20665: Wire-up refiner [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #23519:
URL: https://github.com/apache/kafka/pull/23519#discussion_r4120294506
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -871,6 +990,15 @@ static boolean isCaughtUp(
return endOffset - offset <= acceptableRecoveryLag;
}
+ /**
+ * Whether the member has restored any of the task: it reports a position for it, and that position is not the
+ * cap the client reports before a restore has begun.
+ */
+ private static boolean hasRestoredSomething(final MemberTaskOffsets memberTaskOffsets, final TaskId task) {
+ final Long offset = offsetOf(memberTaskOffsets.taskOffsets(), task);
+ return offset != null && offset != Long.MAX_VALUE;
Review Comment:
Somewhat valid but minor. Will do in follow-ups
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -19556,6 +19558,208 @@ public void testStreamsGroupFallsBackToTargetAssignmentWhenRefinementLosesAnActi
assertEquals(Map.of(memberId, targetAssignment), group.refinedAssignment(group.assignmentEpoch()));
}
+ @Test
+ public void testStreamsGroupStagesAMigrationBehindAWarmupTask() {
+ String groupId = "fooup";
+ String memberA = Uuid.randomUuid().toString();
+ String memberB = Uuid.randomUuid().toString();
+ String subtopology1 = "subtopology1";
+ String fooTopicName = "foo";
+ String changelogTopicName = "changelog";
+ Uuid fooTopicId = Uuid.randomUuid();
+ // Only a task with a changelog can be warmed up, so the subtopology has to be stateful for the refiner to
+ // hold anything back.
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new Subtopology()
+ .setSubtopologyId(subtopology1)
+ .setSourceTopics(List.of(fooTopicName))
+ .setStateChangelogTopics(List.of(new TopicInfo().setName(changelogTopicName)))
+ ));
+
+ // The changelog has to exist, or the topology never becomes ready and the refiner is not consulted at all.
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 3)
+ .addTopic(Uuid.randomUuid(), changelogTopicName, 3)
+ .buildCoordinatorMetadataImage();
+ long metadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage),
+ changelogTopicName, computeTopicHash(changelogTopicName, metadataImage)
+ ));
+
+ // memberA runs all three tasks and the target assignment moves 0_2 to memberB, on a process that holds
+ // nothing of it -- the shape the refiner exists for.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withStreamsGroupAssignmentRefiner(new AssignmentRefinerImpl())
Review Comment:
No, we don't want to do this yet.
--
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-20665: Wire-up refiner [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #23519:
URL: https://github.com/apache/kafka/pull/23519#discussion_r4071647052
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -19556,6 +19558,208 @@ public void testStreamsGroupFallsBackToTargetAssignmentWhenRefinementLosesAnActi
assertEquals(Map.of(memberId, targetAssignment), group.refinedAssignment(group.assignmentEpoch()));
}
+ @Test
+ public void testStreamsGroupStagesAMigrationBehindAWarmupTask() {
+ String groupId = "fooup";
+ String memberA = Uuid.randomUuid().toString();
+ String memberB = Uuid.randomUuid().toString();
+ String subtopology1 = "subtopology1";
+ String fooTopicName = "foo";
+ String changelogTopicName = "changelog";
+ Uuid fooTopicId = Uuid.randomUuid();
+ // Only a task with a changelog can be warmed up, so the subtopology has to be stateful for the refiner to
+ // hold anything back.
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new Subtopology()
+ .setSubtopologyId(subtopology1)
+ .setSourceTopics(List.of(fooTopicName))
+ .setStateChangelogTopics(List.of(new TopicInfo().setName(changelogTopicName)))
+ ));
+
+ // The changelog has to exist, or the topology never becomes ready and the refiner is not consulted at all.
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 3)
+ .addTopic(Uuid.randomUuid(), changelogTopicName, 3)
+ .buildCoordinatorMetadataImage();
+ long metadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage),
+ changelogTopicName, computeTopicHash(changelogTopicName, metadataImage)
+ ));
+
+ // memberA runs all three tasks and the target assignment moves 0_2 to memberB, on a process that holds
+ // nothing of it -- the shape the refiner exists for.
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new MockTaskAssignor("sticky")))
+ .withStreamsGroupAssignmentRefiner(new AssignmentRefinerImpl())
Review Comment:
This explicit injection masks that normal brokers still instantiate `NoOpAssignmentRefiner`: `GroupCoordinatorConfig.java:443` keeps it as `STREAMS_GROUP_ASSIGNMENT_REFINER_CLASS_DEFAULT`, and `GroupCoordinatorConfigTest.java:1244-1248` still enforces that default. Consequently, the refiner completed here is never wired into the default runtime path, contrary to the PR’s stated purpose. Make `AssignmentRefinerImpl` the production default, update the config documentation/test, and cover the default builder path here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java:
##########
@@ -871,6 +990,15 @@ static boolean isCaughtUp(
return endOffset - offset <= acceptableRecoveryLag;
}
+ /**
+ * Whether the member has restored any of the task: it reports a position for it, and that position is not the
+ * cap the client reports before a restore has begun.
+ */
+ private static boolean hasRestoredSomething(final MemberTaskOffsets memberTaskOffsets, final TaskId task) {
+ final Long offset = offsetOf(memberTaskOffsets.taskOffsets(), task);
+ return offset != null && offset != Long.MAX_VALUE;
Review Comment:
`Long.MAX_VALUE` is not a “restore has not begun” marker. The client emits `0L` for an uninitialized changelog position (`ProcessorStateManager.java:459-463`), while offset-sum overflow is saturated to `Long.MAX_VALUE` (`StateDirectory.java:340-342`). Thus a newly initialized warm-up that restored no records is marked as started and protected from preemption, while a large restore whose sum overflowed is marked as not started and can be evicted. Tracking actual progress requires a dedicated marker or comparison with an observed starting position rather than interpreting this aggregate value.
--
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