Rishavbhattarai commented on code in PR #40866:
URL: https://github.com/apache/beam/pull/40866#discussion_r4403738816
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/datastore/DatastoreV1Test.java:
##########
@@ -1388,6 +1360,41 @@ private static List<Query> splitQuery(Query query, int
numSplits) {
* A WriteBatcher for unit tests, which does no timing-based adjustments (so
unit tests have
* consistent results).
*/
+ /**
+ * Runs {@code fn} over {@code inputs} in a single bundle, in-process. The
fns under test hold
+ * Mockito mocks, which can't be serialized, so they can't go through a
{@code TestPipeline}.
+ */
+ static <InputT, OutputT> List<OutputT> processBundle(
+ DoFn<InputT, OutputT> fn, Iterable<InputT> inputs) {
+ List<OutputT> outputs = new ArrayList<>();
+ DoFnRunner<InputT, OutputT> runner =
+ DoFnRunners.simpleRunner(
+ PipelineOptionsFactory.create(),
+ fn,
+ NullSideInputReader.empty(),
+ new WindowedValueMultiReceiver() {
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> void output(TupleTag<T> tag, WindowedValue<T> output)
{
+ outputs.add((OutputT) output.getValue());
+ }
+ },
+ new TupleTag<>(),
+ Collections.emptyList(),
+ null,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.globalDefault(),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+ runner.startBundle();
+ for (InputT input : inputs) {
+ runner.processElement(WindowedValues.valueInGlobalWindow(input));
+ }
+ runner.finishBundle();
+ return outputs;
+ }
+
Review Comment:
Fixed in 994a46b, the Javadoc is back above `FakeWriteBatcher`.
--
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]
