From 05f0ff5e353117894af4ba7dc096c3256d80450b Mon Sep 17 00:00:00 2001
From: zentol <chesnay@apache.org>
Date: Mon, 5 Feb 2018 13:15:29 +0100
Subject: [PATCH] [FLINK-8559][RocksDB] Release resources if snapshot operation
 fails

---
 .../contrib/streaming/state/RocksDBKeyedStateBackend.java | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

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 29a0854f485e5..55073393e70d9 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
@@ -377,7 +377,13 @@ private RunnableFuture<KeyedStateHandle> snapshotIncrementally(
 				checkpointId,
 				checkpointTimestamp);
 
-		snapshotOperation.takeSnapshot();
+		try {
+			snapshotOperation.takeSnapshot();
+		} catch (Exception e) {
+			snapshotOperation.stop();
+			snapshotOperation.releaseResources(true);
+			throw e;
+		}
 
 		return new FutureTask<KeyedStateHandle>(
 			new Callable<KeyedStateHandle>() {