Could this be caused by mixing of configuration settings when running? Running a job with one parallelism, stop/savepointing and then recovering with a different parallelism? I'd assume that's fine and wouldn't put create bad state.
On Tue, Jul 13, 2021 at 12:34 PM Dan Hill <quietgol...@gmail.com> wrote: > I checked my code. Our keys for streams and map state only use either (1) > string, (2) long IDs that don't change or (3) Tuple of 1 and 2. > > I don't know why my current case is breaking. Our job partitions and > parallelism settings have not changed. > > > > On Tue, Jul 13, 2021 at 12:11 PM Dan Hill <quietgol...@gmail.com> wrote: > >> Hey. I just hit a similar error in production when trying to savepoint. >> We also use protobufs. >> >> Has anyone found a better fix to this? >> >> On Fri, Oct 23, 2020 at 5:21 AM Till Rohrmann <trohrm...@apache.org> >> wrote: >> >>> Glad to hear that you solved your problem. Afaik Flink should not read >>> the fields of messages and call hashCode on them. >>> >>> Cheers, >>> Till >>> >>> On Fri, Oct 23, 2020 at 2:18 PM Radoslav Smilyanov < >>> radoslav.smilya...@smule.com> wrote: >>> >>>> Hi Till, >>>> >>>> I found my problem. It was indeed related to a mutable hashcode. >>>> >>>> I was using a protobuf message in the key selector function and one of >>>> the protobuf fields was enum. I checked the implementation of the hashcode >>>> of the generated message and it is using the int value field of the >>>> protobuf message so I assumed that it is ok and it's immutable. >>>> >>>> I replaced the key selector function to use Tuple[Long, Int] (since my >>>> protobuf message has only these two fields where the int parameter stands >>>> for the enum value field). After changing my code to use the Tuple it >>>> worked. >>>> >>>> I am not sure if Flink somehow reads the protobuf message fields and >>>> uses the hashcode of the fields directly since the generated protobuf enum >>>> indeed has a mutable hashcode (Enum.hashcode). >>>> >>>> Nevertheless it's ok with the Tuple key. >>>> >>>> Thanks for your response! >>>> >>>> Best Regards, >>>> Rado >>>> >>>> >>>> On Fri, Oct 23, 2020 at 2:39 PM Till Rohrmann <trohrm...@apache.org> >>>> wrote: >>>> >>>>> Hi Rado, >>>>> >>>>> it is hard to tell the reason w/o a bit more details. Could you share >>>>> with us the complete logs of the problematic run? Also the job you are >>>>> running and the types of the state you are storing in RocksDB and use as >>>>> events in your job are very important. In the linked SO question, the >>>>> problem was a type whose hashcode was not immutable. >>>>> >>>>> Cheers, >>>>> Till >>>>> >>>>> On Wed, Oct 21, 2020 at 6:24 PM Radoslav Smilyanov < >>>>> radoslav.smilya...@smule.com> wrote: >>>>> >>>>>> Hello all, >>>>>> >>>>>> I am running a Flink job that performs data enrichment. My job has 7 >>>>>> kafka consumers that receive messages for dml statements performed for 7 >>>>>> db >>>>>> tables. >>>>>> >>>>>> Job setup: >>>>>> >>>>>> - Flink is run in k8s in a similar way as it is described here >>>>>> >>>>>> <https://ci.apache.org/projects/flink/flink-docs-stable/ops/deployment/kubernetes.html#job-cluster-resource-definitions> >>>>>> . >>>>>> - 1 job manager and 2 task managers >>>>>> - parallelism is set to 4 and 2 task slots >>>>>> - rocksdb as state backend >>>>>> - protobuf for serialization >>>>>> >>>>>> Whenever I try to trigger a savepoint after my state is >>>>>> bootstrapped I get the following error for different operators: >>>>>> >>>>>> Caused by: java.lang.IllegalArgumentException: Key group 0 is not in >>>>>> KeyGroupRange{startKeyGroup=32, endKeyGroup=63}. >>>>>> at >>>>>> org.apache.flink.runtime.state.KeyGroupRangeOffsets.computeKeyGroupIndex(KeyGroupRangeOffsets.java:142) >>>>>> at >>>>>> org.apache.flink.runtime.state.KeyGroupRangeOffsets.setKeyGroupOffset(KeyGroupRangeOffsets.java:104) >>>>>> at >>>>>> org.apache.flink.contrib.streaming.state.snapshot.RocksFullSnapshotStrategy$SnapshotAsynchronousPartCallable.writeKVStateData(RocksFullSnapshotStrategy.java:319) >>>>>> at >>>>>> org.apache.flink.contrib.streaming.state.snapshot.RocksFullSnapshotStrategy$SnapshotAsynchronousPartCallable.writeSnapshotToOutputStream(RocksFullSnapshotStrategy.java:261) >>>>>> >>>>>> Note: key group might vary. >>>>>> >>>>>> I found this >>>>>> <https://stackoverflow.com/questions/49140654/flink-error-key-group-is-not-in-keygrouprange> >>>>>> article >>>>>> in Stackoverflow which relates to such an exception (btw my job graph >>>>>> looks >>>>>> similar to the one described in the article except that my job has more >>>>>> joins). I double checked my hashcodes and I think that they are fine. >>>>>> >>>>>> I tried to reduce the parallelism to 1 with 1 task slot per task >>>>>> manager and this configuration seems to work. This leads me to a >>>>>> direction >>>>>> that it might be some concurrency issue. >>>>>> >>>>>> I would like to understand what is causing the savepoint failure. Do >>>>>> you have any suggestions what I might be missing? >>>>>> >>>>>> Thanks in advance! >>>>>> >>>>>> Best Regards, >>>>>> Rado >>>>>> >>>>>
