[BEAM-13374] ToJson() of a Row with a DateTime field fails with an exception

• Type: Bug
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Affects Version/s: 2.33.0
• Fix Version/s: 2.36.0, 2.37.0
• Component/s: beam-model, sdk-java-core

Language

Java

• Assignee: Andrei Gurau
• Reporter: Sergei Lilichenko
• Created: 02/Dec/21 22:41
• Updated: 21/Apr/22 14:50
• Resolved: 20/Apr/22 16:00

Description

The following code fails with "class org.joda.time.Instant cannot be cast to class org.joda.time.DateTime" exception at ToJson.of() transform. 

The root cause of it appears to be this conversion to Instant.

This bug appeared in a JDBC processing pipeline where a TIMESTAMP column is part of the result set retrieved using JdbcIO.readRows(). 

 

Pipeline p = Pipeline.create();
Schema schema = Schema.of(Field.of("timestamp", FieldType.DATETIME));
p
    .apply("DateTime values", Create.of(new DateTime()))
    .apply("To Row", ParDo.of(new DoFn<DateTime, Row>() {
      @ProcessElement
      public void toRow(@Element DateTime dateTime, OutputReceiver<Row> rowOutputReceiver) {
        rowOutputReceiver.output(
            Row.withSchema(schema)
                .withFieldValue("timestamp", dateTime)
                .build());
      }
    }))
    .setCoder(RowCoder.of(schema))
    .apply("To Json", ToJson.of())
    .apply("Print to Console", ParDo.of(new DoFn<String, Void>() {
      @ProcessElement
      public void print(@Element String expectedJson) {
        System.out.println("JSON: " + expectedJson);
      }
    }));
p.run().waitUntilFinish();