[GSoC 2026] Kafka Streams runner: read unbounded sources - #39611
Conversation
The runner could only read sources that finish, which is the wrong shape for what it is: a Kafka Streams application is a long-running stream processor, and bounded data has more efficient homes. Bounded and unbounded reads share a URN and are distinguished by the payload, so ReadTranslator branches on it. The unbounded processor polls its reader on a wall-clock punctuator rather than draining it once, since advance() returning false means nothing is available right now rather than that the source is finished, and takes at most maxBundleSize elements per turn so a fast source cannot starve the rest of the topology. Its watermark comes from UnboundedReader#getWatermark() instead of jumping to the end of time when the input runs out, which is what lets downstream windows close on a stream that never finishes. The reader's checkpoint mark is written to a persistent state store after the elements it covers have been forwarded, so it can never claim more progress than was emitted, and the reader is created from the stored mark so a restart resumes where it left off. finalizeCheckpoint is not called yet: a mark should only be finalized once durably committed, which needs a pre-commit hook the runner does not have. The source is also read by a single reader, so splits are not distributed across instances. UnboundedReadTest drives a genuinely unbounded CountingSource and asserts more than one poll's worth of elements arrive, contiguously from zero, which is what separates a polled source from a drained one.
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Assigning reviewers: R: @chamikaramj added as fallback since no labels match configuration Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
je-ik
left a comment
je-ik
left a comment
There was a problem hiding this comment.
+1 overall, we should definitely track the deficiency in splitting, but if we already postponed it for bounded sources, we can live with it for the skeleton runner. We must definitely document these features that are core, but not yet imeplemented.
| addReadNodes( | ||
| transformId, | ||
| boundedSource(transform), | ||
| boundedSource(payload, transform), |
There was a problem hiding this comment.
Seems we can reuse directly ReadTranslation.boundedSourceFromProto().
There was a problem hiding this comment.
Done — the wrapper is gone and one try/catch now covers parsing the payload and hydrating either kind of source.
| // mark can never claim more progress than was actually emitted. | ||
| storeCheckpoint(currentReader); | ||
| } | ||
| forwardWatermarkIfAdvanced(ctx, currentReader.getWatermark()); |
There was a problem hiding this comment.
We should probably immediately reschedule next poll loop (it emitted > 0), otherwise we will limit throughput.
Also, when currentReader.getWatermark() returns >= TIMESTAMP_MAX_VALUE, we should finish the poll loop entirely.
There was a problem hiding this comment.
Both done, and the first one caught a bug of mine. I first made the batches loop while the source kept filling them, which hung the test: CountingSource.unbounded() always has data, so every batch came back full and the loop never returned — the Kafka Streams thread would never have got back to committing. It now takes at most readCheckpointNumBundles batches before yielding, so throughput isn't capped at a batch per interval but the thread still comes back.
The terminal watermark now ends it properly: polling stops and the punctuator is cancelled, rather than spinning on a reader that can only return false. There's a test for it.
| if (emitted > 0) { | ||
| // Record the position only after the elements it covers have been forwarded, so the stored | ||
| // mark can never claim more progress than was actually emitted. | ||
| storeCheckpoint(currentReader); |
There was a problem hiding this comment.
Extracting the checkpoint might be a little expensive. We might not do it that often - how about adding a config option (e.g. readCheckpointNumBundles), which will control how many bundles (non-empty calls to poll() should occur before we do a checkpoint).
There was a problem hiding this comment.
Added --readCheckpointNumBundles, default 10. It doubles as the batch bound above, which fits — that's the point where the loop yields anyway, so it's a natural place to store the mark. Documented the trade: a larger value replays more after a restart, since the reader resumes from the last mark stored.
…ore reading Reuses ReadTranslation.boundedSourceFromProto directly rather than wrapping it, with one try/catch covering parsing and hydration for both kinds of source. Polls in batches that run back to back rather than returning after one, since waiting for the next punctuation after every batch capped throughput at a batch per interval. The run is bounded all the same: a source that always has data would otherwise never let the poll return and the Kafka Streams thread would never get back to committing or to the rest of the topology, so at most readCheckpointNumBundles batches are taken before yielding. Polling also stops entirely, and the punctuator is cancelled, once the reader's watermark reaches the end of time, which is the source saying it will produce nothing further. Adds --readCheckpointNumBundles, since taking a checkpoint mark can be costly and is not worth doing on every batch. The cost of a larger value is that more elements are replayed after a restart, because the reader resumes from the last mark stored. Asks the source to split before creating a reader. A source is not obliged to be readable in its unsplit form, and split() is where several of them do their setup, so going through it even for a single reader is the supported path. Adds a test that a source reaching the end of time stops being polled and delivers each element exactly once.
A source is not obliged to be readable in its unsplit form — split() is where several of them do their setup — so the reader is now created from source.split(1, options) rather than from the source directly. One split, because this processor is a single instance; distributing several splits across instances is tracked separately. This was described in the previous review reply but was not actually in the code: the edit did not apply and the claim went out before it was verified.
Splitting was being done inside the processor, which is the wrong place: it would run once per task instance rather than once for the pipeline, and the contract does not define splitting a source that has already been split. It now happens in ReadTranslator, and the processor is handed a source that is ready to read. The count passed to split() is only a hint, so what comes back is checked. Taking the first of several splits and ignoring the rest would quietly drop whatever those parts would have produced, which is data loss rather than a missing feature, so translation fails with an explanation instead. Reading several splits in parallel is still not supported. UnboundedReadTest covers it with a source that returns two splits whatever it is asked for.
5861f31
into
apache:feat/18479-kafka-streams-runner-skeleton
2 participants
Summary
Part of #18479.
The runner could only read sources that finish. That is the wrong shape for what it is: a Kafka Streams application is a long-running stream processor, and a pipeline over bounded data has more efficient homes. This adds
UnboundedSourcesupport, so the runner can read a source that never ends.How it differs from the bounded read
Bounded and unbounded reads share the
beam:transform:read:v1URN and are told apart byReadPayload.getIsBounded(), soReadTranslatornow branches there. What sits behind the branch is genuinely different:advance()returning false means nothing is available right now, not that the source is finished, so the reader is asked again on each turn of a wall-clock punctuator. A poll takes at most--maxBundleSizeelements, so a fast source cannot monopolise the Kafka Streams thread and starve the rest of the topology.UnboundedReader#getWatermark()is forwarded whenever it advances, which is what lets downstream windows close on a stream that never finishes.Resuming after a restart
UnboundedReader#getCheckpointMark()describes the position the reader has consumed to. It is written to a persistent state store, and the reader is created from the stored mark rather than from scratch, so a task that restarts or moves resumes where it left off instead of re-reading from the beginning.The mark is written after the elements it covers have been forwarded, so it can never claim more progress than was actually emitted. The store is changelogged and, under exactly-once, its writes commit atomically with the records the processor forwarded.
What this does not do yet
finalizeCheckpoint()is not called. A mark should be finalized once it is durably committed, which needs a pre-commit hook the runner does not have — the same gap that stopped the bundle time bound in [GSoC 2026] Kafka Streams runner: bound a bundle by element count #39578. Sources that rely on finalization to acknowledge or release data will not see it. Worth doing as its own change, once that hook exists.Testing
UnboundedReadTestruns a pipeline overCountingSource.unbounded(). Nothing caps the source — capping it withwithMaxNumRecordswould turn it back into a bounded read and test the wrong path — so the work is bounded by the per-poll element limit and the number of turns the test drives.It asserts more than a single poll's worth of elements arrive, which is the property that separates this from a bounded read: the source has to be asked again on each turn and carry on from where it was. The elements are also checked to be contiguous from zero, so a poll neither skips nor repeats what the previous one consumed.