Error while deserializing the element
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.(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
Re: Error while deserializing the element
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 于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.(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 >
Re: Error while deserializing the element
Hi Vijay, Yun, I've created a JIRA https://issues.apache.org/jira/browse/FLINK-23886 to track this. Best, JING ZHANG JING ZHANG 于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 于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.(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 >> >
Re: Error while deserializing the element
My previous response in mail list was too brief, sorry about that. For the use of config 'state.backend.rocksdb.timer-service.factory: HEAP', I would like to add some information. Please correct me if anything is wrong. 1. The configure could help to avoid the EOF exception when recover rocksdb queue set from savepoint/checkpoint file. As you could see, if we use HEAP for timers, the timer state and other operator state would not mixed up in one rocksdb instance. So we could avoid the exception. The jobs which adds the configure don't encounter such problem again too. 2. If update the timer store in Rocksdb to store in HEAP, we could not restore the job from savepoint/checkpoint indeed. If the user makes a mistake by recovering job from savepoint/checkpoint file, the job will not throw an exception, but the timer state will be lost. This is unreasonable which we could add a Jira to fix the issue. We suggest our users to use backfill mode to restart the job instead of restart from cp/sp files. 3. When we use 'state.backend.rocksdb.timer-service.factory: HEAP', the job may consume more memory. Therefore, we need to pay close attention to memory usage and increase memory resources if necessary. JING ZHANG 于2021年8月20日周五 下午1:27写道: > Hi Vijay, Yun, > I've created a JIRA https://issues.apache.org/jira/browse/FLINK-23886 to > track this. > > Best, > JING ZHANG > > JING ZHANG 于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 于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.(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.apac