junaiddshaukat commented on code in PR #39761:
URL: https://github.com/apache/beam/pull/39761#discussion_r3791482333
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java:
##########
@@ -157,15 +164,24 @@ public void process(Record<byte[], byte[]> record) {
* Streams thread would never get back to committing or to the rest of the
topology. So at most
* {@link #checkpointEveryNPolls} batches are taken before yielding, which
is also where the
* checkpoint mark is stored, and the next punctuation carries on from there.
+ *
+ * <p>That batch bound is a count, and a count cannot bound the time: how
long an element takes is
+ * decided by the pipeline underneath it, which the source knows nothing
about. Since this runs on
+ * the thread that also serves the rest of the topology, a turn that
overruns the punctuation
+ * interval is already due again when it returns and fires straight away, so
the source keeps the
+ * thread and the stages below it never run — a pipeline reading steadily
and emitting nothing,
+ * rather than one falling behind. {@link #maxPollTimeMs} bounds the turn in
time as well, and
+ * whichever bound is reached first ends it.
*/
private void poll() {
if (exhausted) {
return;
}
ProcessorContext<byte[], KStreamsPayload<?>> ctx =
checkInitialized(context);
UnboundedReader<T> currentReader = ensureReader();
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(maxPollTimeMs);
Review Comment:
All three done.
nanoTime → currentTimeMillis in both places. You are right that millisecond
precision is enough for a bound this size, and it is the cheaper call. The
tests still catch the bound being disabled after the change, so the behaviour
is unchanged.
The constant is renamed ELEMENTS_BETWEEN_DEADLINE_CHECKS. Your reading was
the right one — the value is a count of elements, not a time, and the old name
said the opposite.
On whether it is a Kafka Streams bug: I have taken the claim out of the
comment, because I do not think I can support it. What I can say is what I
measured — a turn of 200 elements took 3ms and held the thread 6% of the time,
a turn of 5000 took 57ms and held it 89% — and that a punctuator is expected to
be quick. Whether Kafka Streams ought to coalesce a punctuation it has already
missed is a fair question, but the runner should not be holding the thread that
long either way, so I would rather fix our side and not assert something about
theirs in a code comment.
I also moved the readMaxPollTimeMs row in the docs into this PR, since it
documents the option this PR adds. It was in #39762 before, which would have
documented an option that did not exist yet.
--
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]
