HADOOP-6920, HADOOP-7292, HADOOP-7306. Porting bugfixes portion of the three patches to yahoo-merge branch.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/yahoo-merge@1126785 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -61,6 +61,9 @@ Trunk (unreleased changes)
 
     HADOOP-7231. Fix synopsis for -count. (Daryn Sharp via eli).
 
+    HADOOP-6920, HADOOP-7292, HADOOP-7306. Porting bugfixes portion of the 
+    three patches to yahoo-merge branch.
+
 Release 0.22.0 - Unreleased
 
   INCOMPATIBLE CHANGES

src/java/org/apache/hadoop/metrics2/impl/MetricsConfig.java

@@ -50,7 +50,6 @@ import org.apache.hadoop.metrics2.filter.GlobFilter;
  * Metrics configuration for MetricsSystemImpl
  */
 class MetricsConfig extends SubsetConfiguration {
-
   static final Log LOG = LogFactory.getLog(MetricsConfig.class);
 
   static final String DEFAULT_FILE_NAME = "hadoop-metrics2.properties";
@@ -123,8 +122,10 @@ class MetricsConfig extends SubsetConfiguration {
         throw new MetricsConfigException(e);
       }
     }
-    throw new MetricsConfigException("Cannot locate configuration: tried "+
-                                     Joiner.on(",").join(fileNames));
+    LOG.warn("Cannot locate configuration: tried "+
+             Joiner.on(",").join(fileNames));
+    // default to an empty configuration
+    return new MetricsConfig(new PropertiesConfiguration(), prefix);
   }
 
   @Override
