MINOR: log start and end offsets for state store restoration
Debug loggin of the start and end offsets used during state store restoration
Author: Damian Guy
Reviewers: Guozhang Wang
Closes from dguy/log-restore-offsets
@@ -118,8 +118,14 @@ public class StoreChangelogReader implements ChangelogReader {
for (final StateRestorer restorer : needsRestoring.values()) {
if (restorer.checkpoint() != StateRestorer.NO_CHECKPOINT) {
consumer.seek(restorer.partition(), restorer.checkpoint());
+ logRestoreOffsets(restorer.partition(),
+ restorer.checkpoint(),
+ endOffsets.get(restorer.partition()));
} else {
consumer.seekToBeginning(Collections.singletonList(restorer.partition()));
+ logRestoreOffsets(restorer.partition(),
+ consumer.position(restorer.partition()),
+ endOffsets.get(restorer.partition()));
}
}
@@ -137,6 +143,13 @@ public class StoreChangelogReader implements ChangelogReader {
}
}
+ private void logRestoreOffsets(final TopicPartition partition, final long checkpoint, final Long aLong) {
+ log.debug("restoring partition {} from offset {} to endOffset {}",
+ partition,
+ checkpoint,
+ aLong);
+ }
+
@Override
public Map<TopicPartition, Long> restoredOffsets() {
final Map<TopicPartition, Long> restoredOffsets = new HashMap<>();
