Gavin Wang created KAFKA-21321:
----------------------------------
Summary: 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
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)
