[FLINK-1701] Serialization problems with TestStreamEnvironment

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Component/s: API / DataStream
• Assignee: Péter Szabó
• Reporter: Péter Szabó
• Created: 13/Mar/15 13:00
• Updated: 28/Feb/19 11:15
• Resolved: 18/Mar/15 20:33

Description

When mapWindowing a stream of tuples of a custom class with TestStreamEnvironment, serialization of StreamDiscretizer fails in StreamConfig.

Example here

The same example runs with StreamExecutionEnvironment without any problem. I'm wondering what is the difference.

StackTrace:
java.lang.RuntimeException: Cannot serialize invokable object class org.apache.flink.streaming.api.invokable.operator.windowing.StreamDiscretizer
at org.apache.flink.streaming.api.StreamConfig.setUserInvokable(StreamConfig.java:173)
at org.apache.flink.streaming.api.StreamingJobGraphGenerator.setVertexConfig(StreamingJobGraphGenerator.java:213)
at org.apache.flink.streaming.api.StreamingJobGraphGenerator.createChain(StreamingJobGraphGenerator.java:129)
at org.apache.flink.streaming.api.StreamingJobGraphGenerator.createChain(StreamingJobGraphGenerator.java:121)
at org.apache.flink.streaming.api.StreamingJobGraphGenerator.createChain(StreamingJobGraphGenerator.java:116)
at org.apache.flink.streaming.api.StreamingJobGraphGenerator.createChain(StreamingJobGraphGenerator.java:121)
at org.apache.flink.streaming.api.StreamingJobGraphGenerator.setChaining(StreamingJobGraphGenerator.java:95)
at org.apache.flink.streaming.api.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:86)
at org.apache.flink.streaming.api.StreamGraph.getJobGraph(StreamGraph.java:510)
at org.apache.flink.streaming.util.TestStreamEnvironment.execute(TestStreamEnvironment.java:59)
at org.apache.flink.streaming.util.TestStreamEnvironment.execute(TestStreamEnvironment.java:54)
at org.apache.flink.streaming.api.ComplexIntegrationTest.complexIntegrationTest7(ComplexIntegrationTest.java:433)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:271)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:70)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:50)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229)
at org.junit.runners.ParentRunner.run(ParentRunner.java:309)
at org.junit.runner.JUnitCore.run(JUnitCore.java:160)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:74)
at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:211)
at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:67)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:134)
Caused by: org.apache.commons.lang3.SerializationException: java.io.NotSerializableException: org.apache.flink.streaming.api.ComplexIntegrationTest
at org.apache.commons.lang3.SerializationUtils.serialize(SerializationUtils.java:157)
at org.apache.commons.lang3.SerializationUtils.serialize(SerializationUtils.java:179)
at org.apache.flink.streaming.api.StreamConfig.setUserInvokable(StreamConfig.java:171)