Merge branch 'apache:trunk' into yarnuijs

Author: riya2305

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/sink/PrometheusMetricsSink.java

@@ -19,7 +19,13 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 import java.util.regex.Matcher;
+
+import org.apache.hadoop.thirdparty.com.google.common.cache.CacheBuilder;
+import org.apache.hadoop.thirdparty.com.google.common.cache.CacheLoader;
+import org.apache.hadoop.thirdparty.com.google.common.cache.LoadingCache;
+import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.UncheckedExecutionException;
 import org.apache.commons.configuration2.SubsetConfiguration;
 import org.apache.hadoop.metrics2.AbstractMetric;
 import org.apache.hadoop.metrics2.MetricType;
@@ -35,13 +41,16 @@
 import java.util.regex.Pattern;
 
 import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Metrics sink for prometheus exporter.
  * <p>
  * Stores the metric data in-memory and return with it on request.
  */
 public class PrometheusMetricsSink implements MetricsSink {
+  private static final Logger LOG = LoggerFactory.getLogger(PrometheusMetricsSink.class);
 
   /**
    * Cached output lines for each metrics.
@@ -62,6 +71,17 @@ public class PrometheusMetricsSink implements MetricsSink {
       Pattern
           .compile("^op=(?<op>\\w+)(.user=(?<user>.*)|)\\.(TotalCount|count)$");
 
+  /**
+   * A fixed cache for Hadoop metric to Prometheus metric name conversion.
+   */
+  private static final int NORMALIZED_NAME_CACHE_MAX_SIZE = 100_000;
+  private static final CacheLoader<String, String> NORMALIZED_NAME_CACHE_LOADER =
+      CacheLoader.from(PrometheusMetricsSink::normalizeImpl);
+  private static final LoadingCache<String, String> NORMALIZED_NAME_CACHE =
+      CacheBuilder.newBuilder()
+          .maximumSize(NORMALIZED_NAME_CACHE_MAX_SIZE)
+          .build(NORMALIZED_NAME_CACHE_LOADER);
+
   public PrometheusMetricsSink() {
   }
 
@@ -83,7 +103,21 @@ public void putMetrics(MetricsRecord metricsRecord) {
 
   /**
    * Convert CamelCase based names to lower-case names where the separator
-   * is the underscore, to follow prometheus naming conventions.
+   * is the underscore, to follow prometheus naming conventions. This method
+   * utilizes a cache to improve performance.
+   *
+   * <p>
+   *  Reference:
+   *  <ul>
+   *    <li>
+   *      <a href="https://prometheus.io/docs/practices/naming/">
+   *        Metrics and Label Naming</a>
+   *    </li>
+   *    <li>
+   *      <a href="https://prometheus.io/docs/instrumenting/exposition_formats/">
+   *        Exposition formats</a>
+   *    </li>
+   *  </ul>
    *
    * @param metricName metricName.
    * @param recordName recordName.
@@ -93,6 +127,22 @@ public String prometheusName(String recordName,
                                String metricName) {
     String baseName = StringUtils.capitalize(recordName)
         + StringUtils.capitalize(metricName);
+    try {
+      return NORMALIZED_NAME_CACHE.get(baseName);
+    } catch (ExecutionException | UncheckedExecutionException e) {
+      // This should not happen since normalization function do not throw any exception
+      // Nevertheless, we can fall back to uncached implementation if it somehow happens.
+      LOG.warn("Exception encountered when loading metric with base name {} from cache, " +
+          "fall back to uncached normalization implementation", baseName, e);
+      return normalizeImpl(baseName);
+    }
+  }
+
+  /**
+   * Underlying Prometheus normalization implementation.
+   * See {@link PrometheusMetricsSink#prometheusName(String, String)} for more information.
+   */
+  private static String normalizeImpl(String baseName) {
     String[] parts = SPLIT_PATTERN.split(baseName);
     String joined =  String.join("_", parts).toLowerCase();
     return DELIMITERS.matcher(joined).replaceAll("_");

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

@@ -57,6 +57,7 @@
 import software.amazon.awssdk.transfer.s3.model.FileUpload;
 import software.amazon.awssdk.transfer.s3.model.UploadFileRequest;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalDirAllocator;
@@ -287,10 +288,11 @@ public boolean inputStreamHasCapability(final String capability) {
    * Initialize dir allocator if not already initialized.
    */
   private void initLocalDirAllocator() {
-    String bufferDir = getConfig().get(BUFFER_DIR) != null
-        ? BUFFER_DIR
-        : HADOOP_TMP_DIR;
-    directoryAllocator = new LocalDirAllocator(bufferDir);
+    String key = BUFFER_DIR;
+    if (StringUtils.isEmpty(getConfig().getTrimmed(key))) {
+      key = HADOOP_TMP_DIR;
+    }
+    directoryAllocator = new LocalDirAllocator(key);
   }
 
   /** Acquire write capacity for rate limiting {@inheritDoc}. */

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

@@ -21,6 +21,8 @@
 
 import java.io.IOException;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import software.amazon.s3.analyticsaccelerator.S3SdkObjectClient;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamConfiguration;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamFactory;
@@ -40,6 +42,8 @@
  *  {@code S3AStore}, if fs.s3a.input.stream.type is set to Analytics.
  */
 public class AnalyticsStreamFactory extends AbstractObjectInputStreamFactory {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(AnalyticsStreamFactory.class);
 
   private S3SeekableInputStreamConfiguration seekableInputStreamConfiguration;
   private LazyAutoCloseableReference<S3SeekableInputStreamFactory>  s3SeekableInputStreamFactory;
@@ -98,7 +102,11 @@ public StreamFactoryRequirements factoryRequirements() {
 
   @Override
   protected void serviceStop() throws Exception {
-    this.s3SeekableInputStreamFactory.close();
+    try {
+      s3SeekableInputStreamFactory.close();
+    } catch (Exception ignored) {
+      LOG.debug("Ignored exception while closing stream factory", ignored);
+    }
     callbacks().incrementFactoryStatistic(ANALYTICS_STREAM_FACTORY_CLOSED);
     super.serviceStop();
   }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AConfiguration.java

@@ -49,6 +49,7 @@
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.contract.ContractTestUtils;
 import org.apache.hadoop.fs.s3a.auth.STSClientFactory;
@@ -63,6 +64,7 @@
 import org.apache.hadoop.util.VersionInfo;
 import org.apache.http.HttpStatus;
 
+import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 import static org.apache.hadoop.fs.contract.ContractTestUtils.skip;
 import static org.apache.hadoop.fs.s3a.Constants.*;
@@ -485,12 +487,29 @@ public void testCloseIdempotent() throws Throwable {
 
   @Test
   public void testDirectoryAllocatorDefval() throws Throwable {
+    removeAllocatorContexts();
     conf = new Configuration();
-    conf.unset(Constants.BUFFER_DIR);
-    fs = S3ATestUtils.createTestFileSystem(conf);
-    File tmp = createTemporaryFileForWriting();
-    assertTrue("not found: " + tmp, tmp.exists());
-    tmp.delete();
+    final String bucketName = getTestBucketName(conf);
+    final String blank = " ";
+    conf.set(Constants.BUFFER_DIR, blank);
+    conf.set(format("fs.s3a.bucket.%s.buffer.dir", bucketName), blank);
+    try {
+      fs = S3ATestUtils.createTestFileSystem(conf);
+      final Configuration fsConf = fs.getConf();
+      Assertions.assertThat(fsConf.get(Constants.BUFFER_DIR))
+          .describedAs("Config option %s", Constants.BUFFER_DIR)
+          .isEqualTo(blank);
+      File tmp = createTemporaryFileForWriting();
+      assertTrue("not found: " + tmp, tmp.exists());
+      tmp.delete();
+    } finally {
+      removeAllocatorContexts();
+    }
+  }
+
+  private static void removeAllocatorContexts() {
+    LocalDirAllocator.removeContext(BUFFER_DIR);
+    LocalDirAllocator.removeContext(HADOOP_TMP_DIR);
   }
 
   /**
@@ -504,13 +523,21 @@ private File createTemporaryFileForWriting() throws IOException {
 
   @Test
   public void testDirectoryAllocatorRR() throws Throwable {
+    removeAllocatorContexts();
     File dir1 = GenericTestUtils.getRandomizedTestDir();
     File dir2 = GenericTestUtils.getRandomizedTestDir();
     dir1.mkdirs();
     dir2.mkdirs();
     conf = new Configuration();
-    conf.set(Constants.BUFFER_DIR, dir1 + ", " + dir2);
+    final String bucketName = getTestBucketName(conf);
+    final String dirs = dir1 + ", " + dir2;
+    conf.set(Constants.BUFFER_DIR, dirs);
+    conf.set(format("fs.s3a.bucket.%s.buffer.dir", bucketName), dirs);
     fs = S3ATestUtils.createTestFileSystem(conf);
+    final Configuration fsConf = fs.getConf();
+    Assertions.assertThat(fsConf.get(Constants.BUFFER_DIR))
+        .describedAs("Config option %s", Constants.BUFFER_DIR)
+        .isEqualTo(dirs);
     File tmp1 = createTemporaryFileForWriting();
     tmp1.delete();
     File tmp2 = createTemporaryFileForWriting();
@@ -552,10 +579,10 @@ public S3AFileSystem run() throws Exception{
   private static <T> T getField(Object target, Class<T> fieldType,
       String fieldName) throws IllegalAccessException {
     Object obj = FieldUtils.readField(target, fieldName, true);
-    assertNotNull(String.format(
+    assertNotNull(format(
         "Could not read field named %s in object with class %s.", fieldName,
         target.getClass().getName()), obj);
-    assertTrue(String.format(
+    assertTrue(format(
         "Unexpected type found for field named %s, expected %s, actual %s.",
         fieldName, fieldType.getName(), obj.getClass().getName()),
         fieldType.isAssignableFrom(obj.getClass()));

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java

@@ -76,7 +76,6 @@
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidUriException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.TrileanConversionException;
 import org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode;
-import org.apache.hadoop.fs.azurebfs.services.AbfsBlobClient;
 import org.apache.hadoop.fs.azurebfs.services.ListResponseData;
 import org.apache.hadoop.fs.azurebfs.enums.Trilean;
 import org.apache.hadoop.fs.azurebfs.extensions.EncryptionContextProvider;
@@ -118,7 +117,6 @@
 import org.apache.hadoop.fs.azurebfs.utils.CRC64;
 import org.apache.hadoop.fs.azurebfs.utils.DateTimeUtils;
 import org.apache.hadoop.fs.azurebfs.utils.EncryptionType;
-import org.apache.hadoop.fs.azurebfs.utils.ListUtils;
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 import org.apache.hadoop.fs.azurebfs.utils.UriUtils;
 import org.apache.hadoop.fs.impl.BackReference;
@@ -1298,13 +1296,8 @@ public String listStatus(final Path path, final String startFrom,
       }
     } while (shouldContinue);
 
-    if (listingClient instanceof AbfsBlobClient) {
-      fileStatuses.addAll(ListUtils.getUniqueListResult(fileStatusList));
-      LOG.debug("ListBlob API returned a total of {} elements including duplicates."
-          + "Number of unique Elements are {}", fileStatusList.size(), fileStatuses.size());
-    } else {
-      fileStatuses.addAll(fileStatusList);
-    }
+    fileStatuses.addAll(listingClient.postListProcessing(
+        relativePath, fileStatusList, tracingContext, uri));
 
     return continuation;
   }