You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2018/02/06 13:13:59 UTC
flink git commit: [FLINK-8559][RocksDB] Release resources if snapshot
operation fails
Repository: flink
Updated Branches:
refs/heads/release-1.4 33efaf74a -> c9322661e
[FLINK-8559][RocksDB] Release resources if snapshot operation fails
This closes #5412.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c9322661
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c9322661
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c9322661
Branch: refs/heads/release-1.4
Commit: c9322661ea7770a2119f51f19bcfd58dc481b4d4
Parents: 33efaf7
Author: zentol <ch...@apache.org>
Authored: Mon Feb 5 13:15:29 2018 +0100
Committer: zentol <ch...@apache.org>
Committed: Tue Feb 6 09:45:28 2018 +0100
----------------------------------------------------------------------
.../contrib/streaming/state/RocksDBKeyedStateBackend.java | 8 +++++++-
1 file changed, 7 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c9322661/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
index 9185ad0..0dd802b 100644
--- a/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
+++ b/flink-contrib/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -374,7 +374,13 @@ public class RocksDBKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
checkpointId,
checkpointTimestamp);
- snapshotOperation.takeSnapshot();
+ try {
+ snapshotOperation.takeSnapshot();
+ } catch (Exception e) {
+ snapshotOperation.stop();
+ snapshotOperation.releaseResources(true);
+ throw e;
+ }
return new FutureTask<KeyedStateHandle>(
new Callable<KeyedStateHandle>() {