Flink错误-密钥组不在KeyGroupRange中

Question

我正在使用RocksDB作为状态后端运行Flink图。对于图形中的一个联接运算符，我得到以下异常。（实际的组号当然因运行而异）。

java.lang.IllegalArgumentException：密钥组45不在KeyGroupRange {startKeyGroup = 0，endKeyGroup = 42}中。

我的接线员也不是如下

Source1 -----> Map1A ---> KeyBy--->\___ >
        \----> Map1B ---> KeyBy--->-----> Join1AB ---->
                                                \____>
Source2 ----->------------KeyBy---> -----------------> Join2,1AB ---->

在Join2,1AB运算符中引发了错误，该运算符将（a）Join1AB的结果与（keyed）source2连接起来。

任何想法可能是什么原因造成的？我在下面有完整的stacktrace，并且我知道这仍然很模糊-但是任何朝着正确方向的指针都值得赞赏。

Caused by: java.lang.IllegalArgumentException: Key group 45 is not in KeyGroupRange{startKeyGroup=0, endKeyGroup=42}.
        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.RocksDBKeyedStateBackend$RocksDBFullSnapshotOperation.writeKVStateData(RocksDBKeyedStateBackend.java:664)
        at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$RocksDBFullSnapshotOperation.writeDBSnapshot(RocksDBKeyedStateBackend.java:521)
        at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$3.performOperation(RocksDBKeyedStateBackend.java:417)
        at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$3.performOperation(RocksDBKeyedStateBackend.java:399)
        at org.apache.flink.runtime.io.async.AbstractAsyncIOCallable.call(AbstractAsyncIOCallable.java:72)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at org.apache.flink.util.FutureUtil.runIfNotDoneAndGet(FutureUtil.java:40)
        at org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:897)
        ... 5 more
    [CIRCULAR REFERENCE:java.lang.IllegalArgumentException: Key group 45 is not in KeyGroupRange{startKeyGroup=0, endKeyGroup=42}.]

编辑：如果我将状态后端更改为文件系统（即FsStateBackend），则将得到以下堆栈跟踪。关键组索引问题。

java.lang.IllegalArgumentException: Key group index out of range of key group range [43, 86).
    at org.apache.flink.runtime.state.heap.NestedMapsStateTable.setMapForKeyGroup(NestedMapsStateTable.java:104)
    at org.apache.flink.runtime.state.heap.NestedMapsStateTable.putAndGetOld(NestedMapsStateTable.java:218)
    at org.apache.flink.runtime.state.heap.NestedMapsStateTable.put(NestedMapsStateTable.java:207)
    at org.apache.flink.runtime.state.heap.NestedMapsStateTable.put(NestedMapsStateTable.java:145)
    at org.apache.flink.runtime.state.heap.HeapValueState.update(HeapValueState.java:72)

<snip user code stack trace>

org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator.processElement1(KeyedCoProcessOperator.java:77)
    at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessor.processInput(StreamTwoInputProcessor.java:242)
    at org.apache.flink.streaming.runtime.tasks.TwoInputStreamTask.run(TwoInputStreamTask.java:91)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:263)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:702)
    at java.lang.Thread.run(Thread.java:745)

Answer 1

问题是我的数据对象（PO​​JO）具有可变的哈希码。具体来说，哈希代码包含枚举。例如，如果我有一个汽车流，其哈希码由汽车年份和汽车类型（枚举）组成，如下所示。

Car {
   private final CarType carType;
   private final int carYear

   public long hashCode() {
     int result = 17;
     result = 31 * result + carYear;
     result = 31 * result + carType.hasCode();  <---- This is mutable!
   }
}

枚举的hashCode本质上是Object.hashCode（）（取决于内存地址）。随后，一台机器（或进程）上的hashCode将与另一台机器（或进程）上的hashCode不同。这也解释了为什么我只在分布式环境中运行而不是本地运行时才遇到此问题。

为了解决这个问题，我将hashCode（）更改为不可变的。做String.hashCode（）的性能很差，所以我可能需要对其进行优化。但是下面对Car的定义将解决此问题。

Car {
   private final CarType carType;
   private final int carYear

   public long hashCode() {
     int result = 17;
     result = 31 * result + carYear;
     result = 31 * result + carType.name().hasCode();  <---- This is IMMUTABLE!
   }
}