MAPREDUCE-7435. Manifest Committer OOM on abfs (#5519)

This modifies the manifest committer so that the list of files
to rename is passed between stages as a file of
writeable entries on the local filesystem.

The map of directories to create is still passed in memory;
this map is built across all tasks, so even if many tasks
created files, if they all write into the same set of directories
the memory needed is O(directories) with the
task count not a factor.

The _SUCCESS file reports on heap size through gauges.
This should give a warning if there are problems.

Contributed by Steve Loughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/IOStatisticsSetters.java

@@ -0,0 +1,75 @@
+/*
+ * 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.statistics;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Setter for IOStatistics entries.
+ * These operations have been in the read/write API
+ * {@code IOStatisticsStore} since IOStatistics
+ * was added; extracting into its own interface allows for
+ * {@link IOStatisticsSnapshot} to also support it.
+ * These are the simple setters, they don't provide for increments,
+ * decrements, calculation of min/max/mean etc.
+ * @since The interface and IOStatisticsSnapshot support was added <i>after</i> Hadoop 3.3.5
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface IOStatisticsSetters extends IOStatistics {
+
+  /**
+   * Set a counter.
+   *
+   * No-op if the counter is unknown.
+   * @param key statistics key
+   * @param value value to set
+   */
+  void setCounter(String key, long value);
+
+  /**
+   * Set a gauge.
+   *
+   * @param key statistics key
+   * @param value value to set
+   */
+  void setGauge(String key, long value);
+
+  /**
+   * Set a maximum.
+   * @param key statistics key
+   * @param value value to set
+   */
+  void setMaximum(String key, long value);
+
+  /**
+   * Set a minimum.
+   * @param key statistics key
+   * @param value value to set
+   */
+  void setMinimum(String key, long value);
+
+  /**
+   * Set a mean statistic to a given value.
+   * @param key statistic key
+   * @param value new value.
+   */
+  void setMeanStatistic(String key, MeanStatistic value);
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/IOStatisticsSnapshot.java

@@ -62,7 +62,8 @@
 @InterfaceAudience.Public
 @InterfaceStability.Evolving
 public final class IOStatisticsSnapshot
-    implements IOStatistics, Serializable, IOStatisticsAggregator {
+    implements IOStatistics, Serializable, IOStatisticsAggregator,
+    IOStatisticsSetters {
 
   private static final long serialVersionUID = -1762522703841538084L;
 
@@ -222,6 +223,33 @@ public synchronized Map<String, MeanStatistic> meanStatistics() {
     return meanStatistics;
   }
 
+  @Override
+  public synchronized void setCounter(final String key, final long value) {
+    counters().put(key, value);
+  }
+
+  @Override
+  public synchronized void setGauge(final String key, final long value) {
+    gauges().put(key, value);
+
+  }
+
+  @Override
+  public synchronized void setMaximum(final String key, final long value) {
+    maximums().put(key, value);
+
+  }
+
+  @Override
+  public synchronized void setMinimum(final String key, final long value) {
+    minimums().put(key, value);
+  }
+
+  @Override
+  public void setMeanStatistic(final String key, final MeanStatistic value) {
+    meanStatistics().put(key, value);
+  }
+
   @Override
   public String toString() {
     return ioStatisticsToString(this);

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/statistics/impl/ForwardingIOStatisticsStore.java → hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/impl/ForwardingIOStatisticsStore.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.hadoop.fs.s3a.statistics.impl;
+package org.apache.hadoop.fs.statistics.impl;
 
 import javax.annotation.Nullable;
 import java.time.Duration;
@@ -25,7 +25,6 @@
 
 import org.apache.hadoop.fs.statistics.IOStatistics;
 import org.apache.hadoop.fs.statistics.MeanStatistic;
-import org.apache.hadoop.fs.statistics.impl.IOStatisticsStore;
 
 /**
  * This may seem odd having an IOStatisticsStore which does nothing

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/impl/IOStatisticsStore.java

@@ -24,13 +24,15 @@
 import org.apache.hadoop.fs.statistics.IOStatistics;
 import org.apache.hadoop.fs.statistics.IOStatisticsAggregator;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
+import org.apache.hadoop.fs.statistics.IOStatisticsSetters;
 import org.apache.hadoop.fs.statistics.MeanStatistic;
 
 /**
  * Interface of an IOStatistics store intended for
  * use in classes which track statistics for reporting.
  */
 public interface IOStatisticsStore extends IOStatistics,
+    IOStatisticsSetters,
     IOStatisticsAggregator,
     DurationTrackerFactory {
 
@@ -56,24 +58,6 @@ default long incrementCounter(String key) {
    */
   long incrementCounter(String key, long value);
 
-  /**
-   * Set a counter.
-   *
-   * No-op if the counter is unknown.
-   * @param key statistics key
-   * @param value value to set
-   */
-  void setCounter(String key, long value);
-
-  /**
-   * Set a gauge.
-   *
-   * No-op if the gauge is unknown.
-   * @param key statistics key
-   * @param value value to set
-   */
-  void setGauge(String key, long value);
-
   /**
    * Increment a gauge.
    * <p>
@@ -85,14 +69,6 @@ default long incrementCounter(String key) {
    */
   long incrementGauge(String key, long value);
 
-  /**
-   * Set a maximum.
-   * No-op if the maximum is unknown.
-   * @param key statistics key
-   * @param value value to set
-   */
-  void setMaximum(String key, long value);
-
   /**
    * Increment a maximum.
    * <p>
@@ -104,16 +80,6 @@ default long incrementCounter(String key) {
    */
   long incrementMaximum(String key, long value);
 
-  /**
-   * Set a minimum.
-   * <p>
-   * No-op if the minimum is unknown.
-   * </p>
-   * @param key statistics key
-   * @param value value to set
-   */
-  void setMinimum(String key, long value);
-
   /**
    * Increment a minimum.
    * <p>
@@ -147,16 +113,6 @@ default long incrementCounter(String key) {
    */
   void addMaximumSample(String key, long value);
 
-  /**
-   * Set a mean statistic to a given value.
-   * <p>
-   * No-op if the key is unknown.
-   * </p>
-   * @param key statistic key
-   * @param value new value.
-   */
-  void setMeanStatistic(String key, MeanStatistic value);
-
   /**
    * Add a sample to the mean statistics.
    * <p>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/impl/IOStatisticsStoreBuilder.java

@@ -67,6 +67,17 @@ public interface IOStatisticsStoreBuilder {
   IOStatisticsStoreBuilder withDurationTracking(
       String... prefixes);
 
+  /**
+   * A value which is tracked with counter/min/max/mean.
+   * Similar to {@link #withDurationTracking(String...)}
+   * but without the failure option and with the same name
+   * across all categories.
+   * @param prefixes prefixes to add.
+   * @return the builder
+   */
+  IOStatisticsStoreBuilder withSampleTracking(
+      String... prefixes);
+
   /**
    * Build the collector.
    * @return a new collector.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/impl/IOStatisticsStoreBuilderImpl.java

@@ -92,6 +92,18 @@ public IOStatisticsStoreBuilderImpl withDurationTracking(
     return this;
   }
 
+  @Override
+  public IOStatisticsStoreBuilderImpl withSampleTracking(
+      final String... prefixes) {
+    for (String p : prefixes) {
+      withCounters(p);
+      withMinimums(p);
+      withMaximums(p);
+      withMeanStatistics(p);
+    }
+    return this;
+  }
+
   @Override
   public IOStatisticsStore build() {
     return new IOStatisticsStoreImpl(counters, gauges, minimums,