This is an automated email from the ASF dual-hosted git repository.
je-ik pushed a commit to branch feat/18479-kafka-streams-runner-skeleton
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to
refs/heads/feat/18479-kafka-streams-runner-skeleton by this push:
new 65a2e400c73 [GSoC 2026] Kafka Streams runner: bound a source poll in
time, not only in elements (#39761)
65a2e400c73 is described below
commit 65a2e400c736ce207a072918f5bd3899237e5ca3
Author: M Junaid Shaukat <[email protected]>
AuthorDate: Sun Aug 16 22:09:28 2026 +0500
[GSoC 2026] Kafka Streams runner: bound a source poll in time, not only in
elements (#39761)
* [GSoC 2026] Kafka Streams runner: bound a source poll in time, not only
in elements
--readMaxPollTimeMs bounds the turn in time as well; whichever bound comes
first ends it.
---
.../kafka/streams/KafkaStreamsPipelineOptions.java | 17 +++++++
.../kafka/streams/translation/ReadTranslator.java | 2 +
.../translation/UnboundedReadProcessor.java | 49 ++++++++++++++++++--
.../streams/translation/UnboundedReadTest.java | 54 ++++++++++++++++++++++
.../en/documentation/runners/kafkastreams.md | 1 +
5 files changed, 118 insertions(+), 5 deletions(-)
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java
index 180e1ef028f..99454b3a658 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java
@@ -58,6 +58,23 @@ public interface KafkaStreamsPipelineOptions extends
PortablePipelineOptions {
void setReadMaxElementsPerPoll(int readMaxElementsPerPoll);
+ @Description(
+ "How long one turn of reading an unbounded source may take, in
milliseconds, before the"
+ + " source yields the Kafka Streams thread. A source is polled from
a punctuator"
+ + " scheduled every 50ms, and the same thread runs the rest of the
topology, so a turn"
+ + " that overruns that interval is already due again when it returns
and fires straight"
+ + " away: the source then holds the thread and the stages below it
are never scheduled,"
+ + " which shows up as a pipeline that reads steadily and emits
nothing at all rather"
+ + " than one that falls behind. Roughly, the source takes this
fraction of a 50ms"
+ + " interval, so the default of 10ms leaves the thread four fifths
of its time."
+ + " --readMaxElementsPerPoll bounds the same turn by count;
whichever bound is reached"
+ + " first ends it, and a count alone cannot bound the time because
how long an element"
+ + " takes depends on the pipeline below.")
+ @Default.Integer(10)
+ int getReadMaxPollTimeMs();
+
+ void setReadMaxPollTimeMs(int readMaxPollTimeMs);
+
@Description(
"How long the consumer group waits before deciding an instance has gone,
in milliseconds."
+ " This is the floor on how quickly work moves to another instance
after one is lost,"
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java
index b6620203b2d..9a727e82a70 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java
@@ -138,6 +138,7 @@ class ReadTranslator implements PTransformTranslator {
Coder<CheckpointT> checkpointCoder =
readableSource.getCheckpointMarkCoder();
int maxElementsPerPoll =
context.getPipelineOptions().getReadMaxElementsPerPoll();
int checkpointEveryNPolls =
context.getPipelineOptions().getReadCheckpointNumBundles();
+ int maxPollTimeMs = context.getPipelineOptions().getReadMaxPollTimeMs();
topology.addSource(
sourceNodeName,
@@ -157,6 +158,7 @@ class ReadTranslator implements PTransformTranslator {
transformId,
maxElementsPerPoll,
checkpointEveryNPolls,
+ maxPollTimeMs,
context.getTerminationTracker()),
sourceNodeName);
topology.addStateStore(
diff --git
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java
index 96aa1bf0c69..4f8c5f10556 100644
---
a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java
+++
b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java
@@ -79,6 +79,9 @@ class UnboundedReadProcessor<T, CheckpointT extends
CheckpointMark>
/** How often the source is polled. */
private static final Duration POLL_INTERVAL = Duration.ofMillis(50);
+ /** Elements read between two checks of whether the turn is out of time. */
+ private static final int ELEMENTS_BETWEEN_DEADLINE_CHECKS = 64;
+
private final UnboundedSource<T, CheckpointT> source;
private final SerializablePipelineOptions options;
// See ReadProcessor: a source produces decoded objects, but the downstream
stage's harness input
@@ -90,6 +93,7 @@ class UnboundedReadProcessor<T, CheckpointT extends
CheckpointMark>
private final String transformId;
private final int maxElementsPerPoll;
private final int checkpointEveryNPolls;
+ private final int maxPollTimeMs;
private @Nullable ProcessorContext<byte[], KStreamsPayload<?>> context;
private @Nullable KeyValueStore<String, byte[]> checkpointStore;
@@ -115,6 +119,7 @@ class UnboundedReadProcessor<T, CheckpointT extends
CheckpointMark>
String transformId,
int maxElementsPerPoll,
int checkpointEveryNPolls,
+ int maxPollTimeMs,
TerminationTracker terminationTracker) {
this.terminationReporter = new TerminationReporter(terminationTracker,
transformId);
this.source = source;
@@ -126,6 +131,7 @@ class UnboundedReadProcessor<T, CheckpointT extends
CheckpointMark>
this.transformId = transformId;
this.maxElementsPerPoll = maxElementsPerPoll;
this.checkpointEveryNPolls = checkpointEveryNPolls;
+ this.maxPollTimeMs = maxPollTimeMs;
}
@Override
@@ -157,6 +163,16 @@ class UnboundedReadProcessor<T, CheckpointT extends
CheckpointMark>
* 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. A punctuator is
+ * expected to be quick, and this one runs on the thread that also serves
the rest of the
+ * topology, so a turn that overruns its own {@link #POLL_INTERVAL} is due
again as soon as it
+ * returns and runs once more instead of the tasks below it. Measured on a
grouping pipeline, a
+ * turn of 200 elements took 3ms and held the thread 6% of the time, while a
turn of 5000 took
+ * 57ms and held it 89%, and the pipeline read tens of millions of elements
while emitting none.
+ * {@link #maxPollTimeMs} bounds the turn in time as well, and whichever
bound is reached first
+ * ends it.
*/
private void poll() {
if (exhausted) {
@@ -164,8 +180,9 @@ class UnboundedReadProcessor<T, CheckpointT extends
CheckpointMark>
}
ProcessorContext<byte[], KStreamsPayload<?>> ctx =
checkInitialized(context);
UnboundedReader<T> currentReader = ensureReader();
+ long deadline = System.currentTimeMillis() + maxPollTimeMs;
for (int batch = 0; batch < checkpointEveryNPolls; batch++) {
- int emitted = readBatch(ctx, currentReader);
+ int emitted = readBatch(ctx, currentReader, deadline);
Instant watermark = currentReader.getWatermark();
forwardWatermarkIfAdvanced(ctx, watermark);
if (!watermark.isBefore(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
@@ -181,24 +198,46 @@ class UnboundedReadProcessor<T, CheckpointT extends
CheckpointMark>
return;
}
if (emitted < maxElementsPerPoll) {
- // Short batch: the source has nothing more for now, so store what was
read and wait for
- // the next punctuation rather than spinning on a reader that keeps
returning false.
+ // Short batch: either the source has nothing more for now, or the
turn ran out of time.
+ // Either way, store what was read and wait for the next punctuation
rather than spinning
+ // on a reader that keeps returning false.
if (emitted > 0) {
storeCheckpoint(currentReader);
}
return;
}
+ if (System.currentTimeMillis() >= deadline) {
+ // A full batch and the turn is out of time: yield with the position
recorded, so the next
+ // punctuation carries on rather than this one running the thread out
from under the rest
+ // of the topology.
+ storeCheckpoint(currentReader);
+ return;
+ }
}
// Yielded on the batch bound rather than on an empty source, so record
the position reached.
storeCheckpoint(currentReader);
}
- /** Forwards up to {@link #maxElementsPerPoll} elements, returning how many
were available. */
+ /**
+ * Forwards up to {@link #maxElementsPerPoll} elements, returning how many
were available.
+ *
+ * <p>Stops early if the turn's deadline passes, since one batch can be long
enough on its own to
+ * overrun it. The clock is read every {@link
#ELEMENTS_BETWEEN_DEADLINE_CHECKS} elements rather
+ * than every element, which bounds the overshoot to that many elements
without putting a clock
+ * read in front of each one.
+ */
private int readBatch(
- ProcessorContext<byte[], KStreamsPayload<?>> ctx, UnboundedReader<T>
currentReader) {
+ ProcessorContext<byte[], KStreamsPayload<?>> ctx,
+ UnboundedReader<T> currentReader,
+ long deadline) {
int emitted = 0;
try {
while (emitted < maxElementsPerPoll) {
+ if (emitted % ELEMENTS_BETWEEN_DEADLINE_CHECKS == 0
+ && emitted > 0
+ && System.currentTimeMillis() >= deadline) {
+ break;
+ }
// start() positions the reader on its first element; advance() moves
to the next. Either
// returning false means nothing is available right now — not that the
source is finished,
// which is the difference from a bounded read.
diff --git
a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadTest.java
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadTest.java
index b668ee972d5..01e1a3146e3 100644
---
a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadTest.java
+++
b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadTest.java
@@ -21,6 +21,7 @@ import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.lessThan;
import java.io.IOException;
import java.time.Duration;
@@ -122,6 +123,59 @@ public class UnboundedReadTest {
}
}
+ /** A pipeline whose source may read {@code elementsPerPoll} elements and
run for {@code ms}. */
+ private static Pipeline pipelineWithPollBounds(int elementsPerPoll, int
maxPollTimeMs) {
+ KafkaStreamsPipelineOptions options =
+
KafkaStreamsTestRunner.testOptions().as(KafkaStreamsPipelineOptions.class);
+ options.setReadMaxElementsPerPoll(elementsPerPoll);
+ options.setReadMaxPollTimeMs(maxPollTimeMs);
+ Pipeline pipeline = Pipeline.create(options);
+ pipeline
+ .apply("read", Read.from(CountingSource.unbounded()))
+ .apply("record", ParDo.of(new RecordFn()));
+ return pipeline;
+ }
+
+ /** Drives one turn of the wall clock and reports how many elements the
source produced. */
+ private static int elementsInOneTurn(Pipeline pipeline) {
+ KafkaStreamsTranslationContext context =
KafkaStreamsTestRunner.translate(pipeline);
+ try (TopologyTestDriver driver =
+ new TopologyTestDriver(
+ context.getTopology(),
KafkaStreamsTestRunner.streamsConfig(pipeline))) {
+ driver.advanceWallClockTime(Duration.ofMillis(100));
+ }
+ return RECEIVED.size();
+ }
+
+ /**
+ * The element bound cannot bound the time a turn takes, because how long an
element takes is
+ * decided by the pipeline below the source. Left on the count alone, a
source with data always
+ * available runs the full count every turn, overruns the punctuation
interval, and is due again
+ * the moment it returns — so it keeps the thread and the rest of the
topology never runs.
+ */
+ @Test
+ public void aPollOutOfTimeYieldsBeforeReachingItsElementBound() {
+ // A turn that is out of time before it starts, so what stops it can only
be the time bound.
+ int elements = elementsInOneTurn(pipelineWithPollBounds(1_000, 0));
+
+ assertThat(
+ "the source should have yielded, not run to its element bound",
+ elements,
+ is(lessThan(1_000)));
+ assertThat("the source should still have made progress", elements,
is(greaterThan(0)));
+ }
+
+ /** The time bound only cuts a turn short; with time to spare the element
bound still applies. */
+ @Test
+ public void aPollWithTimeToSpareReachesItsElementBound() {
+ int elements = elementsInOneTurn(pipelineWithPollBounds(100, 60_000));
+
+ assertThat(
+ "a turn with time to spare should read at least a full batch",
+ elements,
+ is(greaterThan(99)));
+ }
+
@Test
public void aSourceThatReachesTheEndOfTimeStopsBeingPolled() {
// CountingSource.unbounded() with a limit reports the terminal watermark
once it has produced
diff --git a/website/www/site/content/en/documentation/runners/kafkastreams.md
b/website/www/site/content/en/documentation/runners/kafkastreams.md
index bef1f852094..058e85b6d46 100644
--- a/website/www/site/content/en/documentation/runners/kafkastreams.md
+++ b/website/www/site/content/en/documentation/runners/kafkastreams.md
@@ -119,6 +119,7 @@ Named as Java spells them below; from Python the same
options are in snake case,
| `topicReplicationFactor` | `1` | Replication factor for those topics. |
| `maxBundleSize` | `1000` | Elements per bundle, and elements taken per poll
of an unbounded source. |
| `maxBundleTimeMs` | `1000` | Intended cap on how long a bundle may stay
open. **Not applied yet** — see below. |
+| `readMaxPollTimeMs` | `10` | How long one turn of reading an unbounded
source may take before it yields the Kafka Streams thread. A source is polled
every 50ms and shares its thread with the rest of the topology, so a turn that
overruns that interval leaves the stages below it unscheduled; a bound on
elements alone cannot bound the time. |
| `readCheckpointNumBundles` | `10` | Polls of an unbounded source between
stores of its checkpoint mark. Larger values replay more after a restart. |
| `stateDir` | temp directory | Where Kafka Streams keeps local state. |
