junaiddshaukat opened a new pull request, #39249:
URL: https://github.com/apache/beam/pull/39249
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.
--
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]
