Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Content-Type: text/plain; charset=utf-8
Mime-Version: 1.0 (Mac OS X Mail 10.3 \(3273\))
Subject: Re: Keyed CEP checkpoint fails
From: Daiqing Li <lidaiqing1993@gmail.com>
In-Reply-To: <9EC9F61F-FD61-4FFD-9C22-17D47FC711E7@gmail.com>
Date: Thu, 10 Aug 2017 20:40:41 -0400
Cc: FlinkUserGroup <user@flink.apache.org>,
 Kostas Kloudas <k.kloudas@data-artisans.com>
Content-Transfer-Encoding: quoted-printable
Message-Id: <E8582E80-F1E0-4C95-BF7A-99F19D773034@gmail.com>
References: <CACdJq91-h-8N-feaojNuZvm+gOtSL0Q4=6hk=nK9hLamFB+gJA@mail.gmail.com>
 <D10E39E9-B9F7-48B6-A1B6-0BAC164CA783@gmail.com>
 <CACdJq93aViFiDDma8664gRrnbbxqo-huW9JB6CMUcLB69_qXDQ@mail.gmail.com>
 <5795594C-BE41-49B6-91D8-C32BD161E93E@gmail.com>
 <9609722D-BEB4-4313-BEBA-320CC76120C4@gmail.com>
 <9EC9F61F-FD61-4FFD-9C22-17D47FC711E7@gmail.com>
To: Dawid Wysakowicz <wysakowicz.dawid@gmail.com>
archived-at: Fri, 11 Aug 2017 00:40:49 -0000

Hi Dawid,

After rewriting dashcode with Objects.hash for all the fields, I still =
get the same error. One thing special is checkpoints always fail at 428, =
after trying many times. Does it mean anything?
> On Aug 10, 2017, at 9:14 AM, Dawid Wysakowicz =
<wysakowicz.dawid@gmail.com> wrote:
>=20
> Yes, with the information I have, the conclusion would be the same, =
that I think the reason is problem with hashcode. Without some data to =
reproduce it unfortunately I won=E2=80=99t be able to help you further. =
I could just advise you to debug the method SharedBuffer#serialize and =
pay attention to the entryID map.
>=20
>> On 10 Aug 2017, at 14:54, Daiqing Li <lidaiqing1993@gmail.com> wrote:
>>=20
>> Oh sorry, the data in {} is not empty because I hide private =
information about my model. Do you have that same conclusion?
>>> On Aug 10, 2017, at 8:52 AM, Dawid Wysakowicz =
<wysakowicz.dawid@gmail.com> wrote:
>>>=20
>>> You are right, I won=E2=80=99t 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=E2=80=99t 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.
>>>=20
>>> 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.
>>>=20
>>>> On 10 Aug 2017, at 14:29, Daiqing Li <lidaiqing1993@gmail.com> =
wrote:
>>>>=20
>>>> Hi,
>>>>=20
>>>> Here is the code. But I am not sure if you can reproduce the =
problem without data source.
>>>>=20
>>>> Best,
>>>> Daiqing
>>>>=20
>>>> On Thu, Aug 10, 2017 at 8:15 AM, Dawid Wysakowicz =
<wysakowicz.dawid@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=E2=80=99t be able to help you =
without any further info.
>>>>=20
>>>> Regards,
>>>> Dawid
>>>>=20
>>>>> On 10 Aug 2017, at 14:10, Daiqing Li <lidaiqing1993@gmail.com> =
wrote:
>>>>>=20
>>>>> Hi Flink user,
>>>>>=20
>>>>> 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
>>>>>=20
>>>>=20
>>>>=20
>>>> <MilestoneEvent.java><example.java>
>>>=20
>>=20
>=20