[Bug]: Running Java pipeline with JDBCIO when using a window causes a ClassCastException

[sshahar1]

What happened?

I have a pipeline that reads data, windows it, groups, aggregates and write to a relational DB
Code below in Kotlin

pipeline
    .apply(
        "WaitForNewFiles",
        FileIO.match()
            .filepattern(inputPath)
            .continuously(
                // Check for new files every 2 minute
                Duration.standardMinutes(2),
                // Never stop checking for new files
                Watch.Growth.never()
            )
    )
    .apply("GetFiles", FileIO.readMatches())
    .apply(TextIO.readFiles())
    .apply("ToTuple", ParDo.of(toTuple))
    .apply(
        "WaitForLines",
        Window.into<T>(FixedWindows.of(Duration.standardMinutes(1)))
            .triggering(AfterWatermark.pastEndOfWindow())
            .discardingFiredPanes()
            .withAllowedLateness(Duration.ZERO)
    ).apply("Filter", Filter.by(Matcher()))
        .apply("Flatten", ParDo.of(FlattenRecords()))
        .setCoder(AvroCoder.of(MyRecord::class.java))
        .apply("Insert to database",
            JdbcIO.write<MyRecord>()
                .withDataSourceProviderFn(dsProvider)
                .withStatement(cloud_sql_rome_records_insert)
                .withPreparedStatementSetter(preparedSetter()))

Up until upgrading to Apache Beam 2.37 it worked, but after an upgrade we started getting the following Exceptions (and alot of them):

java.lang.ClassCastException: class org.apache.beam.sdk.transforms.windowing.GlobalWindow cannot be cast to class org.apache.beam.sdk.transforms.windowing.IntervalWindow (org.apache.beam.sdk.transforms.windowing.GlobalWindow and org.apache.beam.sdk.transforms.windowing.IntervalWindow are in unnamed module of loader 'app')
at org.apache.beam.sdk.transforms.windowing.IntervalWindow$IntervalWindowCoder.registerByteSizeObserver(IntervalWindow.java:142)
at org.apache.beam.sdk.coders.IterableLikeCoder.registerByteSizeObserver(IterableLikeCoder.java:211)
at org.apache.beam.sdk.coders.IterableLikeCoder.registerByteSizeObserver(IterableLikeCoder.java:60)
at org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.registerByteSizeObserver(WindowedValue.java:640)
at org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.registerByteSizeObserver(WindowedValue.java:558)
at org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$ElementByteSizeObservableCoder.registerByteSizeObserver(IntrinsicMapTaskExecutorFactory.java:403)
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputObjectAndByteCounter.update(OutputObjectAndByteCounter.java:128)
at org.apache.beam.runners.dataflow.worker.DataflowOutputCounter.update(DataflowOutputCounter.java:67)
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:43)
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:285)
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:268)
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$900(SimpleDoFnRunner.java:84)
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnFinishBundleArgumentProvider$Context.output(SimpleDoFnRunner.java:326)
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnFinishBundleArgumentProvider$Context.output(SimpleDoFnRunner.java:321)
at org.apache.beam.sdk.io.jdbc.JdbcIO$1.finish(JdbcIO.java:1613)

The new code introduced in 2.37 in JDBCIO includes:

@FinishBundle
                    public void finish(FinishBundleContext c) {
                      if (outputList != null && outputList.size() > 0) {
                        c.output(outputList, Instant.now(), GlobalWindow.INSTANCE);
                      }
                      outputList = null;
                    }

beam/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcIO.java

Line 1610 in d97c521

@FinishBundle

So, from what I understand the time interval and global interval conflict.

This is a very vague exception obviously and should be addressed in documentation.

Thanks in advance

Issue Priority

Priority: 2

Issue Component

Component: io-java-jdbc

[sshahar1]

I would highly appreciate to hear recommendations or opinions on this issue

[Abacn]

Not familiar with kotlin, but cannot reproduce this with Java code snippet, using either release-2.39.0 or master. Setting a log in the referred method, confirmed that the method was evaluated.

@Test
  public void TestWriteUsingWindow() throws Exception {
    String tableName = DatabaseTestHelper.getTestTableName("UT_WRITE");
    DatabaseTestHelper.createTable(DATA_SOURCE, tableName);

    pipeline.apply("Generate Periodic Impulse",
            PeriodicImpulse.create().startAt(Instant.now()).stopAt(Instant.now().plus(Duration.standardSeconds(14))).withInterval(Duration.standardSeconds(5)))
        .apply("Generate Records", ParDo.of(
            new DoFn<Instant, KV<Integer, String>>() {
              @ProcessElement
              public void processElement(ProcessContext context) {
                ArrayList<KV<Integer, String>> data = getDataToWrite(10);
                for(KV<Integer, String> datum: data) {
                  context.output(datum);
                }
              }
            }))
        .apply(
            "WaitForLines",
            Window.<KV<Integer, String>>into(FixedWindows.of(Duration.standardSeconds(5)))
                .triggering(AfterWatermark.pastEndOfWindow())
                .discardingFiredPanes()
                .withAllowedLateness(Duration.ZERO))
        .apply(getJdbcWrite(tableName));

    pipeline.run();

    assertRowCount(DATA_SOURCE, tableName, 3*10);
    DatabaseTestHelper.deleteTable(DATA_SOURCE, tableName);
  }

The test passed. For Kotlin then likely due to compiler not correctly infer the signature of FinishBundleContext.output here? In DoFn class it has a type of BoundedWindow. Both GlobalWindow and IntervalWindow can be cast to BoundedWindow.

[sshahar1]

Hi @Abacn
Thank you for your response, I'll investigate this observation on my side.
Many thanks

[scott-strong]

Hi @sshahar1 @Abacn,

I ran into the same issue in my java pipeline. We upgraded to Apache Beam 2.39 and started getting the exact error message from above. I can't post the exact pipeline, but we're streaming Pub/Sub events, transforming and windowing, then writing them to MySQL. Below is the windowing function:

Window.<SimplifiedEvent>into(FixedWindows.of(Duration.standardMinutes(options.getWindowDurationMinutes())))
    .discardingFiredPanes()

I'm curious if Sigal has resolved this since this post, or if there have been any other reports of this error message occurring. I know there was a test case where this is passing, but following the stacktrace I don't see how the GlobalWindow would work when passed to the IntervalWindowCoder. Any info is appreciated. Thanks.

[Abacn]

Hi @pabloem do you have an idea of the fix? Seems like here the third parameter is not compatible in the case of fixed window

beam/sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/JdbcIO.java

Line 1642 in 76761db

c.output(outputList, Instant.now(), GlobalWindow.INSTANCE);