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 15:19:48 UTC
(kafka) branch trunk updated: KAFKA-20711: Measure restore-remaining-records in offset slots (#22614)
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 e2aa6d676f2 KAFKA-20711: Measure restore-remaining-records in offset slots (#22614)
e2aa6d676f2 is described below
commit e2aa6d676f260d147302179a7a9cf3af369f01ad
Author: Nick Telford <ni...@gmail.com>
AuthorDate: Tue Aug 11 16:19:40 2026 +0100
KAFKA-20711: Measure restore-remaining-records in offset slots (#22614)
The `restore-remaining-records-total` task metric overshoots and never
reaches zero. The code does initialize the metric based on offset
difference (`endOffset - startOffset`), but counts down based on number
of records restored, not taking "missing offsets" from log compaction,
TX markers, or aborted messages (under EOS) into account.
This fix makes the decrements consistent with the initialization by
expressing the remaining-records metric's progress in changelog offset
slots rather than record counts.
Reviewers: Matthias J. Sax <ma...@confluent.io>
---
.../streams/processor/internals/ReadOnlyTask.java | 2 +-
.../streams/processor/internals/StandbyTask.java | 2 +-
.../processor/internals/StoreChangelogReader.java | 36 ++++++++--
.../streams/processor/internals/StreamTask.java | 6 +-
.../kafka/streams/processor/internals/Task.java | 6 +-
.../processor/internals/StandbyTaskTest.java | 5 +-
.../internals/StoreChangelogReaderTest.java | 77 +++++++++++++++++++++-
.../processor/internals/StreamTaskTest.java | 12 ++--
8 files changed, 126 insertions(+), 20 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
index f4d4e506191..2832f8946e0 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ReadOnlyTask.java
@@ -210,7 +210,7 @@ public class ReadOnlyTask implements Task {
}
@Override
- public void recordRestoration(final Time time, final long numRecords, final boolean initRemaining) {
+ public void recordRestoration(final Time time, final long numRecords, final long numOffsets, final boolean initRemaining) {
throw new UnsupportedOperationException("This task is read-only");
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
index 1b8f23dae98..018814c4eb5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
@@ -96,7 +96,7 @@ public class StandbyTask extends AbstractTask implements Task {
}
@Override
- public void recordRestoration(final Time time, final long numRecords, final boolean initRemaining) {
+ public void recordRestoration(final Time time, final long numRecords, final long numOffsets, final boolean initRemaining) {
if (initRemaining) {
throw new IllegalStateException("Standby task would not record remaining records to restore");
}
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 fb6ffc604b7..d45d6b50a49 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
@@ -141,6 +141,9 @@ public class StoreChangelogReader implements ChangelogReader {
private long restoreStartTimeNs;
+ // the consumer position at which restoration started, so progress can be measured in offset slots
+ private long restoreStartOffset;
+
private ChangelogMetadata(final StateStoreMetadata storeMetadata, final ProcessorStateManager stateManager) {
this.changelogState = ChangelogState.REGISTERED;
this.storeMetadata = storeMetadata;
@@ -668,6 +671,8 @@ public class StoreChangelogReader implements ChangelogReader {
if (numRecords != 0) {
final List<ConsumerRecord<byte[], byte[]>> records = changelogMetadata.bufferedRecords.subList(0, numRecords);
+ // where restoration had reached before this batch; null until the first batch is restored
+ final Long offsetBeforeRestore = storeMetadata.offset();
final OptionalLong optionalLag = restoreConsumer.currentLag(partition);
stateManager.restore(storeMetadata, records, optionalLag);
@@ -680,9 +685,9 @@ public class StoreChangelogReader implements ChangelogReader {
changelogMetadata.bufferedRecords.clear();
}
- task.recordRestoration(time, numRecords, false);
+ final long currentOffset = storeMetadata.offset();
+ recordRestorationProgress(task, changelogMetadata, numRecords, offsetBeforeRestore, currentOffset + 1);
- final Long currentOffset = storeMetadata.offset();
log.trace("Restored {} records from changelog {} to store {}, end offset is {}, current offset is {}",
numRecords, partition, storeName, recordEndOffset(changelogMetadata.restoreEndOffset), currentOffset);
@@ -710,6 +715,10 @@ public class StoreChangelogReader implements ChangelogReader {
log.info("Finished restoring changelog {} to store {} with a total number of {} records",
partition, storeName, changelogMetadata.totalRestored);
+ // account for any offset slots past the last restored record (e.g. trailing transaction
+ // markers) so the remaining-records metric reaches exactly zero on completion
+ recordRestorationProgress(task, changelogMetadata, 0, storeMetadata.offset(), changelogMetadata.restoreEndOffset);
+
changelogMetadata.transitTo(ChangelogState.COMPLETED);
pauseChangelogsFromRestoreConsumer(Collections.singleton(partition));
if (storeMetadata.store() instanceof MeteredStateStore) {
@@ -730,6 +739,23 @@ public class StoreChangelogReader implements ChangelogReader {
return numRecords;
}
+ /**
+ * Record restoration progress: restore-total/restore-rate advance by the records restored
+ * ({@code numRecords}), while the remaining-records metric is decremented by the offset slots
+ * between {@code lastRestoredOffset} (or {@code restoreStartOffset} if null) and {@code restoredToOffset}.
+ * Measuring the latter in offset slots accounts for offsets the restore consumer never returns
+ * (transaction markers, compacted records) so it reaches exactly zero on completion.
+ */
+ private void recordRestorationProgress(final Task task,
+ final ChangelogMetadata changelogMetadata,
+ 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;
+ final long numOffsets = Math.max(restoredToOffset - restoredFromOffset, 0L);
+ task.recordRestoration(time, numRecords, numOffsets, false);
+ }
+
private Set<Task> getTasksFromPartitions(final Map<TaskId, Task> tasks,
final Set<TopicPartition> partitions) {
final Set<Task> result = new HashSet<>();
@@ -1033,6 +1059,8 @@ public class StoreChangelogReader implements ChangelogReader {
throw new StreamsException("Restore consumer get unexpected error trying to get the position " +
" of " + partition, e);
}
+ // remember where restoration began so progress can be measured in offset slots
+ changelogMetadata.restoreStartOffset = startOffset;
if (changelogMetadata.stateManager.taskType() == Task.TaskType.ACTIVE) {
try {
stateRestoreListener.onRestoreStart(partition, storeName, startOffset, changelogMetadata.restoreEndOffset);
@@ -1045,8 +1073,8 @@ public class StoreChangelogReader implements ChangelogReader {
// if the log is truncated between when we get the log end offset and when we get the
// consumer position, then it's possible that the difference become negative and there's actually
// no records to restore; in this case we just initialize the sensor to zero
- final long recordsToRestore = Math.max(changelogMetadata.restoreEndOffset - startOffset, 0L);
- task.recordRestoration(time, recordsToRestore, true);
+ final long offsetsToRestore = Math.max(changelogMetadata.restoreEndOffset - startOffset, 0L);
+ task.recordRestoration(time, 0, offsetsToRestore, true);
changelogMetadata.restoreStartTimeNs = time.nanoseconds();
} else if (changelogMetadata.stateManager.taskType() == TaskType.STANDBY) {
try {
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index b5bf7361510..9bfca78bd96 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -258,12 +258,12 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator,
}
@Override
- public void recordRestoration(final Time time, final long numRecords, final boolean initRemaining) {
+ public void recordRestoration(final Time time, final long numRecords, final long numOffsets, final boolean initRemaining) {
if (initRemaining) {
- maybeRecordSensor(numRecords, time, restoreRemainingSensor);
+ maybeRecordSensor(numOffsets, time, restoreRemainingSensor);
} else {
maybeRecordSensor(numRecords, time, restoreSensor);
- maybeRecordSensor(-1 * numRecords, time, restoreRemainingSensor);
+ maybeRecordSensor(-1 * numOffsets, time, restoreRemainingSensor);
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
index a54377a3948..78aa0bb3343 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
@@ -223,7 +223,11 @@ public interface Task {
void clearTaskTimeout();
- void recordRestoration(final Time time, final long numRecords, final boolean initRemaining);
+ /**
+ * {@code numRecords} feeds restore-total/restore-rate; {@code numOffsets} feeds the offset-based
+ * remaining-records metric (or initialises it when {@code initRemaining} is true).
+ */
+ void recordRestoration(final Time time, final long numRecords, final long numOffsets, final boolean initRemaining);
// task status inquiry
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
index 84d1f218cfc..e784b81c1b1 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
@@ -484,12 +484,13 @@ public class StandbyTaskTest {
assertThat(totalMetric.metricValue(), equalTo(0.0));
assertThat(rateMetric.metricValue(), equalTo(0.0));
- task.recordRestoration(time, 25L, false);
+ // standby tasks have no remaining-records metric, so the offset-slot argument is ignored
+ task.recordRestoration(time, 25L, 30L, false);
assertThat(totalMetric.metricValue(), equalTo(25.0));
assertThat(rateMetric.metricValue(), not(0.0));
- task.recordRestoration(time, 50L, false);
+ task.recordRestoration(time, 50L, 55L, false);
assertThat(totalMetric.metricValue(), equalTo(75.0));
assertThat(rateMetric.metricValue(), not(0.0));
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 ac0f0127bb8..9cc7718218d 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
@@ -54,6 +54,7 @@ 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.EnumSource;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -94,6 +95,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
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.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -663,6 +665,75 @@ public class StoreChangelogReaderTest {
assertNull(callback.storeNameCalledStates.get(RESTORE_BATCH));
}
+ @Test
+ public void shouldDecrementRemainingRecordsToZeroWithOffsetGaps() {
+ // end offset is 10 but the changelog holds only 8 data records (offsets 0..7); offsets 8 and 9
+ // are transaction markers the restore consumer never returns, so remaining-records is initialized
+ // to 10 offset slots and must still decrement to exactly zero once restoration completes
+ setupActiveStateManager();
+ setupStoreMetadata();
+ setupStore();
+ final TaskId taskId = new TaskId(0, 0);
+
+ // null while preparing (seek-to-beginning, so startOffset == 0) and before the first batch;
+ // once a batch is applied it reflects the last restored record's offset (7)
+ when(storeMetadata.offset()).thenReturn(null).thenReturn(null).thenReturn(7L);
+ when(activeStateManager.taskId()).thenReturn(taskId);
+
+ consumer.updateBeginningOffsets(Collections.singletonMap(tp, 0L));
+ adminClient.updateEndOffsets(Collections.singletonMap(tp, 10L));
+
+ final Task mockTask = mock(Task.class);
+
+ changelogReader.register(tp, activeStateManager);
+
+ // first restore initializes the changelog and records the initial remaining (= 10 slots)
+ changelogReader.restore(Collections.singletonMap(taskId, mockTask));
+ assertEquals(0L, consumer.position(tp));
+ assertEquals(StoreChangelogReader.ChangelogState.RESTORING, changelogReader.changelogMetadata(tp).state());
+
+ for (int offset = 0; offset < 8; offset++) {
+ consumer.addRecord(new ConsumerRecord<>(topicName, 0, offset, "key".getBytes(), "value".getBytes()));
+ }
+
+ // second restore applies the 8 data records
+ changelogReader.restore(Collections.singletonMap(taskId, mockTask));
+ assertEquals(8L, changelogReader.changelogMetadata(tp).totalRestored());
+ assertEquals(StoreChangelogReader.ChangelogState.RESTORING, changelogReader.changelogMetadata(tp).state());
+
+ // skip the trailing transaction-marker offsets (8, 9) so the consumer reaches the end
+ consumer.seek(tp, 10L);
+
+ // third restore observes position >= endOffset and completes restoration
+ changelogReader.restore(Collections.singletonMap(taskId, mockTask));
+ assertEquals(StoreChangelogReader.ChangelogState.COMPLETED, changelogReader.changelogMetadata(tp).state());
+ assertEquals(8L, changelogReader.changelogMetadata(tp).totalRestored());
+
+ // remaining-records is driven by numOffsets (init sets it, later calls decrement it); restore-total
+ // is driven by numRecords, a distinct quantity when the changelog has offset gaps
+ final ArgumentCaptor<Long> numRecordsCaptor = ArgumentCaptor.forClass(Long.class);
+ final ArgumentCaptor<Long> numOffsetsCaptor = ArgumentCaptor.forClass(Long.class);
+ final ArgumentCaptor<Boolean> initCaptor = ArgumentCaptor.forClass(Boolean.class);
+ verify(mockTask, atLeastOnce())
+ .recordRestoration(any(), numRecordsCaptor.capture(), numOffsetsCaptor.capture(), initCaptor.capture());
+
+ long initialRemaining = 0L;
+ long decrementedRemaining = 0L;
+ long totalRecords = 0L;
+ for (int i = 0; i < initCaptor.getAllValues().size(); i++) {
+ if (initCaptor.getAllValues().get(i)) {
+ initialRemaining += numOffsetsCaptor.getAllValues().get(i);
+ } else {
+ decrementedRemaining += numOffsetsCaptor.getAllValues().get(i);
+ totalRecords += numRecordsCaptor.getAllValues().get(i);
+ }
+ }
+
+ assertEquals(10L, initialRemaining, "remaining-records metric should be initialized from the offset range");
+ assertEquals(initialRemaining, decrementedRemaining, "remaining-records metric should decrement to exactly zero");
+ assertEquals(8L, totalRecords, "restore-total should count the records actually restored");
+ }
+
@Test
public void shouldRequestPositionAndHandleTimeoutException() {
setupActiveStateManager();
@@ -700,7 +771,7 @@ public class StoreChangelogReaderTest {
assertEquals(10L, (long) changelogReader.changelogMetadata(tp).endOffset());
Mockito.verify(mockTask).clearTaskTimeout();
Mockito.verify(mockTask).maybeInitTaskTimeoutOrThrow(anyLong(), any());
- Mockito.verify(mockTask).recordRestoration(any(), anyLong(), anyBoolean());
+ Mockito.verify(mockTask).recordRestoration(any(), anyLong(), anyLong(), anyBoolean());
clearException.set(true);
Mockito.reset(mockTask);
@@ -798,7 +869,7 @@ public class StoreChangelogReaderTest {
assertEquals(10L, (long) changelogReader.changelogMetadata(tp).endOffset());
assertEquals(6L, consumer.position(tp));
Mockito.verify(mockTask).clearTaskTimeout();
- Mockito.verify(mockTask).recordRestoration(any(), anyLong(), anyBoolean());
+ Mockito.verify(mockTask).recordRestoration(any(), anyLong(), anyLong(), anyBoolean());
}
@Test
@@ -893,7 +964,7 @@ public class StoreChangelogReaderTest {
assertEquals(6L, consumer.position(tp));
if (type == ACTIVE) {
Mockito.verify(mockTask, times(2)).clearTaskTimeout();
- Mockito.verify(mockTask).recordRestoration(any(), anyLong(), anyBoolean());
+ Mockito.verify(mockTask).recordRestoration(any(), anyLong(), anyLong(), anyBoolean());
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index 247711e6d68..8224c34f2b0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -1142,21 +1142,23 @@ public class StreamTaskTest {
assertThat(rateMetric.metricValue(), equalTo(0.0));
assertThat(remainMetric.metricValue(), equalTo(0.0));
- task.recordRestoration(time, 100L, true);
+ // remaining-records is initialized from the offset range
+ task.recordRestoration(time, 0L, 100L, true);
assertThat(remainMetric.metricValue(), equalTo(100.0));
- task.recordRestoration(time, 25L, false);
+ // restore-total counts records; remaining-records is decremented by offset slots (which may differ)
+ task.recordRestoration(time, 25L, 30L, false);
assertThat(totalMetric.metricValue(), equalTo(25.0));
assertThat(rateMetric.metricValue(), not(0.0));
- assertThat(remainMetric.metricValue(), equalTo(75.0));
+ assertThat(remainMetric.metricValue(), equalTo(70.0));
- task.recordRestoration(time, 50L, false);
+ task.recordRestoration(time, 50L, 55L, false);
assertThat(totalMetric.metricValue(), equalTo(75.0));
assertThat(rateMetric.metricValue(), not(0.0));
- assertThat(remainMetric.metricValue(), equalTo(25.0));
+ assertThat(remainMetric.metricValue(), equalTo(15.0));
}
@Test