ramu11 commented on code in PR #23298:
URL: https://github.com/apache/kafka/pull/23298#discussion_r4048628863
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -774,6 +780,42 @@ private int restoreChangelog(final Task task, final
ChangelogMetadata changelogM
return numRecords;
}
+ /**
+ * Advance a standby store's restored offset only once restoration has
actually reached the end
+ * offset — the limit offset for changelogs piggy-backed on a source
topic. A zero-record poll is
+ * not sufficient: retention seeks, source buffering, and incomplete
fetches can leave the consumer
+ * at a non-zero position while records remain to apply.
+ */
+ private void maybeAdvanceStandbyRestoredOffset(final ProcessorStateManager
stateManager,
+ final ChangelogMetadata
changelogMetadata,
+ final StateStoreMetadata
storeMetadata,
+ final TopicPartition
partition) {
+ if (!changelogMetadata.bufferedRecords().isEmpty()) {
+ return;
+ }
+ try {
+ final Long restoreEndOffset = changelogMetadata.restoreEndOffset;
+ if (restoreEndOffset == null) {
+ // dedicated changelog (no restoreEndOffset): lag == 0 with an
empty buffer means the
+ // consumer is at the log-end offset with every fetched record
applied; position() is
+ // 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));
+ }
+ } 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);
+ if (position >= restoreEndOffset) {
Review Comment:
@bbejeck Thanks for the Review, Added a test covering the `position >=
restoreEndOffset` true branch and verifying that `advanceRestoredOffsetTo` is
executed. The new test passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
