addressing comments

Author: virajjasani

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/EditLogTailer.java

@@ -36,6 +36,7 @@
 
 import org.apache.hadoop.thirdparty.com.google.common.collect.Iterators;
 import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.util.Timer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -55,12 +56,10 @@
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.security.SecurityUtil;
 
-import static org.apache.hadoop.util.Time.monotonicNow;
 import static org.apache.hadoop.util.ExitUtil.terminate;
 
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
-import org.apache.hadoop.util.Time;
 
 
 /**
@@ -172,14 +171,21 @@ public class EditLogTailer {
    */
   private final long maxTxnsPerLock;
 
+  /**
+   * Timer instance to be set only using constructor.
+   * Only tests can reassign this by using setTimerForTests().
+   * For source code, this timer instance should be treated as final.
+   */
+  private Timer timer;
+
   public EditLogTailer(FSNamesystem namesystem, Configuration conf) {
     this.tailerThread = new EditLogTailerThread();
     this.conf = conf;
     this.namesystem = namesystem;
+    this.timer = new Timer();
     this.editLog = namesystem.getEditLog();
-    
-    lastLoadTimeMs = monotonicNow();
-    resetLastRollTimeMs();
+    this.lastLoadTimeMs = timer.monotonicNow();
+    this.lastRollTimeMs = timer.monotonicNow();
 
     logRollPeriodMs = conf.getTimeDuration(
         DFSConfigKeys.DFS_HA_LOGROLL_PERIOD_KEY,
@@ -301,7 +307,7 @@ public Void run() throws Exception {
         long editsTailed = 0;
         // Fully tail the journal to the end
         do {
-          long startTime = Time.monotonicNow();
+          long startTime = timer.monotonicNow();
           try {
             NameNode.getNameNodeMetrics().addEditLogTailInterval(
                 startTime - lastLoadTimeMs);
@@ -312,7 +318,7 @@ public Void run() throws Exception {
             throw new IOException(e);
           } finally {
             NameNode.getNameNodeMetrics().addEditLogTailTime(
-                Time.monotonicNow() - startTime);
+                timer.monotonicNow() - startTime);
           }
         } while(editsTailed > 0);
         return null;
@@ -336,7 +342,7 @@ public long doTailEdits() throws IOException, InterruptedException {
         LOG.debug("lastTxnId: " + lastTxnId);
       }
       Collection<EditLogInputStream> streams;
-      long startTime = Time.monotonicNow();
+      long startTime = timer.monotonicNow();
       try {
         streams = editLog.selectInputStreams(lastTxnId + 1, 0,
             null, inProgressOk, true);
@@ -349,7 +355,7 @@ public long doTailEdits() throws IOException, InterruptedException {
         return 0;
       } finally {
         NameNode.getNameNodeMetrics().addEditLogFetchTime(
-            Time.monotonicNow() - startTime);
+            timer.monotonicNow() - startTime);
       }
       if (LOG.isDebugEnabled()) {
         LOG.debug("edit streams to load from: " + streams.size());
@@ -374,7 +380,7 @@ public long doTailEdits() throws IOException, InterruptedException {
       }
 
       if (editsLoaded > 0) {
-        lastLoadTimeMs = monotonicNow();
+        lastLoadTimeMs = timer.monotonicNow();
       }
       lastLoadedTxnId = image.getLastAppliedTxId();
       return editsLoaded;
@@ -395,7 +401,7 @@ public long getLastLoadTimeMs() {
    */
   private boolean tooLongSinceLastLoad() {
     return logRollPeriodMs >= 0 && 
-      (monotonicNow() - lastRollTimeMs) > logRollPeriodMs;
+      (timer.monotonicNow() - lastRollTimeMs) > logRollPeriodMs;
   }
 
   /**
@@ -423,20 +429,38 @@ void triggerActiveLogRoll() {
     try {
       future = rollEditsRpcExecutor.submit(getNameNodeProxy());
       future.get(rollEditsTimeoutMs, TimeUnit.MILLISECONDS);
-      resetLastRollTimeMs();
+      this.lastRollTimeMs = timer.monotonicNow();
       lastRollTriggerTxId = lastLoadedTxnId;
     } catch (ExecutionException | InterruptedException e) {
       LOG.warn("Unable to trigger a roll of the active NN", e);
     } catch (TimeoutException e) {
-      future.cancel(true);
+      if (future != null) {
+        future.cancel(true);
+      }
       LOG.warn(String.format(
           "Unable to finish rolling edits in %d ms", rollEditsTimeoutMs));
     }
   }
 
+  /**
+   * This is only to be used by tests. For source code, the only way to
+   * set timer is by using EditLogTailer constructor.
+   *
+   * @param newTimer Timer instance provided by tests.
+   */
+  @VisibleForTesting
+  void setTimerForTest(final Timer newTimer) {
+    this.timer = newTimer;
+  }
+
+  /**
+   * Used by tests. Return Timer instance used by EditLogTailer.
+   *
+   * @return Return Timer instance used by EditLogTailer.
+   */
   @VisibleForTesting
-  public void resetLastRollTimeMs() {
-    this.lastRollTimeMs = monotonicNow();
+  Timer getTimer() {
+    return timer;
   }
 
   @VisibleForTesting
@@ -498,15 +522,15 @@ private void doWork() {
           // name system lock will be acquired to further block even the block
           // state updates.
           namesystem.cpLockInterruptibly();
-          long startTime = Time.monotonicNow();
+          long startTime = timer.monotonicNow();
           try {
             NameNode.getNameNodeMetrics().addEditLogTailInterval(
                 startTime - lastLoadTimeMs);
             editsTailed = doTailEdits();
           } finally {
             namesystem.cpUnlock();
             NameNode.getNameNodeMetrics().addEditLogTailTime(
-                Time.monotonicNow() - startTime);
+                timer.monotonicNow() - startTime);
           }
           //Update NameDirSize Metric
           if (triggeredLogRoll) {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestEditLogTailer.java

@@ -56,6 +56,7 @@
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.util.Timer;
 import org.slf4j.event.Level;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -397,17 +398,11 @@ public void testStandbyTriggersLogRollsWhenTailInProgressEdits()
 
     // Time in seconds to wait before checking if edit logs are rolled while
     // expecting edit log roll.
-    // Aggressive tests like this one are resource sensitive and bound to
-    // fail with slow system, hence waiting to confirm
-    // curSegmentTxId for 7 seconds makes it more reliable in case
-    // EditLogTailer#doWork() takes few more milliseconds to retrieve result
-    // from EditLogTailer#triggerActiveLogRoll() and lastRollTimeMs is not
-    // updated frequently.
-    final int logRollWaitTime = 7;
+    final int logRollWaitTime = 3;
 
+    final int logRollPeriod = standbyCatchupWaitTime + noLogRollWaitTime + 1;
     Configuration conf = getConf();
-    conf.setInt(DFSConfigKeys.DFS_HA_LOGROLL_PERIOD_KEY,
-        standbyCatchupWaitTime + noLogRollWaitTime + 2);
+    conf.setInt(DFSConfigKeys.DFS_HA_LOGROLL_PERIOD_KEY, logRollPeriod);
     conf.setInt(DFSConfigKeys.DFS_HA_TAILEDITS_PERIOD_KEY, 1);
     conf.setBoolean(DFSConfigKeys.DFS_HA_TAILEDITS_INPROGRESS_KEY, true);
 
@@ -436,33 +431,32 @@ public void testStandbyTriggersLogRollsWhenTailInProgressEdits()
       waitForStandbyToCatchUpWithInProgressEdits(standby, activeTxId,
           standbyCatchupWaitTime);
 
+      long curTime = standby.getNamesystem().getEditLogTailer().getTimer()
+          .monotonicNow();
+      long inSufficientTimeForLogRoll =
+          TimeUnit.SECONDS.toMillis(logRollPeriod / 3);
+      TestTimer testTimer = new TestTimer(curTime + inSufficientTimeForLogRoll);
+      standby.getNamesystem().getEditLogTailer().setTimerForTest(testTimer);
+
       for (int i = DIRS_TO_MAKE / 2; i < DIRS_TO_MAKE; i++) {
         NameNodeAdapter.mkdirs(active, getDirPath(i),
             new PermissionStatus("test", "test",
             new FsPermission((short)00755)), true);
-        // reset lastRollTimeMs in EditLogTailer.
-        active.getNamesystem().getEditLogTailer().resetLastRollTimeMs();
       }
 
-      // We should explicitly update lastRollTimeMs in EditLogTailer
-      // so that our timeout test provided just below can take advantage
-      // of validation: (monotonicNow() - lastRollTimeMs) > logRollPeriodMs
-      // provided in EditLogTailer#tooLongSinceLastLoad().
-      active.getNamesystem().getEditLogTailer().resetLastRollTimeMs();
-
       try {
-        // Aggressive checks like this one are resource sensitive and bound to
-        // fail with slow system, hence waiting to confirm
-        // curSegmentTxId for (noLogRollWaitTime - 1) seconds
-        // makes it more reliable in case EditLogTailer#doWork() takes few
-        // more milliseconds to retrieve result from
-        // EditLogTailer#triggerActiveLogRoll().
-        checkForLogRoll(active, origTxId, noLogRollWaitTime - 1);
+        checkForLogRoll(active, origTxId, noLogRollWaitTime);
         fail("Expected to timeout");
       } catch (TimeoutException e) {
         // expected
       }
 
+      long curTimeNew = standby.getNamesystem().getEditLogTailer().getTimer()
+          .monotonicNow();
+      long sufficientTimeForLogRoll =
+          TimeUnit.SECONDS.toMillis(logRollPeriod * 3);
+      testTimer.setTime(curTimeNew + sufficientTimeForLogRoll);
+
       checkForLogRoll(active, origTxId, logRollWaitTime);
     } finally {
       cluster.shutdown();
@@ -485,7 +479,7 @@ private static void checkForLogRoll(final NameNode active,
       long curSegmentTxId = active.getNamesystem().getFSImage().getEditLog()
           .getCurSegmentTxId();
       return (origTxId != curSegmentTxId);
-    }, 500, TimeUnit.SECONDS.toMillis(maxWaitSec));
+    }, 100, TimeUnit.SECONDS.toMillis(maxWaitSec));
   }
 
   private static MiniDFSCluster createMiniDFSCluster(Configuration conf,
@@ -502,4 +496,23 @@ private static MiniDFSCluster createMiniDFSCluster(Configuration conf,
         .build();
     return cluster;
   }
+
+  private static final class TestTimer extends Timer {
+
+    private volatile long time;
+
+    private TestTimer(long time) {
+      this.time = time;
+    }
+
+    private void setTime(long newTime) {
+      this.time = newTime;
+    }
+
+    @Override
+    public long monotonicNow() {
+      return time;
+    }
+  }
+
 }