junaiddshaukat commented on code in PR #39611:
URL: https://github.com/apache/beam/pull/39611#discussion_r3712628279
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadTranslator.java:
##########
@@ -77,25 +78,99 @@ public void translate(
// Read produces exactly one output PCollection; downstream consumers are
separate PTransforms
// whose inputs reference this PCollection id and are wired by their own
translators.
String outputPCollectionId =
Iterables.getOnlyElement(transform.getOutputsMap().values());
+ RunnerApi.ReadPayload payload = readPayload(transform);
+ // The same URN carries both kinds of source; the payload says which, and
they need different
+ // processors. A bounded source is drained once and ends time; an
unbounded one is polled
+ // forever and moves the watermark as its reader reports progress.
+ if (payload.getIsBounded() == RunnerApi.IsBounded.Enum.UNBOUNDED) {
+ addUnboundedReadNodes(
+ transformId,
+ ReadTranslation.unboundedSourceFromProto(payload),
+ pipeline.getComponents(),
+ outputPCollectionId,
+ context);
+ return;
+ }
addReadNodes(
transformId,
- boundedSource(transform),
+ boundedSource(payload, transform),
Review Comment:
Done — the wrapper is gone and one try/catch now covers parsing the payload
and hydrating either kind of source.
--
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]
