Merge branch 'trunk' into YARN-11180

Author: slfan1989

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/retry/RetryPolicies.java

@@ -181,15 +181,20 @@ public static final RetryPolicy retryByRemoteException(
   }
 
   /**
-   * A retry policy for exceptions other than RemoteException.
+   * <p>
+   * A retry policy where RemoteException and SaslException are not retried, other individual
+   * exception types can have RetryPolicy overrides, and any other exception type without an
+   * override is not retried.
+   * </p>
+   *
    * @param defaultPolicy defaultPolicy.
    * @param exceptionToPolicyMap exceptionToPolicyMap.
    * @return RetryPolicy.
    */
-  public static final RetryPolicy retryOtherThanRemoteException(
+  public static final RetryPolicy retryOtherThanRemoteAndSaslException(
       RetryPolicy defaultPolicy,
       Map<Class<? extends Exception>, RetryPolicy> exceptionToPolicyMap) {
-    return new OtherThanRemoteExceptionDependentRetry(defaultPolicy,
+    return new OtherThanRemoteAndSaslExceptionDependentRetry(defaultPolicy,
         exceptionToPolicyMap);
   }
 
@@ -589,12 +594,12 @@ public RetryAction shouldRetry(Exception e, int retries, int failovers,
     }
   }
 
-  static class OtherThanRemoteExceptionDependentRetry implements RetryPolicy {
+  static class OtherThanRemoteAndSaslExceptionDependentRetry implements RetryPolicy {
 
     private RetryPolicy defaultPolicy;
     private Map<Class<? extends Exception>, RetryPolicy> exceptionToPolicyMap;
 
-    public OtherThanRemoteExceptionDependentRetry(RetryPolicy defaultPolicy,
+    OtherThanRemoteAndSaslExceptionDependentRetry(RetryPolicy defaultPolicy,
         Map<Class<? extends Exception>,
         RetryPolicy> exceptionToPolicyMap) {
       this.defaultPolicy = defaultPolicy;
@@ -605,10 +610,8 @@ public OtherThanRemoteExceptionDependentRetry(RetryPolicy defaultPolicy,
     public RetryAction shouldRetry(Exception e, int retries, int failovers,
         boolean isIdempotentOrAtMostOnce) throws Exception {
       RetryPolicy policy = null;
-      // ignore Remote Exception
-      if (e instanceof RemoteException) {
-        // do nothing
-      } else {
+      // ignore RemoteException and SaslException
+      if (!(e instanceof RemoteException || isSaslFailure(e))) {
         policy = exceptionToPolicyMap.get(e.getClass());
       }
       if (policy == null) {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java

@@ -61,6 +61,7 @@
 
 import javax.net.SocketFactory;
 import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslException;
 import java.io.*;
 import java.net.*;
 import java.nio.ByteBuffer;
@@ -1620,7 +1621,8 @@ private Writable getRpcResponse(final Call call, final Connection connection,
       }
 
       if (call.error != null) {
-        if (call.error instanceof RemoteException) {
+        if (call.error instanceof RemoteException ||
+            call.error instanceof SaslException) {
           call.error.fillInStackTrace();
           throw call.error;
         } else { // local exception

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SaslRpcClient.java

@@ -237,7 +237,14 @@ private SaslClient createSaslClient(SaslAuth authType)
           LOG.debug("client isn't using kerberos");
           return null;
         }
-        String serverPrincipal = getServerPrincipal(authType);
+        final String serverPrincipal;
+        try {
+          serverPrincipal = getServerPrincipal(authType);
+        } catch (IllegalArgumentException ex) {
+          // YARN-11210: getServerPrincipal can throw IllegalArgumentException if Kerberos
+          // configuration is bad, this is surfaced as a non-retryable SaslException
+          throw new SaslException("Bad Kerberos server principal configuration", ex);
+        }
         if (serverPrincipal == null) {
           LOG.debug("protocol doesn't use kerberos");
           return null;

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/retry/TestRetryProxy.java

@@ -291,7 +291,7 @@ public void testRetryOtherThanRemoteException() throws Throwable {
 
     UnreliableInterface unreliable = (UnreliableInterface)
         RetryProxy.create(UnreliableInterface.class, unreliableImpl,
-            retryOtherThanRemoteException(TRY_ONCE_THEN_FAIL,
+            retryOtherThanRemoteAndSaslException(TRY_ONCE_THEN_FAIL,
                 exceptionToPolicyMap));
     // should retry with local IOException.
     unreliable.failsOnceWithIOException();

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/FileChecksumHelper.java

@@ -303,7 +303,8 @@ FileChecksum makeCompositeCrcResult() throws IOException {
       byte[] blockChecksumBytes = blockChecksumBuf.getData();
 
       long sumBlockLengths = 0;
-      for (int i = 0; i < locatedBlocks.size() - 1; ++i) {
+      int i = 0;
+      for (; i < locatedBlocks.size() - 1; ++i) {
         LocatedBlock block = locatedBlocks.get(i);
         // For everything except the last LocatedBlock, we expect getBlockSize()
         // to accurately reflect the number of file bytes digested in the block
@@ -316,19 +317,8 @@ FileChecksum makeCompositeCrcResult() throws IOException {
             "Added blockCrc 0x{} for block index {} of size {}",
             Integer.toString(blockCrc, 16), i, block.getBlockSize());
       }
-
-      // NB: In some cases the located blocks have their block size adjusted
-      // explicitly based on the requested length, but not all cases;
-      // these numbers may or may not reflect actual sizes on disk.
-      long reportedLastBlockSize =
-          blockLocations.getLastLocatedBlock().getBlockSize();
-      long consumedLastBlockLength = reportedLastBlockSize;
-      if (length - sumBlockLengths < reportedLastBlockSize) {
-        LOG.warn(
-            "Last block length {} is less than reportedLastBlockSize {}",
-            length - sumBlockLengths, reportedLastBlockSize);
-        consumedLastBlockLength = length - sumBlockLengths;
-      }
+      LocatedBlock nextBlock = locatedBlocks.get(i);
+      long consumedLastBlockLength = Math.min(length - sumBlockLengths, nextBlock.getBlockSize());
       // NB: blockChecksumBytes.length may be much longer than actual bytes
       // written into the DataOutput.
       int lastBlockCrc = CrcUtil.readInt(

hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/CMakeLists.txt

@@ -24,8 +24,11 @@ set (LIBHDFS_TESTS_DIR ../../libhdfs-tests)
 set (LIBHDFSPP_SRC_DIR ..)
 set (LIBHDFSPP_LIB_DIR ${LIBHDFSPP_SRC_DIR}/lib)
 set (LIBHDFSPP_BINDING_C ${LIBHDFSPP_LIB_DIR}/bindings/c)
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-missing-field-initializers")
-set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-missing-field-initializers")
+
+if (NOT MSVC)
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-missing-field-initializers")
+    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-missing-field-initializers")
+endif (NOT MSVC)
 
 include_directories(
     ${GENERATED_JAVAH}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/OfflineImageViewerPB.java

@@ -80,6 +80,7 @@ public class OfflineImageViewerPB {
       + "    delimiter. The default delimiter is \\t, though this may be\n"
       + "    changed via the -delimiter argument.\n"
       + "    -sp print storage policy, used by delimiter only.\n"
+      + "    -ec print erasure coding policy, used by delimiter only.\n"
       + "  * DetectCorruption: Detect potential corruption of the image by\n"
       + "    selectively loading parts of it and actively searching for\n"
       + "    inconsistencies. Outputs a summary of the found corruptions\n"
@@ -132,6 +133,7 @@ private static Options buildOptions() {
     options.addOption("addr", true, "");
     options.addOption("delimiter", true, "");
     options.addOption("sp", false, "");
+    options.addOption("ec", false, "");
     options.addOption("t", "temp", true, "");
     options.addOption("m", "multiThread", true, "");
 
@@ -228,9 +230,11 @@ public static int run(String[] args) throws Exception {
         break;
       case "DELIMITED":
         boolean printStoragePolicy = cmd.hasOption("sp");
+        boolean printECPolicy = cmd.hasOption("ec");
         try (PBImageDelimitedTextWriter writer =
             new PBImageDelimitedTextWriter(out, delimiter,
-                tempPath, printStoragePolicy, threads, outputFile)) {
+                tempPath, printStoragePolicy, printECPolicy, threads,
+                outputFile, conf)) {
           writer.visit(inputFile);
         }
         break;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/PBImageDelimitedTextWriter.java

@@ -17,9 +17,12 @@
  */
 package org.apache.hadoop.hdfs.tools.offlineImageViewer;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.PermissionStatus;
+import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
+import org.apache.hadoop.hdfs.server.namenode.ErasureCodingPolicyManager;
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.INodeSection.INode;
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.INodeSection.INodeDirectory;
 import org.apache.hadoop.hdfs.server.namenode.FsImageProto.INodeSection.INodeFile;
@@ -46,6 +49,8 @@
 public class PBImageDelimitedTextWriter extends PBImageTextWriter {
   private static final String DATE_FORMAT = "yyyy-MM-dd HH:mm";
   private boolean printStoragePolicy;
+  private boolean printECPolicy;
+  private ErasureCodingPolicyManager ecManager;
 
   static class OutputEntryBuilder {
     private final SimpleDateFormat dateFormatter =
@@ -62,6 +67,7 @@ static class OutputEntryBuilder {
     private long nsQuota = 0;
     private long dsQuota = 0;
     private int storagePolicy = 0;
+    private String ecPolicy = "-";
 
     private String dirPermission = "-";
     private PermissionStatus permissionStatus;
@@ -83,6 +89,13 @@ static class OutputEntryBuilder {
           aclPermission = "+";
         }
         storagePolicy = file.getStoragePolicyID();
+        if (writer.printECPolicy && file.hasErasureCodingPolicyID()) {
+          ErasureCodingPolicy policy = writer.ecManager.
+              getByID((byte) file.getErasureCodingPolicyID());
+          if (policy != null) {
+            ecPolicy = policy.getName();
+          }
+        }
         break;
       case DIRECTORY:
         INodeDirectory dir = inode.getDirectory();
@@ -95,6 +108,12 @@ static class OutputEntryBuilder {
           aclPermission = "+";
         }
         storagePolicy = writer.getStoragePolicy(dir.getXAttrs());
+        if (writer.printECPolicy) {
+          String name= writer.getErasureCodingPolicyName(dir.getXAttrs());
+          if (name != null) {
+            ecPolicy = name;
+          }
+        }
         break;
       case SYMLINK:
         INodeSymlink s = inode.getSymlink();
@@ -134,6 +153,9 @@ public String build() {
       if (writer.printStoragePolicy) {
         writer.append(buffer, storagePolicy);
       }
+      if (writer.printECPolicy) {
+        writer.append(buffer, ecPolicy);
+      }
       return buffer.substring(1);
     }
   }
@@ -146,14 +168,21 @@ public String build() {
   PBImageDelimitedTextWriter(PrintStream out, String delimiter,
                              String tempPath, boolean printStoragePolicy)
       throws IOException {
-    this(out, delimiter, tempPath, printStoragePolicy, 1, "-");
+    this(out, delimiter, tempPath, printStoragePolicy, false, 1, "-", null);
   }
 
   PBImageDelimitedTextWriter(PrintStream out, String delimiter,
-      String tempPath, boolean printStoragePolicy, int threads,
-      String parallelOut) throws IOException {
+                             String tempPath, boolean printStoragePolicy,
+                             boolean printECPolicy, int threads,
+                             String parallelOut, Configuration conf)
+      throws IOException {
     super(out, delimiter, tempPath, threads, parallelOut);
     this.printStoragePolicy = printStoragePolicy;
+    if (printECPolicy && conf != null) {
+      this.printECPolicy = true;
+      ecManager = ErasureCodingPolicyManager.getInstance();
+      ecManager.init(conf);
+    }
   }
 
   @Override
@@ -187,6 +216,9 @@ public String getHeader() {
     if (printStoragePolicy) {
       append(buffer, "StoragePolicyId");
     }
+    if (printECPolicy) {
+      append(buffer, "ErasureCodingPolicy");
+    }
     return buffer.toString();
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/PBImageTextWriter.java

@@ -27,6 +27,8 @@
 import java.io.PrintStream;
 import java.io.RandomAccessFile;
 import java.io.UnsupportedEncodingException;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.util.ArrayList;
@@ -63,6 +65,7 @@
 import org.apache.hadoop.hdfs.server.namenode.INodeId;
 import org.apache.hadoop.hdfs.server.namenode.SerialNumberManager;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.WritableUtils;
 import org.apache.hadoop.util.LimitInputStream;
 import org.apache.hadoop.util.Lists;
 import org.apache.hadoop.util.Time;
@@ -77,6 +80,8 @@
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableList;
 
+import static org.apache.hadoop.hdfs.server.common.HdfsServerConstants.XATTR_ERASURECODING_POLICY;
+
 /**
  * This class reads the protobuf-based fsimage and generates text output
  * for each inode to {@link PBImageTextWriter#out}. The sub-class can override
@@ -1029,4 +1034,23 @@ public static void mergeFiles(String[] srcPaths, String resultPath)
       }
     }
   }
+
+  public String getErasureCodingPolicyName
+      (INodeSection.XAttrFeatureProto xattrFeatureProto) {
+    List<XAttr> xattrs =
+        FSImageFormatPBINode.Loader.loadXAttrs(xattrFeatureProto, stringTable);
+    for (XAttr xattr : xattrs) {
+      if (XATTR_ERASURECODING_POLICY.contains(xattr.getName())){
+        try{
+          ByteArrayInputStream bIn = new ByteArrayInputStream(xattr.getValue());
+          DataInputStream dIn = new DataInputStream(bIn);
+          return WritableUtils.readString(dIn);
+        } catch (IOException ioException){
+          return null;
+        }
+      }
+    }
+    return null;
+  }
+
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestFileChecksum.java

@@ -215,6 +215,39 @@ public void testStripedAndReplicatedFileChecksum() throws Exception {
     }
   }
 
+  /**
+   * Test the corner case of the COMPOSITE_CRC.
+   * For Stripe File, last block size in the file is (int)(blockSize * 0.5),
+   *    but the last block size in the check length is (int)(blockSize * 0.6).
+   * For Replicate File, the last block size in the file is (int)(blockSize * 0.5),
+   *    but the last block size in the check length is ((dataBlocks - 1) * blockSize
+   *    + (int) (blockSize * 0.6))
+   */
+  @Test(timeout = 90000)
+  public void testStripedAndReplicatedFileChecksum2() throws Exception {
+    final int lastBlockSize = (int) (blockSize * 0.5);
+    final int fullStripeLength = dataBlocks * blockSize;
+    final int testFileSize = fullStripeLength + lastBlockSize;
+    prepareTestFiles(testFileSize, new String[] {stripedFile1, replicatedFile});
+
+    final int specialLength = (dataBlocks - 1) * blockSize
+        + (int) (blockSize * 0.6);
+
+    Assert.assertTrue(specialLength % blockSize > lastBlockSize);
+    Assert.assertTrue(specialLength % fullStripeLength > lastBlockSize);
+
+    FileChecksum stripedFileChecksum = getFileChecksum(stripedFile1,
+        specialLength, false);
+    FileChecksum replicatedFileChecksum = getFileChecksum(replicatedFile,
+        specialLength, false);
+
+    if (checksumCombineMode.equals(ChecksumCombineMode.COMPOSITE_CRC.name())) {
+      Assert.assertEquals(replicatedFileChecksum, stripedFileChecksum);
+    } else {
+      Assert.assertNotEquals(replicatedFileChecksum, stripedFileChecksum);
+    }
+  }
+
   @Test(timeout = 90000)
   public void testDifferentBlockSizeReplicatedFileChecksum() throws Exception {
     byte[] fileData = StripedFileTestUtil.generateBytes(fileSize);