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 910bd236c93c24..20bead053bb48f 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; @@ -99,16 +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. - else if (config.containsKey(ConfigConstants.STATE_BACKEND) && - config.containsKey(ConfigConstants.ZOOKEEPER_RECOVERY_PATH)) { - + // Recovery. + else if (recoveryMode == RecoveryMode.ZOOKEEPER) { this.blobStore = new FileSystemBlobStore(config); - } - // Fallback. - else { - this.blobStore = new VoidBlobStore(); + } 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 e7a306b8720b46..efdb003cf34f95 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 { @@ -49,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 79bd28b7aec153..4bcc668d7e4b29 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");