Gavin Wang created KAFKA-21306:
----------------------------------
Summary: Streams group: a static member that rejoins from a new
process can put a task on that process twice
Key: KAFKA-21306
URL: https://issues.apache.org/jira/browse/KAFKA-21306
Project: Kafka
Issue Type: Bug
Components: streams
Reporter: Gavin Wang
Assignee: Gavin Wang
h3. Problem
In a streams group (KIP-1071), one process must never hold the same task twice.
CurrentAssignmentBuilder says "a single process must never run the same task
twice". The reconciler enforces this only when it grants a task: the
isUnreleased*Task checks run on tasks pending assignment, and a task the member
already holds is kept without any check.
A static member that rejoins under a new member ID bypasses that check:
# The rejoin is accepted once the old member has left with epoch -2.
GroupMetadataManager#getOrMaybeCreateStaticStreamsGroupMember copies the old
member, including its assigned tasks.
# The heartbeat then sets the process ID from the request. Nothing compares it
with the old process ID.
# If the rejoin comes from a different process, the copied tasks move to that
process. If another member there already holds one of them as a standby or
warm-up task, that process now holds the task twice.
The new target assignment that the process ID change triggers does not remove
the duplicate. The sticky assignor keeps the active task on the rejoined
member. The standby on the sibling member is revoked, but it stays until that
member acknowledges the revocation.
h3. When it happens
* Kafka Streams with static membership: each stream thread uses
group.instance.id "<id>-<threadIdx>". Member IDs are new on every restart, so a
restarted thread always takes the rejoin path above.
* The process ID changes across a restart when the topology has no persistent
stores (StateDirectory#initializeProcessId creates a random UUID on every
start), or when the state directory is lost or the instance moves to another
host.
* Sequence: an instance with two threads restarts under a new process ID B.
Thread 2 rejoins first and is given the standby of task t, which is allowed
because active t is still on the old process. Thread 1 rejoins and inherits
active t. Process B now holds t both as an active task and as a standby task.
h3. Impact
Both threads of the same instance are told to run t at the same time. This
lasts until the next target assignment revokes the standby and the second
thread acknowledges it: about one heartbeat interval, plus up to the assignment
interval. From reading the client code:
* With persistent stores, the task-directory lock makes the new active task
retry until the standby releases it, so t stalls.
* Without persistent stores, which is exactly the case where the process ID
changes on every restart, the lock is a no-op. Both copies run, which
duplicates memory and restore work.
h3. Reproduction
The test below fails on trunk (dfd3dcab07) at the final assertion: process-B is
both the active owner and a standby holder of s1_0.
{code:java}
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.MetadataImageBuilder;
import org.apache.kafka.coordinator.group.streams.StreamsGroup;
import org.apache.kafka.coordinator.group.streams.StreamsGroupBuilder;
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.apache.kafka.coordinator.group.streams.assignor.StickyTaskAssignor;
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;
import static org.junit.jupiter.api.Assertions.assertFalse;
class StreamsStaticRejoinProcessExclusivityTest {
@Test
public void staticRejoinFromNewProcessMustNotPutATaskOnAProcessTwice() {
// One Kafka Streams instance with two stream threads (static instance
IDs app-1 and app-2) ran on process A,
// with num.standby.replicas=1 and a single stateful task s1_0. Thread
app-1 ran the active task. Both threads
// left with epoch -2, and the instance restarted under a new process
ID B (no persistent stores, or a wiped
// state directory), so each thread rejoins under a new member ID.
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();
GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
.withStreamsGroupTaskAssignors(List.of(new StickyTaskAssignor()))
.withMetadataImage(image)
.withConfig(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_CONFIG, 1)
.withStreamsGroup(new StreamsGroupBuilder("app", 10)
.withMember(streamsGroupMemberBuilderWithDefaults("t1-old",
"app-1")
.setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
.setPreviousMemberEpoch(10)
.setProcessId("process-A")
.setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE, 0, mkTasks("s1",
0)))
.build())
.withMember(streamsGroupMemberBuilderWithDefaults("t2-old",
"app-2")
.setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
.setPreviousMemberEpoch(10)
.setProcessId("process-A")
.build())
.withTargetAssignment("t1-old", mkTasksTuple(TaskRole.ACTIVE,
mkTasks("s1", 0)))
.withTargetAssignment("t2-old", mkTasksTuple(TaskRole.ACTIVE))
.withTopology(StreamsTopology.fromHeartbeatRequest(topology))
.withTargetAssignmentEpoch(10)
.withMetadataHash(computeGroupHash(Map.of(
"in", computeTopicHash("in", image),
"store-changelog", computeTopicHash("store-changelog",
image))))
.withValidatedTopologyEpoch(0)
.withLastAssignmentConfigs(new AssignmentConfigsImpl(1,
List.of())))
.build();
StreamsGroup group = context.groupMetadataManager.streamsGroup("app");
// Thread app-2 rejoins first. The active task is still on process A,
so B may take the standby.
var thread2 = context.streamsGroupHeartbeat(rejoin("t2-new", "app-2",
"process-B", topology));
assertEquals(mkResponseTasks("s1", 0),
thread2.response().data().standbyTasks());
// Thread app-1 rejoins right after. It inherits its old member's
assignment, active s1_0, and moves it to
// process B, where thread app-2 already holds the standby of s1_0.
var thread1 = context.streamsGroupHeartbeat(rejoin("t1-new", "app-1",
"process-B", topology));
assertEquals(mkResponseTasks("s1", 0),
thread1.response().data().activeTasks());
// Fails today: process B now holds s1_0 both as an active task
(t1-new) and as a standby task (t2-new).
assertFalse(
"process-B".equals(group.currentActiveTaskProcessId("s1", 0))
&& group.currentStandbyTaskProcessIds("s1",
0).contains("process-B"),
"process-B holds task s1_0 as an active task and as a standby task
at the same time");
}
private static StreamsGroupHeartbeatRequestData rejoin(String memberId,
String instanceId, String processId,
Topology topology) {
return new StreamsGroupHeartbeatRequestData()
.setGroupId("app")
.setMemberId(memberId)
.setMemberEpoch(JOIN_GROUP_MEMBER_EPOCH)
.setInstanceId(instanceId)
.setProcessId(processId)
.setRebalanceTimeoutMs(1500)
.setTopology(topology)
.setActiveTasks(List.of())
.setStandbyTasks(List.of())
.setWarmupTasks(List.of());
}
}
{code}
h3. Possible fix
When a static member rejoins from a different process, do not keep copied tasks
that the new process already holds in another role. For example, the reconciler
could put those tasks back in pending assignment, so the existing exclusivity
checks apply to them.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
