[GSoC 2026] Kafka Streams runner: bounded Read (primitive) + Create support by junaiddshaukat · Pull Request #39249 · apache/beam

junaiddshaukat · 2026-07-08T19:17:03Z

Summary

Adds bounded primitive Read support to the Kafka Streams runner, which also unblocks Create.of(...) with two or more elements.

Part of #39192 (this is part b -- Create support -- implemented via the Read primitive, as discussed with @je-ik).

Background

Create of 2+ elements expands to Read.from(CreateSource), which in the Fn API path becomes a splittable-DoFn bounded read the runner can't run yet. Per the Slack discussion with je-ik, this uses the deprecated primitive Read instead, and forces the conversion so it doesn't depend on the use_deprecated_read experiment.

What's here:

ReadTranslator -- parses the ReadPayload, deserializes the BoundedSource, and
adds a bootstrap source + ReadProcessor + a fired-state store, same shape as
the Impulse translator.
ReadProcessor -- reads the whole source once on a one-shot wall-clock
punctuator, emitting one data payload per element and a terminal MAX watermark.
Single-instance, no splitting for now.
KafkaStreamsTestRunner.translate applies
SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads
unconditionally, so Read.Bounded (and Create) translate as the primitive Read
regardless of the experiment.
Tests: ReadTest (CountingSource) and CreateTest (Create.of(1, 2, 3)),
both end to end through the in-process EMBEDDED harness.

One thing that bit me -- the wire form. A primitive Read hands the runner decoded Java objects, but the SDK harness input receiver expects each element in the runner-side wire form: a raw object for a coder the runner knows (VarLongCoder worked directly), but a length-prefixed byte[] for one it doesn't (VarIntCoder threw ClassCastException: Integer cannot be cast to [B). Stage-to-stage edges already carry that form because harness outputs are decoded with the runner-side wire coder; the source edge doesn't. So ReadProcessor transcodes each element through the SDK-side wire coder (encode) and back through the runner-side wire coder (decode) -- they're byte-compatible by construction and this also handles nested coders.

Follow-up, not in this PR: the force-conversion is wired into the test translate path (KafkaStreamsTestRunner) only. The production runner's translate path still needs the same conversion (or use_deprecated_read defaulted in its options). I can do that here or as a separate change -- whichever you prefer.

cc: @je-ik

…upport Add a translator for the deprecated primitive Read (beam:transform:read:v1) over a BoundedSource, and force every Read.Bounded -- including the one that Create of two or more elements expands to -- into that primitive read instead of the default BoundedSourceAsSDFWrapperFn splittable-DoFn expansion the runner cannot execute yet (no SDF restriction protocol). - ReadTranslator: parses the ReadPayload, deserializes the BoundedSource, and adds a bootstrap source + ReadProcessor + fired-state store (mirrors Impulse). - ReadProcessor: reads the whole source single-instance on a one-shot wall-clock punctuator, emitting one data payload per element plus a terminal MAX watermark. Each element is transcoded into the runner-side wire form the SDK harness input expects -- a raw object for a model coder, a length-prefixed byte[] for a coder the runner does not know -- so decoded source objects bridge correctly into the fused ExecutableStage. - KafkaStreamsTestRunner.translate applies SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads unconditionally, so Create/Read translate as the primitive Read regardless of the use_deprecated_read experiment. Tests: ReadTest (CountingSource.upTo(5)) and CreateTest (Create.of(1, 2, 3)) run end to end through the in-process EMBEDDED harness.

gemini-code-assist · 2026-07-08T19:17:22Z

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces support for bounded primitive Read operations in the Kafka Streams runner. By implementing a custom translator and processor, the runner can now handle Read.Bounded transforms directly, which also resolves issues with Create transforms containing multiple elements that previously failed due to unsupported splittable-DoFn expansion. The implementation ensures reliable element emission and watermark handling while maintaining compatibility with the SDK harness wire format.

Highlights

Primitive Read Support: Added a ReadTranslator and ReadProcessor to implement bounded primitive Read support in the Kafka Streams runner, bypassing the need for splittable-DoFn expansion.
Create Support: Enabled support for Create.of(...) with two or more elements by leveraging the new primitive Read implementation.
Test Coverage: Added end-to-end tests ReadTest and CreateTest to verify the new functionality within the EMBEDDED harness.

New Features

🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

gemini-code-assist

Code Review

