[
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)
