lukecwik commented on code in PR #24025: URL: https://github.com/apache/beam/pull/24025#discussion_r1035252492
########## sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/FlatMapElements.java: ########## @@ -137,57 +132,90 @@ public <NewInputT> FlatMapElements<NewInputT, OutputT> via( @Experimental(Kind.CONTEXTFUL) public <NewInputT> FlatMapElements<NewInputT, OutputT> via( Contextful<Fn<NewInputT, Iterable<OutputT>>> fn) { - return new FlatMapElements<>( - fn, fn.getClosure(), TypeDescriptors.inputOf(fn.getClosure()), outputType); + return new FlatMapElements<>(fn, TypeDescriptors.inputOf(fn.getClosure()), outputType); } @Override public PCollection<OutputT> expand(PCollection<? extends InputT> input) { checkArgument(fn != null, ".via() is required"); - return input.apply( - "FlatMap", - ParDo.of( - new DoFn<InputT, OutputT>() { - @ProcessElement - public void processElement(ProcessContext c) throws Exception { - Iterable<OutputT> res = - fn.getClosure().apply(c.element(), Fn.Context.wrapProcessContext(c)); - for (OutputT output : res) { - c.output(output); + if (fn instanceof Contextful) { Review Comment: Also, I didn't want to wrap with another PTransform that delineated between these two options -- 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: github-unsubscr...@beam.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org