bbejeck commented on code in PR #23086:
URL: https://github.com/apache/kafka/pull/23086#discussion_r3768924503
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -205,13 +206,135 @@ long calculateRestoreTime(final long restoreEndTimeNs) {
private static final long DEFAULT_OFFSET_UPDATE_MS =
Duration.ofMinutes(5L).toMillis();
+ // Windows the probe reads back from each partition's end. Not 1: under
EOS the last offset is
+ // usually a transaction control record, which is never delivered to a
consumer, so probing
+ // there is a guaranteed empty poll. The first window answers the large
majority; only the
+ // partitions it cannot answer for pay to widen.
+ private static final long[] PROBE_WINDOWS = {128L, 512L, 2048L};
+
+ // One empty poll proves nothing: poll() returns as soon as any fetch
lands, so a partition can
+ // come back empty because another was served first.
+ private static final int PROBE_IDLE_POLLS = 3;
+
+ // Polls' worth of waiting owed to a window before it may be called empty.
A poll returning
+ // immediately has not waited on a fetch, so a run of them is no evidence
-- and widening on
+ // that basis discards the fetch that was about to answer.
+ private static final int PROBE_MIN_WAIT_POLLS = 5;
+
+ // Ceiling per window, so polls that return without waiting cannot spin
the probe.
+ private static final int PROBE_MAX_POLLS = 30;
+
+ // How long a partition whose probe fell back is left alone. A task
corrupted, wiped and
+ // re-registered in a loop would otherwise probe on every iteration; this
bounds it to one probe
+ // per partition per interval. Only a probe that failed arms it, so a
probe that was working is
+ // never suppressed -- suppressing that would sustain the very loop this
guards against.
+ private static final Duration PROBE_RETRY_BACKOFF = Duration.ofSeconds(60);
+
+ // Without a bound the offset lookups inherit default.api.timeout.ms,
which is minutes, and the
+ // probe is an optimisation that must not hold the restore thread that
long. Generous, because
+ // expiry sends every windowed partition in the batch to log start: the
bound is here to cap a
+ // stall, not to react to a slow broker.
+ private static final Duration OFFSET_LOOKUP_TIMEOUT =
Duration.ofSeconds(60);
Review Comment:
Let's actually remove this variable (and it's comment) and go with the
non-overloaded calls then they will use the same timeouts as the others which
is `default.api.timeout.ms` which is 60 seconds
--
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]
