lucasbru commented on code in PR #23339:
URL: https://github.com/apache/kafka/pull/23339#discussion_r4063173094
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -825,7 +843,11 @@ private Map<TopicPartition, Long>
committedOffsetForChangelogs(final Map<TaskId,
}
try {
- // those which do not have a committed offset would default to 0
+ // partitions with no committed offset are mapped to
NO_COMMITTED_OFFSET (-1) rather than 0, so
+ // callers can tell "the group has not committed here" apart from
a genuine committed offset of 0.
+ // Coercing a missing offset straight to 0 here would let a
transiently missing committed offset
+ // (e.g. the group coordinator is momentarily
unavailable/resigning during a broker restart)
Review Comment:
This still implies that this is actually possible, but I thought we had
established it wasn't possible?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -979,7 +1041,13 @@ private void initializeChangelogs(final Map<TaskId, Task>
tasks,
addChangelogsToRestoreConsumer(newPartitionsToRestore.stream().map(metadata ->
metadata.storeMetadata.changelogPartition())
.collect(Collectors.toSet()));
- newPartitionsToRestore.forEach(metadata ->
metadata.transitTo(ChangelogState.RESTORING));
+ newPartitionsToRestore.forEach(metadata -> {
+ final TopicPartition partition =
metadata.storeMetadata.changelogPartition();
+ metadata.transitTo(ChangelogState.RESTORING);
+ // remember that this changelog has begun restoring, so that if
its task is later corrupted and its
+ // store wiped, a subsequently missing committed offset is retried
rather than truncating restore to 0
+ everStartedRestoringChangelogs.add(partition);
Review Comment:
This adds the partition to everStartedRestoringChangelogs unconditionally,
including for a brand-new source changelog that resolved its ceiling to 0 via
the fast path (empty group, never had real data). If that task is later
wiped/corrupted for an unrelated reason and re-registers while the group
genuinely still has no committed offset, resolveRestoreCeilingCommittedOffset
will now take the retry branch instead of the fast path, stalling that task for
up to task.timeout.ms even though there was never anything to protect. Should
this only be added when we're retrying due to a missing offset (i.e. mirror the
committedOffsetMissingSinceMs bookkeeping) rather than on every transition to
RESTORING?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -851,6 +873,44 @@ private Map<TopicPartition, Long>
committedOffsetForChangelogs(final Map<TaskId,
}
}
+ // The committed-offset operand of the restore ceiling min(endOffset,
committedOffset). A dedicated changelog
+ // is unbounded (Long.MAX_VALUE); a source changelog gates a missing
committed offset on whether it has
+ // restored before (see everStartedRestoringChangelogs). Fallback runs
only once the end offset is known, so
+ // a ceiling is never resolved from a half-fetched pair.
Review Comment:
Can you check if this comment makes sense to you (it does not to me).
Should it be a javadoc comment instead?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -890,7 +950,8 @@ private void updateLimitOffsetsForStandbyChangelogs(final
Map<TopicPartition, Lo
metadata.stateManager.changelogAsSource(partition) &&
committedOffsets.containsKey(partition)) {
- final Long newLimit = committedOffsets.get(partition);
+ // no committed offset (NO_COMMITTED_OFFSET) means a standby
limit of 0, i.e. apply nothing yet
+ final Long newLimit =
Math.max(committedOffsets.get(partition), 0L);
Review Comment:
This clamp only covers the case where newLimit ends up below 0, but it
doesn't protect against the monotonicity check a few lines down. If a standby's
restoreEndOffset has already advanced past 0 (say to 100) and the coordinator
transiently returns no committed offset, newLimit becomes 0 while previousLimit
is 100, and `previousLimit > newLimit` throws IllegalStateException, killing
the StreamThread. Isn't that the same transient-coordinator-hiccup case the PR
is trying to make retriable, just hitting standbys instead of actives?
--
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]
