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 >>>>> >>>>
