You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Gavin Wang (Jira)" <ji...@apache.org> on 2026/10/20 20:49:00 UTC
[jira] [Assigned] (KAFKA-21321) Streams group: the assignment refiner does not see a member that joins in the same heartbeat
[ https://issues.apache.org/jira/browse/KAFKA-21321?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Gavin Wang reassigned KAFKA-21321:
----------------------------------
Assignee: Gavin Wang
> Streams group: the assignment refiner does not see a member that joins in the same heartbeat
> --------------------------------------------------------------------------------------------
>
> Key: KAFKA-21321
> URL: https://issues.apache.org/jira/browse/KAFKA-21321
> Project: Kafka
> Issue Type: Bug
> Components: group-coordinator, streams
> Affects Versions: 4.5.0
> Reporter: Gavin Wang
> Assignee: Gavin Wang
> Priority: Major
>
> h3. Problem
> When a streams group heartbeat computes a new target assignment, \{{GroupMetadataManager#refineAssignment}} passes \{{group.members()}} to the refiner ([GroupMetadataManager:4707|https://github.com/apache/kafka/blob/f568b7b2c8802e6635b4d8df93ede88e55a2b84d/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java#L4707]). That map does not contain the member sending the heartbeat yet, because the heartbeat's records are replayed only after the handler returns. The target assignment the refiner gets, however, comes from \{{UpdatedMembersAndTargetAssignmentView}}, which already adds that member, or relabels a replaced static member to its new member ID ([:4583-4595|https://github.com/apache/kafka/blob/f568b7b2c8802e6635b4d8df93ede88e55a2b84d/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java#L4583-L4595]).
> So the refiner treats a target owner that joins in this heartbeat as a member that has left the group, and its result is cached as the epoch's intermediate assignment.
> h3. Effects
> With warm-up tasks enabled (the default), for stateful tasks:
> # *A static member that rejoins under a new member ID gets none of its tasks* when the same heartbeat computes a new target, for example because it rejoins from a new process. The refiner keeps its active tasks on the old member ID, which the replacement removes. The member is told nothing, and no member runs those tasks until the next refinement step.
> # *A new member gets no warm-up task.* A stateful task the target moves to it stays on its old owner without a warm-up, even when warm-up slots are free, until the next refinement step.
> # *A task can run \{{num.standby.replicas + 1}} standbys.* If the new member is the target standby holder of a task whose migration borrows an existing standby, its standby is not held back ([AssignmentRefinerImpl:866-872|https://github.com/apache/kafka/blob/f568b7b2c8802e6635b4d8df93ede88e55a2b84d/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java#L866-L872], [:818-823|https://github.com/apache/kafka/blob/f568b7b2c8802e6635b4d8df93ede88e55a2b84d/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java#L818-L823]). Once delivered it is kept, so the extra standby lasts until the migration completes.
> The refiner's branches for a target owner missing from \{{members}} ([AssignmentRefinerImpl:286-293|https://github.com/apache/kafka/blob/f568b7b2c8802e6635b4d8df93ede88e55a2b84d/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefinerImpl.java#L286-L293]) attribute that case to assignment offloading and member fencing. Neither produces it on trunk: streams assignor offloading is not wired up, and fencing removes the member's target assignment together with the member. So today these branches fire only for a member that joins in the same heartbeat.
> h3. Possible fix
> Pass the view's members to the refiner, together with the view's target assignment.
> h3. Reproduction
> The test below fails on trunk (\{{46dc4b51f7}}) with \{{expected: <[TaskIds(subtopologyId='s1', partitions=[0])]> but was: <[]>}}. Place it in \{{group-coordinator/src/test/java/org/apache/kafka/coordinator/group/}}. Effects 2 and 3 reproduce the same way, with a joining dynamic member.
> {code:java}
> /*
> * Licensed to the Apache Software Foundation (ASF) under one or more
> * contributor license agreements. See the NOTICE file distributed with
> * this work for additional information regarding copyright ownership.
> * The ASF licenses this file to You under the Apache License, Version 2.0
> * (the "License"); you may not use this file except in compliance with
> * the License. You may obtain a copy of the License at
> *
> * http://www.apache.org/licenses/LICENSE-2.0
> *
> * Unless required by applicable law or agreed to in writing, software
> * distributed under the License is distributed on an "AS IS" BASIS,
> * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> * See the License for the specific language governing permissions and
> * limitations under the License.
> */
> package org.apache.kafka.coordinator.group;
> import org.apache.kafka.common.Uuid;
> import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
> import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.Subtopology;
> import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.TopicInfo;
> import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.Topology;
> import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
> import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
> import org.apache.kafka.coordinator.common.runtime.CoordinatorResult;
> import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder;
> import org.apache.kafka.coordinator.group.streams.AssignmentRefinerImpl;
> import org.apache.kafka.coordinator.group.streams.MockTaskAssignor;
> import org.apache.kafka.coordinator.group.streams.StreamsGroupBuilder;
> import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
> import org.apache.kafka.coordinator.group.streams.StreamsTopology;
> import org.apache.kafka.coordinator.group.streams.TaskRole;
> import org.apache.kafka.coordinator.group.streams.assignor.AssignmentConfigsImpl;
> import org.junit.jupiter.api.Test;
> import java.util.List;
> import java.util.Map;
> import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH;
> import static org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH;
> import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.mkResponseTasks;
> import static org.apache.kafka.coordinator.group.StreamsGroupTestUtil.streamsGroupMemberBuilderWithDefaults;
> import static org.apache.kafka.coordinator.group.Utils.computeGroupHash;
> import static org.apache.kafka.coordinator.group.Utils.computeTopicHash;
> import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTasks;
> import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTasksTuple;
> import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch;
> import static org.junit.jupiter.api.Assertions.assertEquals;
> class StreamsGroupRefinerJoiningMemberTest {
> @Test
> public void staticMemberRejoiningFromNewProcessGetsItsStatefulTaskBack() {
> Topology topology = new Topology().setSubtopologies(List.of(new Subtopology()
> .setSubtopologyId("s1")
> .setSourceTopics(List.of("in"))
> .setStateChangelogTopics(List.of(new TopicInfo().setName("store-changelog")))));
> CoordinatorMetadataImage image = new MetadataImageBuilder()
> .addTopic(Uuid.randomUuid(), "in", 1)
> .addTopic(Uuid.randomUuid(), "store-changelog", 1)
> .buildCoordinatorMetadataImage();
> // A static member ran stateful task s1_0 on process A and left with epoch -2. Warm-up tasks are enabled, the
> // default.
> MockTaskAssignor assignor = new MockTaskAssignor("sticky");
> assignor.prepareGroupAssignment(Map.of("member-new", mkTasksTuple(TaskRole.ACTIVE, mkTasks("s1", 0))));
> GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
> .withStreamsGroupTaskAssignors(List.of(assignor))
> .withStreamsGroupAssignmentRefiner(new AssignmentRefinerImpl())
> .withMetadataImage(image)
> .withStreamsGroup(new StreamsGroupBuilder("app", 10)
> .withMember(streamsGroupMemberBuilderWithDefaults("member-old", "instance-1")
> .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
> .setPreviousMemberEpoch(10)
> .setProcessId("process-A")
> .setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE, 10, mkTasks("s1", 0)))
> .build())
> .withTargetAssignment("member-old", mkTasksTuple(TaskRole.ACTIVE, mkTasks("s1", 0)))
> .withTargetAssignmentEpoch(10)
> .withTopology(StreamsTopology.fromHeartbeatRequest(topology))
> .withValidatedTopologyEpoch(0)
> .withMetadataHash(computeGroupHash(Map.of(
> "in", computeTopicHash("in", image),
> "store-changelog", computeTopicHash("store-changelog", image))))
> .withLastAssignmentConfigs(AssignmentConfigsImpl.DEFAULT))
> .build();
> // It rejoins under a new member ID from process B. The process change bumps the group epoch, so the target
> // assignment is computed and refined in this heartbeat.
> CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(
> new StreamsGroupHeartbeatRequestData()
> .setGroupId("app")
> .setMemberId("member-new")
> .setMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
> .setInstanceId("instance-1")
> .setProcessId("process-B")
> .setRebalanceTimeoutMs(1500)
> .setTopology(topology)
> .setActiveTasks(List.of())
> .setStandbyTasks(List.of())
> .setWarmupTasks(List.of()));
> // Fails on trunk: the refiner is given group.members(), which does not contain member-new yet, so it keeps
> // s1_0 on member-old, which the replacement removes. member-new is told nothing, and no member runs s1_0.
> assertEquals(mkResponseTasks("s1", 0), result.response().data().activeTasks());
> }
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)