[FLINK-28367] OffsetDateTime does not work with keyBy

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Affects Version/s: 1.15.0
• Component/s: API / DataStream, API / Type Serialization System

Environment

• Java 1.8 (openjdk 1.8.0_322)
• Scala 2.12.15
• Flink 1.15.0

• Assignee: Unassigned
• Reporter: Henrik Laxhuber
• Created: 03/Jul/22 15:49
• Updated: 05/Jul/22 11:14

Description

Using keyBy incorrectly (de-)serializes java.time.OffsetDateTime types - the offset gets lost and becomes null.

Here's a minimal non-working example:

 

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment

import java.time.OffsetDateTime

object MWE {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    env
      .fromElements("2022-07-03T15:35:48.142Z", "2022-07-03T15:35:48.438Z")
      .map(OffsetDateTime.parse(_))
      .keyBy((t: OffsetDateTime) => t)
      .print()

    env.execute()
  }
} 

 

Expected Output:

2022-07-03T15:35:48.438Z
2022-07-03T15:35:48.142Z

Actual Output:

2022-07-03T15:35:48.438null
2022-07-03T15:35:48.142null

The issue arises whenever keyBy and OffsetDateTime are involved; I believe it could have something to do with the way that flink serializes the state.