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/10 22:46:41 UTC
(kafka) branch trunk updated: KAFKA-20116: Fix reporting task-(end)-offset sum for "streams" protocol (#23013)
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 6c1d5fec60d KAFKA-20116: Fix reporting task-(end)-offset sum for "streams" protocol (#23013)
6c1d5fec60d is described below
commit 6c1d5fec60ddb509558dac34ff7880917e9f9f59
Author: Matthias J. Sax <ma...@confluent.io>
AuthorDate: Mon Aug 10 15:46:33 2026 -0700
KAFKA-20116: Fix reporting task-(end)-offset sum for "streams" protocol (#23013)
This PR ensures that we are assembling the latest task-(end)-offset sum
snapshot only for the "streams" protocol, but not for "classic". For
"classic" the snapshot is never read, so it's unnecessary to maintain
it.
It further ensure that we refresh the task-(end)-offset sum snapshot
even if the main loop is not ready to process active tasks. The async
state-updater thread might still make progress restoring standby and
warmup tasks, so we should continue reporting the most recent metadata.
Reviewers: Bill Bejeck <bb...@apache.org>
---
.../streams/processor/internals/StreamThread.java | 10 ++++-
.../processor/internals/StreamThreadTest.java | 47 +++++++++++++++++++++-
2 files changed, 54 insertions(+), 3 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index 8a74ae6df24..d6a78fc67b5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -1243,6 +1243,10 @@ public class StreamThread extends Thread implements ProcessingThread {
// regardless of streamsGroupReady, as these may throw exceptions that need to be handled.
handleStreamsRebalanceData();
+ // The group coordinator places tasks based on the offsets we report to it, so publish them on every
+ // iteration -- including the ones that return early below, where a task may well still be restoring.
+ taskManager.maybeUpdateTaskOffsetSumSnapshot();
+
if (!streamsGroupReady) {
return;
}
@@ -1268,7 +1272,6 @@ public class StreamThread extends Thread implements ProcessingThread {
if (isStartingRunningOrPartitionAssigned()) {
taskManager.updateLags();
- taskManager.maybeUpdateTaskOffsetSumSnapshot();
/*
* Within an iteration, after processing up to N (N initialized as 1 upon start up) records for each applicable tasks, check the current time:
@@ -1398,6 +1401,10 @@ public class StreamThread extends Thread implements ProcessingThread {
// regardless of streamsGroupReady, as these may throw exceptions that need to be handled.
handleStreamsRebalanceData();
+ // The group coordinator places tasks based on the offsets we report to it, so publish them on every
+ // iteration -- including the ones that return early below, where a task may well still be restoring.
+ taskManager.maybeUpdateTaskOffsetSumSnapshot();
+
if (!streamsGroupReady) {
return;
}
@@ -1416,7 +1423,6 @@ public class StreamThread extends Thread implements ProcessingThread {
if (isRunning()) {
taskManager.updateLags();
- taskManager.maybeUpdateTaskOffsetSumSnapshot();
checkStateUpdater();
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index 6d83f20a60f..a64c85acc0c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -108,6 +108,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
@@ -638,6 +639,48 @@ public class StreamThreadTest {
verify(taskManager, times(2)).maybePurgeCommittedRecords();
}
+ @ParameterizedTest
+ @CsvSource({"true, true", "true, false", "false, true", "false, false"})
+ public void shouldPublishTaskOffsetSumSnapshotOnEveryIteration(final boolean processingThreadsEnabled,
+ final boolean streamsGroupReady) {
+ // The group coordinator places tasks based on the offsets reported to it, so they have to be published on every
+ // iteration, including the ones that return early because the group is not ready -- which is exactly when a
+ // task may still be restoring.
+ final StreamsConfig config = new StreamsConfig(configProps(false, processingThreadsEnabled));
+ final TaskManager taskManager = mock(TaskManager.class);
+ when(mainConsumer.poll(Mockito.any())).thenReturn(ConsumerRecords.empty());
+ final ConsumerGroupMetadata consumerGroupMetadata = mock(ConsumerGroupMetadata.class);
+ when(mainConsumer.groupMetadata()).thenReturn(consumerGroupMetadata);
+ when(consumerGroupMetadata.groupInstanceId()).thenReturn(Optional.empty());
+
+ final StreamsRebalanceData streamsRebalanceData = new StreamsRebalanceData(
+ UUID.randomUUID(), Optional.empty(), Optional.empty(), Map.of(), Map.of(), Map::of, Map::of);
+ final TopologyMetadata topologyMetadata = new TopologyMetadata(internalTopologyBuilder, config);
+ topologyMetadata.buildAndRewriteTopology();
+ thread = new StreamThread(
+ mockTime, config, null,
+ mainConsumer, consumer,
+ changelogReader, null, taskManager, null,
+ new StreamsMetricsImpl(metrics, CLIENT_ID, mockTime),
+ topologyMetadata,
+ PROCESS_ID, CLIENT_ID, new LogContext(""),
+ new AtomicInteger(), new AtomicLong(Long.MAX_VALUE), new LinkedList<>(),
+ null, HANDLER, null,
+ Optional.of(streamsRebalanceData), mock(StreamsMetadataState.class), null, -1L
+ ).updateThreadMetadata(adminClientId(CLIENT_ID));
+ thread.setState(State.STARTING);
+ thread.setState(State.PARTITIONS_ASSIGNED);
+
+ thread.setStreamsGroupReady(streamsGroupReady);
+ if (processingThreadsEnabled) {
+ thread.runOnceWithProcessingThreads();
+ } else {
+ thread.runOnceWithoutProcessingThreads();
+ }
+
+ verify(taskManager).maybeUpdateTaskOffsetSumSnapshot();
+ }
+
@Test
public void shouldNotProcessWhenPartitionRevoked() {
final Properties props = configProps(false, false);
@@ -3447,7 +3490,9 @@ public class StreamThreadTest {
final InOrder inOrder = Mockito.inOrder(mainConsumer, thread.taskManager());
inOrder.verify(mainConsumer).poll(Mockito.any());
inOrder.verify(thread.taskManager()).updateLags();
- inOrder.verify(thread.taskManager()).maybeUpdateTaskOffsetSumSnapshot();
+ // The offset-sum snapshot is only read by the streams-protocol heartbeat thread, so under the classic protocol
+ // it is not published at all.
+ verify(thread.taskManager(), never()).maybeUpdateTaskOffsetSumSnapshot();
}