[
https://issues.apache.org/jira/browse/KAFKA-21306?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Gavin Wang updated KAFKA-21306:
-------------------------------
Affects Version/s: 4.4.0
> 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
> Affects Versions: 4.4.0
> Reporter: Gavin Wang
> Assignee: Gavin Wang
> Priority: Critical
>
> 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)
