You are right, I won’t be able to reproduce this problem without data. One
thing I can tell though that I think the problem is indeed with the hashcode.
Unforunately I don’t know Gson, but one strange thing I noticed is the
exception message: SharedBufferEntry(ValueTimeWrapper({}, 1502298303586, 0),
[SharedBufferEdge(null, 1)], 1). The first {} is your event.toString, which
seems odd as if your event was empty.
Generally speaking as I understand this Exception is thrown because the
hashcode of your event changes during serialization, and access to some
internal temporary cache is broken.
> On 10 Aug 2017, at 14:29, Daiqing Li <lidaiqing1...@gmail.com> wrote:
>
> Hi,
>
> Here is the code. But I am not sure if you can reproduce the problem without
> data source.
>
> Best,
> Daiqing
>
> On Thu, Aug 10, 2017 at 8:15 AM, Dawid Wysakowicz
> <wysakowicz.da...@gmail.com> wrote:
> As @Kostas asked in your previous thread would be possible for you to share
> your code for that job or at least a minimal example to reproduce this
> behaviour. I fear we won’t be able to help you without any further info.
>
> Regards,
> Dawid
>
> > On 10 Aug 2017, at 14:10, Daiqing Li <lidaiqing1...@gmail.com> wrote:
> >
> > Hi Flink user,
> >
> > I am using FsStateBackend and AWS EMR YARN to run a CEP job. I got this
> > exception after running for a while. Could anyone give me some help to
> > debug this? I try parallelism 1, and it has the same problem. I also try
> > reimplemented hashcode and equals method. I use UUID as hashcode right now.
> > 2017-08-09 18:15:04,572 INFO
> > org.apache.flink.runtime.executiongraph.ExecutionGraph -
> > KeyedCEPPatternOperator -> Map -> Sink: Unnamed (3/4)
> > (d4749a4c3469732a2a5edf40b83f88
> > d4) switched from RUNNING to FAILED.
> > AsynchronousException{java.
> > lang.Exception: Could not materialize checkpoint 946 for operator
> > KeyedCEPPatternOperator -> Map -> Sink: Unnamed (3/4).}
> > at org.apache.flink.streaming.
> > runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(
> > StreamTask.java:970)
> > at java.util.concurrent.
> > Executors$RunnableAdapter.
> > call(Executors.java:511)
> > at java.util.concurrent.
> > FutureTask.run(FutureTask.
> > java:266)
> > at java.util.concurrent.
> > ThreadPoolExecutor.runWorker(
> > ThreadPoolExecutor.java:1149)
> > at java.util.concurrent.
> > ThreadPoolExecutor$Worker.run(
> > ThreadPoolExecutor.java:624)
> > at java.lang.Thread.run(Thread.
> > java:748)
> > Caused by: java.lang.Exception: Could not materialize checkpoint 946 for
> > operator KeyedCEPPatternOperator -> Map -> Sink: Unnamed (3/4).
> > ... 6 more
> > Caused by: java.util.concurrent.
> > ExecutionException: java.lang.IllegalStateException: Could not find id for
> > entry: SharedBufferEntry(
> > ValueTimeWrapper({}, 1502298303586, 0), [SharedBufferEdge(null, 1)], 1)
> > at java.util.concurrent.
> > FutureTask.report(FutureTask.
> > java:122)
> > at java.util.concurrent.
> > FutureTask.get(FutureTask.
> > java:192)
> > at org.apache.flink.util.
> > FutureUtil.runIfNotDoneAndGet(
> > FutureUtil.java:43)
> > at org.apache.flink.streaming.
> > runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(
> > StreamTask.java:897)
> > ... 5 more
> > Suppressed: java.lang.Exception: Could not properly cancel managed
> > keyed state future.
> > at org.apache.flink.streaming.
> > api.operators.OperatorSnapshotResult.cancel(OperatorSnapshotResult.java:
> > 90)
> > at org.apache.flink.streaming.
> > runtime.tasks.StreamTask$AsyncCheckpointRunnable.
> > cleanup(StreamTask.java:1023)
> > at org.apache.flink.streaming.
> > runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(
> > StreamTask.java:961)
> > ... 5 more
> >
>
>
> <MilestoneEvent.java><example.java>
signature.asc
Description: Message signed with OpenPGP
