[ https://issues.apache.org/jira/browse/FLINK-5995?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15905681#comment-15905681 ]
ASF GitHub Bot commented on FLINK-5995: --------------------------------------- Github user StephanEwen commented on the issue: https://github.com/apache/flink/pull/3503 That sounds reasonable! On Mar 10, 2017 4:07 PM, "Jincheng Sun" <notificati...@github.com> wrote: > Hi,@StephanEwen <https://github.com/StephanEwen>, Thanks a lot for your > comment. I seriously think about where to add the test more is reasonable, > because the change of this PR is OperatorStateBackend, so I added the > test case injava/ src/test/java/org/apache/flink/runtime/state/ > operatorStateBackendTest.java Is this make sense for you? feel free to > tell me if it is incorrect. > Best, > SunJincheng > > — > You are receiving this because you were mentioned. > Reply to this email directly, view it on GitHub > <https://github.com/apache/flink/pull/3503#issuecomment-285692212>, or mute > the thread > <https://github.com/notifications/unsubscribe-auth/ABpaqtxMMNduF1zFO5Ph5Q_0A70z99njks5rkWctgaJpZM4MY8Na> > . > > Get a Exception when creating the ListStateDescriptor with a TypeInformation > ----------------------------------------------------------------------------- > > Key: FLINK-5995 > URL: https://issues.apache.org/jira/browse/FLINK-5995 > Project: Flink > Issue Type: Bug > Components: DataStream API, State Backends, Checkpointing > Reporter: sunjincheng > Assignee: sunjincheng > > When use OperatorState and creating the ListStateDescriptor with a > TypeInformation,I got a exception. The Exception info is: > {code} > org.apache.flink.runtime.client.JobExecutionException: Job execution failed. > at > org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply$mcV$sp(JobManager.scala:915) > at > org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:858) > at > org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$7.apply(JobManager.scala:858) > at > scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24) > at > scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24) > at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40) > at > akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397) > at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) > at > scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) > at > scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) > at > scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) > Caused by: java.lang.IllegalStateException: Serializer not yet initialized. > at > org.apache.flink.api.common.state.StateDescriptor.getSerializer(StateDescriptor.java:169) > at > org.apache.flink.api.common.state.ListStateDescriptor.getElementSerializer(ListStateDescriptor.java:93) > at > org.apache.flink.runtime.state.DefaultOperatorStateBackend.getOperatorState(DefaultOperatorStateBackend.java:110) > at > org.apache.flink.runtime.state.DefaultOperatorStateBackend.getOperatorState(DefaultOperatorStateBackend.java:91) > at > org.apache.flink.table.runtime.aggregate.UnboundedNonPartitionedProcessingOverProcessFunction.initializeState(UnboundedNonPartitionedProcessingOverProcessFunction.scala:104) > at > org.apache.flink.streaming.api.functions.util.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:178) > at > org.apache.flink.streaming.api.functions.util.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:160) > at > org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:106) > at > org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:242) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.initializeOperators(StreamTask.java:681) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:669) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:251) > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:670) > at java.lang.Thread.run(Thread.java:745) > {code} > So, I'll add `stateDescriptor.initializeSerializerUnlessSet()` call in the > `getOperatorState` method. I appreciate If anyone can give me some advice? -- This message was sent by Atlassian JIRA (v6.3.15#6346)