You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by bb...@apache.org on 2026/10/16 21:33:17 UTC
(kafka) branch trunk updated: MINOR: Don't block on position() when advancing standby restored offsets (#23841)
This is an automated email from the ASF dual-hosted git repository.
bbejeck 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 65566690ca1 MINOR: Don't block on position() when advancing standby restored offsets (#23841)
65566690ca1 is described below
commit 65566690ca1ddb584388c1026e2ff6673e9c6f19
Author: Bill Bejeck <bb...@apache.org>
AuthorDate: Fri Oct 16 17:33:10 2026 -0400
MINOR: Don't block on position() when advancing standby restored offsets (#23841)
Follow-up to KAFKA-14302 (#23298). `maybeAdvanceStandbyRestoredOffset`
called the blocking `restoreConsumer.position()`, which can wait up to
`default.api.timeout.ms` while the position is unresolved. It runs on
every restore loop for every standby changelog, including paused ones
while actives restore, so a single standby could stall the state-updater
thread. Both calls now use `position(partition, Duration.ZERO)`, and a
timeout is logged at debug and retried on the next loop.
Reviewers: TengYao Chi <fr...@apache.org>
---
.../processor/internals/StoreChangelogReader.java | 7 ++-
.../StoreChangelogReaderCaughtUpOffsetTest.java | 70 +++++++++++++++++++---
2 files changed, 68 insertions(+), 9 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
index bcc67231645..f3cb576c759 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
@@ -793,6 +793,8 @@ public class StoreChangelogReader implements ChangelogReader {
if (!changelogMetadata.bufferedRecords().isEmpty()) {
return;
}
+ // position() uses a zero timeout: this runs on every restore loop for every standby changelog
+ // (including paused ones while actives restore), so it must never block on an unresolved position
try {
final Long restoreEndOffset = changelogMetadata.restoreEndOffset;
if (restoreEndOffset == null) {
@@ -801,18 +803,19 @@ public class StoreChangelogReader implements ChangelogReader {
// the next offset to fetch, not last-applied
final OptionalLong lag = restoreConsumer.currentLag(partition);
if (lag.isPresent() && lag.getAsLong() == 0L) {
- stateManager.advanceRestoredOffsetTo(storeMetadata, restoreConsumer.position(partition));
+ stateManager.advanceRestoredOffsetTo(storeMetadata, restoreConsumer.position(partition, Duration.ZERO));
}
} else if (restoreEndOffset > 0L) {
// source changelog: restoreEndOffset is the committed-offset limit; advance only up to
// that limit, never the live position past unapplied/uncommitted records
- final long position = restoreConsumer.position(partition);
+ final long position = restoreConsumer.position(partition, Duration.ZERO);
if (position >= restoreEndOffset) {
stateManager.advanceRestoredOffsetTo(storeMetadata, restoreEndOffset);
}
}
} catch (final TimeoutException timeoutException) {
// Leave the offset unchanged; the next restore loop retries.
+ log.debug("Position of standby changelog partition {} is not yet known, not advancing its restored offset", partition);
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderCaughtUpOffsetTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderCaughtUpOffsetTest.java
index b6d7f9bdd45..f95055710fa 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderCaughtUpOffsetTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderCaughtUpOffsetTest.java
@@ -39,11 +39,13 @@ import org.mockito.ArgumentCaptor;
import java.io.File;
import java.io.IOException;
+import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
@@ -53,8 +55,10 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
public class StoreChangelogReaderCaughtUpOffsetTest {
@@ -229,6 +233,11 @@ public class StoreChangelogReaderCaughtUpOffsetTest {
reader.register(tp, stateManager);
reader.restore(Collections.singletonMap(new TaskId(0, 0), mock(Task.class)));
+ // position 0 is behind the committed limit 7 with nothing applied yet: advertising the
+ // limit now would let a commit skip [0, 7) for good
+ assertNull(stateManager.storeMetadata(tp).offset());
+ assertEquals(Collections.singletonMap(tp, 0L), stateManager.changelogOffsets());
+
final byte[] keyBytes = new byte[] {0x0, 0x0, 0x0, 0x1};
for (long offset = 5L; offset <= 11L; offset++) {
consumer.addRecord(new ConsumerRecord<>(topicName, 0, offset, keyBytes, "value".getBytes()));
@@ -282,16 +291,14 @@ public class StoreChangelogReaderCaughtUpOffsetTest {
final File stateDir = TestUtils.tempDirectory();
final ProcessorStateManager stateManager = newRealStateManager(STANDBY, stateDir, false);
final MockKeyValueStore kvStore = new MockKeyValueStore(storeName, true);
- final MockConsumer<byte[], byte[]> timeoutConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
- @Override
- public long position(final TopicPartition partition) {
- throw new TimeoutException("KABOOM!");
- }
- };
+ final AtomicInteger timedOutPositionCalls = new AtomicInteger();
+ final MockConsumer<byte[], byte[]> timeoutConsumer = newPositionTimeoutConsumer(timedOutPositionCalls);
+ final Task task = mock(Task.class);
try {
stateManager.registerStore(kvStore, kvStore.stateRestoreCallback, null);
stateManager.initializeStoreOffsets(true);
+ // the log is empty from 20_000 on, so the lag is 0 and only the position lookup fails
timeoutConsumer.updateBeginningOffsets(Collections.singletonMap(tp, 20_000L));
timeoutConsumer.updateEndOffsets(Collections.singletonMap(tp, 20_000L));
adminClient.updateEndOffsets(Collections.singletonMap(tp, 20_000L));
@@ -300,10 +307,47 @@ public class StoreChangelogReaderCaughtUpOffsetTest {
new StoreChangelogReader(time, config, logContext, adminClient, timeoutConsumer, callback, standbyListener);
reader.register(tp, stateManager);
reader.transitToUpdateStandby();
- reader.restore(Collections.singletonMap(new TaskId(0, 0), mock(Task.class)));
+ reader.restore(Collections.singletonMap(new TaskId(0, 0), task));
+
+ assertEquals(1, timedOutPositionCalls.get());
+ assertNull(stateManager.storeMetadata(tp).offset());
+ assertEquals(Collections.singletonMap(tp, 0L), stateManager.changelogOffsets());
+ verify(task, never()).maybeInitTaskTimeoutOrThrow(anyLong(), any());
+ } finally {
+ stateManager.close();
+ Utils.delete(stateDir);
+ }
+ }
+
+ @Test
+ public void shouldNotAdvanceStandbySourceChangelogOffsetWhenPositionTimesOut() throws IOException {
+ final File stateDir = TestUtils.tempDirectory();
+ final ProcessorStateManager stateManager = newRealStateManager(STANDBY, stateDir, true);
+ final MockKeyValueStore kvStore = new MockKeyValueStore(storeName, true);
+ final AtomicInteger timedOutPositionCalls = new AtomicInteger();
+ final MockConsumer<byte[], byte[]> timeoutConsumer = newPositionTimeoutConsumer(timedOutPositionCalls);
+ final Task task = mock(Task.class);
+ try {
+ stateManager.registerStore(kvStore, kvStore.stateRestoreCallback, null);
+ stateManager.initializeStoreOffsets(true);
+
+ // a resolved position (20_000) would reach the committed limit and advance the offset
+ timeoutConsumer.updateBeginningOffsets(Collections.singletonMap(tp, 20_000L));
+ timeoutConsumer.updateEndOffsets(Collections.singletonMap(tp, 20_000L));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, 20_000L));
+ adminClient.updateConsumerGroupOffsets(Collections.singletonMap(tp, 10_000L));
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, timeoutConsumer, callback, standbyListener);
+ reader.transitToUpdateStandby();
+ reader.register(tp, stateManager);
+ reader.restore(Collections.singletonMap(new TaskId(0, 0), task));
+ assertEquals(10_000L, (long) reader.changelogMetadata(tp).endOffset());
+ assertEquals(1, timedOutPositionCalls.get());
assertNull(stateManager.storeMetadata(tp).offset());
assertEquals(Collections.singletonMap(tp, 0L), stateManager.changelogOffsets());
+ verify(task, never()).maybeInitTaskTimeoutOrThrow(anyLong(), any());
} finally {
stateManager.close();
Utils.delete(stateDir);
@@ -493,6 +537,18 @@ public class StoreChangelogReaderCaughtUpOffsetTest {
}
}
+ // only the timed position() lookup times out; the untimed one used elsewhere by the reader and by
+ // MockConsumer.currentLag() still resolves, so the failure is isolated to the standby advance
+ private MockConsumer<byte[], byte[]> newPositionTimeoutConsumer(final AtomicInteger timedOutPositionCalls) {
+ return new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized long position(final TopicPartition partition, final Duration timeout) {
+ timedOutPositionCalls.incrementAndGet();
+ throw new TimeoutException("KABOOM!");
+ }
+ };
+ }
+
private ProcessorStateManager newRealStateManager(final Task.TaskType type,
final File stateDir,
final boolean sourceChangelog) {