This pull request introduces support for translating the deprecated primitive Read transform over a BoundedSource in the Kafka Streams runner. It adds ReadProcessor and ReadTranslator to handle reading from bounded sources and emitting elements in the runner-side wire form, and updates KafkaStreamsTestRunner to force bounded reads into this path. The review feedback suggests wrapping PipelineOptions in SerializablePipelineOptions to avoid serialization issues in distributed environments, and creating a reusable EMPTY_BYTE_ARRAY constant to prevent allocating a new empty byte array for every element and watermark.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

gemini-code-assist · 2026-07-08T19:18:56Z

+  private static final Duration PUNCTUATION_DELAY = Duration.ofMillis(50);
+
+  private final BoundedSource<T> source;
+  private final PipelineOptions options;


PipelineOptions is not serializable. Since ReadProcessor is instantiated inside a serialized ProcessorSupplier lambda (or is itself serialized), storing PipelineOptions directly will cause a NotSerializableException in distributed environments. Use SerializablePipelineOptions instead.

Suggested change

private final PipelineOptions options;

private final org.apache.beam.sdk.options.SerializablePipelineOptions options;

gemini-code-assist · 2026-07-08T19:18:57Z

+
+  ReadProcessor(
+      BoundedSource<T> source,
+      PipelineOptions options,


Update the constructor parameter to accept SerializablePipelineOptions.

Suggested change

PipelineOptions options,

org.apache.beam.sdk.options.SerializablePipelineOptions options,

Done. One note, the class is org.apache.beam.runners.core.construction.SerializablePipelineOptions, not sdk.options. Worth flagging that KS builds the topology per instance and doesn't serialize processor suppliers, so this wasn't a live NotSerializableException, but the wrapper is cheap and keeps the captured state serializable.

gemini-code-assist · 2026-07-08T19:18:57Z

+  /** Reads the whole bounded source and forwards each element, in wire form, as a data payload. */
+  private int readAndForward(ProcessorContext<byte[], KStreamsPayload<?>> ctx) {
+    int count = 0;
+    try (BoundedReader<T> reader = source.createReader(options)) {


Retrieve the underlying PipelineOptions using options.get().

Suggested change

try (BoundedReader<T> reader = source.createReader(options)) {

try (BoundedReader<T> reader = source.createReader(options.get())) {

gemini-code-assist · 2026-07-08T19:18:57Z

+    String sourceNodeName = transformId + SOURCE_SUFFIX;
+    String stateStoreName = transformId + STATE_STORE_SUFFIX;
+    String bootstrapTopic = context.getReadBootstrapTopic(transformId);
+    PipelineOptions options = context.getPipelineOptions();


Wrap the PipelineOptions in SerializablePipelineOptions before passing it to the ReadProcessor constructor to ensure serializability.

Suggested change

PipelineOptions options = context.getPipelineOptions();

org.apache.beam.sdk.options.SerializablePipelineOptions options =

new org.apache.beam.sdk.options.SerializablePipelineOptions(context.getPipelineOptions());

gemini-code-assist · 2026-07-08T19:18:57Z

+  static final String FIRED_KEY = "fired";
+
+  /** How soon after {@link #init} the punctuator first fires. */
+  private static final Duration PUNCTUATION_DELAY = Duration.ofMillis(50);


To avoid creating a new empty byte array for every single element in the source (and for watermarks), we should define a reusable static final constant for the empty byte array.

Suggested change

private static final Duration PUNCTUATION_DELAY = Duration.ofMillis(50);

private static final Duration PUNCTUATION_DELAY = Duration.ofMillis(50);

private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];

gemini-code-assist · 2026-07-08T19:18:57Z

+        ctx.forward(
+            new Record<byte[], KStreamsPayload<?>>(
+                new byte[0], KStreamsPayload.data(toRunnerWire(element)), 0L));


Use the reusable EMPTY_BYTE_ARRAY constant instead of allocating a new byte[0] for every element.

Suggested change

ctx.forward(

new Record<byte[], KStreamsPayload<?>>(

new byte[0], KStreamsPayload.data(toRunnerWire(element)), 0L));

ctx.forward(

new Record<byte[], KStreamsPayload<?>>(

EMPTY_BYTE_ARRAY, KStreamsPayload.data(toRunnerWire(element)), 0L));

gemini-code-assist · 2026-07-08T19:18:57Z

+    ctx.forward(
+        new Record<byte[], KStreamsPayload<?>>(
+            new byte[0], KStreamsPayload.<Object>watermark(maxMillis, 0, 1), 0L));


Use the reusable EMPTY_BYTE_ARRAY constant instead of allocating a new byte[0].

Suggested change

ctx.forward(

new Record<byte[], KStreamsPayload<?>>(

new byte[0], KStreamsPayload.<Object>watermark(maxMillis, 0, 1), 0L));

ctx.forward(

new Record<byte[], KStreamsPayload<?>>(

EMPTY_BYTE_ARRAY, KStreamsPayload.<Object>watermark(maxMillis, 0, 1), 0L));

Wrap the options in SerializablePipelineOptions (the Beam idiom for a source reader that holds options) so ReadProcessor's captured state is uniformly serializable alongside the BoundedSource and coders. Kafka Streams builds the topology per instance and does not serialize processor suppliers, so this is not a live NotSerializableException, but the wrapper is cheap and matches how the other portable runners hold options in their source readers.

github-actions · 2026-07-08T20:32:59Z

Assigning reviewers:

R: @jrmccluskey added as fallback since no labels match configuration

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

je-ik

There is a bit misunderstanding how Read works. The flow is:
a) call source.split
b) shuffle the splits so that we read in parallel
c) call createReader only on the split sources

From the implementation point, the initial source.split can be driven by preceding Impulse. We can then fan-out all Impulses together so that we need only single "fake" topic.

je-ik · 2026-07-09T08:05:29Z

  private static final String IMPULSE_BOOTSTRAP_TOPIC_PREFIX = "__beam_impulse_";

+  /** Prefix for the per-transform bootstrap topic a primitive Read reads from. */
+  private static final String READ_BOOTSTRAP_TOPIC_PREFIX = "__beam_read_";


Do we need per-transform topic? Could we just fan-out the single element to all Impulse transforms? It would require analysis of the Pipeline DAG before translation and then maybe add s single ParDo that will do the fan-out? Or it could be done implicitly.

je-ik · 2026-07-09T08:17:07Z

There is a bit misunderstanding how Read works. The flow is: a) call source.split b) shuffle the splits so that we read in parallel c) call createReader only on the split sources

