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/08/31 13:55:24 UTC
(kafka) branch trunk updated: KAFKA-20940: Windowed changelog restore optimization is skipped whenever a checkpointed offset exists (#23285)
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 8702d8e3f58 KAFKA-20940: Windowed changelog restore optimization is skipped whenever a checkpointed offset exists (#23285)
8702d8e3f58 is described below
commit 8702d8e3f5882d9263af120fff26032ef8484aaa
Author: Alan Lau <72...@users.noreply.github.com>
AuthorDate: Mon Aug 31 09:55:15 2026 -0400
KAFKA-20940: Windowed changelog restore optimization is skipped whenever a checkpointed offset exists (#23285)
Jira: https://issues.apache.org/jira/browse/KAFKA-20940
The windowed changelog restore optimization added in KAFKA-13499, which
seeks past already-expired records instead of restoring them, only ran
for partitions with no checkpointed offset; whenever a checkpoint
existed the reader sought straight to it and restored every record after
it, replaying a prefix of data that retention had already expired. This
extends the optimization to checkpointed active windowed changelogs
whose checkpoint is far enough behind the end offset to be worth it,
running them through the same probe but floored at the checkpoint so
restoration never rewinds past records that have already been applied.
When the probe cannot place a partition, or the checkpoint sits below
the log start or the log is empty, restore falls back to the checkpoint
rather than the beginning and leaves the existing truncation and
corruption handling unchanged.
Reviewers: Bill Bejeck <bb...@apache.org>
---
checkstyle/suppressions.xml | 2 +-
.../processor/internals/StoreChangelogReader.java | 140 ++++++-
.../internals/StoreChangelogReaderTest.java | 463 ++++++++++++++++++++-
3 files changed, 576 insertions(+), 29 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 474eb4b4f2a..70c27e0a36e 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -218,7 +218,7 @@
files="(KStreamKStreamJoinTest|KTableKTableForeignKeyJoinIntegrationTest|RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest|RelationalSmokeTest|MockProcessorContextStateStoreTest|IQv2StoreIntegrationTest|StreamsConfigTest).java"/>
<suppress checks="JavaNCSS"
- files="(KStreamKStreamJoinTest|StreamTaskTest|StreamThreadTest|TaskManagerTest|TopologyTestDriverTest).java"/>
+ files="(KStreamKStreamJoinTest|StoreChangelogReaderTest|StreamTaskTest|StreamThreadTest|TaskManagerTest|TopologyTestDriverTest).java"/>
<suppress checks="NPathComplexity"
files="(KStreamKStreamJoinTest|KTableKTableForeignKeyJoinIntegrationTest|RelationalSmokeTest|MockProcessorContextStateStoreTest|TopologyTestDriverTest|IQv2StoreIntegrationTest).java"/>
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 65dd7fa615f..742102c6709 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
@@ -227,6 +227,12 @@ public class StoreChangelogReader implements ChangelogReader {
// probe that is working is never suppressed.
private static final Duration PROBE_RETRY_BACKOFF = Duration.ofSeconds(60);
+ // Smallest stored-offset gap worth a probe: below it, replaying the gap costs about what the probe
+ // spends, and the probe pauses every restoring partition while it runs. Tied to the widest probe
+ // window so the relationship survives retuning.
+ // The 8x is a conservative margin over the probe's cost as measured in load testing.
+ private static final long PROBE_MIN_OFFSET_GAP = 8 * PROBE_WINDOWS[PROBE_WINDOWS.length - 1];
+
private ChangelogReaderState state;
// deliberately outlives unregister: a changelog that is corrupted and re-registered in a loop
@@ -780,7 +786,12 @@ public class StoreChangelogReader implements ChangelogReader {
final long numRecords,
final Long lastRestoredOffset, // this is not a "position" so we need to correct it below
final long restoredToOffset) {
- final long restoredFromOffset = lastRestoredOffset == null ? changelogMetadata.restoreStartOffset : lastRestoredOffset + 1;
+ // clamp to the position restoration started from: after skipping past a stored offset, the first
+ // batch's lastRestoredOffset is the stored offset, which is behind restoreStartOffset; measuring
+ // from the stored offset would decrement the remaining-records metric by the whole skipped gap
+ final long restoredFromOffset = lastRestoredOffset == null
+ ? changelogMetadata.restoreStartOffset
+ : Math.max(changelogMetadata.restoreStartOffset, lastRestoredOffset + 1);
final long numOffsets = Math.max(restoredToOffset - restoredFromOffset, 0L);
task.recordRestoration(time, numRecords, numOffsets, false);
}
@@ -1041,9 +1052,11 @@ public class StoreChangelogReader implements ChangelogReader {
private void prepareChangelogs(final Map<TaskId, Task> tasks,
final Set<ChangelogMetadata> newPartitionsToRestore) {
- // separate those who do not have the current offset loaded from checkpoint
+ // separate those who do not have the current offset loaded from the store
final Set<TopicPartition> newSeekToBeginningPartitions = new HashSet<>();
final Map<TopicPartition, Long> newWindowedPartitionsRetention = new HashMap<>();
+ // for a partition with a stored offset routed through the probe, the stored offset + 1 it may never seek below
+ final Map<TopicPartition, Long> newPartitionsFloor = new HashMap<>();
for (final ChangelogMetadata changelogMetadata : newPartitionsToRestore) {
final StateStoreMetadata storeMetadata = changelogMetadata.storeMetadata;
@@ -1052,13 +1065,23 @@ public class StoreChangelogReader implements ChangelogReader {
final Long endOffset = changelogs.get(partition).restoreEndOffset;
if (currentOffset != null) {
- // the current offset is the offset of the last record, so we should set the position
- // as that offset + 1 as the "next" record to fetch; seek is not a blocking call so
- // there's nothing to capture
- restoreConsumer.seek(partition, currentOffset + 1);
+ // the current offset is the offset of the last record, so the "next" record to fetch is
+ // that offset + 1; it is also the floor the probe may never seek below, since doing so
+ // would re-apply applied records and rewind the stored offset
+ final long floor = currentOffset + 1;
+ if (shouldProbePastStoredOffset(changelogMetadata, partition, floor, endOffset)) {
+ newWindowedPartitionsRetention.put(partition, storeMetadata.retentionPeriod());
+ newPartitionsFloor.put(partition, floor);
+
+ log.debug("Start restoring windowed changelog partition {} by probing to skip expired data " +
+ "past stored offset {} to end offset {}.", partition, currentOffset, recordEndOffset(endOffset));
+ } else {
+ // seek is not a blocking call so there's nothing to capture
+ restoreConsumer.seek(partition, floor);
- log.debug("Start restoring changelog partition {} from current offset {} to end offset {}.",
- partition, currentOffset, recordEndOffset(endOffset));
+ log.debug("Start restoring changelog partition {} from current offset {} to end offset {}.",
+ partition, currentOffset, recordEndOffset(endOffset));
+ }
} else {
final long retentionPeriod = storeMetadata.retentionPeriod();
if (retentionPeriod > 0 && retentionPeriod != Long.MAX_VALUE) {
@@ -1083,7 +1106,7 @@ public class StoreChangelogReader implements ChangelogReader {
}
}
- seekNewPartitions(newWindowedPartitionsRetention, newSeekToBeginningPartitions);
+ seekNewPartitions(newWindowedPartitionsRetention, newSeekToBeginningPartitions, newPartitionsFloor);
for (final ChangelogMetadata changelogMetadata : newPartitionsToRestore) {
final StateStoreMetadata storeMetadata = changelogMetadata.storeMetadata;
@@ -1127,13 +1150,40 @@ public class StoreChangelogReader implements ChangelogReader {
}
}
+ /**
+ * Whether an active windowed changelog with a stored offset should probe to skip expired data instead
+ * of replaying from that offset; the guards below exclude standbys, stores without usable
+ * retention, backed-off partitions, and gaps too small to be worth a probe.
+ */
+ private boolean shouldProbePastStoredOffset(final ChangelogMetadata changelogMetadata,
+ final TopicPartition partition,
+ final long floor,
+ final Long endOffset) {
+ if (changelogMetadata.stateManager.taskType() != TaskType.ACTIVE) {
+ return false;
+ }
+ final long retentionPeriod = changelogMetadata.storeMetadata.retentionPeriod();
+ if (retentionPeriod <= 0 || retentionPeriod == Long.MAX_VALUE) {
+ return false;
+ }
+ if (!probeIsDue(partition)) {
+ return false;
+ }
+ return endOffset != null && endOffset - floor >= PROBE_MIN_OFFSET_GAP;
+ }
+
private void seekNewPartitions(final Map<TopicPartition, Long> windowedPartitionsRetention,
- final Set<TopicPartition> seekToBeginningPartitions) {
+ final Set<TopicPartition> seekToBeginningPartitions,
+ final Map<TopicPartition, Long> newPartitionsFloor) {
// Seek non-windowed partitions to beginning.
if (!seekToBeginningPartitions.isEmpty()) {
restoreConsumer.seekToBeginning(seekToBeginningPartitions);
}
+ // Partitions with a stored offset routed through the probe that fall back are sent to their floor
+ // (stored offset + 1), never to the beginning; collected here and seeked at the end.
+ final Set<TopicPartition> floorFallbackPartitions = new HashSet<>();
+
// Try to optimize windowed partitions by seeking past expired data.
if (!windowedPartitionsRetention.isEmpty()) {
final Set<TopicPartition> allAssigned = restoreConsumer.assignment();
@@ -1148,6 +1198,24 @@ public class StoreChangelogReader implements ChangelogReader {
final Map<TopicPartition, Long> beginningOffsets =
restoreConsumer.beginningOffsets(windowedPartitionsRetention.keySet());
+ // A stored offset below the log start (truncated head) or on an empty/shrunk log cannot be
+ // certified by the skip. Seek it to its floor, pause it, and drop it from the probe, so
+ // the next restore poll -- not the probe's -- takes the InvalidOffsetException ->
+ // TaskCorrupted wipe path a plain seek(stored offset + 1) would; probe from scratch afterward.
+ final Set<TopicPartition> demotedToFloor = new HashSet<>();
+ for (final TopicPartition partition : windowedPartitionsRetention.keySet()) {
+ final Long floor = newPartitionsFloor.get(partition);
+ final Long endOffset = endOffsets.get(partition);
+ if (floor != null &&
+ (floor < beginningOffsets.getOrDefault(partition, 0L) || endOffset == null || endOffset <= 0)) {
+ restoreConsumer.seek(partition, floor);
+ restoreConsumer.pause(Collections.singleton(partition));
+ demotedToFloor.add(partition);
+ }
+ }
+ windowedPartitionsRetention.keySet().removeAll(demotedToFloor);
+
+ // partitions with no stored offset on an empty/unreadable log seek to the beginning
for (final TopicPartition partition : windowedPartitionsRetention.keySet()) {
final Long endOffset = endOffsets.get(partition);
if (endOffset == null || endOffset <= 0) {
@@ -1163,19 +1231,22 @@ public class StoreChangelogReader implements ChangelogReader {
final Set<TopicPartition> unresolved = new HashSet<>(windowedPartitionsRetention.keySet());
runBackwardProbe(unresolved, latestTimestamps, beginningOffsets, endOffsets);
- seekByRetention(latestTimestamps, windowedPartitionsRetention, seekToBeginningPartitions);
+ seekByRetention(latestTimestamps, windowedPartitionsRetention, seekToBeginningPartitions,
+ newPartitionsFloor, floorFallbackPartitions);
} catch (final TimeoutException e) {
log.debug("Could not seek by timestamp for changelog partitions {}, falling back to seek-to-beginning",
windowedPartitionsRetention.keySet(), e);
- seekToBeginningPartitions.addAll(windowedPartitionsRetention.keySet());
+ addProbeFallback(windowedPartitionsRetention.keySet(), newPartitionsFloor,
+ seekToBeginningPartitions, floorFallbackPartitions);
} catch (final KafkaException e) {
log.warn("Failed to seek by timestamp for changelog partitions {}, falling back to seek-to-beginning",
windowedPartitionsRetention.keySet(), e);
- seekToBeginningPartitions.addAll(windowedPartitionsRetention.keySet());
+ addProbeFallback(windowedPartitionsRetention.keySet(), newPartitionsFloor,
+ seekToBeginningPartitions, floorFallbackPartitions);
} finally {
// in the finally, not after the probe: a lookup that times out leaves through a
// catch, and a probe that fails that way must arm the backoff like any other
- recordProbeOutcomes(windowedPartitionsRetention, seekToBeginningPartitions);
+ recordProbeOutcomes(windowedPartitionsRetention, seekToBeginningPartitions, floorFallbackPartitions);
restoreConsumer.pause(allAssigned);
final Set<TopicPartition> toResume = new HashSet<>(allAssigned);
toResume.removeAll(previouslyPaused);
@@ -1190,6 +1261,24 @@ public class StoreChangelogReader implements ChangelogReader {
if (!seekToBeginningPartitions.isEmpty()) {
restoreConsumer.seekToBeginning(seekToBeginningPartitions);
}
+ // Partitions with a stored offset that fell back seek to their floor, never to the beginning.
+ for (final TopicPartition partition : floorFallbackPartitions) {
+ restoreConsumer.seek(partition, newPartitionsFloor.get(partition));
+ }
+ }
+
+ /** A partition the probe could not place goes to its floor if it has a stored offset, else to the beginning. */
+ private static void addProbeFallback(final Collection<TopicPartition> partitions,
+ final Map<TopicPartition, Long> newPartitionsFloor,
+ final Set<TopicPartition> seekToBeginningPartitions,
+ final Set<TopicPartition> floorFallbackPartitions) {
+ for (final TopicPartition partition : partitions) {
+ if (newPartitionsFloor.containsKey(partition)) {
+ floorFallbackPartitions.add(partition);
+ } else {
+ seekToBeginningPartitions.add(partition);
+ }
+ }
}
/**
@@ -1281,7 +1370,9 @@ public class StoreChangelogReader implements ChangelogReader {
private void seekByRetention(final Map<TopicPartition, Long> latestTimestamps,
final Map<TopicPartition, Long> windowedPartitionsRetention,
- final Set<TopicPartition> seekToBeginningPartitions) {
+ final Set<TopicPartition> seekToBeginningPartitions,
+ final Map<TopicPartition, Long> newPartitionsFloor,
+ final Set<TopicPartition> floorFallbackPartitions) {
final Map<TopicPartition, Long> seekTimestamps = new HashMap<>();
for (final Map.Entry<TopicPartition, Long> entry : windowedPartitionsRetention.entrySet()) {
final TopicPartition partition = entry.getKey();
@@ -1296,18 +1387,24 @@ public class StoreChangelogReader implements ChangelogReader {
continue;
}
}
- log.debug("Start restoring changelog partition {} from the beginning.", partition);
- seekToBeginningPartitions.add(partition);
+ log.debug("Start restoring changelog partition {} from its fallback position.", partition);
+ addProbeFallback(Collections.singleton(partition), newPartitionsFloor, seekToBeginningPartitions, floorFallbackPartitions);
}
if (!seekTimestamps.isEmpty()) {
final Map<TopicPartition, OffsetAndTimestamp> offsetsByTimestamp =
restoreConsumer.offsetsForTimes(seekTimestamps);
offsetsByTimestamp.forEach((partition, offsetAndTimestamp) -> {
- if (offsetAndTimestamp != null) {
+ final Long floor = newPartitionsFloor.get(partition);
+ if (offsetAndTimestamp != null && (floor == null || offsetAndTimestamp.offset() > floor)) {
+ // a genuine skip past expired data; floor is null with no stored offset, so any
+ // resolved offset is honoured as before
restoreConsumer.seek(partition, offsetAndTimestamp.offset());
} else {
- seekToBeginningPartitions.add(partition);
+ // no benefit: offsetsForTimes could not resolve, or resolved at or below the
+ // stored-offset floor so the probe would skip nothing. Fall back to the floor (or the
+ // beginning) and arm the backoff so a re-registration does not re-pay for the probe.
+ addProbeFallback(Collections.singleton(partition), newPartitionsFloor, seekToBeginningPartitions, floorFallbackPartitions);
}
});
}
@@ -1315,9 +1412,10 @@ public class StoreChangelogReader implements ChangelogReader {
/** Arms the backoff for partitions the probe could not place, and clears it for those it did. */
private void recordProbeOutcomes(final Map<TopicPartition, Long> windowedPartitionsRetention,
- final Set<TopicPartition> seekToBeginningPartitions) {
+ final Set<TopicPartition> seekToBeginningPartitions,
+ final Set<TopicPartition> floorFallbackPartitions) {
for (final TopicPartition partition : windowedPartitionsRetention.keySet()) {
- if (seekToBeginningPartitions.contains(partition)) {
+ if (seekToBeginningPartitions.contains(partition) || floorFallbackPartitions.contains(partition)) {
probeFailedAtMs.put(partition, time.milliseconds());
} else {
probeFailedAtMs.remove(partition);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java
index 53d964090fc..7a1481b45c5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java
@@ -40,6 +40,7 @@ import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.errors.TaskCorruptedException;
import org.apache.kafka.streams.processor.StandbyUpdateListener.SuspendReason;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
@@ -66,6 +67,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
@@ -98,6 +100,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -1610,7 +1613,7 @@ public class StoreChangelogReaderTest {
final StateStore store = mock(StateStore.class);
when(meta.changelogPartition()).thenReturn(tps[i]);
when(meta.store()).thenReturn(store);
- when(meta.offset()).thenReturn(null, 0L); // no checkpoint, then a value once restoring
+ when(meta.offset()).thenReturn(null, 0L); // no stored offset, then a value once restoring
when(meta.retentionPeriod()).thenReturn(shortRetentionMs);
when(store.name()).thenReturn(storeName);
when(manager.storeMetadata(tps[i])).thenReturn(meta);
@@ -1699,7 +1702,7 @@ public class StoreChangelogReaderTest {
final StateStore store = mock(StateStore.class);
when(meta.changelogPartition()).thenReturn(tps[i]);
when(meta.store()).thenReturn(store);
- when(meta.offset()).thenReturn(null, 0L); // no checkpoint, then a value once restoring
+ when(meta.offset()).thenReturn(null, 0L); // no stored offset, then a value once restoring
when(meta.retentionPeriod()).thenReturn(shortRetentionMs);
when(store.name()).thenReturn(storeName);
when(manager.storeMetadata(tps[i])).thenReturn(meta);
@@ -1788,7 +1791,7 @@ public class StoreChangelogReaderTest {
final StateStore store = mock(StateStore.class);
when(meta.changelogPartition()).thenReturn(tps[i]);
when(meta.store()).thenReturn(store);
- when(meta.offset()).thenReturn(null, 0L); // no checkpoint, then a value once restoring
+ when(meta.offset()).thenReturn(null, 0L); // no stored offset, then a value once restoring
when(meta.retentionPeriod()).thenReturn(shortRetentionMs);
when(store.name()).thenReturn(storeName);
when(manager.storeMetadata(tps[i])).thenReturn(meta);
@@ -1883,7 +1886,7 @@ public class StoreChangelogReaderTest {
final StateStore store = mock(StateStore.class);
when(meta.changelogPartition()).thenReturn(tps[i]);
when(meta.store()).thenReturn(store);
- when(meta.offset()).thenReturn(null, 0L); // no checkpoint, then a value once restoring
+ when(meta.offset()).thenReturn(null, 0L); // no stored offset, then a value once restoring
when(meta.retentionPeriod()).thenReturn(shortRetentionMs);
when(store.name()).thenReturn(storeName);
when(manager.storeMetadata(tps[i])).thenReturn(meta);
@@ -2110,7 +2113,7 @@ public class StoreChangelogReaderTest {
final StateStore store = mock(StateStore.class);
when(meta.changelogPartition()).thenReturn(tp);
when(meta.store()).thenReturn(store);
- when(meta.offset()).thenReturn(null, 0L); // no checkpoint, then a value once restoring
+ when(meta.offset()).thenReturn(null, 0L); // no stored offset, then a value once restoring
when(meta.retentionPeriod()).thenReturn(retentionMs);
when(store.name()).thenReturn(storeName);
when(manager.storeMetadata(tp)).thenReturn(meta);
@@ -2126,7 +2129,7 @@ public class StoreChangelogReaderTest {
}
@Test
- public void shouldSeekByTimestampForWindowedStoreWithoutCheckpoint() {
+ public void shouldSeekByTimestampForWindowedStoreWithoutStoredOffset() {
final long retentionMs = Duration.ofHours(2).toMillis();
final long offsetForTimestamp = 42L;
final long latestRecordTimestamp = 10_000_000L;
@@ -2225,7 +2228,7 @@ public class StoreChangelogReaderTest {
}
@Test
- public void shouldSeekToBeginningForNonWindowedStoreWithoutCheckpoint() {
+ public void shouldSeekToBeginningForNonWindowedStoreWithoutStoredOffset() {
final StateStoreMetadata kvStoreMetadata = mock(StateStoreMetadata.class);
final ProcessorStateManager kvStateManager = mock(ProcessorStateManager.class);
final StateStore kvStore = mock(StateStore.class);
@@ -2252,6 +2255,452 @@ public class StoreChangelogReaderTest {
assertEquals(0L, consumer.position(tp), "Non-windowed store should seek to beginning, not by timestamp");
}
+ private ProcessorStateManager windowedActiveManager(final StateStoreMetadata meta,
+ final TopicPartition partition,
+ final TaskId taskId,
+ final long retentionMs) {
+ final ProcessorStateManager manager = mock(ProcessorStateManager.class);
+ final StateStore windowStore = mock(StateStore.class);
+ when(meta.changelogPartition()).thenReturn(partition);
+ when(meta.store()).thenReturn(windowStore);
+ when(meta.retentionPeriod()).thenReturn(retentionMs);
+ when(windowStore.name()).thenReturn(storeName);
+ when(manager.storeMetadata(partition)).thenReturn(meta);
+ when(manager.taskType()).thenReturn(ACTIVE);
+ when(manager.taskId()).thenReturn(taskId);
+ return manager;
+ }
+
+ private static ConsumerRecord<byte[], byte[]> changelogRecord(final TopicPartition partition,
+ final long offset,
+ final long timestamp) {
+ return new ConsumerRecord<>(partition.topic(), partition.partition(), offset, timestamp,
+ TimestampType.CREATE_TIME, 0, 0, new byte[0], new byte[0], new RecordHeaders(), Optional.empty());
+ }
+
+ /**
+ * With a stored offset far behind a large changelog, the reader used to replay the whole gap. It
+ * should instead probe for observed stream time and skip data the store discards on restore,
+ * just as it does when restoring from scratch.
+ */
+ @Test
+ public void shouldSeekPastStoredOffsetByTimestampWhenGapIsLarge() {
+ final long retentionMs = Duration.ofHours(2).toMillis();
+ final long storedOffset = 100L;
+ final long endOffset = 100_000L; // gap far exceeds PROBE_MIN_OFFSET_GAP
+ final long offsetForTimestamp = 50_000L; // well past the stored offset
+
+ final MockConsumer<byte[], byte[]> timestampConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(final Map<TopicPartition, Long> timestampsToSearch) {
+ final Map<TopicPartition, OffsetAndTimestamp> result = new HashMap<>();
+ timestampsToSearch.forEach((key, value) -> result.put(key, new OffsetAndTimestamp(offsetForTimestamp, value)));
+ return result;
+ }
+ };
+ timestampConsumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ timestampConsumer.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ timestampConsumer.schedulePollTask(() -> timestampConsumer.addRecord(changelogRecord(tp, endOffset - 1, 10_000_000L)));
+
+ final TaskId taskId = new TaskId(0, 0);
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ when(meta.offset()).thenReturn(storedOffset);
+ final ProcessorStateManager manager = windowedActiveManager(meta, tp, taskId, retentionMs);
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, timestampConsumer, callback, standbyListener);
+ reader.register(tp, manager);
+ reader.restore(Collections.singletonMap(taskId, mock(Task.class)));
+
+ assertEquals(offsetForTimestamp, timestampConsumer.position(tp),
+ "a stored offset far behind a large changelog should skip expired data, not replay from the stored offset");
+ assertEquals(offsetForTimestamp, callback.restoreStartOffset,
+ "restoration should be reported as starting at the skipped-to offset");
+ }
+
+ /**
+ * After skipping past a stored offset, the first restored batch begins at the skip target, not at
+ * the stored offset. Measuring the batch from the stored offset would decrement the remaining-records
+ * metric by the whole skipped gap and drive it negative.
+ */
+ @Test
+ public void shouldNotDecrementRemainingRecordsBelowZeroWhenSkippingPastStoredOffset() {
+ final long retentionMs = Duration.ofHours(2).toMillis();
+ final long storedOffset = 100L;
+ final long endOffset = 100_000L;
+ final long offsetForTimestamp = 50_000L;
+ final long batchRecordOffset = 60_000L;
+
+ final MockConsumer<byte[], byte[]> timestampConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(final Map<TopicPartition, Long> timestampsToSearch) {
+ final Map<TopicPartition, OffsetAndTimestamp> result = new HashMap<>();
+ timestampsToSearch.forEach((key, value) -> result.put(key, new OffsetAndTimestamp(offsetForTimestamp, value)));
+ return result;
+ }
+ };
+ timestampConsumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ timestampConsumer.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ // the probe consumes the record at the head; the batch record sits below the probe position
+ // and is only returned by the main poll after the skip
+ timestampConsumer.schedulePollTask(() -> {
+ timestampConsumer.addRecord(changelogRecord(tp, endOffset - 1, 10_000_000L));
+ timestampConsumer.addRecord(changelogRecord(tp, batchRecordOffset, 10_000_000L));
+ });
+
+ // storeMetadata.offset() advances as records are restored, mirroring ProcessorStateManager.restore
+ final long[] storeOffset = {storedOffset};
+ final TaskId taskId = new TaskId(0, 0);
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ when(meta.offset()).thenAnswer(invocation -> storeOffset[0]);
+ final ProcessorStateManager manager = windowedActiveManager(meta, tp, taskId, retentionMs);
+ doAnswer(invocation -> {
+ final List<ConsumerRecord<byte[], byte[]>> restored = invocation.getArgument(1);
+ if (!restored.isEmpty()) {
+ storeOffset[0] = restored.get(restored.size() - 1).offset();
+ }
+ return null;
+ }).when(manager).restore(any(), any(), any());
+
+ final Task task = mock(Task.class);
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, timestampConsumer, callback, standbyListener);
+ reader.register(tp, manager);
+ reader.restore(Collections.singletonMap(taskId, task));
+
+ final ArgumentCaptor<Long> numOffsets = ArgumentCaptor.forClass(Long.class);
+ final ArgumentCaptor<Boolean> initRemaining = ArgumentCaptor.forClass(Boolean.class);
+ verify(task, atLeastOnce()).recordRestoration(any(), anyLong(), numOffsets.capture(), initRemaining.capture());
+
+ long initialized = 0L;
+ long decremented = 0L;
+ for (int i = 0; i < initRemaining.getAllValues().size(); i++) {
+ if (initRemaining.getAllValues().get(i)) {
+ initialized += numOffsets.getAllValues().get(i);
+ } else {
+ decremented += numOffsets.getAllValues().get(i);
+ }
+ }
+ assertEquals(offsetForTimestamp, callback.restoreStartOffset,
+ "restoration should be reported as starting at the skipped-to offset");
+ assertTrue(decremented <= initialized,
+ "remaining-records was initialised to " + initialized + " but decremented by " + decremented
+ + "; measuring the first batch from the stored offset instead of the skip target drives it negative");
+ }
+
+ /**
+ * When little of the gap is expired, offsetsForTimes returns an offset at or before the
+ * stored offset. Seeking there would re-apply already-applied records and rewind the stored offset, so
+ * the seek must be floored at the stored offset.
+ */
+ @Test
+ public void shouldNotSeekBelowStoredOffsetWhenTimestampOffsetIsBehindIt() {
+ final long retentionMs = Duration.ofHours(2).toMillis();
+ final long storedOffset = 100L;
+ final long endOffset = 100_000L;
+ final long offsetForTimestamp = 90L; // behind the stored offset: almost all of the gap is live
+
+ final Map<TopicPartition, Long> probed = new HashMap<>();
+ final MockConsumer<byte[], byte[]> timestampConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(final Map<TopicPartition, Long> timestampsToSearch) {
+ probed.putAll(timestampsToSearch);
+ final Map<TopicPartition, OffsetAndTimestamp> result = new HashMap<>();
+ timestampsToSearch.forEach((key, value) -> result.put(key, new OffsetAndTimestamp(offsetForTimestamp, value)));
+ return result;
+ }
+ };
+ timestampConsumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ timestampConsumer.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ timestampConsumer.schedulePollTask(() -> timestampConsumer.addRecord(changelogRecord(tp, endOffset - 1, 10_000_000L)));
+
+ final TaskId taskId = new TaskId(0, 0);
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ when(meta.offset()).thenReturn(storedOffset);
+ final ProcessorStateManager manager = windowedActiveManager(meta, tp, taskId, retentionMs);
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, timestampConsumer, callback, standbyListener);
+ reader.register(tp, manager);
+ reader.restore(Collections.singletonMap(taskId, mock(Task.class)));
+
+ assertTrue(probed.containsKey(tp), "a large gap should route the stored-offset partition through the probe");
+ assertEquals(storedOffset + 1, timestampConsumer.position(tp),
+ "when offsetsForTimes lands at or before the stored offset, restoration must resume at the stored offset, never rewind below it");
+ }
+
+ /**
+ * A probe that resolves at or below the stored offset skipped nothing, so it must arm the backoff
+ * just like a probe that never answered: an immediate re-registration should replay from the
+ * stored offset without probing again, rather than re-paying for a probe that cannot help.
+ */
+ @Test
+ public void shouldBackOffWhenTimestampOffsetClampsToStoredOffset() {
+ final long retentionMs = Duration.ofHours(2).toMillis();
+ final long storedOffset = 100L;
+ final long endOffset = 100_000L;
+ final long offsetForTimestamp = 90L; // at or below the stored offset: nothing in the gap expired
+ final int[] probeRounds = {0};
+
+ final MockConsumer<byte[], byte[]> probeConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(final Map<TopicPartition, Long> timestampsToSearch) {
+ probeRounds[0]++;
+ final Map<TopicPartition, OffsetAndTimestamp> result = new HashMap<>();
+ timestampsToSearch.forEach((key, value) -> result.put(key, new OffsetAndTimestamp(offsetForTimestamp, value)));
+ return result;
+ }
+ };
+ probeConsumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ probeConsumer.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ // the probe resolves a stream time from the head record, then offsetsForTimes clamps to the stored offset
+ probeConsumer.schedulePollTask(() -> probeConsumer.addRecord(changelogRecord(tp, endOffset - 1, 10_000_000L)));
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, probeConsumer, callback, standbyListener);
+ final TaskId taskId = new TaskId(0, 0);
+
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ when(meta.offset()).thenReturn(storedOffset);
+ reader.register(tp, windowedActiveManager(meta, tp, taskId, retentionMs));
+ reader.restore(Collections.singletonMap(taskId, mock(Task.class)));
+
+ assertEquals(storedOffset + 1, probeConsumer.position(tp),
+ "a probe that clamps to the stored offset must resume there, not rewind below it");
+ assertEquals(1, probeRounds[0], "the first registration probes once");
+
+ reader.unregister(Collections.singleton(tp));
+
+ final StateStoreMetadata meta2 = mock(StateStoreMetadata.class);
+ when(meta2.offset()).thenReturn(storedOffset);
+ reader.register(tp, windowedActiveManager(meta2, tp, taskId, retentionMs));
+ reader.restore(Collections.singletonMap(taskId, mock(Task.class)));
+
+ assertEquals(1, probeRounds[0],
+ "a probe that skipped nothing must arm the backoff, so re-registering immediately replays from the stored offset without probing");
+ assertEquals(storedOffset + 1, probeConsumer.position(tp), "the backed-off registration replays from the stored offset");
+ }
+
+ /**
+ * A small stored offset gap is not worth a probe: replaying it costs about what the probe spends,
+ * and the probe pauses every restoring partition while it runs. Common restores should pay no
+ * new RPCs.
+ */
+ @Test
+ public void shouldNotProbeWhenStoredOffsetGapIsSmall() {
+ final long retentionMs = Duration.ofHours(2).toMillis();
+ final long storedOffset = 100L;
+ final long endOffset = 200L; // gap far below PROBE_MIN_OFFSET_GAP
+
+ final MockConsumer<byte[], byte[]> noProbeConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized Map<TopicPartition, Long> endOffsets(final Collection<TopicPartition> partitions) {
+ throw new AssertionError("a small stored offset gap must not trigger a probe");
+ }
+
+ @Override
+ public synchronized Map<TopicPartition, Long> beginningOffsets(final Collection<TopicPartition> partitions) {
+ throw new AssertionError("a small stored offset gap must not trigger a probe");
+ }
+
+ @Override
+ public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(final Map<TopicPartition, Long> timestampsToSearch) {
+ throw new AssertionError("a small stored offset gap must not trigger a probe");
+ }
+ };
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+
+ final TaskId taskId = new TaskId(0, 0);
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ when(meta.offset()).thenReturn(storedOffset);
+ final ProcessorStateManager manager = windowedActiveManager(meta, tp, taskId, retentionMs);
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, noProbeConsumer, callback, standbyListener);
+ reader.register(tp, manager);
+ reader.restore(Collections.singletonMap(taskId, mock(Task.class)));
+
+ assertEquals(storedOffset + 1, noProbeConsumer.position(tp), "a small gap should replay from the stored offset, cheaply");
+ }
+
+ /**
+ * A probe that never answers must send a stored-offset partition back to its stored offset, not to
+ * the beginning, and arm the backoff so a re-registration loop does not probe on every iteration.
+ */
+ @Test
+ public void shouldFallBackToStoredOffsetAndBackOffWhenProbeNeverAnswers() {
+ final long retentionMs = Duration.ofHours(2).toMillis();
+ final long storedOffset = 100L;
+ final long endOffset = 100_000L;
+ final int[] probeRounds = {0};
+
+ final MockConsumer<byte[], byte[]> probeConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized Map<TopicPartition, Long> beginningOffsets(final Collection<TopicPartition> partitions) {
+ probeRounds[0]++; // only the probe looks these up
+ return super.beginningOffsets(partitions);
+ }
+ };
+ probeConsumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ probeConsumer.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ // no records are scheduled, so the probe never answers and must fall back
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, probeConsumer, callback, standbyListener);
+ final TaskId taskId = new TaskId(0, 0);
+
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ when(meta.offset()).thenReturn(storedOffset);
+ reader.register(tp, windowedActiveManager(meta, tp, taskId, retentionMs));
+ reader.restore(Collections.singletonMap(taskId, mock(Task.class)));
+
+ assertEquals(storedOffset + 1, probeConsumer.position(tp),
+ "a probe that never answers must fall back to the stored offset, not the beginning");
+ assertEquals(1, probeRounds[0], "the first registration probes once");
+
+ reader.unregister(Collections.singleton(tp));
+
+ final StateStoreMetadata meta2 = mock(StateStoreMetadata.class);
+ when(meta2.offset()).thenReturn(storedOffset);
+ reader.register(tp, windowedActiveManager(meta2, tp, taskId, retentionMs));
+ reader.restore(Collections.singletonMap(taskId, mock(Task.class)));
+
+ assertEquals(1, probeRounds[0],
+ "the probe just fell back, so re-registering immediately must replay from the stored offset without probing");
+ assertEquals(storedOffset + 1, probeConsumer.position(tp), "the backed-off registration replays from the stored offset");
+ }
+
+ /**
+ * When a probe falls back for a mixed set, each partition must land on its own safe position: a
+ * partition with no stored offset at the beginning, a stored-offset one at its stored offset.
+ */
+ @Test
+ public void shouldFallBackNoStoredOffsetToBeginningAndStoredOffsetToItsFloor() {
+ final long retentionMs = Duration.ofHours(2).toMillis();
+ final long storedOffset = 100L;
+ final long endOffset = 100_000L;
+
+ final MockConsumer<byte[], byte[]> probeConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name());
+ final Map<TopicPartition, Long> begins = new HashMap<>();
+ begins.put(tp, 0L);
+ begins.put(tp1, 0L);
+ final Map<TopicPartition, Long> ends = new HashMap<>();
+ ends.put(tp, endOffset);
+ ends.put(tp1, endOffset);
+ probeConsumer.updateBeginningOffsets(begins);
+ probeConsumer.updateEndOffsets(ends);
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp1, endOffset));
+ // no records are scheduled, so both probes fall back
+
+ final TaskId taskId0 = new TaskId(0, 0);
+ final TaskId taskId1 = new TaskId(0, 1);
+ final StateStoreMetadata meta0 = mock(StateStoreMetadata.class);
+ when(meta0.offset()).thenReturn(null);
+ final StateStoreMetadata meta1 = mock(StateStoreMetadata.class);
+ when(meta1.offset()).thenReturn(storedOffset);
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, probeConsumer, callback, standbyListener);
+ reader.register(tp, windowedActiveManager(meta0, tp, taskId0, retentionMs));
+ reader.register(tp1, windowedActiveManager(meta1, tp1, taskId1, retentionMs));
+
+ final Map<TaskId, Task> tasks = new HashMap<>();
+ tasks.put(taskId0, mock(Task.class));
+ tasks.put(taskId1, mock(Task.class));
+ reader.restore(tasks);
+
+ assertEquals(0L, probeConsumer.position(tp), "the partition with no stored offset falls back to the beginning");
+ assertEquals(storedOffset + 1, probeConsumer.position(tp1),
+ "the stored-offset partition falls back to its stored offset, not the beginning");
+ }
+
+ /**
+ * A stored offset below the current log start means records in (stored offset, logStart) were deleted
+ * before they could be verified. The skip cannot certify them, so the partition takes the
+ * deterministic InvalidOffsetException -> TaskCorruptedException wipe path, exactly as a plain
+ * seek to the stored offset would.
+ */
+ @Test
+ public void shouldMarkTaskCorruptedWhenStoredOffsetIsBelowTheLogStart() {
+ final long retentionMs = Duration.ofHours(2).toMillis();
+ final long storedOffset = 100L;
+ final long logStart = 5_000L; // (stored offset, logStart) has been truncated
+ final long endOffset = 100_000L;
+
+ final MockConsumer<byte[], byte[]> probeConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name());
+ probeConsumer.updateBeginningOffsets(Collections.singletonMap(tp, logStart));
+ probeConsumer.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, endOffset));
+ // a record must be queued so the out-of-range position is discovered on the next poll
+ probeConsumer.schedulePollTask(() -> probeConsumer.addRecord(changelogRecord(tp, logStart, 10_000_000L)));
+
+ final TaskId taskId = new TaskId(0, 0);
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ when(meta.offset()).thenReturn(storedOffset);
+ final ProcessorStateManager manager = windowedActiveManager(meta, tp, taskId, retentionMs);
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, probeConsumer, callback, standbyListener);
+ reader.register(tp, manager);
+
+ assertThrows(TaskCorruptedException.class,
+ () -> reader.restore(Collections.singletonMap(taskId, mock(Task.class))),
+ "a stored offset below the log start cannot be verified and must trigger the wipe path");
+ }
+
+ /**
+ * The skip optimisation is active-only: a standby with a stored offset replays its gap and is never
+ * probed.
+ */
+ @Test
+ public void shouldNotProbeStandbyWithStoredOffset() {
+ final long storedOffset = 100L;
+
+ final MockConsumer<byte[], byte[]> noProbeConsumer = new MockConsumer<>(AutoOffsetResetStrategy.EARLIEST.name()) {
+ @Override
+ public synchronized Map<TopicPartition, Long> endOffsets(final Collection<TopicPartition> partitions) {
+ throw new AssertionError("standbys must not probe");
+ }
+
+ @Override
+ public synchronized Map<TopicPartition, Long> beginningOffsets(final Collection<TopicPartition> partitions) {
+ throw new AssertionError("standbys must not probe");
+ }
+
+ @Override
+ public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(final Map<TopicPartition, Long> timestampsToSearch) {
+ throw new AssertionError("standbys must not probe");
+ }
+ };
+
+ final TaskId taskId = new TaskId(0, 0);
+ final StateStoreMetadata meta = mock(StateStoreMetadata.class);
+ final ProcessorStateManager manager = mock(ProcessorStateManager.class);
+ final StateStore windowStore = mock(StateStore.class);
+ when(meta.changelogPartition()).thenReturn(tp);
+ when(meta.store()).thenReturn(windowStore);
+ when(meta.offset()).thenReturn(storedOffset);
+ when(windowStore.name()).thenReturn(storeName);
+ when(manager.storeMetadata(tp)).thenReturn(meta);
+ when(manager.taskType()).thenReturn(STANDBY);
+ when(manager.taskId()).thenReturn(taskId);
+ when(manager.changelogAsSource(tp)).thenReturn(false);
+
+ final StoreChangelogReader reader =
+ new StoreChangelogReader(time, config, logContext, adminClient, noProbeConsumer, callback, standbyListener);
+ reader.register(tp, manager);
+ reader.restore(Collections.singletonMap(taskId, mock(Task.class)));
+
+ assertEquals(storedOffset + 1, noProbeConsumer.position(tp),
+ "a standby replays its stored offset gap; the skip optimisation is active-only");
+ }
+
private void assignPartition(final long messages,
final TopicPartition topicPartition) {
consumer.updatePartitions(