[jira] [Commented] (FLINK-9233) Merging state may cause runtime exception when windows trigger onMerge

["Hai Zhou (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-9233?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16447234#comment-16447234 ] 

Hai Zhou commented on FLINK-9233:
---------------------------------

I am not quite sure that upgrading the rocksdb version to 5.9 or higher can solve this problem.

[~StephanEwen], [~aljoscha], Do you have any ideas?

> Merging state may cause runtime exception when windows  trigger onMerge
> -----------------------------------------------------------------------
>
>                 Key: FLINK-9233
>                 URL: https://issues.apache.org/jira/browse/FLINK-9233
>             Project: Flink
>          Issue Type: Bug
>          Components: State Backends, Checkpointing
>    Affects Versions: 1.4.0
>            Reporter: Hai Zhou
>            Priority: Major
>
> the main logic of my flink job is as follows:
> {code:java}
> clickStream.coGroup(exposureStream).where(...).equalTo(...)
> .window(EventTimeSessionWindows.withGap())
> .trigger(new SessionMatchTrigger)
> .evictor()
> .apply();
> {code}
> {code:java}
> SessionMatchTrigger{
>     ReducingStateDescriptor  stateDesc = new ReducingStateDescriptor()
> ...
>     public boolean canMerge() {
>         return true;
>     }
>     public void onMerge(TimeWindow window, OnMergeContext ctx) {
>         ctx.mergePartitionedState(this.stateDesc);
>         ctx.registerEventTimeTimer(window.maxTimestamp());
>     }
> ....
> }
> {code}
> {panel:title=detailed error logs}
> java.lang.RuntimeException: Error while merging state.
>  at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator$Context.mergePartitionedState(WindowOperator.java:895)
>  at com.package.trigger.SessionMatchTrigger.onMerge(SessionMatchTrigger.java:56)
>  at com.package.trigger.SessionMatchTrigger.onMerge(SessionMatchTrigger.java:14)
>  at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator$Context.onMerge(WindowOperator.java:939)
>  at org.apache.flink.streaming.runtime.operators.windowing.EvictingWindowOperator$1.merge(EvictingWindowOperator.java:141)
>  at org.apache.flink.streaming.runtime.operators.windowing.EvictingWindowOperator$1.merge(EvictingWindowOperator.java:120)
>  at org.apache.flink.streaming.runtime.operators.windowing.MergingWindowSet.addWindow(MergingWindowSet.java:212)
>  at org.apache.flink.streaming.runtime.operators.windowing.EvictingWindowOperator.processElement(EvictingWindowOperator.java:119)
>  at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:207)
>  at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:69)
>  at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:264)
>  at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
>  at java.lang.Thread.run(Thread.java:745)
>  Caused by: java.lang.Exception: Error while merging state in RocksDB
>  at org.apache.flink.contrib.streaming.state.RocksDBReducingState.mergeNamespaces(RocksDBReducingState.java:186)
>  at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator$Context.mergePartitionedState(WindowOperator.java:887)
>  ... 12 more
>  Caused by: java.lang.IllegalArgumentException: Illegal value provided for SubCode.
>  at org.rocksdb.Status$SubCode.getSubCode(Status.java:109)
>  at org.rocksdb.Status.<init>(Status.java:30)
>  at org.rocksdb.RocksDB.delete(Native Method)
>  at org.rocksdb.RocksDB.delete(RocksDB.java:1110)
>  at org.apache.flink.contrib.streaming.state.RocksDBReducingState.mergeNamespaces(RocksDBReducingState.java:143)
>  ... 13 more
> {panel}
>  
> I found the reason of this error. 
>  Due to Java's
> {RocksDB.Status.SubCode}
> was out of sync with
> {include/rocksdb/status.h:SubCode}
> .
>  When running out of disc space this led to an
> {IllegalArgumentException}
> because of an invalid status code, rather than just returning the corresponding status code without an exception.
>  more details:<[https://github.com/facebook/rocksdb/pull/3050]>



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)