Merge branch 'trunk' into YARN-10885

.asf.yaml

@@ -22,4 +22,4 @@ notifications:
   commits:      common-commits@hadoop.apache.org
   issues:       common-issues@hadoop.apache.org
   pullrequests: common-issues@hadoop.apache.org
-  jira_options: link label worklog
+  jira_options: comment link label

hadoop-common-project/hadoop-common/pom.xml

@@ -1151,7 +1151,7 @@
                 <id>src-test-compile-protoc-legacy</id>
                 <phase>generate-test-sources</phase>
                 <goals>
-                  <goal>compile</goal>
+                  <goal>test-compile</goal>
                 </goals>
                 <configuration>
                   <skip>false</skip>
@@ -1160,7 +1160,7 @@
                     com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}
                   </protocArtifact>
                   <includeDependenciesInDescriptorSet>false</includeDependenciesInDescriptorSet>
-                  <protoSourceRoot>${basedir}/src/test/proto</protoSourceRoot>
+                  <protoTestSourceRoot>${basedir}/src/test/proto</protoTestSourceRoot>
                   <outputDirectory>${project.build.directory}/generated-test-sources/java</outputDirectory>
                   <clearOutputDirectory>false</clearOutputDirectory>
                   <includes>

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

@@ -480,13 +480,13 @@ public class CommonConfigurationKeys extends CommonConfigurationKeysPublic {
    * Thread-level IOStats Support.
    * {@value}
    */
-  public static final String THREAD_LEVEL_IOSTATISTICS_ENABLED =
-      "fs.thread.level.iostatistics.enabled";
+  public static final String IOSTATISTICS_THREAD_LEVEL_ENABLED =
+      "fs.iostatistics.thread.level.enabled";
 
   /**
    * Default value for Thread-level IOStats Support is true.
    */
-  public static final boolean THREAD_LEVEL_IOSTATISTICS_ENABLED_DEFAULT =
+  public static final boolean IOSTATISTICS_THREAD_LEVEL_ENABLED_DEFAULT =
       true;
 
 }

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

@@ -256,9 +256,8 @@ public <T extends FileSystem & Renewable> void removeRenewAction(
       try {
         action.cancel();
       } catch (InterruptedException ie) {
-        LOG.error("Interrupted while canceling token for " + fs.getUri()
-            + "filesystem");
-        LOG.debug("Exception in removeRenewAction: {}", ie);
+        LOG.error("Interrupted while canceling token for {} filesystem.", fs.getUri());
+        LOG.debug("Exception in removeRenewAction.", ie);
       }
     }
   }

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

@@ -80,4 +80,14 @@ static void setThreadIOStatisticsContext(
     IOStatisticsContextIntegration.setThreadIOStatisticsContext(
         statisticsContext);
   }
+
+  /**
+   * Static probe to check if the thread-level IO statistics enabled.
+   *
+   * @return if the thread-level IO statistics enabled.
+   */
+  static boolean enabled() {
+    return IOStatisticsContextIntegration.isIOStatisticsThreadLevelEnabled();
+  }
+
 }

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

@@ -29,8 +29,8 @@
 import org.apache.hadoop.fs.impl.WeakReferenceThreadMap;
 import org.apache.hadoop.fs.statistics.IOStatisticsContext;
 
