[FLINK-2747] TypeExtractor does not correctly analyze Scala Immutables (AnyVal)

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 0.10.0, 1.0.0
• Component/s: API / Type Serialization System
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Aljoscha Krettek
• Created: 23/Sep/15 12:04
• Updated: 09/Feb/22 22:37

Description

This example program only works correctly if Kryo is force-enabled.

object Test {
  class Id(val underlying: Int) extends AnyVal

  class X(var id: Id) {
    def this() { this(new Id(0)) }
  }

  class MySource extends SourceFunction[X] {
    def run(ctx: SourceFunction.SourceContext[X]) {
      ctx.collect(new X(new Id(1)))
    }
    def cancel() {}
  }

  def main(args: Array[String]) {
    val env = StreamExecutionContext.getExecutionContext
    env.addSource(new MySource).print
    env.execute("Test")
  }
}

The program fails with this:

Caused by: java.lang.RuntimeException: Cannot instantiate class.
	at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.createInstance(PojoSerializer.java:227)
	at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.deserialize(PojoSerializer.java:421)
	at org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer.deserialize(StreamRecordSerializer.java:110)
	at org.apache.flink.streaming.runtime.streamrecord.StreamRecordSerializer.deserialize(StreamRecordSerializer.java:41)
	at org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate.read(NonReusingDeserializationDelegate.java:55)
	at org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer.getNextRecord(SpillingAdaptiveSpanningRecordDeserializer.java:125)
	at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:136)
	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:198)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:579)
	at java.lang.Thread.run(Thread.java:745)