HADOOP-19478. S3A: pull out new configuration load/probes under S3AStore

New service under S3AStoreImpl: StoreConfigurationService

Change-Id: I99c8305574492c05170274dcc363bfba4857981b

Author: steveloughran

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

@@ -1531,6 +1531,14 @@ private Constants() {
    */
   public static final String FS_S3A_CREATE_OVERWRITE_SUPPORTED = "fs.s3a.create.overwrite.supported";
 
+  /**
+   * If conditional create is available, should it be used in
+   * createFile() operations to check for file existence?
+   * If set, this disables probes for directories.
+   * Value {@value}.
+   */
+  public static final String FS_S3A_CONDITIONAL_CREATE_FOR_FILES = "fs.s3a.conditional.create.for.files";
+
   /**
    * Create a multipart file, always: {@value}.
    * <p>

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3AStoreBuilder.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.fs.s3a.S3AStorageStatistics;
 import org.apache.hadoop.fs.s3a.S3AStore;
 import org.apache.hadoop.fs.s3a.audit.AuditSpanS3A;
+import org.apache.hadoop.fs.s3a.impl.store.StoreConfigurationService;
 import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
 import org.apache.hadoop.fs.store.audit.AuditSpanSource;
@@ -51,6 +52,8 @@ public class S3AStoreBuilder {
 
   private AuditSpanSource<AuditSpanS3A> auditSpanSource;
 
+  private StoreConfigurationService storeConfigurationService;
+
   /**
    * The original file system statistics: fairly minimal but broadly
    * collected so it is important to pick up.
@@ -117,6 +120,16 @@ public S3AStoreBuilder withFsStatistics(final FileSystem.Statistics value) {
     return this;
   }
 
+  /**
+   * Set the store configuration service.
+   * @param value new value
+   * @return the builder
+   */
+  public S3AStoreBuilder withStoreConfigurationService(final StoreConfigurationService value) {
+    storeConfigurationService = value;
+    return this;
+  }
+
   public S3AStore build() {
     return new S3AStoreImpl(storeContextFactory,
         clientManager,
@@ -127,6 +140,6 @@ public S3AStore build() {
         readRateLimiter,
         writeRateLimiter,
         auditSpanSource,
-        fsStatistics);
+        fsStatistics, storeConfigurationService);
   }
 }

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3AStoreImpl.java

@@ -79,6 +79,7 @@
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStreamFactory;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectReadParameters;
 import org.apache.hadoop.fs.s3a.impl.streams.StreamFactoryRequirements;
+import org.apache.hadoop.fs.s3a.impl.store.StoreConfigurationService;
 import org.apache.hadoop.fs.s3a.statistics.S3AStatisticsContext;
 import org.apache.hadoop.fs.statistics.DurationTracker;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
@@ -193,10 +194,12 @@ public class S3AStoreImpl
    */
   private ObjectInputStreamFactory objectInputStreamFactory;
 
+  private final StoreConfigurationService storeConfigurationService;
+
   /**
    * Constructor to create S3A store.
    * Package private, as {@link S3AStoreBuilder} creates them.
-   * */
+   */
   S3AStoreImpl(StoreContextFactory storeContextFactory,
       ClientManager clientManager,
       DurationTrackerFactory durationTrackerFactory,
@@ -206,7 +209,8 @@ public class S3AStoreImpl
       RateLimiting readRateLimiter,
       RateLimiting writeRateLimiter,
       AuditSpanSource<AuditSpanS3A> auditSpanSource,
-      @Nullable FileSystem.Statistics fsStatistics) {
+      @Nullable FileSystem.Statistics fsStatistics,
+      StoreConfigurationService storeConfigurationService) {
     super("S3AStore");
     this.auditSpanSource = requireNonNull(auditSpanSource);
     this.clientManager = requireNonNull(clientManager);
@@ -224,6 +228,8 @@ public class S3AStoreImpl
     this.bucket = requireNonNull(storeContext.getBucket());
     this.requestFactory = requireNonNull(storeContext.getRequestFactory());
     addService(clientManager);
+    this.storeConfigurationService = requireNonNull(storeConfigurationService);
+    addService(storeConfigurationService);
   }
 
   /**

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/store/StoreConfigurationFlags.java

@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.impl.store;
+
+import org.apache.hadoop.conf.Configuration;
+
+import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CONDITIONAL_CREATE_FOR_FILES;
+import static org.apache.hadoop.fs.s3a.Constants.FS_S3A_CREATE_OVERWRITE_SUPPORTED;
+
+/**
+ * Store configuration flags.
+ */
+public enum StoreConfigurationFlags {
+
+  ConditionCreateAvailable(FS_S3A_CREATE_OVERWRITE_SUPPORTED,
+      true),
+  UseConditionalCreateForFiles(FS_S3A_CONDITIONAL_CREATE_FOR_FILES,
+      false);
+
+  /**
+   * Key name; read from the configuration, and
+   * for the capability probe unless the arity 3
+   * constructor is used.
+   */
+  private final String key;
+
+  /**
+   * Capability to probe for in {@link #hasCapability(String)}.
+   */
+  private final String capability;
+
+  /**
+   * Default value when reading from the configuration.
+   */
+  private final boolean defaultValue;
+
+  StoreConfigurationFlags(String key, boolean defaultValue) {
+    this(key, "", defaultValue);
+  }
+
+  StoreConfigurationFlags(String key,
+      String capability,
+      boolean defaultValue) {
+    this.key = key;
+    this.capability = capability;
+    this.defaultValue = defaultValue;
+  }
+
+  public String getKey() {
+    return key;
+  }
+
+  public String getCapability() {
+    return capability;
+  }
+
+  /**
+   * Read from the the configuration, falling
+   * back to the default value.
+   * @param conf configuration.
+   * @return the evaluated value.
+   */
+  public boolean evaluate(Configuration conf) {
+    return conf.getBoolean(key, defaultValue);
+  }
+
+  /**
+   * Does this enum's key match the supplied key.
+   * @param k key to probe for
+   * @return true if there is a match.
+   */
+  public boolean keyMatches(String k) {
+    return key.equals(k);
+  }
+
+  /**
+   * Does this enum's capability match the supplied key?
+   * @param k key to probe for
+   * @return true if there is a match.
+   */
+  public boolean hasCapability(String k) {
+    return !capability.isEmpty() &&
+        capability.equals(k);
+  }
+
+
+}

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/store/StoreConfigurationService.java