-import static org.apache.hadoop.fs.CommonConfigurationKeys.THREAD_LEVEL_IOSTATISTICS_ENABLED;
-import static org.apache.hadoop.fs.CommonConfigurationKeys.THREAD_LEVEL_IOSTATISTICS_ENABLED_DEFAULT;
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_THREAD_LEVEL_ENABLED;
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_THREAD_LEVEL_ENABLED_DEFAULT;
 
 /**
  * A Utility class for IOStatisticsContext, which helps in creating and
@@ -76,8 +76,17 @@ public final class IOStatisticsContextIntegration {
     // Work out if the current context has thread level IOStatistics enabled.
     final Configuration configuration = new Configuration();
     isThreadIOStatsEnabled =
-        configuration.getBoolean(THREAD_LEVEL_IOSTATISTICS_ENABLED,
-            THREAD_LEVEL_IOSTATISTICS_ENABLED_DEFAULT);
+        configuration.getBoolean(IOSTATISTICS_THREAD_LEVEL_ENABLED,
+            IOSTATISTICS_THREAD_LEVEL_ENABLED_DEFAULT);
+  }
+
+  /**
+   * Static probe to check if the thread-level IO statistics enabled.
+   *
+   * @return if the thread-level IO statistics enabled.
+   */
+  public static boolean isIOStatisticsThreadLevelEnabled() {
+    return isThreadIOStatsEnabled;
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/AlreadyClosedException.java

@@ -0,0 +1,33 @@
+/*
+ * 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.io.compress;
+
+import java.io.IOException;
+
+/**
+ * An exception class for when a closed compressor/decopressor is being used
+ * {@link org.apache.hadoop.io.compress.Compressor}
+ * {@link org.apache.hadoop.io.compress.Decompressor}
+ */
+public class AlreadyClosedException extends IOException {
+
+  public AlreadyClosedException(String message) {
+    super(message);
+  }
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/CodecPool.java

@@ -205,6 +205,7 @@ public static void returnCompressor(Compressor compressor) {
     }
     // if the compressor can't be reused, don't pool it.
     if (compressor.getClass().isAnnotationPresent(DoNotPool.class)) {
+      compressor.end();
       return;
     }
     compressor.reset();
@@ -225,6 +226,7 @@ public static void returnDecompressor(Decompressor decompressor) {
     }
     // if the decompressor can't be reused, don't pool it.
     if (decompressor.getClass().isAnnotationPresent(DoNotPool.class)) {
+      decompressor.end();
       return;
     }
     decompressor.reset();

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zlib/BuiltInGzipCompressor.java

@@ -24,6 +24,7 @@
 import java.util.zip.GZIPOutputStream;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.compress.AlreadyClosedException;
 import org.apache.hadoop.io.compress.Compressor;
 import org.apache.hadoop.io.compress.DoNotPool;
 import org.apache.hadoop.util.DataChecksum;
@@ -83,6 +84,10 @@ public int compress(byte[] b, int off, int len) throws IOException {
       throw new IOException("compress called on finished compressor");
     }
 
+    if (state == BuiltInGzipDecompressor.GzipStateLabel.ENDED) {
+      throw new AlreadyClosedException("compress called on closed compressor");
+    }
+
     int compressedBytesWritten = 0;
 
     // If we are not within uncompressed data yet, output the header.
@@ -139,6 +144,8 @@ public long getBytesWritten() {
   @Override
   public void end() {
     deflater.end();
+
+    state = BuiltInGzipDecompressor.GzipStateLabel.ENDED;
   }
 
   @Override

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zlib/BuiltInGzipDecompressor.java

@@ -23,6 +23,7 @@
 import java.util.zip.DataFormatException;
 import java.util.zip.Inflater;
 
+import org.apache.hadoop.io.compress.AlreadyClosedException;
 import org.apache.hadoop.io.compress.Decompressor;
 import org.apache.hadoop.io.compress.DoNotPool;
 import org.apache.hadoop.util.DataChecksum;
@@ -109,7 +110,11 @@ public enum GzipStateLabel {
      * Immediately after the trailer (and potentially prior to the next gzip
      * member/substream header), without reset() having been called.
      */
-    FINISHED;
+    FINISHED,
+    /**
+     * Immediately after end() has been called.
+     */
+    ENDED;
   }
 
   /**
@@ -186,6 +191,10 @@ public synchronized int decompress(byte[] b, int off, int len)
   throws IOException {
     int numAvailBytes = 0;
 
+    if (state == GzipStateLabel.ENDED) {
+      throw new AlreadyClosedException("decompress called on closed decompressor");
+    }
+
     if (state != GzipStateLabel.DEFLATE_STREAM) {
       executeHeaderState();
 
@@ -476,6 +485,8 @@ public synchronized void reset() {
   @Override
   public synchronized void end() {
     inflater.end();
+
+    state = GzipStateLabel.ENDED;
   }
 
   /**