We are running a simple Data Flow Job that reads data from a Spanner DB and just displays the Changed Record's Mod Type. When we deploy the Job in Dataflow it always fails stating, the below error,
Error message from worker: java.lang.UnsupportedOperationException: BundleFinalizer unsupported by non-portable Dataflow.\n org.apache.beam.runners.dataflow.worker.SplittableProcessFnFactory$SplittableDoFnRunnerFactory.lambda$createRunner$2(SplittableProcessFnFactory.java:172)\n org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.OutputAndTimeBoundedSplittableProcessElementInvoker$1.bundleFinalizer(OutputAndTimeBoundedSplittableProcessElementInvoker.java:206)\n org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.ReadChangeStreamPartitionDoFn$DoFnInvoker.invokeProcessElement(Unknown Source)\n org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.OutputAndTimeBoundedSplittableProcessElementInvoker.invokeProcessElement(OutputAndTimeBoundedSplittableProcessElementInvoker.java:125)\n org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SplittableParDoViaKeyedWorkItems$ProcessFn.processElement(SplittableParDoViaKeyedWorkItems.java:567)
Below is the source code,
public static void main(String[] args) {
PipelineOptions pipelineOptions = PipelineOptionsFactory.fromArgs(args).withValidation().create();
Pipeline pipeline = Pipeline.create(pipelineOptions);
SpannerConfig spannerConfig = SpannerConfig.create().withProjectId("prj-test-1234")
.withInstanceId("dbinstance1").withDatabaseId("dbtestnew");
PCollection<String> changeRecord = pipeline
.apply(SpannerIO.readChangeStream().withSpannerConfig(spannerConfig)
.withChangeStreamName("dbteststream").withMetadataDatabase("testmetadata"))
.apply(ParDo.of(new DoFn<DataChangeRecord, String>() {
@ProcessElement
public void process(ProcessContext context) {
System.out.println("context " + context.element().getModType());
context.output(context.element().getModType().name());
}
}));
pipeline.run();
}