Github user StefanRRichter commented on a diff in the pull request: https://github.com/apache/flink/pull/5239#discussion_r159883287 --- Diff: flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java --- @@ -831,11 +834,12 @@ private StreamStateHandle materializeStateData(Path filePath) throws Exception { return result; } finally { - if (inputStream != null && closeableRegistry.unregisterCloseable(inputStream)) { + + if (closeableRegistry.unregisterCloseable(inputStream)) { inputStream.close(); } - if (outputStream != null && closeableRegistry.unregisterCloseable(outputStream)) { + if (closeableRegistry.unregisterCloseable(outputStream)) { --- End diff -- Yes, that is what it is used for. In case of cancelation, we want all blocking ops to terminate asap using this mechanism.
---