HBASE-25913 Introduce EnvironmentEdge.Clock and Clock.currentTimeAdvancing

- Use a per region Clock in HRegion to ensure the time advances

- Update unit tests as needed

Author: apurtell

hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupDelete.java

@@ -28,9 +28,8 @@
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
-import org.apache.hadoop.hbase.util.EnvironmentEdge;
+import org.apache.hadoop.hbase.util.BaseEnvironmentEdge;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
-import org.apache.hadoop.hbase.util.HashedBytes;
 import org.apache.hadoop.util.ToolRunner;
 import org.junit.Assert;
 import org.junit.ClassRule;
@@ -113,21 +112,7 @@ public void testBackupDeleteCommand() throws Exception {
   public void testBackupPurgeOldBackupsCommand() throws Exception {
     LOG.info("test backup delete (purge old backups) on a single table with data: command-line");
     List<TableName> tableList = Lists.newArrayList(table1);
-    EnvironmentEdgeManager.injectEdge(new EnvironmentEdge() {
-      // time - 2 days
-      @Override
-      public long currentTime() {
-        return System.currentTimeMillis() - 2 * 24 * 3600 * 1000 ;
-      }
-      @Override
-      public Clock getClock(HashedBytes name) {
-        return null;
-      }
-      @Override
-      public boolean removeClock(Clock clock) {
-        return false;
-      }
-    });
+    EnvironmentEdgeManager.injectEdge(new TwoDaysBackEnvironmentEdge());
     String backupId = fullTableBackup(tableList);
     assertTrue(checkSucceeded(backupId));
 
@@ -169,4 +154,13 @@ public boolean removeClock(Clock clock) {
     LOG.info(baos.toString());
     assertTrue(output.indexOf("Deleted 1 backups") >= 0);
   }
+
+  static class TwoDaysBackEnvironmentEdge extends BaseEnvironmentEdge {
+    // time - 2 days
+    @Override
+    public long currentTime() {
+      return System.currentTimeMillis() - 2 * 24 * 3600 * 1000 ;
+    }
+  }
+
 }

hbase-common/src/main/java/org/apache/hadoop/hbase/util/BaseEnvironmentEdge.java

@@ -60,7 +60,8 @@ public long currentTime() {
     }
     @Override
     public long currentTimeAdvancing() {
-      throw new UnsupportedOperationException("Default clock does not implement currentTimeAdvancing()");
+      throw new UnsupportedOperationException(
+        "Default clock does not implement currentTimeAdvancing()");
     }
     @Override
     public void get() {

hbase-common/src/main/java/org/apache/hadoop/hbase/util/EnvironmentEdge.java

@@ -41,13 +41,13 @@ public interface EnvironmentEdge {
   /**
    * Get the clock associated with the given identifier.
    * @param name clock identifier
-   * @returns the clock instance for the given identifier
+   * @return the clock instance for the given identifier
    */
   Clock getClock(HashedBytes name);
 
   /**
-   * Remove the clock with the given identifier.
-   * @param clockId clock identifier
+   * Release the reference to and possible remove this clock.
+   * @param clock the clock
    * @return true if the clock was removed, false if it did not exist
    */
   boolean removeClock(Clock clock);
@@ -64,7 +64,6 @@ public interface Clock {
 
     /**
      * Returns the current time using a named clock.
-     * @param clockId clock identifier
      * @return The current time, according to the given named clock.
      */
     long currentTime();
@@ -74,7 +73,6 @@ public interface Clock {
      * at least one tick before returning.
      * <p>
      * This method may block the current thread's execution or cause it to yield.
-     * @param clockId clock identifier
      * @return The current time, according to the given named clock.
      * @throws InterruptedException if interrupted while waiting for the clock to advance
      */

hbase-common/src/main/java/org/apache/hadoop/hbase/util/HashedBytes.java

@@ -36,7 +36,12 @@ public class HashedBytes {
 
   public HashedBytes(byte[] bytes) {
     this.bytes = bytes;
-    hashCode = Bytes.hashCode(bytes);
+    hashCode = Bytes.hashCode(this.bytes);
+  }
+
+  public HashedBytes(byte[]... bytes) {
+    this.bytes = Bytes.add(bytes);
+    hashCode = Bytes.hashCode(this.bytes);
   }
 
   public byte[] getBytes() {
@@ -50,10 +55,12 @@ public int hashCode() {
 
   @Override
   public boolean equals(Object obj) {
-    if (this == obj)
+    if (this == obj) {
       return true;
-    if (obj == null || getClass() != obj.getClass())
+    }
+    if (obj == null || getClass() != obj.getClass()) {
       return false;
+    }
     HashedBytes other = (HashedBytes) obj;
     return (hashCode == other.hashCode) && Arrays.equals(bytes, other.bytes);
   }

hbase-common/src/main/java/org/apache/hadoop/hbase/util/IncrementingEnvironmentEdge.java

@@ -26,6 +26,40 @@
 @InterfaceAudience.Private
 public class IncrementingEnvironmentEdge extends BaseEnvironmentEdge {
 
+  class ManualIncrementingClock implements EnvironmentEdge.Clock {
+
+    private HashedBytes name;
+
+    public ManualIncrementingClock(HashedBytes name) {
+      this.name = name;
+    }
+
+    @Override
+    public HashedBytes getName() {
+      return name;
+    }
+
+    @Override
+    public long currentTime() {
+      return System.currentTimeMillis() + timeIncrement;
+    }
+
+    @Override
+    public long currentTimeAdvancing() {
+      return System.currentTimeMillis() + timeIncrement;
+    }
+
+    @Override
+    public void get() {
+    }
+
+    @Override
+    public boolean remove() {
+      return true;
+    }
+
+  }
+
   private long timeIncrement;
 
   /**
@@ -63,4 +97,14 @@ public synchronized long incrementTime(long amount) {
     return timeIncrement;
   }
 
+  @Override
+  public Clock getClock(HashedBytes name) {
+    return new ManualIncrementingClock(name);
+  }
+
+  @Override
+  public boolean removeClock(Clock clock) {
+    return true;
+  }
+
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/util/ManualEnvironmentEdge.java renamed to hbase-common/src/main/java/org/apache/hadoop/hbase/util/ManualEnvironmentEdge.java

@@ -27,6 +27,40 @@
 @InterfaceAudience.Private
 public class ManualEnvironmentEdge extends BaseEnvironmentEdge {
 
+  class ManualFixedClock implements EnvironmentEdge.Clock {
+
+    private HashedBytes name;
+
+    public ManualFixedClock(HashedBytes name) {
+      this.name = name;
+    }
+
+    @Override
+    public HashedBytes getName() {
+      return name;
+    }
+
+    @Override
+    public long currentTime() {
+      return value;
+    }
+
+    @Override
+    public long currentTimeAdvancing() {
+      return value;
+    }
+
+    @Override
+    public void get() {
+    }
+
+    @Override
+    public boolean remove() {
+      return true;
+    }
+
+  }
+
   protected long value;
 
   public ManualEnvironmentEdge() {
@@ -51,4 +85,14 @@ public long currentTime() {
     return this.value;
   }
 
+  @Override
+  public Clock getClock(HashedBytes name) {
+    return new ManualFixedClock(name);
+  }
+
+  @Override
+  public boolean removeClock(Clock clock) {
+    return true;
+  }
+
 }

hbase-common/src/main/java/org/apache/hadoop/hbase/util/clock/BoundedIncrementSpinAdvancingClock.java

@@ -50,12 +50,11 @@ public long currentTimeAdvancing() throws InterruptedException {
     // If we have advanced too far, now we have to wait for the system clock to
     // catch up.
     if (currentAdvance.incrementAndGet() >= MAX_ADVANCE) {
-      long now;
       while (true) {
-        now = System.currentTimeMillis();
+        long now = System.currentTimeMillis();
         if (now > lastTime.get()) {
-          final long updateTime = now;
-          return lastTime.updateAndGet(x -> update(updateTime));
+          lastTime.set(update(now));
+          return now;
         }
         spin();
       }
@@ -76,15 +75,15 @@ protected long advance(long last) {
     // System clock hasn't moved forward. Increment our notion of current tick to keep
     // the time advancing.
     currentAdvance.incrementAndGet();
-    return super.advance(last);
+    return last + 1;
   }
 
   // Broken out to inlinable method for subclassing and instrumentation.
   protected long update(long now) {
     // System clock has moved ahead of our notion of current tick. Move us forward to match.
     // We do that just by returning the current time, which was passed in to us as 'n'.
     currentAdvance.set(0);
-    return super.update(now);
+    return now;
   }
 
   // Broken out to inlinable method for subclassing and instrumentation.

hbase-common/src/main/java/org/apache/hadoop/hbase/util/clock/BoundedIncrementYieldAdvancingClock.java

@@ -32,7 +32,7 @@
 @InterfaceAudience.Private
 public class BoundedIncrementYieldAdvancingClock extends IncrementAdvancingClock {
 
-  static final int MAX_ADVANCE = 1000; // one second in milliseconds
+  static final int MAX_ADVANCE = 1000;
 
   protected AtomicLong currentAdvance = new AtomicLong(0);
 
@@ -50,12 +50,11 @@ public long currentTimeAdvancing() throws InterruptedException {
     // If we have advanced too far, now we have to wait for the system clock to
     // catch up.
     if (currentAdvance.incrementAndGet() >= MAX_ADVANCE) {
-      long now;
       while (true) {
-        now = System.currentTimeMillis();
+        long now = System.currentTimeMillis();
         if (now > lastTime.get()) {
-          final long updateTime = now;
-          return lastTime.updateAndGet(x -> update(updateTime));
+          lastTime.set(update(now));
+          return now;
         }
         spin();
       }
@@ -76,15 +75,15 @@ protected long advance(long last) {
     // System clock hasn't moved forward. Increment our notion of current tick to keep
     // the time advancing.
     currentAdvance.incrementAndGet();
-    return super.advance(last);
+    return last + 1;
   }
 
   // Broken out to inlinable method for subclassing and instrumentation.
   protected long update(long now) {
     // System clock has moved ahead of our notion of current tick. Move us forward to match.
     // We do that just by returning the current time, which was passed in to us as 'n'.
     currentAdvance.set(0);
-    return super.update(now);
+    return now;
   }
 
   // Broken out to inlinable method for subclassing and instrumentation.

hbase-common/src/main/java/org/apache/hadoop/hbase/util/clock/IncrementAdvancingClock.java

@@ -20,8 +20,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
-import org.apache.hadoop.hbase.util.HashedBytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdge.Clock;
+import org.apache.hadoop.hbase.util.HashedBytes;
 import org.apache.yetus.audience.InterfaceAudience;
 
 /**

hbase-common/src/main/java/org/apache/hadoop/hbase/util/clock/SpinAdvancingClock.java

@@ -20,8 +20,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
-import org.apache.hadoop.hbase.util.HashedBytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdge.Clock;
+import org.apache.hadoop.hbase.util.HashedBytes;
 import org.apache.yetus.audience.InterfaceAudience;
 
 /**