Merge branch 'trunk' into dev/HADOOP-16854-oomfix

Author: bilaharith

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

@@ -38,6 +38,7 @@
 @InterfaceStability.Stable
 public class BytesWritable extends BinaryComparable
     implements WritableComparable<BinaryComparable> {
+  private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
   private static final int LENGTH_BYTES = 4;
 
   private static final byte[] EMPTY_BYTES = new byte[0];
@@ -126,7 +127,7 @@ public int getSize() {
   public void setSize(int size) {
     if (size > getCapacity()) {
       // Avoid overflowing the int too early by casting to a long.
-      long newSize = Math.min(Integer.MAX_VALUE, (3L * size) / 2L);
+      long newSize = Math.min(MAX_ARRAY_SIZE, (3L * size) / 2L);
       setCapacity((int) newSize);
     }
     this.size = size;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/NetUtils.java

@@ -638,6 +638,22 @@ public static String getHostNameOfIP(String ipPort) {
     }
   }
 
+  /**
+   * Attempt to normalize the given string to "host:port"
+   * if it like "ip:port".
+   *
+   * @param ipPort maybe lik ip:port or host:port.
+   * @return host:port
+   */
+  public static String normalizeIP2HostName(String ipPort) {
+    if (null == ipPort || !ipPortPattern.matcher(ipPort).matches()) {
+      return ipPort;
+    }
+
+    InetSocketAddress address = createSocketAddr(ipPort);
+    return getHostPortString(address);
+  }
+
   /**
    * Return hostname without throwing exception.
    * The returned hostname String format is "hostname".

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/NetworkTopology.java

@@ -949,6 +949,7 @@ public <T extends Node> void sortByDistanceUsingNetworkLocation(Node reader,
    * <p>
    * As an additional twist, we also randomize the nodes at each network
    * distance. This helps with load balancing when there is data skew.
+   * And it helps choose node with more fast storage type.
    *
    * @param reader    Node where data will be read
    * @param nodes     Available replicas with the requested data

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/CompressDecompressTester.java

@@ -126,7 +126,7 @@ private void addPair(T compressor, E decompressor, String name) {
     builder.add(new TesterPair<T, E>(name, compressor, decompressor));
   }
 
-  public void test() throws InstantiationException, IllegalAccessException {
+  public void test() throws Exception {
     pairs = builder.build();
     pairs = assertionDelegate.filterOnAssumeWhat(pairs);
 
@@ -287,47 +287,45 @@ private boolean checkSetInputArrayIndexOutOfBoundsException(
 
       @Override
       public void assertCompression(String name, Compressor compressor,
-          Decompressor decompressor, byte[] rawData) {
+          Decompressor decompressor, byte[] rawData) throws Exception {
 
         int cSize = 0;
         int decompressedSize = 0;
-        byte[] compressedResult = new byte[rawData.length];
+        // Snappy compression can increase data size
+        int maxCompressedLength = 32 + rawData.length + rawData.length/6;
+        byte[] compressedResult = new byte[maxCompressedLength];
         byte[] decompressedBytes = new byte[rawData.length];
-        try {
-          assertTrue(
-              joiner.join(name, "compressor.needsInput before error !!!"),
-              compressor.needsInput());
-          assertTrue(
+        assertTrue(
+            joiner.join(name, "compressor.needsInput before error !!!"),
+            compressor.needsInput());
+        assertEquals(
               joiner.join(name, "compressor.getBytesWritten before error !!!"),
-              compressor.getBytesWritten() == 0);
-          compressor.setInput(rawData, 0, rawData.length);
-          compressor.finish();
-          while (!compressor.finished()) {
-            cSize += compressor.compress(compressedResult, 0,
-                compressedResult.length);
-          }
-          compressor.reset();
-
-          assertTrue(
-              joiner.join(name, "decompressor.needsInput() before error !!!"),
-              decompressor.needsInput());
-          decompressor.setInput(compressedResult, 0, cSize);
-          assertFalse(
-              joiner.join(name, "decompressor.needsInput() after error !!!"),
-              decompressor.needsInput());
-          while (!decompressor.finished()) {
-            decompressedSize = decompressor.decompress(decompressedBytes, 0,
-                decompressedBytes.length);
-          }
-          decompressor.reset();
-          assertTrue(joiner.join(name, " byte size not equals error !!!"),
-              decompressedSize == rawData.length);
-          assertArrayEquals(
-              joiner.join(name, " byte arrays not equals error !!!"), rawData,
-              decompressedBytes);
-        } catch (Exception ex) {
-          fail(joiner.join(name, ex.getMessage()));
+            0, compressor.getBytesWritten());
+        compressor.setInput(rawData, 0, rawData.length);
+        compressor.finish();
+        while (!compressor.finished()) {
+          cSize += compressor.compress(compressedResult, 0,
+              compressedResult.length);
+        }
+        compressor.reset();
+
+        assertTrue(
+            joiner.join(name, "decompressor.needsInput() before error !!!"),
+            decompressor.needsInput());
+        decompressor.setInput(compressedResult, 0, cSize);
+        assertFalse(
+            joiner.join(name, "decompressor.needsInput() after error !!!"),
+            decompressor.needsInput());
+        while (!decompressor.finished()) {
+          decompressedSize = decompressor.decompress(decompressedBytes, 0,
+              decompressedBytes.length);
         }
+        decompressor.reset();
+        assertEquals(joiner.join(name, " byte size not equals error !!!"),
+            rawData.length, decompressedSize);
+        assertArrayEquals(
+            joiner.join(name, " byte arrays not equals error !!!"), rawData,
+            decompressedBytes);
       }
     }),
 
@@ -519,6 +517,6 @@ abstract static class TesterCompressionStrategy {
     protected final Logger logger = Logger.getLogger(getClass());
 
     abstract void assertCompression(String name, Compressor compressor,
-        Decompressor decompressor, byte[] originalRawData);
+        Decompressor decompressor, byte[] originalRawData) throws Exception;
   }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/snappy/TestSnappyCompressorDecompressor.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.io.compress.snappy;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -44,11 +45,16 @@
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import static org.junit.Assume.*;
 
 public class TestSnappyCompressorDecompressor {
 
+  public static final Logger LOG =
+      LoggerFactory.getLogger(TestSnappyCompressorDecompressor.class);
+
   @Before
   public void before() {
     assumeTrue(SnappyCodec.isNativeCodeLoaded());
@@ -167,40 +173,41 @@ public void testSnappyDecompressorCompressAIOBException() {
   }    
 
   @Test
-  public void testSnappyCompressDecompress() {
+  public void testSnappyCompressDecompress() throws Exception {
     int BYTE_SIZE = 1024 * 54;
     byte[] bytes = BytesGenerator.get(BYTE_SIZE);
     SnappyCompressor compressor = new SnappyCompressor();
-    try {
-      compressor.setInput(bytes, 0, bytes.length);
-      assertTrue("SnappyCompressDecompress getBytesRead error !!!",
-          compressor.getBytesRead() > 0);
-      assertTrue(
-          "SnappyCompressDecompress getBytesWritten before compress error !!!",
-          compressor.getBytesWritten() == 0);
-
-      byte[] compressed = new byte[BYTE_SIZE];
-      int cSize = compressor.compress(compressed, 0, compressed.length);
-      assertTrue(
-          "SnappyCompressDecompress getBytesWritten after compress error !!!",
-          compressor.getBytesWritten() > 0);
-
-      SnappyDecompressor decompressor = new SnappyDecompressor(BYTE_SIZE);
-      // set as input for decompressor only compressed data indicated with cSize
-      decompressor.setInput(compressed, 0, cSize);
-      byte[] decompressed = new byte[BYTE_SIZE];
-      decompressor.decompress(decompressed, 0, decompressed.length);
-
-      assertTrue("testSnappyCompressDecompress finished error !!!",
-          decompressor.finished());
-      Assert.assertArrayEquals(bytes, decompressed);
-      compressor.reset();
-      decompressor.reset();
-      assertTrue("decompressor getRemaining error !!!",
-          decompressor.getRemaining() == 0);
-    } catch (Exception e) {
-      fail("testSnappyCompressDecompress ex error!!!");
-    }
+    compressor.setInput(bytes, 0, bytes.length);
+    assertTrue("SnappyCompressDecompress getBytesRead error !!!",
+        compressor.getBytesRead() > 0);
+    assertEquals(
+        "SnappyCompressDecompress getBytesWritten before compress error !!!",
+        0, compressor.getBytesWritten());
+
+    // snappy compression may increase data size.
+    // This calculation comes from "Snappy::MaxCompressedLength(size_t)"
+    int maxSize = 32 + BYTE_SIZE + BYTE_SIZE / 6;
+    byte[] compressed = new byte[maxSize];
+    int cSize = compressor.compress(compressed, 0, compressed.length);
+    LOG.info("input size: {}", BYTE_SIZE);
+    LOG.info("compressed size: {}", cSize);
+    assertTrue(
+        "SnappyCompressDecompress getBytesWritten after compress error !!!",
+        compressor.getBytesWritten() > 0);
+
+    SnappyDecompressor decompressor = new SnappyDecompressor();
+    // set as input for decompressor only compressed data indicated with cSize
+    decompressor.setInput(compressed, 0, cSize);
+    byte[] decompressed = new byte[BYTE_SIZE];
+    decompressor.decompress(decompressed, 0, decompressed.length);
+
+    assertTrue("testSnappyCompressDecompress finished error !!!",
+        decompressor.finished());
+    Assert.assertArrayEquals(bytes, decompressed);
+    compressor.reset();
+    decompressor.reset();
+    assertEquals("decompressor getRemaining error !!!",
+        0, decompressor.getRemaining());
   }
 
   @Test
@@ -278,7 +285,38 @@ public void testSnappyBlockCompression() {
       fail("testSnappyBlockCompression ex error !!!");
     }
   }
-  
+
+  @Test
+  // The buffer size is smaller than the input.
+  public void testSnappyCompressDecompressWithSmallBuffer() throws Exception {
+    int inputSize = 1024 * 50;
+    int bufferSize = 512;
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    byte[] buffer = new byte[bufferSize];
+    byte[] input = BytesGenerator.get(inputSize);
+
+    SnappyCompressor compressor = new SnappyCompressor();
+    compressor.setInput(input, 0, inputSize);
+    compressor.finish();
+    while (!compressor.finished()) {
+      int len = compressor.compress(buffer, 0, buffer.length);
+      out.write(buffer, 0, len);
+    }
+    byte[] compressed = out.toByteArray();
+    assertThat(compressed).hasSizeGreaterThan(0);
+    out.reset();
+
+    SnappyDecompressor decompressor = new SnappyDecompressor();
+    decompressor.setInput(compressed, 0, compressed.length);
+    while (!decompressor.finished()) {
+      int len = decompressor.decompress(buffer, 0, buffer.length);
+      out.write(buffer, 0, len);
+    }
+    byte[] decompressed = out.toByteArray();
+
+    assertThat(decompressed).isEqualTo(input);
+  }
+
   private void compressDecompressLoop(int rawDataSize) throws IOException {
     byte[] rawData = BytesGenerator.get(rawDataSize);    
     byte[] compressedResult = new byte[rawDataSize+20];

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/net/TestSocketIOWithTimeout.java

@@ -40,7 +40,7 @@
 import static org.junit.Assert.*;
 
 /**
- * This tests timout out from SocketInputStream and
+ * This tests timeout out from SocketInputStream and
  * SocketOutputStream using pipes.
  * 
  * Normal read and write using these streams are tested by pretty much

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

@@ -3556,7 +3556,8 @@ public RemoteIterator<OpenFileEntry> listOpenFiles(
 
   public RemoteIterator<OpenFileEntry> listOpenFiles(
       EnumSet<OpenFilesType> openFilesTypes, String path) throws IOException {
-    return dfs.listOpenFiles(openFilesTypes, path);
+    Path absF = fixRelativePart(new Path(path));
+    return dfs.listOpenFiles(openFilesTypes, getPathName(absF));
   }
 
 

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

@@ -343,7 +343,7 @@ interface Failover {
         PREFIX + "connection.retries.on.timeouts";
     int     CONNECTION_RETRIES_ON_SOCKET_TIMEOUTS_DEFAULT = 0;
     String  RANDOM_ORDER = PREFIX + "random.order";
-    boolean RANDOM_ORDER_DEFAULT = false;
+    boolean RANDOM_ORDER_DEFAULT = true;
     String  RESOLVE_ADDRESS_NEEDED_KEY = PREFIX + "resolve-needed";
     boolean RESOLVE_ADDRESS_NEEDED_DEFAULT = false;
     String RESOLVE_SERVICE_KEY = PREFIX + "resolver.impl";

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

@@ -27,6 +27,7 @@
 import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.net.InetSocketAddress;
+import java.nio.channels.UnresolvedAddressException;
 import java.util.List;
 
 import com.google.common.io.ByteArrayDataOutput;
@@ -823,7 +824,7 @@ private BlockReaderPeer nextTcpPeer() throws IOException {
           datanode);
       LOG.trace("nextTcpPeer: created newConnectedPeer {}", peer);
       return new BlockReaderPeer(peer, false);
-    } catch (IOException e) {
+    } catch (IOException | UnresolvedAddressException e) {
       LOG.trace("nextTcpPeer: failed to create newConnectedPeer connected to"
           + "{}", datanode);
       throw e;

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

@@ -158,7 +158,7 @@ public ExtendedBlock getBlock() {
    * {@link org.apache.hadoop.hdfs.protocol.LocatedBlock#updateCachedStorageInfo}
    * to update the cached Storage ID/Type arrays.
    */
-  public DatanodeInfo[] getLocations() {
+  public DatanodeInfoWithStorage[] getLocations() {
     return locs;
   }