je-ik commented on code in PR #39761:
URL: https://github.com/apache/beam/pull/39761#discussion_r3791286447
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java:
##########
@@ -79,6 +80,9 @@ class UnboundedReadProcessor<T, CheckpointT extends
CheckpointMark>
/** How often the source is polled. */
private static final Duration POLL_INTERVAL = Duration.ofMillis(50);
+ /** How many elements pass between checks of the turn's deadline. */
Review Comment:
I don't fully understand this comment, is it _INTERVAL or number of elements?
##########
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:
Why do we user nanoTime() here? It is more expensive than
currentTimeMillis() and tje millisecond precision should be okay, no?
##########
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
Review Comment:
This actually sounds like a bug in Kafka Streams, no?
--
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]