From the implementation point, the initial source.split can be driven by preceding Impulse. We can then fan-out all Impulses together so that we need only single "fake" topic.

Ah! I see, we plan to add the splitting in follow-up issue.

je-ik

Because we plan to add the splitting in follow up, we will address parallelism and the ability to seed the Pipeline from single topic there.

github-actions Bot added the runners label Jul 8, 2026

junaiddshaukat changed the title ~~[GSoC 2026] Kafka Streams runner #39192: bounded Read (primitive) + Create support~~ [GSoC 2026] Kafka Streams runner: bounded Read (primitive) + Create support Jul 8, 2026

gemini-code-assist Bot reviewed Jul 8, 2026

View reviewed changes

github-actions Bot added the Next Action: Reviewers label Jul 8, 2026

je-ik requested changes Jul 9, 2026

View reviewed changes

je-ik approved these changes Jul 9, 2026

View reviewed changes

je-ik merged commit 27c4522 into apache:feat/18479-kafka-streams-runner-skeleton Jul 9, 2026
3 checks passed

	private final PipelineOptions options;
	private final org.apache.beam.sdk.options.SerializablePipelineOptions options;

	PipelineOptions options,
	org.apache.beam.sdk.options.SerializablePipelineOptions options,

	try (BoundedReader<T> reader = source.createReader(options)) {
	try (BoundedReader<T> reader = source.createReader(options.get())) {

	PipelineOptions options = context.getPipelineOptions();
	org.apache.beam.sdk.options.SerializablePipelineOptions options =
	new org.apache.beam.sdk.options.SerializablePipelineOptions(context.getPipelineOptions());

	private static final Duration PUNCTUATION_DELAY = Duration.ofMillis(50);
	private static final Duration PUNCTUATION_DELAY = Duration.ofMillis(50);
	private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];

Uh oh!

Conversation

junaiddshaukat commented Jul 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Background

What's here:

Uh oh!

gemini-code-assist Bot commented Jul 8, 2026

Summary of Changes

Highlights

Footnotes

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

junaiddshaukat Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Jul 8, 2026

Uh oh!

je-ik left a comment

Choose a reason for hiding this comment

Uh oh!

je-ik Jul 9, 2026

Choose a reason for hiding this comment

Uh oh!

je-ik commented Jul 9, 2026

Uh oh!

je-ik left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

junaiddshaukat commented Jul 8, 2026 •

edited

Loading