You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ap...@apache.org on 2017/07/17 17:59:26 UTC
[93/94] [abbrv] hbase git commit: HBASE-14070 - Core HLC (Sai Teja
Ranuva) Rebased by Amit Patel
HBASE-14070 - Core HLC (Sai Teja Ranuva) Rebased by Amit Patel
Core HLC. Adds HLC framework (i.e., Clock, Timestamp API) and tests
Not used by core (yet).
Signed-off-by: Michael Stack <st...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/d53fbc74
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/d53fbc74
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/d53fbc74
Branch: refs/heads/HBASE-14070.HLC
Commit: d53fbc74ba6e3b4489b82bb48ced6dfe470c8b10
Parents: 2d5a0fb
Author: Amit Patel <ia...@gmail.com>
Authored: Thu Jul 6 13:01:23 2017 -0700
Committer: Apekshit Sharma <ap...@apache.org>
Committed: Mon Jul 17 10:56:08 2017 -0700
----------------------------------------------------------------------
.../apache/hadoop/hbase/HTableDescriptor.java | 10 +
.../hadoop/hbase/client/TableDescriptor.java | 9 +
.../hbase/client/TableDescriptorBuilder.java | 40 ++
.../java/org/apache/hadoop/hbase/Clock.java | 408 +++++++++++++++++++
.../java/org/apache/hadoop/hbase/ClockType.java | 38 ++
.../apache/hadoop/hbase/SettableTimestamp.java | 2 +-
.../org/apache/hadoop/hbase/TimestampType.java | 309 ++++++++++++++
.../java/org/apache/hadoop/hbase/TestClock.java | 293 +++++++++++++
.../apache/hadoop/hbase/TestTimestampType.java | 207 ++++++++++
.../master/procedure/ModifyTableProcedure.java | 4 +
.../hadoop/hbase/regionserver/HRegion.java | 78 +++-
.../hbase/regionserver/HRegionServer.java | 24 ++
.../hadoop/hbase/regionserver/HStore.java | 11 +-
.../hadoop/hbase/regionserver/Region.java | 6 +
.../regionserver/RegionServerServices.java | 4 +
.../apache/hadoop/hbase/regionserver/Store.java | 7 +
.../hadoop/hbase/regionserver/StoreScanner.java | 102 ++++-
.../DropDeletesCompactionScanQueryMatcher.java | 19 +-
.../querymatcher/LegacyScanQueryMatcher.java | 3 +-
.../MajorCompactionScanQueryMatcher.java | 5 +-
.../MinorCompactionScanQueryMatcher.java | 3 +-
.../NormalUserScanQueryMatcher.java | 3 +-
.../querymatcher/RawScanQueryMatcher.java | 3 +-
.../querymatcher/ScanQueryMatcher.java | 24 +-
.../StripeCompactionScanQueryMatcher.java | 5 +-
.../hbase/security/access/AccessController.java | 10 +-
.../hadoop/hbase/HBaseTestingUtility.java | 28 ++
.../hadoop/hbase/MockRegionServerServices.java | 5 +
.../hadoop/hbase/TestClockWithCluster.java | 115 ++++++
.../coprocessor/TestIncrementTimeRange.java | 39 +-
.../hadoop/hbase/mapreduce/TestCopyTable.java | 31 +-
.../hadoop/hbase/master/MockRegionServer.java | 7 +
.../regionserver/TestCompactingMemStore.java | 46 ++-
.../hbase/regionserver/TestDefaultMemStore.java | 59 ++-
.../hadoop/hbase/regionserver/TestHRegion.java | 74 +++-
.../regionserver/TestHRegionReplayEvents.java | 3 +
.../regionserver/TestRegionSplitPolicy.java | 4 +
.../hbase/regionserver/TestStoreScanner.java | 171 +++++---
.../hbase/regionserver/TestWALLockup.java | 3 +
.../TestCompactionScanQueryMatcher.java | 3 +-
.../querymatcher/TestUserScanQueryMatcher.java | 12 +-
.../regionserver/wal/AbstractTestWALReplay.java | 47 ++-
.../access/TestCellACLWithMultipleVersions.java | 180 +++++---
.../hbase/util/TestCoprocessorScanPolicy.java | 30 +-
.../apache/hadoop/hbase/util/TestTableName.java | 5 +-
45 files changed, 2260 insertions(+), 229 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java
index c09d434..c865255 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java
@@ -70,6 +70,7 @@ public class HTableDescriptor implements TableDescriptor, Comparable<HTableDescr
public static final long DEFAULT_MEMSTORE_FLUSH_SIZE = TableDescriptorBuilder.DEFAULT_MEMSTORE_FLUSH_SIZE;
public static final int DEFAULT_REGION_REPLICATION = TableDescriptorBuilder.DEFAULT_REGION_REPLICATION;
public static final boolean DEFAULT_REGION_MEMSTORE_REPLICATION = TableDescriptorBuilder.DEFAULT_REGION_MEMSTORE_REPLICATION;
+ public static final ClockType DEFAULT_CLOCK_TYPE = TableDescriptorBuilder.DEFAULT_CLOCK_TYPE;
protected final ModifyableTableDescriptor delegatee;
/**
@@ -382,6 +383,15 @@ public class HTableDescriptor implements TableDescriptor, Comparable<HTableDescr
}
/**
+ * Returns the clock type for the table.
+ * @return clock type for the table.
+ */
+ @Override
+ public ClockType getClockType() {
+ return delegatee.getClockType();
+ }
+
+ /**
* Returns the size of the memstore after which a flush to filesystem is triggered.
*
* @return memory cache flush size for each hregion, -1 if not set.
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptor.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptor.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptor.java
index 65e64f8..abd6ee9 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptor.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptor.java
@@ -24,6 +24,8 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
+
+import org.apache.hadoop.hbase.ClockType;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.util.Bytes;
@@ -106,6 +108,13 @@ public interface TableDescriptor {
Durability getDurability();
/**
+ * Returns the clock type setting for the table.
+ *
+ * @return clock type setting for the table.
+ */
+ ClockType getClockType();
+
+ /**
* Returns an unmodifiable collection of all the {@link ColumnFamilyDescriptor} of
* all the column families of the table.
*
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java
index 44d5c99..2e8d3e9 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java
@@ -36,6 +36,7 @@ import java.util.stream.Stream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.ClockType;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
@@ -117,6 +118,14 @@ public class TableDescriptorBuilder {
= new Bytes(Bytes.toBytes("DURABILITY"));
/**
+ * {@link ClockType} setting for the table.
+ */
+ @InterfaceAudience.Private
+ static final String CLOCK_TYPE = "CLOCK_TYPE";
+ private static final Bytes CLOCK_TYPE_KEY
+ = new Bytes(Bytes.toBytes(CLOCK_TYPE));
+
+ /**
* The number of region replicas for the table.
*/
@InterfaceAudience.Private
@@ -149,6 +158,12 @@ public class TableDescriptorBuilder {
*/
private static final Durability DEFAULT_DURABLITY = Durability.USE_DEFAULT;
+ /**
+ * Default clock type for HTD is SYSTEM
+ */
+ @InterfaceAudience.Private
+ public static final ClockType DEFAULT_CLOCK_TYPE = ClockType.SYSTEM;
+
@InterfaceAudience.Private
public static final String PRIORITY = "PRIORITY";
private static final Bytes PRIORITY_KEY
@@ -338,6 +353,11 @@ public class TableDescriptorBuilder {
return this;
}
+ public TableDescriptorBuilder setClockType(ClockType clockType) {
+ desc.setClockType(clockType);
+ return this;
+ }
+
public TableDescriptorBuilder setFlushPolicyClassName(String clazz) {
desc.setFlushPolicyClassName(clazz);
return this;
@@ -694,6 +714,24 @@ public class TableDescriptorBuilder {
}
/**
+ * Sets the {@link ClockType} for the table. This defaults to DEFAULT_CLOCK_TYPE.
+ * @param clockType
+ * @return the modifyable TD
+ */
+ public ModifyableTableDescriptor setClockType(ClockType clockType) {
+ return setValue(CLOCK_TYPE_KEY, clockType.name());
+ }
+
+ /**
+ * Returns the clock type for the table.
+ * @return the clock type for the table.
+ */
+ @Override
+ public ClockType getClockType() {
+ return getOrDefault(CLOCK_TYPE_KEY, ClockType::valueOf, DEFAULT_CLOCK_TYPE);
+ }
+
+ /**
* Get the name of the table
*
* @return TableName
@@ -1469,6 +1507,8 @@ public class TableDescriptorBuilder {
public int getColumnFamilyCount() {
return families.size();
}
+
+
}
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-common/src/main/java/org/apache/hadoop/hbase/Clock.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/Clock.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/Clock.java
new file mode 100644
index 0000000..abc6252
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/Clock.java
@@ -0,0 +1,408 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.apache.hadoop.hbase.util.AtomicUtils.updateMax;
+
+/**
+ * A clock is an implementation of an algorithm to get timestamps corresponding to one of the
+ * {@link TimestampType}s for the current time. Different clock implementations can have
+ * different semantics associated with them. Every such clock should be able to map its
+ * representation of time to one of the {link TimestampType}s.
+ * HBase has traditionally been using the {@link java.lang.System#currentTimeMillis()} to
+ * timestamp events in HBase. {@link java.lang.System#currentTimeMillis()} does not give any
+ * guarantees about monotonicity of time. We will keep this implementation of clock in place for
+ * backward compatibility and call it SYSTEM clock.
+ * It is easy to provide monotonically non decreasing time semantics by keeping track of the last
+ * timestamp given by the clock and updating it on receipt of external message. This
+ * implementation of clock is called SYSTEM_MONOTONIC.
+ * SYSTEM Clock and SYSTEM_MONOTONIC clock as described above, both being physical clocks, they
+ * cannot track causality. Hybrid Logical Clocks(HLC), as described in
+ * <a href="http://www.cse.buffalo.edu/tech-reports/2014-04.pdf">HLC Paper</a>, helps tracking
+ * causality using a
+ * <a href="http://research.microsoft.com/en-us/um/people/lamport/pubs/time-clocks.pdf">Logical
+ * Clock</a> but always keeps the logical time close to the wall time or physical time. It kind
+ * of has the advantages of both the worlds. One such advantage being getting consistent
+ * snapshots in physical time as described in the paper. Hybrid Logical Clock has an additional
+ * advantage that it is always monotonically increasing.
+ * Note: It is assumed that any physical clock implementation has millisecond resolution else the
+ * {@link TimestampType} implementation has to changed to accommodate it. It is decided after
+ * careful discussion to go with millisecond resolution in the HLC design document attached in the
+ * issue <a href="https://issues.apache.org/jira/browse/HBASE-14070">HBASE-14070 </a>.
+ */
+
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public interface Clock {
+ long DEFAULT_MAX_CLOCK_SKEW_IN_MS = 30000;
+
+ /**
+ * This is a method to get the current time.
+ *
+ * @return Timestamp of current time in 64 bit representation corresponding to the particular
+ * clock
+ */
+ long now() throws RuntimeException;
+
+ /**
+ * This is a method to update the current time with the passed timestamp.
+ * @param timestamp
+ * @return Timestamp of current time in 64 bit representation corresponding to the particular
+ * clock
+ */
+ long update(long timestamp) throws RuntimeException;
+
+ /**
+ * @return true if the clock implementation gives monotonically non decreasing timestamps else
+ * false.
+ */
+ boolean isMonotonic();
+
+ /**
+ * @return true if the clock implementation gives monotonically increasing timestamps else false.
+ */
+ boolean isMonotonicallyIncreasing();
+
+ /**
+ * @return {@link org.apache.hadoop.hbase.TimestampType}
+ */
+ TimestampType getTimestampType();
+
+ /**
+ * @return {@link org.apache.hadoop.hbase.ClockType}
+ */
+ ClockType getClockType();
+
+
+ /**
+ * Indicates that Physical Time or Logical Time component has overflowed. This extends
+ * RuntimeException.
+ */
+ @SuppressWarnings("serial")
+ class ClockException extends RuntimeException {
+ public ClockException(String msg) {
+ super(msg);
+ }
+ }
+
+ //////////////////////////////////////////////////////////////////
+ // Physical Clock
+ //////////////////////////////////////////////////////////////////
+
+ interface PhysicalClock {
+ /**
+ * This is a method to get the current time.
+ *
+ * @return Timestamp of current time in 64 bit representation corresponding to the particular
+ * clock
+ */
+ long now() throws RuntimeException;
+
+ /**
+ * This is a method to get the unit of the physical time used by the clock
+ *
+ * @return A {@link TimeUnit}
+ */
+ TimeUnit getTimeUnit();
+ }
+
+ class JavaMillisPhysicalClock implements PhysicalClock {
+ @Override
+ public long now() {
+ return EnvironmentEdgeManager.currentTime();
+ }
+
+ @Override
+ public TimeUnit getTimeUnit() {
+ return TimeUnit.MILLISECONDS;
+ }
+ }
+
+ JavaMillisPhysicalClock DEFAULT_JAVA_MILLIS_PHYSICAL_CLOCK =
+ new JavaMillisPhysicalClock();
+
+ //////////////////////////////////////////////////////////////////
+ // Implementation of clocks
+ //////////////////////////////////////////////////////////////////
+
+ /**
+ * System clock is an implementation of clock which doesn't give any monotonic guarantees.
+ */
+ class System implements Clock, PhysicalClock {
+ private final PhysicalClock physicalClock = DEFAULT_JAVA_MILLIS_PHYSICAL_CLOCK;
+ private final ClockType clockType = ClockType.SYSTEM;
+ private final TimestampType timestampType = TimestampType.PHYSICAL;
+
+ @Override
+ public long now() {
+ return physicalClock.now();
+ }
+
+ @Override
+ public long update(long timestamp) {
+ return physicalClock.now();
+ }
+
+ @Override
+ public boolean isMonotonic() {
+ return false;
+ }
+
+ @Override
+ public boolean isMonotonicallyIncreasing() {
+ return false;
+ }
+
+ public TimeUnit getTimeUnit() {
+ return physicalClock.getTimeUnit();
+ }
+
+ @Override
+ public TimestampType getTimestampType() { return timestampType; }
+
+ @Override
+ public ClockType getClockType() { return clockType; }
+ }
+
+ /**
+ * System clock is an implementation of clock which guarantees monotonically non-decreasing
+ * timestamps.
+ */
+ class SystemMonotonic implements Clock, PhysicalClock {
+ private final long maxClockSkewInMs;
+ private final PhysicalClock physicalClock;
+ private final AtomicLong physicalTime = new AtomicLong();
+ private final ClockType clockType = ClockType.SYSTEM_MONOTONIC;
+ private final TimestampType timestampType = TimestampType.PHYSICAL;
+
+ public SystemMonotonic(PhysicalClock physicalClock, long maxClockSkewInMs) {
+ this.physicalClock = physicalClock;
+ this.maxClockSkewInMs = maxClockSkewInMs > 0 ?
+ maxClockSkewInMs : DEFAULT_MAX_CLOCK_SKEW_IN_MS;
+ }
+
+ public SystemMonotonic() {
+ this.physicalClock = DEFAULT_JAVA_MILLIS_PHYSICAL_CLOCK;
+ this.maxClockSkewInMs = DEFAULT_MAX_CLOCK_SKEW_IN_MS;
+ }
+
+ @Override
+ public long now() {
+ long systemTime = physicalClock.now();
+ updateMax(physicalTime, systemTime);
+ return physicalTime.get();
+ }
+
+ public long update(long targetTimestamp) throws ClockException {
+ final long systemTime = physicalClock.now();
+ if (maxClockSkewInMs > 0 && (targetTimestamp - systemTime) > maxClockSkewInMs) {
+ throw new ClockException(
+ "Received event with timestamp:" + getTimestampType().toString(targetTimestamp)
+ + " which is greater than allowed clock skew ");
+ }
+ final long oldPhysicalTime = systemTime > targetTimestamp ? systemTime : targetTimestamp;
+ updateMax(physicalTime, oldPhysicalTime);
+ return physicalTime.get();
+ }
+
+ @Override
+ public boolean isMonotonic() {
+ return true;
+ }
+
+ @Override
+ public boolean isMonotonicallyIncreasing() {
+ return false;
+ }
+
+ public TimeUnit getTimeUnit() {
+ return physicalClock.getTimeUnit();
+ }
+
+ @VisibleForTesting
+ void setPhysicalTime(long time) {
+ physicalTime.set(time);
+ }
+
+ @Override
+ public TimestampType getTimestampType() { return timestampType; }
+
+ @Override
+ public ClockType getClockType() { return clockType; }
+ }
+
+ class HLC implements Clock, PhysicalClock {
+ private final PhysicalClock physicalClock;
+ private final long maxClockSkew;
+ private final long maxPhysicalTime;
+ private final long maxLogicalTime;
+ private long physicalTime;
+ private long logicalTime;
+ private final ClockType clockType = ClockType.HLC;
+ private final TimestampType timestampType = TimestampType.HYBRID;
+
+ public HLC(PhysicalClock physicalClock, long maxClockSkew) {
+ this.physicalClock = physicalClock;
+ this.maxClockSkew = maxClockSkew > 0 ? maxClockSkew : DEFAULT_MAX_CLOCK_SKEW_IN_MS;
+ this.maxPhysicalTime = timestampType.getMaxPhysicalTime();
+ this.maxLogicalTime = timestampType.getMaxLogicalTime();
+ this.physicalTime = 0;
+ this.logicalTime = 0;
+ }
+
+ public HLC() {
+ this.physicalClock = DEFAULT_JAVA_MILLIS_PHYSICAL_CLOCK;
+ this.maxClockSkew = DEFAULT_MAX_CLOCK_SKEW_IN_MS;
+ this.maxPhysicalTime = timestampType.getMaxPhysicalTime();
+ this.maxLogicalTime = timestampType.getMaxLogicalTime();
+ this.physicalTime = 0;
+ this.logicalTime = 0;
+ }
+
+ @Override
+ public synchronized long now() throws ClockException {
+ final long systemTime = physicalClock.now();
+
+ checkPhysicalTimeOverflow(systemTime, maxPhysicalTime);
+ checkLogicalTimeOverflow(logicalTime, maxLogicalTime);
+
+ if (systemTime <= physicalTime) {
+ logicalTime++;
+ } else if (systemTime > physicalTime) {
+ logicalTime = 0;
+ physicalTime = systemTime;
+ }
+
+ return toTimestamp();
+ }
+
+ /**
+ * Updates {@link HLC} with the given timestamp received from elsewhere (possibly
+ * some other node). Returned timestamp is strict greater than msgTimestamp and local
+ * timestamp.
+ *
+ * @param timestamp timestamp from the external message.
+ * @return a hybrid timestamp of HLC that is strictly greater than local timestamp and
+ * msgTimestamp
+ * @throws ClockException
+ */
+ @Override
+ public synchronized long update(long timestamp)
+ throws ClockException {
+ final long targetPhysicalTime = timestampType.getPhysicalTime(timestamp);
+ final long targetLogicalTime = timestampType.getLogicalTime(timestamp);
+ final long oldPhysicalTime = physicalTime;
+ final long systemTime = physicalClock.now();
+
+ physicalTime = Math.max(Math.max(oldPhysicalTime, targetPhysicalTime), systemTime);
+ checkPhysicalTimeOverflow(systemTime, maxPhysicalTime);
+
+ if (targetPhysicalTime - systemTime > maxClockSkew) {
+ throw new ClockException("Received event with timestamp:" +
+ timestampType.toString(timestamp) + " which is greater than allowed clock skew");
+ }
+ if (physicalTime == oldPhysicalTime && oldPhysicalTime == targetPhysicalTime) {
+ logicalTime = Math.max(logicalTime, targetLogicalTime) + 1;
+ } else if (physicalTime == targetPhysicalTime) {
+ logicalTime = targetLogicalTime + 1;
+ } else if (physicalTime == oldPhysicalTime) {
+ logicalTime++;
+ } else {
+ logicalTime = 0;
+ }
+
+ checkLogicalTimeOverflow(logicalTime, maxLogicalTime);
+ return toTimestamp();
+ }
+
+ @Override
+ public boolean isMonotonic() {
+ return true;
+ }
+
+ @Override
+ public boolean isMonotonicallyIncreasing() {
+ return true;
+ }
+
+ public TimeUnit getTimeUnit() {
+ return physicalClock.getTimeUnit();
+ }
+
+ private long toTimestamp() {
+ return timestampType.toTimestamp(getTimeUnit(), physicalTime, logicalTime);
+ }
+
+ @VisibleForTesting
+ synchronized void setLogicalTime(long logicalTime) {
+ this.logicalTime = logicalTime;
+ }
+
+ @VisibleForTesting
+ synchronized void setPhysicalTime(long physicalTime) {
+ this.physicalTime = physicalTime;
+ }
+
+ @Override
+ public TimestampType getTimestampType() { return timestampType; }
+
+ @Override
+ public ClockType getClockType() { return clockType; }
+ }
+
+ //////////////////////////////////////////////////////////////////
+ // Utility functions
+ //////////////////////////////////////////////////////////////////
+
+ // Only for testing.
+ @VisibleForTesting
+ static Clock getDummyClockOfGivenClockType(ClockType clockType) {
+ if (clockType == ClockType.HLC) {
+ return new Clock.HLC();
+ } else if (clockType == ClockType.SYSTEM_MONOTONIC) {
+ return new Clock.SystemMonotonic();
+ } else {
+ return new Clock.System();
+ }
+ }
+
+ static void checkLogicalTimeOverflow(long logicalTime, long maxLogicalTime) {
+ if (logicalTime >= maxLogicalTime) {
+ // highly unlikely to happen, when it happens, we throw exception for the above layer to
+ // handle it the way they wish to.
+ throw new Clock.ClockException(
+ "Logical Time Overflowed: " + logicalTime + "max " + "logical time: " + maxLogicalTime);
+ }
+ }
+
+ static void checkPhysicalTimeOverflow(long physicalTime, long maxPhysicalTime) {
+ if (physicalTime >= maxPhysicalTime) {
+ // Extremely unlikely to happen, if this happens upper layers may have to kill the server.
+ throw new Clock.ClockException(
+ "Physical Time overflowed: " + physicalTime + " and max physical time:" + maxPhysicalTime);
+ }
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-common/src/main/java/org/apache/hadoop/hbase/ClockType.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/ClockType.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/ClockType.java
new file mode 100644
index 0000000..aee8e43
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/ClockType.java
@@ -0,0 +1,38 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+
+@InterfaceAudience.Private
+public enum ClockType {
+ SYSTEM{
+ public TimestampType timestampType() {
+ return TimestampType.PHYSICAL;
+ }
+ }, SYSTEM_MONOTONIC {
+ public TimestampType timestampType() {
+ return TimestampType.PHYSICAL;
+ }
+ }, HLC {
+ public TimestampType timestampType() {
+ return TimestampType.HYBRID;
+ }
+ };
+ abstract public TimestampType timestampType();
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-common/src/main/java/org/apache/hadoop/hbase/SettableTimestamp.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/SettableTimestamp.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/SettableTimestamp.java
index 8637db2..41d3a0a 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/SettableTimestamp.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/SettableTimestamp.java
@@ -26,8 +26,8 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience;
* Note : Server side Cell implementations in write path must implement this.
* @deprecated as of 2.0 and will be removed in 3.0. Use {@link ExtendedCell} instead
*/
+@Deprecated // Co Processors SHOULD NOT use this if the clock type of the tables is HLC
@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC)
-@Deprecated
public interface SettableTimestamp {
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-common/src/main/java/org/apache/hadoop/hbase/TimestampType.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/TimestampType.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/TimestampType.java
new file mode 100644
index 0000000..e2e2604
--- /dev/null
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/TimestampType.java
@@ -0,0 +1,309 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase;
+
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.classification.InterfaceStability;
+import org.apache.commons.lang.time.FastDateFormat;
+
+import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * {@link TimestampType} is an enum to represent different ways of encoding time in HBase using
+ * 64 bits. Time is usually encoded as a 64-bit long in {@link org.apache.hadoop.hbase.Cell}
+ * timestamps and is used for sorting {@link org.apache.hadoop.hbase.Cell}s, ordering writes etc.
+ * It has methods which help in constructing or interpreting the 64 bit timestamp and getter
+ * methods to read the hard coded constants of the particular {@link TimestampType}.
+ *
+ * <p>
+ * Enum {@link TimestampType} is dumb in a way. It doesn't have any logic other than interpreting
+ * the 64 bits. Any monotonically increasing or monotonically non-decreasing semantics of the
+ * timestamps are the responsibility of the clock implementation generating the particular
+ * timestamps. There can be several clock implementations, and each such implementation can map
+ * its representation of the timestamp to one of the available Timestamp types i.e.
+ * {@link #HYBRID} or {@link #PHYSICAL}. In essence, the {@link TimestampType} is only used
+ * internally by the Clock implementations and thus never exposed to the user. The user has to
+ * know only the different available clock types. So, for the user, timestamp types do not exist.
+ * </p>
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public enum TimestampType {
+ /**
+ * Hybrid is a Timestamp type used to encode both physical time and logical time components
+ * into a single. 64 bits long integer. It has methods to decipher the 64 bits hybrid timestamp
+ * and also to construct the hybrid timestamp.
+ */
+ HYBRID {
+ /**
+ * Hard coded 44-bits for physical time, with most significant bit carrying the sign i.e 0
+ * as we are dealing with positive integers and the remaining 43 bits are to be interpreted as
+ * system time in milli seconds. See
+ * <a href="https://issues.apache.org/jira/browse/HBASE-14070">HBASE-14070 </a> for
+ * understanding the choice of going with the millisecond resolution for physical time.
+ * Thus allowing us to represent all the dates between unix epoch (1970) and year 2248 with
+ * signed timestamp comparison with 44 bits for physical time assuming a millisecond
+ * resolution with signed long integers. Picking 42 bits to represent the physical time has
+ * the problem of representing time until 2039 only, with signed integers, might cause Y2k39
+ * bug hoping HBase to be around till then. The trade-off here is with the year until we can
+ * represent the physical time vs if we are able capture all the events in the worst case
+ * (read: leap seconds etc) without the logical component of the timestamp overflowing. With
+ * 20 bits for logical time, one can represent upto one million events at the same
+ * millisecond. In case of leap seconds, the no of events happening in the same second is very
+ * unlikely to exceed one million.
+ */
+ @SuppressWarnings("unused")
+ private static final int BITS_FOR_PHYSICAL_TIME = 44;
+
+ /**
+ * Remaining 20-bits for logical time, allowing values up to 1,048,576. Logical Time is the
+ * least significant part of the 64 bit timestamp, so unsigned comparison can be used for LT.
+ */
+
+ private static final int BITS_FOR_LOGICAL_TIME = 20;
+
+ /**
+ * Max value for physical time in the {@link #HYBRID} timestamp representation, inclusive.
+ * This assumes signed comparison.
+ */
+ private static final long PHYSICAL_TIME_MAX_VALUE = 0x7ffffffffffL;
+
+ /**
+ * Max value for logical time in the {@link #HYBRID} timestamp representation
+ */
+ static final long LOGICAL_TIME_MAX_VALUE = 0xfffffL;
+
+ public long toEpochTimeMillisFromTimestamp(long timestamp) {
+ return getPhysicalTime(timestamp);
+ }
+
+ public long fromEpochTimeMillisToTimestamp(long timestamp) {
+ return toTimestamp(TimeUnit.MILLISECONDS, timestamp, 0);
+ }
+
+ public long toTimestamp(TimeUnit timeUnit, long physicalTime, long logicalTime) {
+ physicalTime = TimeUnit.MILLISECONDS.convert(physicalTime, timeUnit);
+ return (physicalTime << BITS_FOR_LOGICAL_TIME) + logicalTime;
+ }
+
+ public long getPhysicalTime(long timestamp) {
+ return timestamp >>> BITS_FOR_LOGICAL_TIME; // assume unsigned timestamp
+ }
+
+ long getLogicalTime(long timestamp) {
+ return timestamp & LOGICAL_TIME_MAX_VALUE;
+ }
+
+ public long getMaxPhysicalTime() {
+ return PHYSICAL_TIME_MAX_VALUE;
+ }
+
+ public long getMaxLogicalTime() {
+ return LOGICAL_TIME_MAX_VALUE;
+ }
+
+ int getBitsForLogicalTime() {
+ return BITS_FOR_LOGICAL_TIME;
+ }
+
+ /**
+ * Returns whether the given timestamp is "likely" of {@link #HYBRID} {@link TimestampType}.
+ * Timestamp implementations can use the full range of 64bits long to represent physical and
+ * logical components of time. However, this method returns whether the given timestamp is a
+ * likely representation depending on heuristics for the clock implementation.
+ *
+ * Hybrid timestamps are checked whether they belong to Hybrid range assuming
+ * that Hybrid timestamps will only have > 0 logical time component for timestamps
+ * corresponding to years after 2016. This method will return false if lt > 0 and year is
+ * before 2016. Due to left shifting for Hybrid time, all millisecond-since-epoch timestamps
+ * from years 1970-10K fall into
+ * year 1970 when interpreted as Hybrid timestamps. Thus, {@link #isLikelyOfType(long)} will
+ * return false for timestamps which are in the year 1970 and logical time = 0 when
+ * interpreted as of type Hybrid Time.
+ *
+ * <p>
+ * <b>Note that </b> this method uses heuristics which may not hold
+ * if system timestamps are intermixed from client side and server side or timestamp
+ * sources other than system clock are used.
+ * </p>
+ * @param timestamp {@link #HYBRID} Timestamp
+ * @return true if the timestamp is likely to be of the corresponding {@link TimestampType}
+ * else false
+ */
+ public boolean isLikelyOfType(long timestamp) {
+ final long physicalTime = getPhysicalTime(timestamp);
+ final long logicalTime = getLogicalTime(timestamp);
+
+ // heuristic 1: Up until year 2016 (1451635200000), lt component cannot be non-zero.
+ if (physicalTime < 1451635200000L && logicalTime != 0) {
+ return false;
+ } else if (physicalTime < 31536000000L) {
+ // heuristic 2: Even if logical time = 0, physical time after left shifting by 20 bits,
+ // will be before year 1971(31536000000L), as after left shifting by 20, all epoch ms
+ // timestamps from wall time end up in year less than 1971, even for epoch time for the
+ // year 10000. This assumes Hybrid time is not used to represent timestamps for year 1970
+ // UTC.
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Returns a string representation for Physical Time and Logical Time components. The format is:
+ * <code>yyyy-MM-dd HH:mm:ss:SSS(Physical Time),Logical Time</code>
+ * Physical Time is converted to UTC time and not to local time for uniformity.
+ * Example: 2015-07-17 16:56:35:891(1437177395891), 0
+ * @param timestamp A {@link #HYBRID} Timestamp
+ * @return A date time string formatted as mentioned in the method description
+ */
+ public String toString(long timestamp) {
+ long physicalTime = getPhysicalTime(timestamp);
+ long logicalTime = getLogicalTime(timestamp);
+ return new StringBuilder().append(dateFormat.format(physicalTime)).append("(")
+ .append(physicalTime).append(")").append(", ").append(logicalTime).toString();
+ }
+ },
+
+ /**
+ * Physical is a Timestamp type used to encode the physical time in 64 bits.
+ * It has helper methods to decipher the 64 bit encoding of physical time.
+ */
+ PHYSICAL {
+ public long toEpochTimeMillisFromTimestamp(long timestamp) {
+ return timestamp;
+ }
+
+ public long fromEpochTimeMillisToTimestamp(long timestamp) {
+ return timestamp;
+ }
+
+ public long toTimestamp(TimeUnit timeUnit, long physicalTime, long logicalTime) {
+ return TimeUnit.MILLISECONDS.convert(physicalTime, timeUnit);
+ }
+
+ public long getPhysicalTime(long timestamp) {
+ return timestamp;
+ }
+
+ long getLogicalTime(long timestamp) {
+ return 0;
+ }
+
+ public long getMaxPhysicalTime() {
+ return Long.MAX_VALUE;
+ }
+
+ public long getMaxLogicalTime() {
+ return 0;
+ }
+
+ int getBitsForLogicalTime() {
+ return 0;
+ }
+
+ public boolean isLikelyOfType(long timestamp) {
+ // heuristic: the timestamp should be up to year 3K (32503680000000L).
+ return timestamp < 32503680000000L;
+ }
+
+ /**
+ * Returns a string representation for Physical Time and Logical Time components. The format is:
+ * <code>yyyy-MM-dd HH:mm:ss:SSS(Physical Time)</code>
+ * Physical Time is converted to UTC time and not to local time for uniformity.
+ * Example: 2015-07-17 16:56:35:891(1437177395891), 0
+ * @param timestamp epoch time in milliseconds
+ * @return A date time string formatted as mentioned in the method description
+ */
+ public String toString(long timestamp) {
+ return new StringBuilder().append(dateFormat.format(timestamp)).append("(")
+ .append(timestamp).append(")").append(", ").append("0").toString();
+ }
+ };
+
+ /**
+ * This is used internally by the enum methods of Hybrid and Physical Timestamp types to
+ * convert the
+ * timestamp to the format set here. UTC timezone instead of local time zone for convenience
+ * and uniformity
+ */
+ private static final FastDateFormat dateFormat =
+ FastDateFormat.getInstance("yyyy-MM-dd'T'HH:mm:ss:SSS", TimeZone.getTimeZone("UTC"));
+
+ /**
+ * Converts the given timestamp to the unix epoch timestamp with millisecond resolution.
+ * Returned timestamp is compatible with System.currentTimeMillis().
+ * @param timestamp {@link #HYBRID} or {@link #PHYSICAL} Timestamp
+ * @return number of milliseconds from epoch
+ */
+ abstract public long toEpochTimeMillisFromTimestamp(long timestamp);
+
+ /**
+ * Converts the given time in milliseconds to the corresponding {@link TimestampType}
+ * representation.
+ * @param timeInMillis epoch time in {@link TimeUnit#MILLISECONDS}
+ * @return a timestamp representation corresponding to {@link TimestampType}.
+ */
+ abstract public long fromEpochTimeMillisToTimestamp(long timeInMillis);
+
+ /**
+ * Converts the given physical clock in the given {@link TimeUnit} to a 64-bit timestamp
+ * @param timeUnit a time unit as in the enum {@link TimeUnit}
+ * @param physicalTime physical time
+ * @param logicalTime logical time
+ * @return a timestamp in 64 bits
+ */
+ abstract public long toTimestamp(TimeUnit timeUnit, long physicalTime, long logicalTime);
+
+ /**
+ * Extracts and returns the physical time from the timestamp
+ * @param timestamp {@link #HYBRID} or {@link #PHYSICAL} Timestamp
+ * @return physical time in {@link TimeUnit#MILLISECONDS}
+ */
+ abstract public long getPhysicalTime(long timestamp);
+
+ /**
+ * Extracts and returns the logical time from the timestamp
+ * @param timestamp {@link #HYBRID} or {@link #PHYSICAL} Timestamp
+ * @return logical time
+ */
+ abstract long getLogicalTime(long timestamp);
+
+ /**
+ * @return the maximum possible physical time in {@link TimeUnit#MILLISECONDS}
+ */
+ abstract public long getMaxPhysicalTime();
+
+ /**
+ * @return the maximum possible logical time
+ */
+ abstract public long getMaxLogicalTime();
+
+ /**
+ * @return number of least significant bits allocated for logical time
+ */
+ abstract int getBitsForLogicalTime();
+
+ /**
+ * @param timestamp epoch time in milliseconds
+ * @return True if the timestamp generated by the clock is of type {@link #PHYSICAL} else False
+ */
+ abstract public boolean isLikelyOfType(long timestamp);
+
+ public abstract String toString(long timestamp);
+
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-common/src/test/java/org/apache/hadoop/hbase/TestClock.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/TestClock.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/TestClock.java
new file mode 100644
index 0000000..2476c51
--- /dev/null
+++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/TestClock.java
@@ -0,0 +1,293 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.junit.rules.ExpectedException;
+
+import java.util.ArrayList;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+@Category(SmallTests.class)
+public class TestClock {
+
+ static final Clock.PhysicalClock PHYSICAL_CLOCK = mock(Clock.PhysicalClock.class);
+ static final long MAX_CLOCK_SKEW_IN_MS = 1000;
+
+ @Rule
+ public final ExpectedException exception = ExpectedException.none();
+
+ /**
+ * Wrapper class around any Clock.
+ * On calls to now() and update(), timestamps returned by the underlying clock are stored in an
+ * array. Call assertMonotonic() to assert that the timestamps returned were monotonic or not.
+ */
+ private class MonotonicityCheckerClock {
+ final boolean strictlyIncreasing;
+ final Clock clock;
+ final ArrayList<Long> timestamps = new ArrayList<>();
+
+ MonotonicityCheckerClock(Clock clock, boolean strictlyIncreasing) {
+ this.clock = clock;
+ this.strictlyIncreasing = strictlyIncreasing;
+ }
+
+ long now() {
+ long ts = clock.now();
+ timestamps.add(ts);
+ return ts;
+ }
+
+ long update(long timestamp) {
+ long ts = clock.update(timestamp);
+ timestamps.add(ts);
+ return ts;
+ }
+
+ void assertMonotonic() {
+ assertTrue(timestamps.size() > 0);
+ long prev = 0;
+ for (long timestamp : timestamps) {
+ if (strictlyIncreasing) {
+ // This simple comparison works correctly for all types of clocks we have currently.
+ assertTrue(timestamps.toString(), timestamp > prev);
+ } else {
+ assertTrue(timestamps.toString(), timestamp >= prev);
+ }
+ prev = timestamp;
+ }
+ }
+ }
+
+ @Before
+ public void setUp() {
+ when(PHYSICAL_CLOCK.getTimeUnit()).thenReturn(TimeUnit.MILLISECONDS);
+ }
+
+ // All Clocks Tests
+
+ @Test
+ public void testSystemMonotonicNow() {
+ MonotonicityCheckerClock systemMonotonic =
+ new MonotonicityCheckerClock(new Clock.SystemMonotonic(PHYSICAL_CLOCK, 30000), false);
+
+ // case 1: Set time and assert
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ assertEquals(100, systemMonotonic.now());
+
+ // case 2: Go back in time and check monotonic property.
+ when(PHYSICAL_CLOCK.now()).thenReturn(99L);
+ assertEquals(100, systemMonotonic.now());
+
+ // case 3: system time goes ahead compared to previous timestamp.
+ when(PHYSICAL_CLOCK.now()).thenReturn(101L);
+ assertEquals(101, systemMonotonic.now());
+
+ systemMonotonic.assertMonotonic();
+ }
+
+ @Test
+ public void testSystemMonotonicUpdate() {
+ MonotonicityCheckerClock systemMonotonic =
+ new MonotonicityCheckerClock(new Clock.SystemMonotonic(PHYSICAL_CLOCK, 30000), false);
+
+ // Sets internal time
+ when(PHYSICAL_CLOCK.now()).thenReturn(99L);
+ assertEquals(99, systemMonotonic.now());
+
+ // case 1: Message timestamp is greater than current System Monotonic Time,
+ // physical time at 100 still.
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ assertEquals(102, systemMonotonic.update(102));
+
+ // case 2: Message timestamp is greater than current System Monotonic Time,
+ // physical time at 100 still.
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ assertEquals(103, systemMonotonic.update(103));
+
+ // case 3: Message timestamp is less than current System Monotonic Time, greater than current
+ // physical time which is 100.
+ assertEquals(103, systemMonotonic.update(101));
+
+ // case 4: Message timestamp is less than current System Monotonic Time, less than current
+ // physical time which is 100.
+ assertEquals(103, systemMonotonic.update(99));
+
+ // case 5: Message timestamp<System monotonic time and both less than current Physical Time
+ when(PHYSICAL_CLOCK.now()).thenReturn(106L);
+ assertEquals(106, systemMonotonic.update(102));
+
+ // case 6: Message timestamp>System monotonic time and both less than current Physical Time
+ when(PHYSICAL_CLOCK.now()).thenReturn(109L);
+ assertEquals(109, systemMonotonic.update(108));
+
+ systemMonotonic.assertMonotonic();
+ }
+
+ @Test
+ public void testSystemMonotonicUpdateMaxClockSkew() throws Clock.ClockException {
+ final long time = 100L;
+ Clock.SystemMonotonic systemMonotonic =
+ new Clock.SystemMonotonic(PHYSICAL_CLOCK, MAX_CLOCK_SKEW_IN_MS);
+
+ // Set Current Time.
+ when(PHYSICAL_CLOCK.now()).thenReturn(time);
+ systemMonotonic.now();
+
+ // Shouldn't throw ClockException
+ systemMonotonic.update(time + MAX_CLOCK_SKEW_IN_MS - 1);
+
+ exception.expect(Clock.ClockException.class);
+ systemMonotonic.update(time + MAX_CLOCK_SKEW_IN_MS + 1);
+ }
+
+ // All Hybrid Logical Clock Tests
+
+ private void assertHLCTime(long hlcTime, long expectedPhysicalTime, long expectedLogicalTime) {
+ assertEquals(expectedPhysicalTime, TimestampType.HYBRID.getPhysicalTime(hlcTime));
+ assertEquals(expectedLogicalTime, TimestampType.HYBRID.getLogicalTime(hlcTime));
+ }
+
+ @Test
+ public void testHLCNow() throws Clock.ClockException {
+ MonotonicityCheckerClock hybridLogicalClock =
+ new MonotonicityCheckerClock(new Clock.HLC(PHYSICAL_CLOCK, 30000), true);
+
+ // case 1: Test if it returns correct time based on current physical time.
+ // Remember, initially logical time = 0
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ assertHLCTime(hybridLogicalClock.now(), 100, 0);
+
+ // case 2: physical time doesn't change, logical time should increment.
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ assertHLCTime(hybridLogicalClock.now(), 100, 1);
+
+ // case 3: physical time doesn't change still, logical time should increment again
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ assertHLCTime(hybridLogicalClock.now(), 100, 2);
+
+ // case 4: physical time moves forward, logical time should reset to 0.
+ when(PHYSICAL_CLOCK.now()).thenReturn(101L);
+ assertHLCTime(hybridLogicalClock.now(), 101, 0);
+
+ // case 5: Monotonic increasing check, physical time goes back.
+ when(PHYSICAL_CLOCK.now()).thenReturn(99L);
+ assertHLCTime(hybridLogicalClock.now(), 101, 1);
+
+ hybridLogicalClock.assertMonotonic();
+ }
+
+ @Test
+ public void testHLCNowLogicalTimeOverFlow() {
+ Clock.HLC hybridLogicalClock = new Clock.HLC(PHYSICAL_CLOCK, 100);
+
+ // Set Current Time.
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ hybridLogicalClock.setPhysicalTime(100);
+ hybridLogicalClock.setLogicalTime(TimestampType.HYBRID.getMaxLogicalTime());
+
+ exception.expect(Clock.ClockException.class);
+ hybridLogicalClock.now();
+ }
+
+ @Test
+ public void testHLCUpdate() throws Clock.ClockException {
+ long messageTimestamp;
+ MonotonicityCheckerClock hybridLogicalClock =
+ new MonotonicityCheckerClock(new Clock.HLC(PHYSICAL_CLOCK, 100), true);
+
+ // Set Current Time.
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ hybridLogicalClock.now();
+
+ // case 1: Message physical timestamp is lower than current physical time.
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS, 99, 1);
+ when(PHYSICAL_CLOCK.now()).thenReturn(101L);
+ assertHLCTime(hybridLogicalClock.update(messageTimestamp), 101, 0);
+
+ // case 2: Message physical timestamp is greater than HLC physical time.
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS, 105 , 3);
+ when(PHYSICAL_CLOCK.now()).thenReturn(102L);
+ assertHLCTime(hybridLogicalClock.update(messageTimestamp), 105, 4);
+
+ // case 3: Message timestamp is less than HLC timestamp
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS, 104 , 4);
+ when(PHYSICAL_CLOCK.now()).thenReturn(103L);
+ assertHLCTime(hybridLogicalClock.update(messageTimestamp), 105, 5);
+
+ //case 4: Message timestamp with same physical time as HLC, but lower logical time
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS, 105 , 2);
+ when(PHYSICAL_CLOCK.now()).thenReturn(101L);
+ assertHLCTime(hybridLogicalClock.update(messageTimestamp), 105, 6);
+
+ //case 5: Message timestamp with same physical time as HLC, but higher logical time
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS, 105 , 8);
+ when(PHYSICAL_CLOCK.now()).thenReturn(102L);
+ assertHLCTime(hybridLogicalClock.update(messageTimestamp), 105, 9);
+
+ //case 6: Actual Physical Time greater than message physical timestamp and HLC physical time.
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS, 105 , 10);
+ when(PHYSICAL_CLOCK.now()).thenReturn(110L);
+ assertHLCTime(hybridLogicalClock.update(messageTimestamp), 110, 0);
+
+ hybridLogicalClock.assertMonotonic();
+ }
+
+ @Test
+ public void testHLCUpdateLogicalTimeOverFlow() throws Clock.ClockException {
+ long messageTimestamp;
+ Clock.HLC hybridLogicalClock = new Clock.HLC(PHYSICAL_CLOCK, 100);
+
+ // Set Current Time.
+ when(PHYSICAL_CLOCK.now()).thenReturn(100L);
+ hybridLogicalClock.now();
+
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS, 100,
+ TimestampType.HYBRID.getMaxLogicalTime());
+ exception.expect(Clock.ClockException.class);
+ hybridLogicalClock.update(messageTimestamp);
+ }
+
+ @Test
+ public void testHLCUpdateMaxClockSkew() throws Clock.ClockException {
+ final long time = 100;
+ long messageTimestamp;
+ Clock.HLC hybridLogicalClock = new Clock.HLC(PHYSICAL_CLOCK, MAX_CLOCK_SKEW_IN_MS);
+
+ // Set Current Time.
+ when(PHYSICAL_CLOCK.now()).thenReturn(time);
+ hybridLogicalClock.now();
+
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS,
+ time + MAX_CLOCK_SKEW_IN_MS - 1, 0);
+ hybridLogicalClock.update(messageTimestamp);
+
+ messageTimestamp = TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS,
+ time + MAX_CLOCK_SKEW_IN_MS + 1, 0);
+ exception.expect(Clock.ClockException.class);
+ hybridLogicalClock.update(messageTimestamp);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-common/src/test/java/org/apache/hadoop/hbase/TestTimestampType.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/test/java/org/apache/hadoop/hbase/TestTimestampType.java b/hbase-common/src/test/java/org/apache/hadoop/hbase/TestTimestampType.java
new file mode 100644
index 0000000..b1a661b
--- /dev/null
+++ b/hbase-common/src/test/java/org/apache/hadoop/hbase/TestTimestampType.java
@@ -0,0 +1,207 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.text.ParseException;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.*;
+
+/**
+ * Tests for TimestampType enum.
+ */
+@Category(SmallTests.class)
+public class TestTimestampType {
+ private static final Log LOG = LogFactory.getLog(TestTimestampType.class);
+
+ private static final long PHYSICAL_TIME = 1234567890123L;
+ private static final long LOGICAL_TIME = 12;
+
+ private static final long HLC_TIME = TimestampType.HYBRID.toTimestamp(
+ TimeUnit.MILLISECONDS, PHYSICAL_TIME, LOGICAL_TIME);
+ private static final long PHYSICAL_CLOCK_TIME = TimestampType.PHYSICAL.toTimestamp(
+ TimeUnit.MILLISECONDS, PHYSICAL_TIME, LOGICAL_TIME);
+
+ @Test
+ public void testFromToEpoch() {
+ for (TimestampType timestamp : TimestampType.values()) {
+ long wallTime = System.currentTimeMillis();
+ long converted = timestamp.toEpochTimeMillisFromTimestamp(
+ timestamp.fromEpochTimeMillisToTimestamp(wallTime));
+
+ assertEquals(wallTime, converted);
+ }
+ }
+
+ /* Tests for HLC Clock */
+ @Test
+ public void testHybridMaxValues() {
+ // assert 44-bit Physical Time with signed comparison (actual 43 bits)
+ assertEquals(
+ (1L << (63-TimestampType.HYBRID.getBitsForLogicalTime())) - 1,
+ TimestampType.HYBRID.getMaxPhysicalTime());
+
+ // assert 20-bit Logical Time
+ assertEquals(
+ (1L << TimestampType.HYBRID.getBitsForLogicalTime()) - 1,
+ TimestampType.HYBRID.getMaxLogicalTime());
+
+ // assert that maximum representable timestamp is Long.MAX_VALUE (assuming signed comparison).
+ assertEquals(
+ Long.MAX_VALUE,
+ TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS,
+ TimestampType.HYBRID.getMaxPhysicalTime(),
+ TimestampType.HYBRID.getMaxLogicalTime())
+ );
+ }
+
+ @Test
+ public void testHybridTimeComponents() {
+ assertEquals(PHYSICAL_TIME, TimestampType.HYBRID.getPhysicalTime(HLC_TIME));
+ assertEquals(LOGICAL_TIME, TimestampType.HYBRID.getLogicalTime(HLC_TIME));
+ }
+
+ @Test
+ public void testHybridToString() {
+ assertEquals("2009-02-13T23:31:30:123(1234567890123), 12",
+ TimestampType.HYBRID.toString(HLC_TIME));
+ }
+
+ @Test
+ public void testHybridToTimestamp() {
+ long expected = (PHYSICAL_TIME << TimestampType.HYBRID.getBitsForLogicalTime()) + LOGICAL_TIME;
+ // test millisecond
+ assertEquals(HLC_TIME, expected);
+
+ // test nanosecond
+ long ts = TimestampType.HYBRID.toTimestamp(TimeUnit.NANOSECONDS,
+ TimeUnit.MILLISECONDS.toNanos(PHYSICAL_TIME), LOGICAL_TIME);
+ assertEquals(ts, expected);
+ }
+
+ @Test
+ public void testHybridIsLikelyOfType() throws ParseException {
+ ZonedDateTime date = LocalDateTime.of(1970, 1, 1, 1, 1).atZone(ZoneId.of("UTC"));
+ // test timestamps of Hybrid type from year 1971 to 2248 where lt = 0
+ for (int year = 1971; year <= 2248; year += 1) {
+ date = date.withYear(year);
+ // Hybrid type ts with pt = date and lt = 0
+ long ts = TimestampType.HYBRID.toTimestamp(TimeUnit.SECONDS, date.toEpochSecond(), 0);
+ assertTrue("Year = " + year, TimestampType.HYBRID.isLikelyOfType(ts));
+ }
+
+ // test timestamps of Hybrid type from year 2016 to 2348 where lt > 0
+ for (int year = 2016; year <= 2248; year += 1) {
+ date = date.withYear(year);
+
+ // Hybrid type ts with pt = date and lt = 123
+ long ts = TimestampType.HYBRID.toTimestamp(TimeUnit.SECONDS, date.toEpochSecond(), 123);
+ assertTrue("Year = " + year, TimestampType.HYBRID.isLikelyOfType(ts));
+ }
+
+ // test that timestamps from different years are not Hybrid type
+ for (int year = 1970; year <= 10000 ;year += 10) {
+ date = date.withYear(year);
+ final long ts = date.toEpochSecond() * 1000;
+ assertFalse("Year = " + year, TimestampType.HYBRID.isLikelyOfType(ts));
+ }
+
+ // test that timestamps up to 2016 are not Hybrid even if lt = 0
+ for (int year = 1970; year <= 2016; year += 1) {
+ date = date.withYear(year);
+
+ // reset lt = 0
+ long ts = (((date.toEpochSecond() * 1000)
+ >> TimestampType.HYBRID.getBitsForLogicalTime())
+ << TimestampType.HYBRID.getBitsForLogicalTime());
+ assertFalse("Year = " + year, TimestampType.HYBRID.isLikelyOfType(ts));
+ }
+
+ // test that timestamps from currentTime epoch are not Hybrid type
+ long systemTimeNow = System.currentTimeMillis();
+ LOG.info(TimestampType.PHYSICAL.toString(systemTimeNow));
+ LOG.info(TimestampType.PHYSICAL.toString((TimestampType.HYBRID.getPhysicalTime(systemTimeNow))));
+ assertFalse(TimestampType.HYBRID.isLikelyOfType(systemTimeNow));
+ }
+
+
+ @Test
+ public void testPhysicalMaxValues() {
+ assertEquals( (1L << 63)- 1, TimestampType.PHYSICAL.getMaxPhysicalTime());
+
+ assertEquals(0, TimestampType.PHYSICAL.getMaxLogicalTime());
+ }
+
+ @Test
+ public void testPhysicalClockPhysicalAndLogicalTime() {
+ assertEquals(PHYSICAL_TIME, TimestampType.PHYSICAL.getPhysicalTime(PHYSICAL_CLOCK_TIME));
+ assertEquals(0, TimestampType.PHYSICAL.getLogicalTime(PHYSICAL_CLOCK_TIME));
+ }
+
+ @Test
+ public void testPhysicalToString() {
+ assertEquals("2009-02-13T23:31:30:123(1234567890123), 0",
+ TimestampType.PHYSICAL.toString(PHYSICAL_CLOCK_TIME));
+ }
+
+ @Test
+ public void testPhysicalToTimestamp() {
+ // test millisecond
+ assertEquals(PHYSICAL_CLOCK_TIME, PHYSICAL_TIME);
+
+ // test nanosecond
+ long ts = TimestampType.PHYSICAL.toTimestamp(TimeUnit.NANOSECONDS,
+ TimeUnit.MILLISECONDS.toNanos(PHYSICAL_TIME), LOGICAL_TIME);
+ assertEquals(ts, PHYSICAL_TIME);
+ }
+
+ @Test
+ public void testPhysicalIsLikelyOfType() throws ParseException {
+ final ZonedDateTime date = LocalDateTime.of(1970, 1, 1, 1, 1).atZone(ZoneId.of("UTC"));
+
+ // test that timestamps from 1970 to 3K epoch are of Physical type
+ for (int year = 1970; year < 3000 ;year += 10) {
+ // Start date 1970 to 10000
+ date.withYear(year);
+ final long ts = date.toEpochSecond() * 1000;
+ assertTrue("Year = " + year, TimestampType.PHYSICAL.isLikelyOfType(ts));
+ }
+
+ // test that timestamps from currentTime epoch are of Physical type
+ long systemTimeNow = System.currentTimeMillis();
+ assertTrue(TimestampType.PHYSICAL.isLikelyOfType(systemTimeNow));
+
+ // test timestamps of Hybrid type from year 1970 to 2248 are not of Physical type
+ for (int year = 1970; year <= 2248; year += 1) {
+ date.withYear(year);
+ // Hybrid type ts with pt = date and lt = 0
+ long ts = TimestampType.HYBRID.toTimestamp(TimeUnit.SECONDS, date.toEpochSecond(), 0);
+ assertFalse(TimestampType.PHYSICAL.isLikelyOfType(ts));
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyTableProcedure.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyTableProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyTableProcedure.java
index 20a6a03..6d42c06 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyTableProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/ModifyTableProcedure.java
@@ -256,6 +256,10 @@ public class ModifyTableProcedure
throw new IOException("REGION_REPLICATION change is not supported for enabled tables");
}
}
+ // do not allow changing of clock type.
+ if (modifiedHTableDescriptor.getClockType() != unmodifiedHTableDescriptor.getClockType()) {
+ throw new IOException("Clock Type change is not supported for tables");
+ }
// Find out whether all column families in unmodifiedHTableDescriptor also exists in
// the modifiedHTableDescriptor. This is to determine whether we are safe to rollback.
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index b02b042..a55be97 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -92,6 +92,8 @@ import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HConstants.OperationStatusCode;
import org.apache.hadoop.hbase.HDFSBlocksDistribution;
import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.ClockType;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.NamespaceDescriptor;
@@ -100,6 +102,8 @@ import org.apache.hadoop.hbase.RegionTooBusyException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.Tag;
import org.apache.hadoop.hbase.TagUtil;
+import org.apache.hadoop.hbase.Clock;
+import org.apache.hadoop.hbase.TimestampType;
import org.apache.hadoop.hbase.UnknownScannerException;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.client.Append;
@@ -380,6 +384,24 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
return minimumReadPoint;
}
+ @Override
+ public Clock getClock() {
+ if (this.clock == null) {
+ return this.getRegionServerServices().getRegionServerClock(
+ getTableDescriptor().getClockType());
+ }
+ return this.clock;
+ }
+
+ /**
+ * Only for the purpose of testing
+ * @param clock
+ */
+ @VisibleForTesting
+ public void setClock(Clock clock) {
+ this.clock = clock;
+ }
+
/*
* Data structure of write state flags used coordinating flushes,
* compactions and closes.
@@ -617,6 +639,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
private final ConcurrentMap<Store, Long> lastStoreFlushTimeMap = new ConcurrentHashMap<>();
final RegionServerServices rsServices;
+ private Clock clock;
private RegionServerAccounting rsAccounting;
private long flushCheckInterval;
// flushPerChanges is to prevent too many changes in memstore
@@ -775,6 +798,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
? DEFAULT_DURABILITY
: htd.getDurability();
if (rsServices != null) {
+ this.clock = rsServices.getRegionServerClock(htd.getClockType());
this.rsAccounting = this.rsServices.getRegionServerAccounting();
// don't initialize coprocessors if not running within a regionserver
// TODO: revisit if coprocessors should load in other cases
@@ -789,6 +813,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
recoveringRegions.put(encodedName, this);
}
} else {
+ this.clock = new Clock.System();
this.metricsRegionWrapper = null;
this.metricsRegion = null;
}
@@ -2796,8 +2821,31 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
return getScanner(scan, additionalScanners, HConstants.NO_NONCE, HConstants.NO_NONCE);
}
+ /**
+ * Clients use physical timestamps when setting time ranges. Tables that use HLCs must map the
+ * physical timestamp to HLC time
+ */
+ private void mapTimeRangesWithRespectToClock(Scan scan) {
+ TimeRange tr = scan.getTimeRange();
+ if (tr.isAllTime()) {
+ return;
+ }
+ TimestampType timestampType = getClock().getTimestampType();
+ // Clip time range max to prevent overflow when converting from epoch time to timestamp time
+ long trMaxClipped = Math.min(tr.getMax(), timestampType.getMaxPhysicalTime());
+ try {
+ scan.setTimeRange(timestampType.fromEpochTimeMillisToTimestamp(tr.getMin()),
+ timestampType.fromEpochTimeMillisToTimestamp(trMaxClipped));
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
private RegionScanner getScanner(Scan scan, List<KeyValueScanner> additionalScanners,
long nonceGroup, long nonce) throws IOException {
+ if (getClock().getClockType() == ClockType.HLC) {
+ mapTimeRangesWithRespectToClock(scan);
+ }
startRegionOperation(Operation.SCAN);
try {
// Verify families are all valid
@@ -3219,7 +3267,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
try {
// STEP 1. Try to acquire as many locks as we can, and ensure we acquire at least one.
int numReadyToWrite = 0;
- long now = EnvironmentEdgeManager.currentTime();
+ long now = clock.now();
while (lastIndexExclusive < batchOp.operations.length) {
if (checkBatchOp(batchOp, lastIndexExclusive, familyMaps, now, observedExceptions)) {
lastIndexExclusive++;
@@ -3257,7 +3305,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
// STEP 2. Update any LATEST_TIMESTAMP timestamps
// We should record the timestamp only after we have acquired the rowLock,
// otherwise, newer puts/deletes are not guaranteed to have a newer timestamp
- now = EnvironmentEdgeManager.currentTime();
+ now = clock.now();
byte[] byteNow = Bytes.toBytes(now);
// Nothing to put/delete -- an exception in the above such as NoSuchColumnFamily?
@@ -3756,8 +3804,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
// non-decreasing (see HBASE-14070) we should make sure that the mutation has a
// larger timestamp than what was observed via Get. doBatchMutate already does this, but
// there is no way to pass the cellTs. See HBASE-14054.
- long now = EnvironmentEdgeManager.currentTime();
- long ts = Math.max(now, cellTs); // ensure write is not eclipsed
+ long now = clock.now();
+ long ts = clock.isMonotonic() ? now : Math.max(now, cellTs); // ensure write is not eclipsed
byte[] byteTs = Bytes.toBytes(ts);
if (mutation != null) {
if (mutation instanceof Put) {
@@ -4038,7 +4086,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
if (timestampSlop == HConstants.LATEST_TIMESTAMP) {
return;
}
- long maxTs = now + timestampSlop;
+ long maxTs = clock.getTimestampType().getPhysicalTime(now) + timestampSlop;
for (List<Cell> kvs : familyMap.values()) {
assert kvs instanceof RandomAccess;
int listSize = kvs.size();
@@ -7101,7 +7149,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
// Short circuit the read only case
if (processor.readOnly()) {
try {
- long now = EnvironmentEdgeManager.currentTime();
+ long now = clock.now();
doProcessRowWithTimeout(processor, now, this, null, null, timeout);
processor.postProcess(this, walEdit, true);
} finally {
@@ -7131,7 +7179,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
// STEP 3. Region lock
lock(this.updatesLock.readLock(), acquiredRowLocks.isEmpty() ? 1 : acquiredRowLocks.size());
locked = true;
- long now = EnvironmentEdgeManager.currentTime();
+ long now = clock.now();
// STEP 4. Let the processor scan the rows, generate mutations and add waledits
doProcessRowWithTimeout(processor, now, this, mutations, walEdit, timeout);
if (!mutations.isEmpty()) {
@@ -7440,7 +7488,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
final List<Cell> results)
throws IOException {
WALEdit walEdit = null;
- long now = EnvironmentEdgeManager.currentTime();
+ long now = clock.now();
final boolean writeToWAL = effectiveDurability != Durability.SKIP_WAL;
// Process a Store/family at a time.
for (Map.Entry<byte [], List<Cell>> entry: mutation.getFamilyCellMap().entrySet()) {
@@ -7556,7 +7604,11 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
long ts = now;
if (currentValue != null) {
tags = TagUtil.carryForwardTags(tags, currentValue);
- ts = Math.max(now, currentValue.getTimestamp() + 1);
+ // Ensure that timestamp of increment operation is greater than existing cell timestamp
+ if (this.getClock().getClockType() == ClockType.SYSTEM ||
+ this.getClock().getClockType() == ClockType.SYSTEM_MONOTONIC) {
+ ts = Math.max(now, currentValue.getTimestamp() + 1);
+ }
newValue += getLongValue(currentValue);
}
// Now make up the new Cell. TODO: FIX. This is carnel knowledge of how KeyValues are made...
@@ -7582,7 +7634,11 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
byte [] row = mutation.getRow();
if (currentValue != null) {
tags = TagUtil.carryForwardTags(tags, currentValue);
- ts = Math.max(now, currentValue.getTimestamp() + 1);
+ // Ensure that timestamp of increment operation is greater than existing cell timestamp
+ if (this.getClock().getClockType() == ClockType.SYSTEM ||
+ this.getClock().getClockType() == ClockType.SYSTEM_MONOTONIC) {
+ ts = Math.max(now, currentValue.getTimestamp() + 1);
+ }
tags = TagUtil.carryForwardTTLTag(tags, mutation.getTTL());
byte[] tagBytes = TagUtil.fromList(tags);
// Allocate an empty cell and copy in all parts.
@@ -7685,7 +7741,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
public static final long FIXED_OVERHEAD = ClassSize.align(
ClassSize.OBJECT +
ClassSize.ARRAY +
- 49 * ClassSize.REFERENCE + 2 * Bytes.SIZEOF_INT +
+ 50 * ClassSize.REFERENCE + 2 * Bytes.SIZEOF_INT +
(15 * Bytes.SIZEOF_LONG) +
6 * Bytes.SIZEOF_BOOLEAN);
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
index 986d6d4..dc7c70e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
@@ -75,6 +75,8 @@ import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HBaseInterfaceAudience;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.Clock;
+import org.apache.hadoop.hbase.ClockType;
import org.apache.hadoop.hbase.HealthCheckChore;
import org.apache.hadoop.hbase.MetaTableAccessor;
import org.apache.hadoop.hbase.NotServingRegionException;
@@ -331,6 +333,13 @@ public class HRegionServer extends HasThread implements
// debugging and unit tests.
private volatile boolean abortRequested;
+ // Region server contains instances of all three clock clocks. Regions have a set
+ // clock type so depending on the clock type needed by a region, the appropriate
+ // one can be accessed.
+ final protected Clock hybridLogicalClock;
+ final protected Clock systemMonotonicClock;
+ final protected Clock systemClock;
+
ConcurrentMap<String, Integer> rowlocks = new ConcurrentHashMap<>();
// A state before we go into stopped state. At this stage we're closing user
@@ -588,6 +597,10 @@ public class HRegionServer extends HasThread implements
this.abortRequested = false;
this.stopped = false;
+ this.hybridLogicalClock = new Clock.HLC();
+ this.systemMonotonicClock = new Clock.SystemMonotonic();
+ this.systemClock = new Clock.System();
+
rpcServices = createRpcServices();
this.startcode = System.currentTimeMillis();
if (this instanceof HMaster) {
@@ -2074,6 +2087,17 @@ public class HRegionServer extends HasThread implements
}
@Override
+ public Clock getRegionServerClock(ClockType clockType) {
+ if (clockType.equals(ClockType.HLC)){
+ return this.hybridLogicalClock;
+ } else if (clockType.equals(ClockType.SYSTEM_MONOTONIC)) {
+ return this.systemMonotonicClock;
+ } else {
+ return this.systemClock;
+ }
+ }
+
+ @Override
public Connection getConnection() {
return getClusterConnection();
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
index c914ab6..4b2b460 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
@@ -92,6 +92,8 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.StringUtils.TraditionalBinaryPrefix;
+import org.apache.hadoop.hbase.TimestampType;
+import org.apache.hadoop.hbase.Clock;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
@@ -341,10 +343,10 @@ public class HStore implements Store {
/**
* @param family
- * @return TTL in seconds of the specified family
+ * @return TTL in milli seconds of the specified family
*/
public static long determineTTLFromFamily(final ColumnFamilyDescriptor family) {
- // HCD.getTimeToLive returns ttl in seconds. Convert to milliseconds.
+ // HColumnDescriptor.getTimeToLive returns ttl in seconds. Convert to milliseconds.
long ttl = family.getTimeToLive();
if (ttl == HConstants.FOREVER) {
// Default is unlimited ttl.
@@ -403,6 +405,11 @@ public class HStore implements Store {
}
@Override
+ public Clock getClock() {
+ return region.getClock();
+ }
+
+ @Override
@Deprecated
public long getSnapshotSize() {
MemstoreSize size = getSizeOfSnapshot();
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java
index 4c188fe..ea50943 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java
@@ -24,6 +24,7 @@ import java.util.Map;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.Clock;
import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.HBaseInterfaceAudience;
import org.apache.hadoop.hbase.HDFSBlocksDistribution;
@@ -82,6 +83,11 @@ public interface Region extends ConfigurationObserver {
/** @return table descriptor for this region */
TableDescriptor getTableDescriptor();
+ /** @return clock of the Region Server corresponding the clock type used by the
+ * table contained in this region.
+ */
+ Clock getClock();
+
/** @return true if region is available (not closed and not closing) */
boolean isAvailable();
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java
index 5afa652..5c37136 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java
@@ -39,6 +39,8 @@ import org.apache.hadoop.hbase.quotas.RegionServerRpcQuotaManager;
import org.apache.hadoop.hbase.quotas.RegionServerSpaceQuotaManager;
import org.apache.hadoop.hbase.regionserver.throttle.ThroughputController;
import org.apache.hadoop.hbase.wal.WAL;
+import org.apache.hadoop.hbase.Clock;
+import org.apache.hadoop.hbase.ClockType;
import org.apache.zookeeper.KeeperException;
import com.google.protobuf.Service;
@@ -58,6 +60,8 @@ public interface RegionServerServices extends OnlineRegions, FavoredNodesForRegi
* default (common) WAL */
WAL getWAL(HRegionInfo regionInfo) throws IOException;
+ Clock getRegionServerClock(ClockType clockType);
+
/** @return the List of WALs that are used by this server
* Doesn't include the meta WAL
*/
http://git-wip-us.apache.org/repos/asf/hbase/blob/d53fbc74/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
index fd9de9b..82edc7a 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.hbase.CellComparator;
import org.apache.hadoop.hbase.HBaseInterfaceAudience;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.Clock;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.classification.InterfaceStability;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
@@ -346,6 +347,12 @@ public interface Store extends HeapSize, StoreConfigInformation, PropagatingConf
MemstoreSize getSizeToFlush();
/**
+ * @return clock of the Region Server corresponding the clock type used by the
+ * table referred to by this store.
+ */
+ Clock getClock();
+
+ /**
* Returns the memstore snapshot size
* @return size of the memstore snapshot
* @deprecated Since 2.0 and will be removed in 3.0. Use {@link #getSizeOfSnapshot()} instead.