You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ab...@apache.org on 2023/08/30 21:51:38 UTC
[kafka] branch 3.5 updated: HOTFIX: avoid placement of unnecessary transient standby tasks & improve assignor logging (#14149)
This is an automated email from the ASF dual-hosted git repository.
ableegoldman pushed a commit to branch 3.5
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.5 by this push:
new 1966f51d621 HOTFIX: avoid placement of unnecessary transient standby tasks & improve assignor logging (#14149)
1966f51d621 is described below
commit 1966f51d6212a1e4efa14003ad1c708b35d7b0db
Author: A. Sophie Blee-Goldman <ab...@gmail.com>
AuthorDate: Wed Aug 30 13:29:38 2023 -0700
HOTFIX: avoid placement of unnecessary transient standby tasks & improve assignor logging (#14149)
Minor fix to avoid creating unnecessary standby tasks, especially when these may be surprising or unexpected as in the case of an application with num.standby.replicas = 0 and warmup replicas disabled.
The "bug" here was introduced during the fix for an issue with cooperative rebalancing and in-memory stores. The fundamental problem is that in-memory stores cannot be unassigned from a consumer for any period, however temporary, without being closed and losing all the accumulated state. This caused some grief when the new HA task assignor would assign an active task to a node based on the readiness of the standby version of that task, but would have to remove the active task from the [...]
To fix this, we simply began to place standby tasks on the intended recipient of an active task awaiting revocation by another consumer. However, the fix was a bit of an overreach, as we assigned these temporary standby tasks in all cases, regardless of whether there had previously been a standby version of that task. We can narrow this down without sacrificing any of the intended functionality by only assigning this kind of standby task where the consumer had previously owned some ve [...]
Also breaks up some of the long log lines in the StreamsPartitionAssignor and expands the summary info while moving it all to the front of the line (following reports of missing info due to truncation of long log lines in larger applications)
---
.../internals/StreamsPartitionAssignor.java | 52 +++++++++++++--------
.../internals/assignment/ClientState.java | 6 ++-
.../internals/StreamsPartitionAssignorTest.java | 54 ++++++++++++++++++++++
3 files changed, 91 insertions(+), 21 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
index 54ab6e8fa0e..2da8f1e47c2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
@@ -618,18 +618,24 @@ public class StreamsPartitionAssignor implements ConsumerPartitionAssignor, Conf
final boolean lagComputationSuccessful =
populateClientStatesMap(clientStates, clientMetadataMap, taskForPartition, changelogTopics);
- log.info("{} members participating in this rebalance: \n{}.",
- clientStates.size(),
- clientStates.entrySet().stream()
- .sorted(comparingByKey())
- .map(entry -> entry.getKey() + ": " + entry.getValue().consumers())
- .collect(Collectors.joining(Utils.NL)));
+
+ log.info("{} client nodes and {} consumers participating in this rebalance: \n{}.",
+ clientStates.size(),
+ clientStates.values().stream().map(ClientState::capacity).reduce(Integer::sum),
+ clientStates.entrySet().stream()
+ .sorted(comparingByKey())
+ .map(entry -> entry.getKey() + ": " + entry.getValue().consumers())
+ .collect(Collectors.joining(Utils.NL)));
final Set<TaskId> allTasks = partitionsForTask.keySet();
statefulTasks.addAll(changelogTopics.statefulTaskIds());
- log.debug("Assigning tasks {} including stateful {} to clients {} with number of replicas {}",
- allTasks, statefulTasks, clientStates, numStandbyReplicas());
+ log.info("Assigning stateful tasks: {}\n"
+ + "and stateless tasks: {}",
+ statefulTasks,
+ allTasks.stream().filter(t -> !statefulTasks.contains(t)));
+ log.debug("Assigning tasks and {} standby replicas to client nodes {}",
+ numStandbyReplicas(), clientStates);
final TaskAssignor taskAssignor = createTaskAssignor(lagComputationSuccessful);
@@ -638,15 +644,17 @@ public class StreamsPartitionAssignor implements ConsumerPartitionAssignor, Conf
statefulTasks,
assignmentConfigs);
- log.info("{} assigned tasks {} including stateful {} to {} clients as: \n{}.",
- allTasks.size(),
- allTasks,
- statefulTasks,
- clientStates.size(),
- clientStates.entrySet().stream()
- .sorted(comparingByKey())
- .map(entry -> entry.getKey() + "=" + entry.getValue().currentAssignment())
- .collect(Collectors.joining(Utils.NL)));
+ // Break this up into multiple logs to make sure the summary info gets through, which helps avoid
+ // info loss for example due to long line truncation with large apps
+ log.info("Assigned {} total tasks including {} stateful tasks to {} client nodes.",
+ allTasks.size(),
+ statefulTasks.size(),
+ clientStates.size());
+ log.info("Assignment of tasks to nodes: {}",
+ clientStates.entrySet().stream()
+ .sorted(comparingByKey())
+ .map(entry -> entry.getKey() + "=" + entry.getValue().currentAssignment())
+ .collect(Collectors.joining(Utils.NL)));
return probingRebalanceNeeded;
}
@@ -1063,9 +1071,13 @@ public class StreamsPartitionAssignor implements ConsumerPartitionAssignor, Conf
}
for (final TaskId task : revokedTasks) {
- if (allStatefulTasks.contains(task)) {
- log.info("Adding removed stateful active task {} as a standby for {} before it is revoked in followup rebalance",
- task, consumer);
+ // If this task is stateful and already owned by the consumer, but can't (yet) be assigned as an active
+ // task during this rebalance as it must be revoked from another consumer first, place a temporary
+ // standby task here until it can receive the active task to avoid closing the state store (and losing
+ // all of the accumulated state in the case of in-memory stores)
+ if (clientState.previouslyOwnedStandby(task) && allStatefulTasks.contains(task)) {
+ log.info("Adding removed stateful active task {} as a standby for {} until it is revoked and can "
+ + "be transitioned to active in a followup rebalance", task, consumer);
// This has no effect on the assignment, as we'll never consult the ClientState again, but
// it does perform a useful assertion that the it's legal to assign this task as a standby to this instance
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java
index b8ba4ce27e1..33eb8d343ac 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java
@@ -95,7 +95,7 @@ public class ClientState {
this.clientTags = unmodifiableMap(clientTags);
}
- int capacity() {
+ public int capacity() {
return capacity;
}
@@ -228,6 +228,10 @@ public class ClientState {
);
}
+ public boolean previouslyOwnedStandby(final TaskId task) {
+ return previousStandbyTasks.taskIds().contains(task);
+ }
+
public int assignedTaskCount() {
return activeTaskCount() + standbyTaskCount();
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
index a03970811eb..5e5a067bd63 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
@@ -678,6 +678,60 @@ public class StreamsPartitionAssignorTest {
assertEquals(expectedInfo11TaskIds, info11.activeTasks());
}
+ @Test
+ public void shouldNotAssignTemporaryStandbyTask() {
+ builder.addSource(null, "source1", null, null, null, "topic1");
+
+ final List<PartitionInfo> localInfos = asList(
+ new PartitionInfo("topic1", 0, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic1", 1, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic1", 2, Node.noNode(), new Node[0], new Node[0]),
+ new PartitionInfo("topic1", 3, Node.noNode(), new Node[0], new Node[0])
+ );
+
+ final Cluster localMetadata = new Cluster(
+ "cluster",
+ Collections.singletonList(Node.noNode()),
+ localInfos,
+ emptySet(),
+ emptySet()
+ );
+
+ final List<String> topics = singletonList("topic1");
+
+ createMockTaskManager(mkSet(TASK_0_0, TASK_0_1, TASK_0_2, TASK_0_3), emptySet());
+ configureDefaultPartitionAssignor();
+
+ subscriptions.put("consumer10",
+ new Subscription(
+ topics,
+ getInfo(UUID_1, mkSet(TASK_0_0, TASK_0_2), emptySet()).encode(),
+ asList(t1p0, t1p2)
+ ));
+ subscriptions.put("consumer11",
+ new Subscription(
+ topics,
+ getInfo(UUID_1, mkSet(TASK_0_1, TASK_0_3), emptySet()).encode(),
+ asList(t1p1, t1p3)
+ ));
+ subscriptions.put("consumer20",
+ new Subscription(
+ topics,
+ getInfo(UUID_2, emptySet(), emptySet()).encode(),
+ emptyList()
+ ));
+
+ final Map<String, Assignment> assignments = partitionAssignor.assign(localMetadata, new GroupSubscription(subscriptions)).groupAssignment();
+
+ // neither active nor standby tasks should be assigned to consumer 3, which will have to wait until
+ // the followup cooperative rebalance to get the active task(s) it was assigned (and does not need
+ // a standby copy before that since it previously had no tasks at all)
+ final AssignmentInfo info20 = AssignmentInfo.decode(assignments.get("consumer20").userData());
+ assertTrue(info20.activeTasks().isEmpty());
+ assertTrue(info20.standbyTasks().isEmpty());
+
+ }
+
@Test
public void testAssignEmptyMetadata() {
builder.addSource(null, "source1", null, null, null, "topic1");