@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a.impl.store;
+
+import java.util.Arrays;
+import java.util.EnumSet;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.service.AbstractService;
+
+/**
+ * A service which handles store configurations.
+ * New configuration options should be added here.
+ * <p>
+ *   The goal is to pull configuration flags and variables
+ *   out of S3AFileSystem but not reimplement the
+ *   same structure in S3AStore.
+ *   Instead, configuration flags, numbers etc can
+ *   be managed here.
+ *   Maybe in future reflection could be used to
+ *   build up the config, as done in ABFS.
+ * <p>
+ * Usage.
+ * <ol>
+ *   <li>Instantiate.</li>
+ *   <li>Call {@link #init(Configuration)} to trigger config reading</lib>
+ *   <li>Read loaded options.</li>
+ * </ol>
+ * The start and close operations are (currently) no-ops.
+ */
+public class StoreConfigurationService extends AbstractService
+  implements StreamCapabilities {
+
+  private EnumSet<StoreConfigurationFlags> configurationFlags;
+
+  public StoreConfigurationService(final String name) {
+    super(name);
+  }
+
+  public StoreConfigurationService() {
+    super("StoreConfigurationService");
+  }
+
+  /**
+   * Initialize the service by reading in configuration
+   * settings.
+   * @param conf configuration
+   * @throws Exception parser failures.
+   */
+  @Override
+  protected void serviceInit(final Configuration conf) throws Exception {
+    super.serviceInit(conf);
+    configurationFlags = EnumSet.noneOf(StoreConfigurationFlags.class);
+    Arrays.stream(StoreConfigurationFlags.values())
+        .filter(v -> v.evaluate(conf))
+        .forEach(configurationFlags::add);
+  }
+
+  /**
+   * Is a configuration flag set?
+   * @param flag flag to probe for.
+   * @return true iff the flag is set
+   */
+  public boolean isFlagSet(StoreConfigurationFlags flag) {
+    return configurationFlags.contains(flag);
+  }
+
+  /**
+   * Get a clone of the flags.
+   * @return a copy of the flags.
+   */
+  public EnumSet<StoreConfigurationFlags> getConfigurationFlags() {
+    return configurationFlags.clone();
+  }
+
+  /**
+   * Does one of the flags have this capability?
+   * @param capability what to probe for
+   * @return true if the capability is implellmented.
+   */
+  @Override
+  public boolean hasCapability(String capability) {
+    return configurationFlags.stream()
+        .anyMatch(f -> f.keyMatches(capability));
+  }
+
+  /**
+   * Set a flag.
+   * This is NOT thread safe.
+   * @param flag flag to set
+   * @return true if the flag enumset changed state.
+   */
+  public boolean setFlag(StoreConfigurationFlags flag) {
+    return configurationFlags.add(flag);
+  }
+
+  /**
+   * Clear a flag.
+   * This is NOT thread safe.
+   * @param flag flag to clear
+   * @return true if the flag enumset changed state.
+   */
+  public boolean clearFlag(StoreConfigurationFlags flag) {
+    return configurationFlags.remove(flag);
+  }
+
+}