Merge branch 'apache:trunk' into trunk

Author: PradhanPrerak

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

@@ -3581,7 +3581,15 @@ public static Class<? extends FileSystem> getFileSystemClass(String scheme,
       throw new UnsupportedFileSystemException("No FileSystem for scheme "
           + "\"" + scheme + "\"");
     }
-    LOGGER.debug("FS for {} is {}", scheme, clazz);
+    if (LOGGER.isDebugEnabled()) {
+      LOGGER.debug("FS for {} is {}", scheme, clazz);
+      final String jarLocation = ClassUtil.findContainingJar(clazz);
+      if (jarLocation != null) {
+        LOGGER.debug("Jar location for {} : {}", clazz, jarLocation);
+      } else {
+        LOGGER.debug("Class location for {} : {}", clazz, ClassUtil.findClassLocation(clazz));
+      }
+    }
     return clazz;
   }
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/ClassUtil.java

@@ -36,13 +36,25 @@ public class ClassUtil {
    * @return a jar file that contains the class, or null.
    */
   public static String findContainingJar(Class<?> clazz) {
-    ClassLoader loader = clazz.getClassLoader();
-    String classFile = clazz.getName().replaceAll("\\.", "/") + ".class";
+    return findContainingResource(clazz.getClassLoader(), clazz.getName(), "jar");
+  }
+
+  /**
+   * Find the absolute location of the class.
+   *
+   * @param clazz the class to find.
+   * @return the class file with absolute location, or null.
+   */
+  public static String findClassLocation(Class<?> clazz) {
+    return findContainingResource(clazz.getClassLoader(), clazz.getName(), "file");
+  }
+
+  private static String findContainingResource(ClassLoader loader, String clazz, String resource) {
+    String classFile = clazz.replaceAll("\\.", "/") + ".class";
     try {
-      for(final Enumeration<URL> itr = loader.getResources(classFile);
-          itr.hasMoreElements();) {
+      for (final Enumeration<URL> itr = loader.getResources(classFile); itr.hasMoreElements();) {
         final URL url = itr.nextElement();
-        if ("jar".equals(url.getProtocol())) {
+        if (resource.equals(url.getProtocol())) {
           String toReturn = url.getPath();
           if (toReturn.startsWith("file:")) {
             toReturn = toReturn.substring("file:".length());

hadoop-common-project/hadoop-common/src/site/markdown/NativeLibraries.md.vm

@@ -104,7 +104,7 @@ The bin/hadoop script ensures that the native hadoop library is on the library p
 During runtime, check the hadoop log files for your MapReduce tasks.
 
 * If everything is all right, then: `DEBUG util.NativeCodeLoader - Trying to load the custom-built native-hadoop library...` `INFO util.NativeCodeLoader - Loaded the native-hadoop library`
-* If something goes wrong, then: `INFO util.NativeCodeLoader - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable`
+* If something goes wrong, then: `WARN util.NativeCodeLoader - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable`
 
 Check
 -----

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FSMainOperationsBaseTest.java

@@ -102,7 +102,9 @@ public void setUp() throws Exception {
 
   @After
   public void tearDown() throws Exception {
-    fSys.delete(new Path(getAbsoluteTestRootPath(fSys), new Path("test")), true);
+    if (fSys != null) {
+      fSys.delete(new Path(getAbsoluteTestRootPath(fSys), new Path("test")), true);
+    }
   }
 
 
@@ -192,7 +194,7 @@ public void testWorkingDirectory() throws Exception {
 
   @Test
   public void testWDAbsolute() throws IOException {
-    Path absoluteDir = new Path(fSys.getUri() + "/test/existingDir");
+    Path absoluteDir = getTestRootPath(fSys, "test/existingDir");
     fSys.mkdirs(absoluteDir);
     fSys.setWorkingDirectory(absoluteDir);
     Assert.assertEquals(absoluteDir, fSys.getWorkingDirectory());

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java

@@ -81,6 +81,12 @@ public abstract class FileContextMainOperationsBaseTest  {
   protected final FileContextTestHelper fileContextTestHelper =
     createFileContextHelper();
 
+  /**
+   * Create the test helper.
+   * Important: this is invoked during the construction of the base class,
+   * so is very brittle.
+   * @return a test helper.
+   */
   protected FileContextTestHelper createFileContextHelper() {
     return new FileContextTestHelper();
   }
@@ -107,7 +113,7 @@ public boolean accept(Path file) {
 
   private static final byte[] data = getFileData(numBlocks,
       getDefaultBlockSize());
-  
+
   @Before
   public void setUp() throws Exception {
     File testBuildData = GenericTestUtils.getRandomizedTestDir();

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestFSMainOperationsLocalFileSystem.java

@@ -21,23 +21,13 @@
 import java.io.IOException;
 
 import org.apache.hadoop.conf.Configuration;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
 
 public class TestFSMainOperationsLocalFileSystem extends FSMainOperationsBaseTest {
 
   @Override
   protected FileSystem createFileSystem() throws IOException {
     return FileSystem.getLocal(new Configuration());
   }
-    
-  @Override
-  @Before
-  public void setUp() throws Exception {
-    super.setUp();
-  }
 
   static Path wd = null;
   @Override
@@ -46,19 +36,5 @@ protected Path getDefaultWorkingDirectory() throws IOException {
       wd = FileSystem.getLocal(new Configuration()).getWorkingDirectory();
     return wd;
   }
-  
-  @Override
-  @After
-  public void tearDown() throws Exception {
-    super.tearDown();
-  }
-  
-  @Test
-  @Override
-  public void testWDAbsolute() throws IOException {
-    Path absoluteDir = getTestRootPath(fSys, "test/existingDir");
-    fSys.mkdirs(absoluteDir);
-    fSys.setWorkingDirectory(absoluteDir);
-    Assert.assertEquals(absoluteDir, fSys.getWorkingDirectory());
-  }
+
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestFSMainOperationsLocalFileSystem.java

@@ -53,14 +53,5 @@ public void tearDown() throws Exception {
     super.tearDown();
     ViewFileSystemTestSetup.tearDown(this, fcTarget);
   }
-  
-  @Test
-  @Override
-  public void testWDAbsolute() throws IOException {
-    Path absoluteDir = getTestRootPath(fSys, "test/existingDir");
-    fSys.mkdirs(absoluteDir);
-    fSys.setWorkingDirectory(absoluteDir);
-    Assert.assertEquals(absoluteDir, fSys.getWorkingDirectory());
 
-  }
 }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestClassUtil.java

@@ -20,21 +20,47 @@
 
 import java.io.File;
 
-import org.junit.Assert;
+import org.apache.hadoop.fs.viewfs.ViewFileSystem;
 
-import org.apache.log4j.Logger;
+import org.assertj.core.api.Assertions;
 import org.junit.Test;
 
 public class TestClassUtil {
+
   @Test(timeout=10000)
   public void testFindContainingJar() {
-    String containingJar = ClassUtil.findContainingJar(Logger.class);
-    Assert.assertNotNull("Containing jar not found for Logger", 
-        containingJar);
+    String containingJar = ClassUtil.findContainingJar(Assertions.class);
+    Assertions
+        .assertThat(containingJar)
+        .describedAs("Containing jar for %s", Assertions.class)
+        .isNotNull();
     File jarFile = new File(containingJar);
-    Assert.assertTrue("Containing jar does not exist on file system ",
-        jarFile.exists());
-    Assert.assertTrue("Incorrect jar file " + containingJar,
-        jarFile.getName().matches("reload4j.*[.]jar"));
+    Assertions
+        .assertThat(jarFile)
+        .describedAs("Containing jar %s", jarFile)
+        .exists();
+    Assertions
+        .assertThat(jarFile.getName())
+        .describedAs("Containing jar name %s", jarFile.getName())
+        .matches("assertj-core.*[.]jar");
+  }
+
+  @Test(timeout = 10000)
+  public void testFindContainingClass() {
+    String classFileLocation = ClassUtil.findClassLocation(ViewFileSystem.class);
+    Assertions
+        .assertThat(classFileLocation)
+        .describedAs("Class path for %s", ViewFileSystem.class)
+        .isNotNull();
+    File classFile = new File(classFileLocation);
+    Assertions
+        .assertThat(classFile)
+        .describedAs("Containing class file %s", classFile)
+        .exists();
+    Assertions
+        .assertThat(classFile.getName())
+        .describedAs("Containing class file name %s", classFile.getName())
+        .matches("ViewFileSystem.class");
   }
+
 }

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

@@ -33,6 +33,11 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeoutException;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -120,4 +125,37 @@ private CombinedHostsFileReader() {
     }
     return allDNs;
   }
+
+  /**
+   * Wrapper to call readFile with timeout via Future Tasks.
+   * If timeout is reached, it will throw IOException
+   * @param hostsFile the input json file to read from
+   * @param readTimeout timeout for FutureTask execution in milliseconds
+   * @return the set of DatanodeAdminProperties
+   * @throws IOException
+   */
+  public static DatanodeAdminProperties[]
+      readFileWithTimeout(final String hostsFile, final int readTimeout) throws IOException {
+    FutureTask<DatanodeAdminProperties[]> futureTask = new FutureTask<>(
+        new Callable<DatanodeAdminProperties[]>() {
+          @Override
+          public DatanodeAdminProperties[] call() throws Exception {
+            return readFile(hostsFile);
+        }
+      });
+
+    Thread thread = new Thread(futureTask);
+    thread.start();
+
+    try {
+      return futureTask.get(readTimeout, TimeUnit.MILLISECONDS);
+    } catch (TimeoutException e) {
+      futureTask.cancel(true);
+      LOG.error("refresh File read operation timed out");
+      throw new IOException("host file read operation timed out");
+    } catch (InterruptedException | ExecutionException e) {
+      LOG.error("File read operation interrupted : " + e.getMessage());
+      throw new IOException("host file read operation timed out");
+    }
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -757,6 +757,8 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
       "dfs.namenode.hosts.provider.classname";
   public static final String  DFS_HOSTS = "dfs.hosts";
   public static final String  DFS_HOSTS_EXCLUDE = "dfs.hosts.exclude";
+  public static final String  DFS_HOSTS_TIMEOUT = "dfs.hosts.timeout";
+  public static final int     DFS_HOSTS_TIMEOUT_DEFAULT = 0;
   public static final String  DFS_NAMENODE_AUDIT_LOGGERS_KEY = "dfs.namenode.audit.loggers";
   public static final String  DFS_NAMENODE_DEFAULT_AUDIT_LOGGER_NAME = "default";
   public static final String  DFS_NAMENODE_AUDIT_LOG_TOKEN_TRACKING_ID_KEY = "dfs.namenode.audit.log.token.tracking.id";

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/CombinedHostFileManager.java

@@ -179,12 +179,15 @@ public Configuration getConf() {
 
   @Override
   public void refresh() throws IOException {
-    refresh(conf.get(DFSConfigKeys.DFS_HOSTS, ""));
+    refresh(conf.get(DFSConfigKeys.DFS_HOSTS, ""),
+        conf.getInt(DFSConfigKeys.DFS_HOSTS_TIMEOUT, DFSConfigKeys.DFS_HOSTS_TIMEOUT_DEFAULT)
+    );
   }
-  private void refresh(final String hostsFile) throws IOException {
+  private void refresh(final String hostsFile, final int readTimeout) throws IOException {
     HostProperties hostProps = new HostProperties();
-    DatanodeAdminProperties[] all =
-        CombinedHostsFileReader.readFile(hostsFile);
+    DatanodeAdminProperties[] all = readTimeout != DFSConfigKeys.DFS_HOSTS_TIMEOUT_DEFAULT
+        ? CombinedHostsFileReader.readFileWithTimeout(hostsFile, readTimeout)
+        : CombinedHostsFileReader.readFile(hostsFile);
     for(DatanodeAdminProperties properties : all) {
       InetSocketAddress addr = parseEntry(hostsFile,
           properties.getHostName(), properties.getPort());

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -1131,7 +1131,14 @@
   not permitted to connect to the namenode.  The full pathname of the
   file must be specified.  If the value is empty, no hosts are
   excluded.</description>
-</property> 
+</property>
+
+<property>
+  <name>dfs.hosts.timeout</name>
+  <value>0</value>
+  <description>Specifies a timeout (in milliseconds) for reading the dfs.hosts file.
+  A value of zero indicates no timeout to be set.</description>
+</property>
 
 <property>
   <name>dfs.namenode.max.objects</name>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/util/TestCombinedHostsFileReader.java

@@ -19,14 +19,21 @@
 
 import java.io.File;
 import java.io.FileWriter;
+import java.io.IOException;
+import java.util.concurrent.Callable;
 
 import org.apache.hadoop.hdfs.protocol.DatanodeAdminProperties;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.junit.Before;
 import org.junit.After;
 import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.mockito.Mock;
+import org.mockito.Mockito;
 
 import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.when;
 
 /**
  * Test for JSON based HostsFileReader.
@@ -44,8 +51,12 @@ public class TestCombinedHostsFileReader {
   private final File legacyFile =
       new File(TESTCACHEDATADIR, "legacy.dfs.hosts.json");
 
+  @Mock
+  private Callable<DatanodeAdminProperties[]> callable;
+
   @Before
   public void setUp() throws Exception {
+    callable = Mockito.mock(Callable.class);
   }
 
   @After
@@ -87,4 +98,50 @@ public void testEmptyCombinedHostsFileReader() throws Exception {
         CombinedHostsFileReader.readFile(newFile.getAbsolutePath());
     assertEquals(0, all.length);
   }
+
+  /*
+   * When timeout is enabled, test for success when reading file within timeout
+   * limits
+   */
+  @Test
+  public void testReadFileWithTimeoutSuccess() throws Exception {
+
+    DatanodeAdminProperties[] all = CombinedHostsFileReader.readFileWithTimeout(
+        jsonFile.getAbsolutePath(), 1000);
+    assertEquals(7, all.length);
+  }
+
+  /*
+   * When timeout is enabled, test for IOException when reading file exceeds
+   * timeout limits
+   */
+  @Test(expected = IOException.class)
+  public void testReadFileWithTimeoutTimeoutException() throws Exception {
+    when(callable.call()).thenAnswer(new Answer<Void>() {
+      @Override
+      public Void answer(InvocationOnMock invocation) throws Throwable {
+        Thread.sleep(2000);
+        return null;
+      }
+    });
+
+    CombinedHostsFileReader.readFileWithTimeout(
+        jsonFile.getAbsolutePath(), 1);
+  }
+
+  /*
+   * When timeout is enabled, test for IOException when execution is interrupted
+   */
+  @Test(expected = IOException.class)
+  public void testReadFileWithTimeoutInterruptedException() throws Exception {
+    when(callable.call()).thenAnswer(new Answer<Void>() {
+      @Override
+      public Void answer(InvocationOnMock invocation) throws Throwable {
+        throw new InterruptedException();
+      }
+    });
+
+    CombinedHostsFileReader.readFileWithTimeout(
+        jsonFile.getAbsolutePath(), 1);
+  }
 }