Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
From: Stefan Richter <s.richter@data-artisans.com>
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable
Mime-Version: 1.0 (Mac OS X Mail 10.2 \(3259\))
Subject: Re: Previously working job fails on Flink 1.2.0
Date: Mon, 20 Feb 2017 15:22:49 +0100
References: <681a0422-0ede-cf4c-02d9-f58f09044052@hausmann-family.de>
To: user@flink.apache.org
In-Reply-To: <681a0422-0ede-cf4c-02d9-f58f09044052@hausmann-family.de>
Message-Id: <A4DE5764-1310-4324-9BD3-A53E0B2CC3E8@data-artisans.com>
archived-at: Mon, 20 Feb 2017 14:23:11 -0000

Hi,

Flink 1.2 is partitioning all keys into key-groups, the atomic units for =
rescaling. This partitioning is done by hash partitioning and is also in =
sync with the routing of tuples to operator instances (each parallel =
instance of a keyed operator is responsible for some range of key =
groups). This exception means that Flink detected a tuple in the state =
backend of a parallel operator instance that should not be there =
because, by its key hash, it belongs to a different key-group. Or =
phrased differently, this tuple belongs to a different parallel operator =
instance. If this is a Flink bug or user code bug is very hard to tell, =
the log also does not provide additional insights. I could see this =
happen in case that your keys are mutable and your code makes some =
changes to the object that change the hash code. Another question is =
also: did you migrate your job from Flink 1.1.3 through an old savepoint =
or did you do a fresh start. Other than that, I can recommend to check =
your code for mutating of keys. If this fails deterministically, you =
could also try to set a breakpoint for the line of the exception and =
take a look if the key that is about to be inserted is somehow special.

Best,
Stefan=20


> Am 20.02.2017 um 14:32 schrieb Steffen Hausmann =
<steffen@hausmann-family.de>:
>=20
> Hi there,
>=20
> I=E2=80=99m having problems running a job on Flink 1.2.0 that =
successfully executes on Flink 1.1.3. The job is supposed to read events =
from a Kinesis stream and to send outputs to Elasticsearch and it =
actually initiates successfully on a Flink 1.2.0 cluster running on =
YARN, but as soon as I start to ingest events into the Kinesis stream, =
the job fails (see the attachment for more information):
>=20
> java.lang.RuntimeException: Unexpected key group index. This indicates =
a bug.
>=20
> at =
org.apache.flink.runtime.state.heap.StateTable.set(StateTable.java:57)
>=20
> at =
org.apache.flink.runtime.state.heap.HeapListState.add(HeapListState.java:9=
8)
>=20
> at =
org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.proc=
essElement(WindowOperator.java:372)
>=20
> at =
org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(St=
reamInputProcessor.java:185)
>=20
> at =
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputSt=
reamTask.java:63)
>=20
> at =
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java=
:272)
>=20
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:655)
>=20
> at java.lang.Thread.run(Thread.java:745)
>=20
> Any ideas what=E2=80=99s going wrong here? The job executes =
successfully when it=E2=80=99s compiled against the Flink 1.1.3 =
artifacts and run on a Flink 1.1.3 cluster. Does this indicate a bug in =
my code or is this rather a bug in Flink? How can I further debug this?
>=20
> Any guidance is highly appreciated.
>=20
> Thanks,
>=20
> Steffen
>=20
> <log>