Hi Vijay, Yun, I've created a JIRA https://issues.apache.org/jira/browse/FLINK-23886 to track this.
Best, JING ZHANG JING ZHANG <beyond1...@gmail.com> 于2021年8月20日周五 下午1:19写道： > Hi Vijay, > I have encountered the same problem several times in online production > Flink jobs, but I have not found the root cause of the exception yet. > We have walk around the exception by adding the following parameter, hope > it could help you. > state.backend.rocksdb.timer-service.factory: HEAP > > I would invite Yun Tang who is an expert on the topic to look into the > problem, we could also create a JIRA to track the issue. > > [1] > https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/state/state_backends/#timers-heap-vs-rocksdb > > Best, > JING ZHANG > > vijayakumar palaniappan <vijayakuma...@gmail.com> 于2021年8月19日周四 上午8:02写道： > >> Setup Specifics: >> Version: 1.6.2 >> RocksDB Map State >> Timers stored in rocksdb >> >> When we have this job running for long periods of time like > 30 days, if >> for some reason the job restarts, we encounter "Error while >> deserializing the element". Is this a known issue fixed in later >> versions? I see some changes to code for FLINK-10175, but we don't use >> any queryable state >> >> Below is the stack trace >> >> org.apache.flink.util.FlinkRuntimeException: Error while deserializing >> the element. >> >> at >> org.apache.flink.contrib.streaming.state.RocksDBCachingPriorityQueueSet.deserializeElement(RocksDBCachingPriorityQueueSet.java:389) >> >> at >> org.apache.flink.contrib.streaming.state.RocksDBCachingPriorityQueueSet.peek(RocksDBCachingPriorityQueueSet.java:146) >> >> at >> org.apache.flink.contrib.streaming.state.RocksDBCachingPriorityQueueSet.peek(RocksDBCachingPriorityQueueSet.java:56) >> >> at >> org.apache.flink.runtime.state.heap.KeyGroupPartitionedPriorityQueue$InternalPriorityQueueComparator.comparePriority(KeyGroupPartitionedPriorityQueue.java:274) >> >> at >> org.apache.flink.runtime.state.heap.KeyGroupPartitionedPriorityQueue$InternalPriorityQueueComparator.comparePriority(KeyGroupPartitionedPriorityQueue.java:261) >> >> at >> org.apache.flink.runtime.state.heap.HeapPriorityQueue.isElementPriorityLessThen(HeapPriorityQueue.java:164) >> >> at >> org.apache.flink.runtime.state.heap.HeapPriorityQueue.siftUp(HeapPriorityQueue.java:121) >> >> at >> org.apache.flink.runtime.state.heap.HeapPriorityQueue.addInternal(HeapPriorityQueue.java:85) >> >> at >> org.apache.flink.runtime.state.heap.AbstractHeapPriorityQueue.add(AbstractHeapPriorityQueue.java:73) >> >> at >> org.apache.flink.runtime.state.heap.KeyGroupPartitionedPriorityQueue.<init>(KeyGroupPartitionedPriorityQueue.java:89) >> >> at >> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend$RocksDBPriorityQueueSetFactory.create(RocksDBKeyedStateBackend.java:2792) >> >> at >> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.create(RocksDBKeyedStateBackend.java:450) >> >> at >> org.apache.flink.streaming.api.operators.InternalTimeServiceManager.createTimerPriorityQueue(InternalTimeServiceManager.java:121) >> >> at >> org.apache.flink.streaming.api.operators.InternalTimeServiceManager.registerOrGetTimerService(InternalTimeServiceManager.java:106) >> >> at >> org.apache.flink.streaming.api.operators.InternalTimeServiceManager.getInternalTimerService(InternalTimeServiceManager.java:87) >> >> at >> org.apache.flink.streaming.api.operators.AbstractStreamOperator.getInternalTimerService(AbstractStreamOperator.java:764) >> >> at >> org.apache.flink.streaming.api.operators.KeyedProcessOperator.open(KeyedProcessOperator.java:61) >> >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:424) >> >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:290) >> >> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711) >> >> at java.lang.Thread.run(Thread.java:748) >> >> Caused by: java.io.EOFException >> >> at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:290) >> >> at org.apache.flink.types.StringValue.readString(StringValue.java:769) >> >> at >> org.apache.flink.api.common.typeutils.base.StringSerializer.deserialize(StringSerializer.java:69) >> >> at >> org.apache.flink.api.common.typeutils.base.StringSerializer.deserialize(StringSerializer.java:28) >> >> at >> org.apache.flink.api.java.typeutils.runtime.RowSerializer.deserialize(RowSerializer.java:179) >> >> at >> org.apache.flink.api.java.typeutils.runtime.RowSerializer.deserialize(RowSerializer.java:46) >> >> at >> org.apache.flink.streaming.api.operators.TimerSerializer.deserialize(TimerSerializer.java:168) >> >> at >> org.apache.flink.streaming.api.operators.TimerSerializer.deserialize(TimerSerializer.java:45) >> >> at >> org.apache.flink.contrib.streaming.state.RocksDBCachingPriorityQueueSet.deserializeElement(RocksDBCachingPriorityQueueSet.java:387) >> >> ... 20 more >> >> -- >> Thanks, >> -Vijay >> >
