[FLINK-7266] [core] Prevent attempt for parent directory deletion for…

… object stores This closes apache#4397
laubersder · Nov 17, 2017 · b00f1b3 · b00f1b3
1 parent f29f805
commit b00f1b3
Showing 1 changed file with 6 additions and 4 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FileStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FileStateHandle.java
@@ -20,6 +20,7 @@
 
 import org.apache.flink.core.fs.FSDataInputStream;
 import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.FileSystemKind;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.state.StreamStateHandle;
 import org.apache.flink.util.FileUtils;
@@ -77,14 +78,15 @@ public FSDataInputStream openInputStream() throws IOException {
  */
  @Override
  public void discardState() throws Exception {
-
  FileSystem fs = getFileSystem();
 
  fs.delete(filePath, false);
 
- try {
- FileUtils.deletePathIfEmpty(fs, filePath.getParent());
- } catch (Exception ignored) {}
+ if (fs.getKind() == FileSystemKind.FILE_SYSTEM) {
+ try {
+ FileUtils.deletePathIfEmpty(fs, filePath.getParent());
+ } catch (Exception ignored) {}
+ }
  }
 
  /**