Merge branch 'apache:trunk' into HADOOP-18361

.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,16 @@ 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;
 
+  public static final String HADOOP_SECURITY_RESOLVER_IMPL =
+      "hadoop.security.resolver.impl";
+
 }

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/VectoredReadUtils.java

@@ -210,6 +210,7 @@ public static List<? extends FileRange> validateNonOverlappingAndReturnSortedRan
       if (sortedRanges[i].getOffset() < prev.getOffset() + prev.getLength()) {
         throw new UnsupportedOperationException("Overlapping ranges are not supported");
       }
+      prev = sortedRanges[i];
     }
     return Arrays.asList(sortedRanges);
   }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/BlockCache.java

@@ -0,0 +1,70 @@
+/*
+ * 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.impl.prefetch;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Provides functionality necessary for caching blocks of data read from FileSystem.
+ */
+public interface BlockCache extends Closeable {
+
+  /**
+   * Indicates whether the given block is in this cache.
+   *
+   * @param blockNumber the id of the given block.
+   * @return true if the given block is in this cache, false otherwise.
+   */
+  boolean containsBlock(int blockNumber);
+
+  /**
+   * Gets the blocks in this cache.
+   *
+   * @return the blocks in this cache.
+   */
+  Iterable<Integer> blocks();
+
+  /**
+   * Gets the number of blocks in this cache.
+   *
+   * @return the number of blocks in this cache.
+   */
+  int size();
+
+  /**
+   * Gets the block having the given {@code blockNumber}.
+   *
+   * @param blockNumber the id of the desired block.
+   * @param buffer contents of the desired block are copied to this buffer.
+   * @throws IOException if there is an error reading the given block.
+   */
+  void get(int blockNumber, ByteBuffer buffer) throws IOException;
+
+  /**
+   * Puts the given block in this cache.
+   *
+   * @param blockNumber the id of the given block.
+   * @param buffer contents of the given block to be added to this cache.
+   * @throws IOException if there is an error writing the given block.
+   */
+  void put(int blockNumber, ByteBuffer buffer) throws IOException;
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/prefetch/BlockData.java

@@ -0,0 +1,250 @@
+/*
+ * 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.impl.prefetch;
+
+import static org.apache.hadoop.fs.impl.prefetch.Validate.checkNotNegative;
+import static org.apache.hadoop.fs.impl.prefetch.Validate.checkPositiveInteger;
+import static org.apache.hadoop.fs.impl.prefetch.Validate.checkWithinRange;
+
+/**
+ * Holds information about blocks of data in a file.
+ */
+public final class BlockData {
+
+  // State of each block of data.
+  enum State {
+
+    /** Data is not yet ready to be read from this block (still being prefetched). */
+    NOT_READY,
+
+    /** A read of this block has been enqueued in the prefetch queue. */
+    QUEUED,
+
+    /** A read of this block has been enqueued in the prefetch queue. */
+    READY,
+
+    /** This block has been cached in the local disk cache. */
+    CACHED
+  }
+
+  /**
+   * State of all blocks in a file.
+   */
+  private State[] state;
+
+  /**
+   * The size of a file.
+   */
+  private final long fileSize;
+
+  /**
+   * The file is divided into blocks of this size.
+   */
+  private final int blockSize;
+
+  /**
+   * The file has these many blocks.
+   */
+  private final int numBlocks;
+
+  /**
+   * Constructs an instance of {@link BlockData}.
+   * @param fileSize the size of a file.
+   * @param blockSize the file is divided into blocks of this size.
+   * @throws IllegalArgumentException if fileSize is negative.
+   * @throws IllegalArgumentException if blockSize is negative.
+   * @throws IllegalArgumentException if blockSize is zero or negative.
+   */
+  public BlockData(long fileSize, int blockSize) {
+    checkNotNegative(fileSize, "fileSize");
+    if (fileSize == 0) {
+      checkNotNegative(blockSize, "blockSize");
+    } else {
+      checkPositiveInteger(blockSize, "blockSize");
+    }
+
+    this.fileSize = fileSize;
+    this.blockSize = blockSize;
+    this.numBlocks =
+        (fileSize == 0)
+            ? 0
+            : ((int) (fileSize / blockSize)) + (fileSize % blockSize > 0
+                ? 1
+                : 0);
+    this.state = new State[this.numBlocks];
+    for (int b = 0; b < this.numBlocks; b++) {
+      setState(b, State.NOT_READY);
+    }
+  }
+
+  /**
+   * Gets the size of each block.
+   * @return the size of each block.
+   */
+  public int getBlockSize() {
+    return blockSize;
+  }
+
+  /**
+   * Gets the size of the associated file.
+   * @return the size of the associated file.
+   */
+  public long getFileSize() {
+    return fileSize;
+  }
+
+  /**
+   * Gets the number of blocks in the associated file.
+   * @return the number of blocks in the associated file.
+   */
+  public int getNumBlocks() {
+    return numBlocks;
+  }
+
+  /**
+   * Indicates whether the given block is the last block in the associated file.
+   * @param blockNumber the id of the desired block.
+   * @return true if the given block is the last block in the associated file, false otherwise.
+   * @throws IllegalArgumentException if blockNumber is invalid.
+   */
+  public boolean isLastBlock(int blockNumber) {
+    if (fileSize == 0) {
+      return false;
+    }
+
+    throwIfInvalidBlockNumber(blockNumber);
+
+    return blockNumber == (numBlocks - 1);
+  }
+
+  /**
+   * Gets the id of the block that contains the given absolute offset.
+   * @param offset the absolute offset to check.
+   * @return the id of the block that contains the given absolute offset.
+   * @throws IllegalArgumentException if offset is invalid.
+   */
+  public int getBlockNumber(long offset) {
+    throwIfInvalidOffset(offset);
+
+    return (int) (offset / blockSize);
+  }
+
+  /**
+   * Gets the size of the given block.
+   * @param blockNumber the id of the desired block.
+   * @return the size of the given block.
+   */
+  public int getSize(int blockNumber) {
+    if (fileSize == 0) {
+      return 0;
+    }
+
+    if (isLastBlock(blockNumber)) {
+      return (int) (fileSize - (((long) blockSize) * (numBlocks - 1)));
+    } else {
+      return blockSize;
+    }
+  }
+
+  /**
+   * Indicates whether the given absolute offset is valid.
+   * @param offset absolute offset in the file..
+   * @return true if the given absolute offset is valid, false otherwise.
+   */
+  public boolean isValidOffset(long offset) {
+    return (offset >= 0) && (offset < fileSize);
+  }
+
+  /**
+   * Gets the start offset of the given block.
+   * @param blockNumber the id of the given block.
+   * @return the start offset of the given block.
+   * @throws IllegalArgumentException if blockNumber is invalid.
+   */
+  public long getStartOffset(int blockNumber) {
+    throwIfInvalidBlockNumber(blockNumber);
+
+    return blockNumber * (long) blockSize;
+  }
+
+  /**
+   * Gets the relative offset corresponding to the given block and the absolute offset.
+   * @param blockNumber the id of the given block.
+   * @param offset absolute offset in the file.
+   * @return the relative offset corresponding to the given block and the absolute offset.
+   * @throws IllegalArgumentException if either blockNumber or offset is invalid.
+   */
+  public int getRelativeOffset(int blockNumber, long offset) {
+    throwIfInvalidOffset(offset);
+
+    return (int) (offset - getStartOffset(blockNumber));
+  }
+
+  /**
+   * Gets the state of the given block.
+   * @param blockNumber the id of the given block.
+   * @return the state of the given block.
+   * @throws IllegalArgumentException if blockNumber is invalid.
+   */
+  public State getState(int blockNumber) {
+    throwIfInvalidBlockNumber(blockNumber);
+
+    return state[blockNumber];
+  }
+
+  /**
+   * Sets the state of the given block to the given value.
+   * @param blockNumber the id of the given block.
+   * @param blockState the target state.
+   * @throws IllegalArgumentException if blockNumber is invalid.
+   */
+  public void setState(int blockNumber, State blockState) {
+    throwIfInvalidBlockNumber(blockNumber);
+
+    state[blockNumber] = blockState;
+  }
+
+  // Debug helper.
+  public String getStateString() {
+    StringBuilder sb = new StringBuilder();
+    int blockNumber = 0;
+    while (blockNumber < numBlocks) {
+      State tstate = getState(blockNumber);
+      int endBlockNumber = blockNumber;
+      while ((endBlockNumber < numBlocks) && (getState(endBlockNumber)
+          == tstate)) {
+        endBlockNumber++;
+      }
+      sb.append(
+          String.format("[%03d ~ %03d] %s%n", blockNumber, endBlockNumber - 1,
+              tstate));
+      blockNumber = endBlockNumber;
+    }
+    return sb.toString();
+  }
+
+  private void throwIfInvalidBlockNumber(int blockNumber) {
+    checkWithinRange(blockNumber, "blockNumber", 0, numBlocks - 1);
+  }
+
+  private void throwIfInvalidOffset(long offset) {
+    checkWithinRange(offset, "offset", 0, fileSize - 1);
+  }
+}