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

["Stephan Ewen (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-2747?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16205131#comment-16205131 ] 

Stephan Ewen commented on FLINK-2747:
-------------------------------------

One way we can fix this is by obeying the {{enableObjectReuse()}} flags in the Network Stack deserializers as well.
For immutable ty

A more involved fix is to detect in the {{TypeInformation}} the immutable types that cannot be instantiated empty, and force non-reuse mode for them.

> TypeExtractor does not correctly analyze Scala Immutables (AnyVal)
> ------------------------------------------------------------------
>
>                 Key: FLINK-2747
>                 URL: https://issues.apache.org/jira/browse/FLINK-2747
>             Project: Flink
>          Issue Type: Bug
>          Components: Type Serialization System
>    Affects Versions: 0.10.0, 1.0.0
>            Reporter: Aljoscha Krettek
>             Fix For: 1.0.0
>
>
> This example program only works correctly if Kryo is force-enabled.
> {code}
> 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")
>   }
> }
> {code}
> The program fails with this:
> {code}
> 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)
> {code}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)