From 3b2b6a823c6b2931d9e95d1c8f1f759af2dd436a Mon Sep 17 00:00:00 2001 From: Ufuk Celebi Date: Tue, 1 Mar 2016 12:23:31 +0100 Subject: [PATCH 1/2] [FLINK-3556] [runtime] Remove false check in HA blob store configuration --- .../apache/flink/runtime/blob/BlobServer.java | 18 ++++++++++-------- .../runtime/blob/FileSystemBlobStore.java | 2 ++ 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java index 910bd236c93c2..f74c4c9e2375a 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java @@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.IllegalConfigurationException; import org.apache.flink.runtime.jobmanager.RecoveryMode; import org.apache.flink.util.NetUtils; import org.slf4j.Logger; @@ -101,14 +102,15 @@ public BlobServer(Configuration config) throws IOException { } // Recovery. Check that everything has been setup correctly. This is not clean, but it's // better to resolve this with some upcoming changes to the state backend setup. - else if (config.containsKey(ConfigConstants.STATE_BACKEND) && - config.containsKey(ConfigConstants.ZOOKEEPER_RECOVERY_PATH)) { - - this.blobStore = new FileSystemBlobStore(config); - } - // Fallback. - else { - this.blobStore = new VoidBlobStore(); + else if (recoveryMode == RecoveryMode.ZOOKEEPER) { + if (config.containsKey(ConfigConstants.ZOOKEEPER_RECOVERY_PATH)) { + this.blobStore = new FileSystemBlobStore(config); + } else { + throw new IllegalConfigurationException("Missing '" + ConfigConstants.ZOOKEEPER_RECOVERY_PATH + + "' configuration key. Please configure a path for recovery files."); + } + } else { + throw new IllegalConfigurationException("Unexpected recovery mode '" + recoveryMode + "'."); } // configure the maximum number of concurrent connections diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java index e7a306b8720b4..1bd443f6b2530 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java @@ -40,6 +40,8 @@ /** * Blob store backed by {@link FileSystem}. + * + *

This is used in addition to the local blob storage. */ class FileSystemBlobStore implements BlobStore { From 345ee4798ec0091fcde98d976e1aadef4d9366f3 Mon Sep 17 00:00:00 2001 From: Ufuk Celebi Date: Tue, 1 Mar 2016 14:32:04 +0100 Subject: [PATCH 2/2] [pr-comments] Check config in FileSystemBlobStore --- .../org/apache/flink/runtime/blob/BlobServer.java | 12 +++--------- .../flink/runtime/blob/FileSystemBlobStore.java | 15 ++++++--------- .../apache/flink/runtime/util/ZooKeeperUtils.java | 1 - 3 files changed, 9 insertions(+), 19 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java index f74c4c9e2375a..20bead053bb48 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServer.java @@ -100,17 +100,11 @@ public BlobServer(Configuration config) throws IOException { if (recoveryMode == RecoveryMode.STANDALONE) { this.blobStore = new VoidBlobStore(); } - // Recovery. Check that everything has been setup correctly. This is not clean, but it's - // better to resolve this with some upcoming changes to the state backend setup. + // Recovery. else if (recoveryMode == RecoveryMode.ZOOKEEPER) { - if (config.containsKey(ConfigConstants.ZOOKEEPER_RECOVERY_PATH)) { - this.blobStore = new FileSystemBlobStore(config); - } else { - throw new IllegalConfigurationException("Missing '" + ConfigConstants.ZOOKEEPER_RECOVERY_PATH - + "' configuration key. Please configure a path for recovery files."); - } + this.blobStore = new FileSystemBlobStore(config); } else { - throw new IllegalConfigurationException("Unexpected recovery mode '" + recoveryMode + "'."); + throw new IllegalConfigurationException("Unexpected recovery mode '" + recoveryMode + "."); } // configure the maximum number of concurrent connections diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java index 1bd443f6b2530..efdb003cf34f9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java @@ -41,7 +41,7 @@ /** * Blob store backed by {@link FileSystem}. * - *

This is used in addition to the local blob storage. + *

This is used in addition to the local blob storage */ class FileSystemBlobStore implements BlobStore { @@ -51,18 +51,15 @@ class FileSystemBlobStore implements BlobStore { private final String basePath; FileSystemBlobStore(Configuration config) throws IOException { - String stateBackendBasePath = config.getString( - ConfigConstants.ZOOKEEPER_RECOVERY_PATH, ""); + String recoveryPath = config.getString(ConfigConstants.ZOOKEEPER_RECOVERY_PATH, null); - if (stateBackendBasePath.equals("")) { + if (recoveryPath == null) { throw new IllegalConfigurationException(String.format("Missing configuration for " + - "file system state backend recovery path. Please specify via " + - "'%s' key.", ConfigConstants.ZOOKEEPER_RECOVERY_PATH)); + "file system state backend recovery path. Please specify via " + + "'%s' key.", ConfigConstants.ZOOKEEPER_RECOVERY_PATH)); } - stateBackendBasePath += "/blob"; - - this.basePath = stateBackendBasePath; + this.basePath = recoveryPath + "/blob"; FileSystem.get(new Path(basePath).toUri()).mkdirs(new Path(basePath)); LOG.info("Created blob directory {}.", basePath); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java index 79bd28b7aec15..4bcc668d7e4b2 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java @@ -225,7 +225,6 @@ public static CompletedCheckpointStore createCompletedCheckpoints( ConfigConstants.ZOOKEEPER_CHECKPOINTS_PATH, ConfigConstants.DEFAULT_ZOOKEEPER_CHECKPOINTS_PATH); - StateStorageHelper stateStorage = createFileSystemStateStorage( configuration, "completedCheckpoint");