@@ -159,6 +160,7 @@ class MetricsConfig extends SubsetConfiguration {
   Iterable<String> keys() {
     return new Iterable<String>() {
       @SuppressWarnings("unchecked")
+      @Override
       public Iterator<String> iterator() {
         return (Iterator<String>) getKeys();
       }

src/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java

@@ -154,7 +154,7 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource {
       case NORMAL:
         try { start(); }
         catch (MetricsConfigException e) {
-          // Usually because hadoop-metrics2.properties is missing
+          // Configuration errors (e.g., typos) should not be fatal.
           // We can always start the metrics system later via JMX.
           LOG.warn("Metrics system not started: "+ e.getMessage());
           LOG.debug("Stacktrace: ", e);
@@ -213,8 +213,8 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource {
     MetricsInfo si = sb.info();
     String name2 = name == null ? si.name() : name;
     final String finalDesc = desc == null ? si.description() : desc;
-    final String finalName = DefaultMetricsSystem.sourceName(name2);
-    checkArgument(!allSources.containsKey(finalName));
+    final String finalName = // be friendly to non-metrics tests
+        DefaultMetricsSystem.sourceName(name2, !monitoring);
     allSources.put(finalName, s);
     LOG.debug(finalName +", "+ finalDesc);
     if (monitoring) {
@@ -281,6 +281,7 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource {
     callbacks.add((Callback) Proxy.newProxyInstance(
         callback.getClass().getClassLoader(), new Class<?>[] { Callback.class },
         new InvocationHandler() {
+          @Override
           public Object invoke(Object proxy, Method method, Object[] args)
               throws Throwable {
             try {
@@ -531,7 +532,10 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource {
   @Override
   public synchronized boolean shutdown() {
     LOG.debug("refCount="+ refCount);
-    if (refCount <= 0) LOG.debug("Redundant shutdown", new Throwable());
+    if (refCount <= 0) {
+      LOG.debug("Redundant shutdown", new Throwable());
+      return true; // already shutdown
+    }
     if (--refCount > 0) return false;
     if (monitoring) {
       try { stop(); }

src/java/org/apache/hadoop/metrics2/lib/DefaultMetricsSystem.java

@@ -32,7 +32,6 @@ import org.apache.hadoop.metrics2.impl.MetricsSystemImpl;
 @InterfaceAudience.Public
 @InterfaceStability.Evolving
 public enum DefaultMetricsSystem {
-
   INSTANCE; // the singleton
 
   private MetricsSystem impl = new MetricsSystemImpl();
@@ -103,8 +102,8 @@ public enum DefaultMetricsSystem {
   }
 
   @InterfaceAudience.Private
-  public static String sourceName(String name) {
-    return INSTANCE.newSourceName(name);
+  public static String sourceName(String name, boolean dupOK) {
+    return INSTANCE.newSourceName(name, dupOK);
   }
 
   synchronized ObjectName newObjectName(String name) {
@@ -118,9 +117,13 @@ public enum DefaultMetricsSystem {
     }
   }
 
-  synchronized String newSourceName(String name) {
-    if (sourceNames.map.containsKey(name) && !miniClusterMode) {
-      throw new MetricsException(name +" already exists!");
+  synchronized String newSourceName(String name, boolean dupOK) {
+    if (sourceNames.map.containsKey(name)) {
+      if (dupOK) {
+        return name;
+      } else if (!miniClusterMode) {
+        throw new MetricsException("Metrics source "+ name +" already exists!");
+      }
     }
     return sourceNames.uniqueName(name);
   }

src/test/core/org/apache/hadoop/metrics2/impl/TestMetricsConfig.java

@@ -32,7 +32,6 @@ import static org.apache.hadoop.metrics2.impl.ConfigUtil.*;
  * Test metrics configuration
  */
 public class TestMetricsConfig {
-
   static final Log LOG = LogFactory.getLog(TestMetricsConfig.class);
 
   /**
@@ -110,18 +109,11 @@ public class TestMetricsConfig {
   }
 
   /**
-   * Should throw if missing config files
+   * Should not throw if missing config files
    */
   @Test public void testMissingFiles() {
-    try {
-      MetricsConfig.create("JobTracker");
-    }
-    catch (MetricsConfigException e) {
-      assertTrue("expected the 'cannot locate configuration' exception",
-                 e.getMessage().startsWith("Cannot locate configuration"));
-      return;
-    }
-    fail("should've thrown");
+    MetricsConfig config = MetricsConfig.create("JobTracker", "non-existent.properties");
+    assertTrue(config.isEmpty());
   }
 
   /**
@@ -148,5 +140,4 @@ public class TestMetricsConfig {
   public static String getTestFilename(String basename) {
     return "build/classes/"+ basename +".properties";
   }
-
 }

src/test/core/org/apache/hadoop/metrics2/impl/TestMetricsSystemImpl.java

@@ -34,9 +34,12 @@ import com.google.common.collect.Iterables;
 import org.apache.commons.configuration.SubsetConfiguration;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.metrics2.MetricsException;
 import static org.apache.hadoop.test.MoreAsserts.*;
 import org.apache.hadoop.metrics2.MetricsRecord;
 import org.apache.hadoop.metrics2.MetricsSink;
+import org.apache.hadoop.metrics2.MetricsSource;
+import org.apache.hadoop.metrics2.MetricsSystem;
 import org.apache.hadoop.metrics2.MetricsTag;
 import org.apache.hadoop.metrics2.annotation.*;
 import static org.apache.hadoop.metrics2.lib.Interns.*;
@@ -101,6 +104,28 @@ public class TestMetricsSystemImpl {
     assertEquals("output", mr1, mr2);
   }
 
+  @Test public void testRegisterDups() {
+    MetricsSystem ms = new MetricsSystemImpl();
+    TestSource ts1 = new TestSource("ts1");
+    TestSource ts2 = new TestSource("ts2");
+    ms.register("ts1", "", ts1);
+    MetricsSource s1 = ms.getSource("ts1");
+    assertNotNull(s1);
+    // should work when metrics system is not started
+    ms.register("ts1", "", ts2);
+    MetricsSource s2 = ms.getSource("ts1");
+    assertNotNull(s2);
+    assertNotSame(s1, s2);
+  }
+
+  @Test(expected=MetricsException.class) public void testRegisterDupError() {
+    MetricsSystem ms = new MetricsSystemImpl("test");
+    TestSource ts = new TestSource("ts");
+    ms.register(ts);
+    ms.register(ts);
+  }
+
+
   private void checkMetricsRecords(List<MetricsRecord> recs) {
     LOG.debug(recs);
     MetricsRecord r = recs.get(0);

src/test/core/org/apache/hadoop/metrics2/impl/TestSinkQueue.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.metrics2.impl;
 
 import java.util.ConcurrentModificationException;
+import java.util.concurrent.CountDownLatch;
 
 import org.junit.Test;
 import static org.junit.Assert.*;
@@ -32,8 +33,7 @@ import static org.apache.hadoop.metrics2.impl.SinkQueue.*;
  * Test the half-blocking metrics sink queue
  */
 public class TestSinkQueue {
-
-  private final Log LOG = LogFactory.getLog(TestSinkQueue.class);
+  private static final Log LOG = LogFactory.getLog(TestSinkQueue.class);
 
   /**
    * Test common use case
@@ -48,7 +48,7 @@ public class TestSinkQueue {
 
     assertTrue("should enqueue", q.enqueue(2));
     q.consume(new Consumer<Integer>() {
-      public void consume(Integer e) {
+      @Override public void consume(Integer e) {
         assertEquals("element", 2, (int) e);
       }
     });
@@ -64,6 +64,11 @@ public class TestSinkQueue {
    * @throws Exception
    */
   @Test public void testEmptyBlocking() throws Exception {
+    testEmptyBlocking(0);
+    testEmptyBlocking(100);
+  }
+
+  private void testEmptyBlocking(int awhile) throws Exception {
     final SinkQueue<Integer> q = new SinkQueue<Integer>(2);
     final Runnable trigger = mock(Runnable.class);
     // try consuming emtpy equeue and blocking
@@ -72,7 +77,7 @@ public class TestSinkQueue {
         try {
           assertEquals("element", 1, (int) q.dequeue());
           q.consume(new Consumer<Integer>() {
-            public void consume(Integer e) {
+            @Override public void consume(Integer e) {
               assertEquals("element", 2, (int) e);
               trigger.run();
             }
@@ -84,7 +89,10 @@ public class TestSinkQueue {
       }
     };
     t.start();
-    Thread.yield(); // Let the other block
+    // Should work with or without sleep
+    if (awhile > 0) {
+      Thread.sleep(awhile);
+    }
     q.enqueue(1);
     q.enqueue(2);
     t.join();
@@ -104,7 +112,7 @@ public class TestSinkQueue {
 
     q.enqueue(3);
     q.consume(new Consumer<Integer>() {
-      public void consume(Integer e) {
+      @Override public void consume(Integer e) {
         assertEquals("element", 3, (int) e);
       }
     });
@@ -127,7 +135,7 @@ public class TestSinkQueue {
     final Runnable trigger = mock(Runnable.class);
     q.consumeAll(new Consumer<Integer>() {
       private int expected = 0;
-      public void consume(Integer e) {
+      @Override public void consume(Integer e) {
         assertEquals("element", expected++, (int) e);
         trigger.run();
       }
@@ -147,7 +155,7 @@ public class TestSinkQueue {
 
     try {
       q.consume(new Consumer<Integer>() {
-        public void consume(Integer e) {
+        @Override public void consume(Integer e) {
           throw ex;
         }
       });
@@ -196,22 +204,22 @@ public class TestSinkQueue {
     assertEquals("queue back", 2, (int) q.back());
     assertTrue("should drop", !q.enqueue(3)); // should not block
     shouldThrowCME(new Fun() {
-      public void run() {
+      @Override public void run() {
         q.clear();
       }
     });
     shouldThrowCME(new Fun() {
-      public void run() throws Exception {
+      @Override public void run() throws Exception {
         q.consume(null);
       }
     });
     shouldThrowCME(new Fun() {
-      public void run() throws Exception {
+      @Override public void run() throws Exception {
         q.consumeAll(null);
       }
     });
     shouldThrowCME(new Fun() {
-      public void run() throws Exception {
+      @Override public void run() throws Exception {
         q.dequeue();
       }
     });
@@ -229,21 +237,26 @@ public class TestSinkQueue {
       LOG.info(e);
       return;
     }
-    fail("should've thrown");
+    LOG.error("should've thrown CME");
+    fail("should've thrown CME");
   }
 
   private SinkQueue<Integer> newSleepingConsumerQueue(int capacity,
-                                                      int... values) {
+      int... values) throws Exception {
     final SinkQueue<Integer> q = new SinkQueue<Integer>(capacity);
     for (int i : values) {
       q.enqueue(i);
     }
+    final CountDownLatch barrier = new CountDownLatch(1);
     Thread t = new Thread() {
       @Override public void run() {
         try {
+          Thread.sleep(10); // causes failure without barrier
           q.consume(new Consumer<Integer>() {
+            @Override
             public void consume(Integer e) throws InterruptedException {
               LOG.info("sleeping");
+              barrier.countDown();
               Thread.sleep(1000 * 86400); // a long time
             }
           });
@@ -256,7 +269,7 @@ public class TestSinkQueue {
     t.setName("Sleeping consumer");
     t.setDaemon(true);  // so jvm can exit
     t.start();
-    Thread.yield(); // Let the consumer consume
+    barrier.await();
     LOG.debug("Returning new sleeping consumer queue");
     return q;
   }
@@ -264,5 +277,4 @@ public class TestSinkQueue {
   static interface Fun {
     void run() throws Exception;
   }
-
 }