You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2026/08/11 06:11:05 UTC
(kafka) branch trunk updated: MINOR: improve "streams" default rebalance listener (#22991)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 5087159021d MINOR: improve "streams" default rebalance listener (#22991)
5087159021d is described below
commit 5087159021dace51b87a94df85c2527099fcf986
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Mon Aug 10 23:10:57 2026 -0700
MINOR: improve "streams" default rebalance listener (#22991)
If the broker side assignor makes an error on task assignment, it could
crash the client with a "duplicate taskId" error.
This PR adds a proper task-id deduplication guard plus new WARN logs to
let the user know that it received some bad assignment.
Reviewers: Bill Bejeck <bb...@apache.org>
---
.../internals/DefaultStreamsRebalanceListener.java | 53 ++++++++++++++--
.../DefaultStreamsRebalanceListenerTest.java | 71 +++++++++++++++++++++-
2 files changed, 119 insertions(+), 5 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListener.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListener.java
index 1cdfcba6810..8237e60b66f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListener.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListener.java
@@ -28,8 +28,11 @@ import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.slf4j.Logger;
import java.util.Collection;
+import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -87,20 +90,26 @@ public class DefaultStreamsRebalanceListener implements StreamsRebalanceListener
@Override
public void onTasksAssigned(final StreamsRebalanceData.Assignment assignment) {
+ log.debug("Received broker assignment {}", assignment);
+
final long start = time.milliseconds();
+ final StreamsRebalanceData.Assignment runningAssignment = deduplicateTasks(assignment);
final Map<TaskId, Set<TopicPartition>> activeTasksWithPartitions =
- pairWithTopicPartitions(assignment.activeTasks().stream());
+ pairWithTopicPartitions(runningAssignment.activeTasks().stream());
final Map<TaskId, Set<TopicPartition>> standbyTasksWithPartitions =
- pairWithTopicPartitions(Stream.concat(assignment.standbyTasks().stream(), assignment.warmupTasks().stream()));
+ pairWithTopicPartitions(Stream.concat(
+ runningAssignment.standbyTasks().stream(),
+ runningAssignment.warmupTasks().stream()
+ ));
- log.info("Processing new assignment {} from Streams Rebalance Protocol", assignment);
+ log.info("Processing new assignment {} from Streams Rebalance Protocol", runningAssignment);
try {
streamThread.setStreamsGroupReady(assignment.isGroupReady());
taskManager.handleAssignment(activeTasksWithPartitions, standbyTasksWithPartitions);
streamThread.setState(StreamThread.State.PARTITIONS_ASSIGNED);
taskManager.handleRebalanceComplete();
- streamsRebalanceData.setReconciledAssignment(assignment);
+ streamsRebalanceData.setReconciledAssignment(runningAssignment);
} finally {
tasksAssignedSensor.record(time.milliseconds() - start);
}
@@ -117,6 +126,42 @@ public class DefaultStreamsRebalanceListener implements StreamsRebalanceListener
}
}
+ private StreamsRebalanceData.Assignment deduplicateTasks(final StreamsRebalanceData.Assignment assignment) {
+ // The overlaps below are only used to warn about them, so they are kept sorted to give the log a
+ // deterministic and readable task order.
+ final SortedSet<StreamsRebalanceData.TaskId> standbyAndWarmup = new TreeSet<>(assignment.standbyTasks());
+ standbyAndWarmup.retainAll(assignment.warmupTasks());
+ if (!standbyAndWarmup.isEmpty()) {
+ log.warn("Tasks {} were assigned as standby tasks and as warm-up tasks. A standby task and a warm-up " +
+ "task are the same thing on the client, so the two assignments are merged into a single task each.",
+ standbyAndWarmup);
+ }
+
+ final SortedSet<StreamsRebalanceData.TaskId> activeAndReplica = new TreeSet<>(assignment.standbyTasks());
+ activeAndReplica.addAll(assignment.warmupTasks());
+ activeAndReplica.retainAll(assignment.activeTasks());
+ if (!activeAndReplica.isEmpty()) {
+ log.warn("Tasks {} were assigned as active tasks and also as standby or warm-up tasks. The standby and " +
+ "warm-up assignment is ignored, and the tasks are run as active tasks.", activeAndReplica);
+ }
+
+ // The client runs one task per task id, so it reports one role for it. A task assigned as both standby and
+ // warm-up is reported as a standby: the standby is a permanent part of the assignment, while the warm-up is
+ // only injected on top of it and goes away again.
+ final Set<StreamsRebalanceData.TaskId> standbyTasks = new HashSet<>(assignment.standbyTasks());
+ standbyTasks.removeAll(assignment.activeTasks());
+ final Set<StreamsRebalanceData.TaskId> warmupTasks = new HashSet<>(assignment.warmupTasks());
+ warmupTasks.removeAll(assignment.activeTasks());
+ warmupTasks.removeAll(standbyTasks);
+
+ return new StreamsRebalanceData.Assignment(
+ assignment.activeTasks(),
+ standbyTasks,
+ warmupTasks,
+ assignment.isGroupReady()
+ );
+ }
+
private Map<TaskId, Set<TopicPartition>> pairWithTopicPartitions(final Stream<StreamsRebalanceData.TaskId> taskIdStream) {
return taskIdStream
.collect(Collectors.toMap(
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListenerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListenerTest.java
index 2e3e5cdfe7a..fe4db906ba4 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListenerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListenerTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.internals.StreamsRebalanceData;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.metrics.RebalanceListenerMetrics;
@@ -32,6 +33,7 @@ import org.mockito.InOrder;
import org.mockito.MockedStatic;
import org.slf4j.LoggerFactory;
+import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -40,6 +42,7 @@ import java.util.UUID;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
@@ -54,6 +57,9 @@ import static org.mockito.Mockito.when;
public class DefaultStreamsRebalanceListenerTest {
private static final String THREAD_ID = "test-thread-id";
+ private static final String MERGED_WARNING = "merged into a single task";
+ private static final String IGNORED_WARNING = "is ignored";
+
private final TaskManager taskManager = mock(TaskManager.class);
private final StreamThread streamThread = mock(StreamThread.class);
private final StreamsMetricsImpl streamsMetrics = mock(StreamsMetricsImpl.class);
@@ -89,6 +95,25 @@ public class DefaultStreamsRebalanceListenerTest {
}
}
+ private StreamsRebalanceData rebalanceDataWithSubtopologies(final String... subtopologyIds) {
+ final StreamsRebalanceData streamsRebalanceData = mock(StreamsRebalanceData.class);
+ final Map<String, StreamsRebalanceData.Subtopology> subtopologies = new HashMap<>();
+ for (final String subtopologyId : subtopologyIds) {
+ subtopologies.put(subtopologyId, new StreamsRebalanceData.Subtopology(
+ Set.of("source" + subtopologyId), Set.of(), Map.of(), Map.of(), Set.of()));
+ }
+ when(streamsRebalanceData.subtopologies()).thenReturn(subtopologies);
+ return streamsRebalanceData;
+ }
+
+ private String warningContaining(final LogCaptureAppender appender, final String fragment) {
+ return appender.getMessages("WARN").stream()
+ .filter(message -> message.contains(fragment))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError(
+ "No WARN containing \"" + fragment + "\". Messages: " + appender.getMessages("WARN")));
+ }
+
@ParameterizedTest
@EnumSource(StreamThread.State.class)
void testOnTasksRevoked(final StreamThread.State state) {
@@ -180,7 +205,12 @@ public class DefaultStreamsRebalanceListenerTest {
false
);
- assertDoesNotThrow(() -> defaultStreamsRebalanceListener.onTasksAssigned(assignment));
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(DefaultStreamsRebalanceListener.class)) {
+ assertDoesNotThrow(() -> defaultStreamsRebalanceListener.onTasksAssigned(assignment));
+
+ assertTrue(appender.getMessages("WARN").isEmpty(),
+ "No warning should be logged for an assignment without overlapping roles: " + appender.getMessages("WARN"));
+ }
final InOrder inOrder = inOrder(taskManager, streamThread, streamsRebalanceData);
inOrder.verify(streamThread).setStreamsGroupReady(false);
@@ -196,6 +226,45 @@ public class DefaultStreamsRebalanceListenerTest {
inOrder.verify(streamsRebalanceData).setReconciledAssignment(assignment);
}
+ @Test
+ void shouldDropStandbyAndWarmupForTaskThatIsAlsoAssignedAsActive() {
+ final StreamsRebalanceData streamsRebalanceData = rebalanceDataWithSubtopologies("1", "2", "3", "4");
+ createRebalanceListenerWithRebalanceData(streamsRebalanceData);
+
+ final StreamsRebalanceData.Assignment assignment = new StreamsRebalanceData.Assignment(
+ Set.of(new StreamsRebalanceData.TaskId("1", 0)),
+ Set.of(new StreamsRebalanceData.TaskId("1", 0), new StreamsRebalanceData.TaskId("2", 0), new StreamsRebalanceData.TaskId("3", 0)),
+ Set.of(new StreamsRebalanceData.TaskId("1", 0), new StreamsRebalanceData.TaskId("3", 0), new StreamsRebalanceData.TaskId("4", 0)),
+ true
+ );
+
+ try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(DefaultStreamsRebalanceListener.class)) {
+ assertDoesNotThrow(() -> defaultStreamsRebalanceListener.onTasksAssigned(assignment));
+
+ final String mergedWarning = warningContaining(appender, MERGED_WARNING);
+ assertTrue(mergedWarning.contains("[1_0, 3_0]"), "Only the merged task should be named: " + mergedWarning);
+ final String ignoredWarning = warningContaining(appender, IGNORED_WARNING);
+ assertTrue(ignoredWarning.contains("[1_0]"), "Only the ignored task should be named: " + ignoredWarning);
+ }
+
+ verify(taskManager).handleAssignment(
+ Map.of(new TaskId(1, 0), Set.of(new TopicPartition("source1", 0))),
+ Map.of(
+ new TaskId(2, 0), Set.of(new TopicPartition("source2", 0)),
+ new TaskId(3, 0), Set.of(new TopicPartition("source3", 0)),
+ new TaskId(4, 0), Set.of(new TopicPartition("source4", 0)))
+ );
+
+ // The next heartbeat reports the reconciled assignment as the owned tasks, and the client runs one task per
+ // task id: 1_0 is reported as active only, and 3_0 as a standby only, since the standby outlives the warm-up.
+ verify(streamsRebalanceData).setReconciledAssignment(new StreamsRebalanceData.Assignment(
+ Set.of(new StreamsRebalanceData.TaskId("1", 0)),
+ Set.of(new StreamsRebalanceData.TaskId("2", 0), new StreamsRebalanceData.TaskId("3", 0)),
+ Set.of(new StreamsRebalanceData.TaskId("4", 0)),
+ true
+ ));
+ }
+
@Test
void testOnTasksAssignedWithException() {
final Exception exception = new RuntimeException("sample exception");