You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2026/03/11 16:20:58 UTC
(kafka) branch trunk updated: KAFKA-20269 [3/N]: Refactor assignment update for delayed streams assignments (#21696)
This is an automated email from the ASF dual-hosted git repository.
dajac 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 ff010a402d1 KAFKA-20269 [3/N]: Refactor assignment update for delayed streams assignments (#21696)
ff010a402d1 is described below
commit ff010a402d1148be06e00da4b1b3d4c95fdae698
Author: Sean Quah <sq...@confluent.io>
AuthorDate: Wed Mar 11 16:20:52 2026 +0000
KAFKA-20269 [3/N]: Refactor assignment update for delayed streams assignments (#21696)
Refactor the streams target assignment update method to return both the
target assignment epoch and target assignment. When assignment batching
or assignment offload are implemented, the target assignment update
method may return the last target assignment, depending on timings and
the group coordinator config.
Reviewers: Lucas Brutschy <lb...@confluent.io>, David Jacot
<dj...@confluent.io>
---
.../coordinator/group/GroupMetadataManager.java | 94 +++++++++++++---------
1 file changed, 58 insertions(+), 36 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 3d7ecdbfe2c..ffcd909a2b9 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
@@ -299,6 +299,23 @@ public class GroupMetadataManager {
group.targetAssignment(member.memberId())
);
}
+
+ private static UpdateTargetAssignmentResult<TasksTuple> fromLastTargetAssignment(
+ StreamsGroup group,
+ Optional<StreamsGroupMember> member
+ ) {
+ if (member.isPresent()) {
+ return new UpdateTargetAssignmentResult<>(
+ group.assignmentEpoch(),
+ group.targetAssignment(member.get().memberId())
+ );
+ } else {
+ return new UpdateTargetAssignmentResult<>(
+ group.assignmentEpoch(),
+ TasksTuple.EMPTY
+ );
+ }
+ }
}
public static class Builder {
@@ -2086,36 +2103,16 @@ public class GroupMetadataManager {
// 4. Update the target assignment if the group epoch is larger than the target assignment epoch or a static member
// replaces an existing static member.
// The delta between the existing and the new target assignment is persisted to the partition.
- int targetAssignmentEpoch;
- TasksTuple targetAssignment;
- if (groupEpoch > group.assignmentEpoch()) {
- boolean initialDelayActive = timer.isScheduled(streamsInitialRebalanceKey(groupId));
- if (initialDelayActive) {
- // During initial rebalance delay, return empty assignment to first joining members.
- targetAssignmentEpoch = Math.max(1, group.assignmentEpoch());
- targetAssignment = TasksTuple.EMPTY;
-
- returnedStatus.add(
- new Status()
- .setStatusCode(StreamsGroupHeartbeatResponse.Status.ASSIGNMENT_DELAYED.code())
- .setStatusDetail("Assignment delayed due to the configured initial rebalance delay.")
- );
- } else {
- targetAssignment = updateStreamsTargetAssignment(
- group,
- groupEpoch,
- Optional.of(updatedMember),
- updatedConfiguredTopology,
- metadataImage,
- records,
- currentAssignmentConfigs
- );
- targetAssignmentEpoch = groupEpoch;
- }
- } else {
- targetAssignmentEpoch = group.assignmentEpoch();
- targetAssignment = group.targetAssignment(updatedMember.memberId());
- }
+ UpdateTargetAssignmentResult<TasksTuple> updateTargetAssignmentResult = maybeUpdateStreamsTargetAssignment(
+ group,
+ groupEpoch,
+ Optional.of(updatedMember),
+ updatedConfiguredTopology,
+ metadataImage,
+ records,
+ Optional.of(returnedStatus),
+ currentAssignmentConfigs
+ );
// 5. Reconcile the member's assignment with the target assignment if the member is not
// fully reconciled yet.
@@ -2125,8 +2122,8 @@ public class GroupMetadataManager {
group::currentActiveTaskProcessId,
group::currentStandbyTaskProcessIds,
group::currentWarmupTaskProcessIds,
- targetAssignmentEpoch,
- targetAssignment,
+ updateTargetAssignmentResult.targetAssignmentEpoch(),
+ updateTargetAssignmentResult.targetAssignment(),
ownedActiveTasks,
ownedStandbyTasks,
ownedWarmupTasks,
@@ -3940,17 +3937,38 @@ public class GroupMetadataManager {
* @param updatedMember The updated member (optional).
* @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.
* @return The new target assignment for the updated member, or EMPTY if no member specified.
*/
- private TasksTuple updateStreamsTargetAssignment(
+ private UpdateTargetAssignmentResult<TasksTuple> maybeUpdateStreamsTargetAssignment(
StreamsGroup group,
int groupEpoch,
Optional<StreamsGroupMember> updatedMember,
ConfiguredTopology configuredTopology,
CoordinatorMetadataImage metadataImage,
List<CoordinatorRecord> records,
+ Optional<List<Status>> returnedStatus,
Map<String, String> assignmentConfigs
) {
+ boolean initialDelayActive = timer.isScheduled(streamsInitialRebalanceKey(group.groupId()));
+ if (initialDelayActive) {
+ returnedStatus.ifPresent(statusList -> statusList.add(
+ new Status()
+ .setStatusCode(StreamsGroupHeartbeatResponse.Status.ASSIGNMENT_DELAYED.code())
+ .setStatusDetail("Assignment delayed due to the configured initial rebalance delay.")
+ ));
+
+ return new UpdateTargetAssignmentResult<>(
+ group.assignmentEpoch(),
+ TasksTuple.EMPTY
+ );
+ }
+
+ if (group.assignmentEpoch() >= groupEpoch) {
+ // The assignment is up to date.
+ return UpdateTargetAssignmentResult.fromLastTargetAssignment(group, updatedMember);
+ }
+
TaskAssignor assignor = streamsGroupAssignor(group.groupId());
try {
org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder assignmentResultBuilder =
@@ -3986,8 +4004,11 @@ public class GroupMetadataManager {
records.addAll(assignmentResult.records());
- return updatedMember.map(member -> assignmentResult.targetAssignment().get(member.memberId()))
- .orElse(TasksTuple.EMPTY);
+ return new UpdateTargetAssignmentResult<>(
+ groupEpoch,
+ updatedMember.map(member -> assignmentResult.targetAssignment().get(member.memberId()))
+ .orElse(TasksTuple.EMPTY)
+ );
} catch (TaskAssignorException ex) {
String msg = String.format("Failed to compute a new target assignment for epoch %d: %s",
groupEpoch, ex.getMessage());
@@ -4024,13 +4045,14 @@ public class GroupMetadataManager {
}
List<CoordinatorRecord> records = new ArrayList<>();
- updateStreamsTargetAssignment(
+ maybeUpdateStreamsTargetAssignment(
group,
group.groupEpoch(),
Optional.empty(),
group.configuredTopology().get(),
metadataImage,
records,
+ Optional.empty(),
group.lastAssignmentConfigs()
);