This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 6fd96a3d391 KAFKA-20665: Bump group epoch for warm-up refinement steps
(#22748)
6fd96a3d391 is described below
commit 6fd96a3d3916b69eb1c7e0b488fc8882c592af1a
Author: Matthias J. Sax <[email protected]>
AuthorDate: Tue Jul 28 16:17:39 2026 -0700
KAFKA-20665: Bump group epoch for warm-up refinement steps (#22748)
Replace the streams-heartbeat bumpGroupEpoch boolean with an
AssignmentUpdate enum (NONE/RECOMPUTE/REFINE). RECOMPUTE keeps today's
behavior (re-run the assignor for a new final target). REFINE only
advances the assignment epoch of the unchanged final target via the
target-assignment metadata record, without re-running the assignor, so
members re-reconcile toward the next intermediate (warm-up) step.
A REFINE bump is triggered by hasHotWarmupTask(): a caught-up warm-up
(lag within acceptable.recovery.lag). Each refinement step needs its own
assignment epoch because members only pick up a changed assignment when
the epoch advances, and a new assignment epoch requires a group-epoch
bump.
Reviewers: Alieh Saeedi <[email protected]>, Sean Quah
<[email protected]>, David Jacot <[email protected]>
---
.../coordinator/group/GroupMetadataManager.java | 97 +++++++++++++++++-----
.../coordinator/group/streams/TasksTuple.java | 26 ++++++
.../group/GroupMetadataManagerTest.java | 77 +++++++++++++++++
.../coordinator/group/streams/TasksTupleTest.java | 47 +++++++++++
4 files changed, 226 insertions(+), 21 deletions(-)
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index eaba792fdac..e801efc8722 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -2142,8 +2142,12 @@ public class GroupMetadataManager {
StreamsGroupMember updatedMember = updatedMemberBuilder.build();
// If the member is new or has changed, a
StreamsGroupMemberMetadataValue record is written to the __consumer_offsets
partition
- // to persist the change, and bump the group epoch later.
- boolean bumpGroupEpoch = hasStreamsMemberMetadataChanged(groupId,
instanceId, member, updatedMember, records);
+ // to persist the change, and bump the group epoch later. We track
whether (and why) the group epoch is bumped:
+ // RECOMPUTE re-runs the assignor to produce a new final target
assignment; REFINE only advances the assignment epoch to move
+ // the in-memory refinement (warm-up promotion) forward without
recomputing the final target.
+ AssignmentUpdate assignmentUpdate =
hasStreamsMemberMetadataChanged(groupId, instanceId, member, updatedMember,
records)
+ ? AssignmentUpdate.RECOMPUTE
+ : AssignmentUpdate.NONE;
// 2. Initialize/Update the group topology.
// If the topology is new or has changed, a StreamsGroupTopologyValue
record is written to the __consumer_offsets partition to persist
@@ -2166,7 +2170,7 @@ public class GroupMetadataManager {
if (metadataHash != group.metadataHash()) {
log.info("[GroupId {}][MemberId {}] Computed new metadata
hash: {}.",
groupId, memberId, metadataHash);
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
reconfigureTopology = true;
}
@@ -2204,21 +2208,37 @@ public class GroupMetadataManager {
// We validated a topology that was not validated before, so bump the
group epoch as we may have to reassign tasks.
if (validatedTopologyEpoch != group.validatedTopologyEpoch()) {
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
}
// Check if assignment configurations have changed
Map<String, String> currentAssignmentConfigs =
streamsGroupAssignmentConfigs(groupId);
Map<String, String> storedAssignmentConfigs =
group.lastAssignmentConfigs();
- if (!bumpGroupEpoch &&
!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
+ if (assignmentUpdate == AssignmentUpdate.NONE &&
!currentAssignmentConfigs.equals(storedAssignmentConfigs)) {
log.info("[GroupId {}][MemberId {}] Assignment configurations
changed to {}. Triggering rebalance.",
groupId, memberId, currentAssignmentConfigs);
- bumpGroupEpoch = true;
+ assignmentUpdate = AssignmentUpdate.RECOMPUTE;
+ }
+
+ TasksTuple refinedAssignment = null;
+ if (assignmentUpdate == AssignmentUpdate.NONE && group.state() ==
StreamsGroup.StreamsGroupState.STABLE) {
+ // We are not computing a new target assignment later, thus we try
to refine the current target assignment
+ // into an intermediate assignment (with warm-up tasks) the member
should be reconciled towards.
+ refinedAssignment = maybeRefineAssignment(// no-op for now
+ updatedMember,
+ group.targetAssignment(),
+ group.taskOffsets(),
+ streamsGroupNumWarmupReplicas(groupId),
+ streamsGroupAcceptableRecoveryLag(groupId)
+ );
+ if (!refinedAssignment.sameTasks(updatedMember.assignedTasks())) {
+ assignmentUpdate = AssignmentUpdate.REFINED;
+ }
}
// Actually bump the group epoch
int groupEpoch = group.groupEpoch();
- if (bumpGroupEpoch) {
+ if (assignmentUpdate != AssignmentUpdate.NONE) {
groupEpoch += 1;
records.add(newStreamsGroupMetadataRecord(
groupId,
@@ -2259,18 +2279,22 @@ public class GroupMetadataManager {
metadataImage,
records,
Optional.of(returnedStatus),
- currentAssignmentConfigs
+ currentAssignmentConfigs,
+ assignmentUpdate == AssignmentUpdate.REFINED
);
- // 4b. Refine the target assignment into the intermediate assignment
(with warm-up tasks) the member should be
- // reconciled toward. Runs on every heartbeat, before reconciliation.
No-op for now.
- TasksTuple refinedTarget = refine(
- updatedMember,
- updateTargetAssignmentResult.targetAssignment,
- group.taskOffsets(),
- streamsGroupNumWarmupReplicas(group.groupId()),
- streamsGroupAcceptableRecoveryLag(group.groupId())
- );
+ // 4b. If we did not already refine above -- ie, we computed a new
target assignment, or the group is not
+ // yet reconciled (a rebalance is still in progress) -- refine the
target assignment into an intermediate
+ // assignment (with warm-up tasks) the member should be reconciled
towards.
+ if (refinedAssignment == null) {
+ refinedAssignment = maybeRefineAssignment(// no-op for now
+ updatedMember,
+ updateTargetAssignmentResult.targetAssignment,
+ group.taskOffsets(),
+ streamsGroupNumWarmupReplicas(group.groupId()),
+ streamsGroupAcceptableRecoveryLag(group.groupId())
+ );
+ }
// 5. Reconcile the member's assignment with the (refined) target
assignment if the member is not
// fully reconciled yet.
@@ -2281,7 +2305,7 @@ public class GroupMetadataManager {
group::currentStandbyTaskProcessIds,
group::currentWarmupTaskProcessIds,
updateTargetAssignmentResult.targetAssignmentEpoch(),
- refinedTarget,
+ refinedAssignment,
ownedActiveTasks,
ownedStandbyTasks,
ownedWarmupTasks,
@@ -4336,6 +4360,22 @@ public class GroupMetadataManager {
}
}
+ /**
+ * How a streams-group heartbeat updates the group epoch and target
assignment.
+ * <ul>
+ * <li>{@code NONE} - nothing changed; the group epoch is not
bumped.</li>
+ * <li>{@code RECOMPUTE} - the group changed (membership, topology or
configuration); bump the group epoch and re-run the
+ * assignor to produce a new final target assignment.</li>
+ * <li>{@code REFINE} - a warm-up task has caught up; bump the group
epoch to advance the in-memory refinement one step
+ * WITHOUT re-running the assignor (the final target assignment is
unchanged).</li>
+ * </ul>
+ */
+ private enum AssignmentUpdate {
+ NONE,
+ RECOMPUTE,
+ REFINED
+ }
+
/**
* Updates the target assignment according to the updated member and
metadata image.
*
@@ -4345,6 +4385,8 @@ public class GroupMetadataManager {
* @param metadataImage The metadata image.
* @param records The list to accumulate any new records.
* @param returnedStatus A mutable collection of status to be
returned in the response.
+ * @param refineOnly If true, only advance the assignment epoch
of the unchanged final target (warm-up refinement step)
+ * instead of re-running the assignor.
* @return The target assignment epoch and the full per-member target
assignment.
*/
private UpdateTargetAssignmentResult<Map<String, TasksTuple>>
maybeUpdateStreamsTargetAssignment(
@@ -4355,7 +4397,8 @@ public class GroupMetadataManager {
CoordinatorMetadataImage metadataImage,
List<CoordinatorRecord> records,
Optional<List<Status>> returnedStatus,
- Map<String, String> assignmentConfigs
+ Map<String, String> assignmentConfigs,
+ boolean refineOnly
) {
boolean initialDelayActive =
timer.isScheduled(streamsInitialRebalanceKey(group.groupId()));
if (initialDelayActive) {
@@ -4387,6 +4430,17 @@ public class GroupMetadataManager {
return new UpdateTargetAssignmentResult<>(group.assignmentEpoch(),
updatedMembersAndTargetAssignment.targetAssignment());
}
+ // The second condition rules out taking the shortcut while an
assignment is still pending (deferred by the
+ // assignment interval, or offloaded): the caller only refines a
STABLE group, where both epochs are equal, but
+ // if that ever changes, closing the epoch gap here would drop the
pending assignor run instead of deferring to it.
+ if (refineOnly && group.assignmentEpoch() >= group.groupEpoch()) {
+ // A refinement step only advances the assignment epoch of the
unchanged final target so that members
+ // re-reconcile toward the next in-memory intermediate assignment.
The assignor is not run and the per-member target
+ // assignment records are left untouched; only the (small)
target-assignment metadata record carrying the epoch is written.
+
records.add(newStreamsGroupTargetAssignmentMetadataRecord(group.groupId(),
groupEpoch, group.assignmentTimestamp()));
+ return new UpdateTargetAssignmentResult<>(groupEpoch,
updatedMembersAndTargetAssignment.targetAssignment());
+ }
+
boolean canComputeNextTargetAssignment =
canComputeNextTargetAssignment(
group.assignmentTimestamp(),
streamsGroupAssignmentIntervalMs(group.groupId()),
@@ -4470,7 +4524,7 @@ public class GroupMetadataManager {
*
* @return The member's intermediate assignment tuple.
*/
- private static TasksTuple refine(
+ private static TasksTuple maybeRefineAssignment(
final StreamsGroupMember member,
final Map<String, TasksTuple> targetAssignment,
final Map<String, MemberTaskOffsets> taskOffsets,
@@ -4516,7 +4570,8 @@ public class GroupMetadataManager {
metadataImage,
records,
Optional.empty(),
- group.lastAssignmentConfigs()
+ group.lastAssignmentConfigs(),
+ false
);
return new CoordinatorResult<>(records, null);
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTuple.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTuple.java
index 5e2e6e04d70..c0f105a94d0 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTuple.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTuple.java
@@ -79,6 +79,32 @@ public record TasksTuple(Map<String, Set<Integer>>
activeTasks,
);
}
+ /**
+ * Compares this tuple's task sets to those of a {@link
TasksTupleWithEpochs}, ignoring the assignment epochs
+ * carried by the active tasks of {@code other}. Used to decide whether
the intermediate assignment produced by
+ * the refiner differs from the assignment a member is currently
reconciled to (an active task set is considered
+ * unchanged even if only its epochs would differ).
+ *
+ * @param other Another task tuple with epochs.
+ * @return true if the active, standby and warm-up task sets are equal
(active-task epochs are not compared).
+ */
+ public boolean sameTasks(TasksTupleWithEpochs other) {
+ if (!warmupTasks.equals(other.warmupTasks()) ||
!standbyTasks.equals(other.standbyTasks())) {
+ return false;
+ }
+ Map<String, Map<Integer, Integer>> otherActiveTasks =
other.activeTasksWithEpochs();
+ if (activeTasks.size() != otherActiveTasks.size()) {
+ return false;
+ }
+ for (Map.Entry<String, Set<Integer>> entry : activeTasks.entrySet()) {
+ Map<Integer, Integer> otherPartitions =
otherActiveTasks.get(entry.getKey());
+ if (otherPartitions == null ||
!entry.getValue().equals(otherPartitions.keySet())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
/**
* Creates a {{@link TasksTuple}} from a
* {{@link
org.apache.kafka.coordinator.group.generated.StreamsGroupTargetAssignmentMemberValue}}.
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 32c60e5384c..b5dfd129c16 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -123,6 +123,7 @@ import
org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMe
import
org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue.Endpoint;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataValue;
+import
org.apache.kafka.coordinator.group.generated.StreamsGroupTargetAssignmentMemberKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.coordinator.group.modern.Assignment;
@@ -18994,6 +18995,82 @@ public class GroupMetadataManagerTest {
assertEquals(MemberTaskOffsets.EMPTY, group.taskOffsets(memberId));
}
+ @Test
+ public void
testStreamsGroupHeartbeatRefinementStepBumpsEpochWithoutRunningAssignor() {
+ String groupId = "fooup";
+ String memberId = Uuid.randomUuid().toString();
+ String subtopology1 = "subtopology1";
+ String fooTopicName = "foo";
+ Uuid fooTopicId = Uuid.randomUuid();
+ Topology topology = new Topology().setSubtopologies(List.of(
+ new
Subtopology().setSubtopologyId(subtopology1).setSourceTopics(List.of(fooTopicName))
+ ));
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 4)
+ .buildCoordinatorMetadataImage();
+ long groupMetadataHash = computeGroupHash(Map.of(
+ fooTopicName, computeTopicHash(fooTopicName, metadataImage)
+ ));
+
+ // maybeRefineAssignment() still returns the target unchanged, so the
refinement predicate can only fire when a
+ // member's assigned tasks differ from its target. The setup below
forces that: assigned {0,1,2} vs target
+ // {0,1,2,3}, with all epochs equal so the group still reports STABLE.
Real reconciliation never leaves those two
+ // out of sync at the same epoch; a real refiner will make this case
reachable properly. Metadata, topology and
+ // configs all match, so refinement is the only possible reason to
bump the group epoch.
+ MockTaskAssignor assignor = new MockTaskAssignor("sticky");
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, 10)
+ .withMember(streamsGroupMemberBuilderWithDefaults(memberId)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+
.setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE, 10,
+ TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2)))
+ .build())
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topology))
+ .withTargetAssignment(memberId,
TaskAssignmentTestUtil.mkTasksTuple(TaskRole.ACTIVE,
+ TaskAssignmentTestUtil.mkTasks(subtopology1, 0, 1, 2, 3)))
+ .withTargetAssignmentEpoch(10)
+ .withTargetAssignmentTimestamp(12345L)
+ .withMetadataHash(groupMetadataHash)
+ .withValidatedTopologyEpoch(0)
+ .withLastAssignmentConfigs(Map.of("num.standby.replicas", "0"))
+ )
+ .build();
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord>
result = context.streamsGroupHeartbeat(
+ new StreamsGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(memberId)
+ .setMemberEpoch(10)
+ .setProcessId("process-id")
+ .setRebalanceTimeoutMs(1500)
+ .setActiveTasks(List.of(new
StreamsGroupHeartbeatRequestData.TaskIds()
+ .setSubtopologyId(subtopology1)
+ .setPartitions(List.of(0, 1, 2))))
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of()));
+
+ // The refinement step bumped the group/assignment epoch to 11; the
member reconciles toward it.
+ assertEquals(11, result.response().data().memberEpoch());
+ assertTrue(result.records().stream()
+ .filter(r -> r.key() instanceof StreamsGroupMetadataKey)
+ .map(r -> (StreamsGroupMetadataValue) r.value().message())
+ .anyMatch(v -> v.epoch() == 11));
+
+ // A refinement step only advances the assignment epoch of the
unchanged final target. The target-assignment
+ // metadata record carries the bumped epoch and PRESERVES the
assignment timestamp (12345L) rather than
+ // resetting it to the current time, so the assignment interval is not
restarted.
+ assertTrue(result.records().contains(
+
StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataRecord(groupId,
11, 12345L)));
+
+ // The assignor was not run: the final target is unchanged, so no
per-member target-assignment record is written.
+ assertTrue(result.records().stream()
+ .noneMatch(r -> r.key() instanceof
StreamsGroupTargetAssignmentMemberKey));
+ }
+
@Test
public void testStreamsGroupHeartbeatStoresTaskOffsetsWithoutPersisting() {
String groupId = "fooup";
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TasksTupleTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TasksTupleTest.java
index 94eee2484e3..4d4ed6a3542 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TasksTupleTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TasksTupleTest.java
@@ -166,6 +166,53 @@ public class TasksTupleTest {
assertTrue(tuple1.containsAny(tuple5));
}
+ @Test
+ public void testSameTasks() {
+ TasksTuple tuple = new TasksTuple(
+ Map.of(SUBTOPOLOGY_1, Set.of(1, 2, 3)),
+ Map.of(SUBTOPOLOGY_2, Set.of(4, 5, 6)),
+ Map.of(SUBTOPOLOGY_3, Set.of(7, 8, 9))
+ );
+
+ // Same task sets, but the active tasks carry (different) epochs ->
still the same.
+ assertTrue(tuple.sameTasks(new TasksTupleWithEpochs(
+ Map.of(SUBTOPOLOGY_1, Map.of(1, 5, 2, 7, 3, 9)),
+ Map.of(SUBTOPOLOGY_2, Set.of(4, 5, 6)),
+ Map.of(SUBTOPOLOGY_3, Set.of(7, 8, 9))
+ )));
+
+ // Different active task set -> not the same.
+ assertFalse(tuple.sameTasks(new TasksTupleWithEpochs(
+ Map.of(SUBTOPOLOGY_1, Map.of(1, 5, 2, 7)),
+ Map.of(SUBTOPOLOGY_2, Set.of(4, 5, 6)),
+ Map.of(SUBTOPOLOGY_3, Set.of(7, 8, 9))
+ )));
+
+ // Extra active subtopology (differing key set) -> not the same.
+ assertFalse(tuple.sameTasks(new TasksTupleWithEpochs(
+ Map.of(SUBTOPOLOGY_1, Map.of(1, 5, 2, 7, 3, 9), SUBTOPOLOGY_2,
Map.of(1, 1)),
+ Map.of(SUBTOPOLOGY_2, Set.of(4, 5, 6)),
+ Map.of(SUBTOPOLOGY_3, Set.of(7, 8, 9))
+ )));
+
+ // Different standby task set -> not the same.
+ assertFalse(tuple.sameTasks(new TasksTupleWithEpochs(
+ Map.of(SUBTOPOLOGY_1, Map.of(1, 5, 2, 7, 3, 9)),
+ Map.of(SUBTOPOLOGY_2, Set.of(4, 5)),
+ Map.of(SUBTOPOLOGY_3, Set.of(7, 8, 9))
+ )));
+
+ // Different warm-up task set -> not the same.
+ assertFalse(tuple.sameTasks(new TasksTupleWithEpochs(
+ Map.of(SUBTOPOLOGY_1, Map.of(1, 5, 2, 7, 3, 9)),
+ Map.of(SUBTOPOLOGY_2, Set.of(4, 5, 6)),
+ Map.of(SUBTOPOLOGY_3, Set.of(7, 8))
+ )));
+
+ // Two empty tuples are the same.
+ assertTrue(new TasksTuple(Map.of(), Map.of(),
Map.of()).sameTasks(TasksTupleWithEpochs.EMPTY));
+ }
+
@Test
public void testIsEmpty() {
TasksTuple emptyTuple = new TasksTuple(Map.of(), Map.of(), Map.of());
