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/08/14 19:48:08 UTC
hbase git commit: HBASE-18510 Update clock on replaying recovered
edits (Sai Teja Ranuva)
Repository: hbase
Updated Branches:
refs/heads/HBASE-14070.HLC 93c0d71c4 -> 58a51bcb7
HBASE-18510 Update clock on replaying recovered edits (Sai Teja Ranuva)
- Added clock updating to HRegion#replayRecoveredEdits
- Added TestHRegion#testHybridLogicalClockUpdatesOnRecoveryEditReplay and
TestHRegion#testSystemMonotonicClockUpdatesOnRecoveryEditReplay
- Added comments clarifying expected logical and physical time in tests
- Changed TimestampType#getLogicalTime to be public
Change-Id: Ieba31b70fa2bdd3fa8080b521958039cb288badc
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/58a51bcb
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/58a51bcb
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/58a51bcb
Branch: refs/heads/HBASE-14070.HLC
Commit: 58a51bcb72250c3173d418a6d3fffb0c1f0d446c
Parents: 93c0d71
Author: Amit Patel <ia...@gmail.com>
Authored: Thu Jul 27 11:45:13 2017 -0700
Committer: Apekshit Sharma <ap...@apache.org>
Committed: Mon Aug 14 12:40:43 2017 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hbase/TimestampType.java | 10 +-
.../hadoop/hbase/regionserver/HRegion.java | 1 +
.../hadoop/hbase/regionserver/TestHRegion.java | 108 +++++++++++++++++++
3 files changed, 116 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/58a51bcb/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
index 2f0a229..effbde2 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/TimestampType.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/TimestampType.java
@@ -16,6 +16,7 @@
package org.apache.hadoop.hbase;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.classification.InterfaceStability;
import org.apache.commons.lang.time.FastDateFormat;
@@ -105,7 +106,8 @@ public enum TimestampType {
return timestamp >>> BITS_FOR_LOGICAL_TIME; // assume unsigned timestamp
}
- long getLogicalTime(long timestamp) {
+ @VisibleForTesting
+ public long getLogicalTime(long timestamp) {
return timestamp & LOGICAL_TIME_MAX_VALUE;
}
@@ -200,7 +202,8 @@ public enum TimestampType {
return timestamp;
}
- long getLogicalTime(long timestamp) {
+ @VisibleForTesting
+ public long getLogicalTime(long timestamp) {
return 0;
}
@@ -281,7 +284,8 @@ public enum TimestampType {
* @param timestamp {@link #HYBRID} or {@link #PHYSICAL} Timestamp
* @return logical time
*/
- abstract long getLogicalTime(long timestamp);
+ @VisibleForTesting
+ abstract public long getLogicalTime(long timestamp);
/**
* @return the maximum possible physical time in {@link TimeUnit#MILLISECONDS}
http://git-wip-us.apache.org/repos/asf/hbase/blob/58a51bcb/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 2b82e4c..56f8bb6 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
@@ -4434,6 +4434,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
continue;
}
CellUtil.setSequenceId(cell, currentReplaySeqId);
+ clock.update(cell.getTimestamp());
restoreEdit(store, cell, memstoreSize);
editsCount++;
http://git-wip-us.apache.org/repos/asf/hbase/blob/58a51bcb/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
index 7350473..a7e590b 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
@@ -720,6 +720,114 @@ public class TestHRegion {
}
}
+ public void verifyClockUpdatesOnRecoveryEditReplay(Clock clock, List<Long> seqIds,
+ List<Long> editTimestamps, long recoverSeqId, long expectedTimestamp) throws Exception {
+ byte[] family = Bytes.toBytes("family");
+ region = initHRegion(tableName, method, CONF, family);
+ final WALFactory wals = new WALFactory(CONF, null, method);
+ region.setClock(clock);
+
+ try {
+ Path regiondir = region.getRegionFileSystem().getRegionDir();
+ FileSystem fs = region.getRegionFileSystem().getFileSystem();
+ byte[] regionName = region.getRegionInfo().getEncodedNameAsBytes();
+ Path recoveredEditsDir = WALSplitter.getRegionDirRecoveredEditsDir(regiondir);
+
+ for (int i = 0; i < seqIds.size(); i++) {
+ long editTimestamp = editTimestamps.get(i);
+ long seqId = seqIds.get(i);
+
+ Path recoveredEdits = new Path(recoveredEditsDir, String.format("%019d", seqId));
+ fs.create(recoveredEdits);
+ WALProvider.Writer writer = wals.createRecoveredEditsWriter(fs, recoveredEdits);
+
+ WALEdit edit = new WALEdit();
+ edit.add(new KeyValue(row, family, Bytes.toBytes(seqId), editTimestamp, KeyValue.Type.Put,
+ Bytes.toBytes(seqId)));
+ writer.append(new WAL.Entry(new WALKey(regionName, tableName, seqId, editTimestamp,
+ HConstants.DEFAULT_CLUSTER_ID), edit));
+
+ writer.close();
+ }
+ MonitoredTask status = TaskMonitor.get().createStatus(method);
+ Map<byte[], Long> maxSeqIdInStores = new TreeMap<>(Bytes.BYTES_COMPARATOR);
+ for (Store store : region.getStores()) {
+ maxSeqIdInStores.put(store.getColumnFamilyName().getBytes(), recoverSeqId - 1);
+ }
+
+ region.replayRecoveredEditsIfAny(regiondir, maxSeqIdInStores, null, status);
+
+ assertEquals(expectedTimestamp, clock.now());
+
+ } finally {
+ HBaseTestingUtility.closeRegionAndWAL(this.region);
+ this.region = null;
+ wals.close();
+ }
+ }
+
+ @Test
+ public void testHybridLogicalClockUpdatesOnRecoveryEditReplay() throws Exception {
+ long systemTime = Clock.SYSTEM_CLOCK.now();
+ Clock mockSystemClock = mock(Clock.class);
+ when(mockSystemClock.now()).thenReturn(systemTime);
+ when(mockSystemClock.getTimeUnit()).thenReturn(TimeUnit.MILLISECONDS);
+ when(mockSystemClock.isMonotonic()).thenReturn(true);
+ Clock hlClock = new Clock.HLC(new Clock.SystemMonotonic(mockSystemClock));
+
+ long maxSeqId = 1050;
+ long minSeqId = 1000;
+ List<Long> seqIds = new ArrayList<>();
+ List<Long> editTimestamps = new ArrayList<>();
+
+ // Add edits with timestamps that are less than or equal to systemTime
+ for (long i = minSeqId; i <= maxSeqId; i += 10) {
+ seqIds.add(i);
+ editTimestamps.add(TimestampType.HYBRID
+ .fromEpochTimeMillisToTimestamp(systemTime - (maxSeqId - i)));
+ }
+
+ // Recover edits at sequence id 1030, 1040, and 1050
+ //
+ // Clock is updated three times with each update incrementing the logical time by one
+ // (from 0, 1, then 2). Then memstore gets flushed and clock.now() is called when
+ // constructing the StoreScanner for determining which rows to flush, incrementing
+ // the logical time from 2 to 3. Logical time advances by one again from 3 to 4 when
+ // calling clock.now() to assert that it equals the expected timestamp
+ verifyClockUpdatesOnRecoveryEditReplay(hlClock, seqIds, editTimestamps,
+ 1030L, TimestampType.HYBRID.toTimestamp(TimeUnit.MILLISECONDS, systemTime, 4));
+ }
+
+ @Test
+ public void testSystemMonotonicClockUpdatesOnRecoveryEditReplay() throws Exception {
+ long systemTime = Clock.SYSTEM_CLOCK.now();
+ Clock mockSystemClock = mock(Clock.class);
+ when(mockSystemClock.now()).thenReturn(systemTime);
+ when(mockSystemClock.getTimeUnit()).thenReturn(TimeUnit.MILLISECONDS);
+ when(mockSystemClock.isMonotonic()).thenReturn(true);
+ Clock systemMonotonicClock = new Clock.SystemMonotonic(mockSystemClock);
+
+ long maxSeqId = 1050;
+ long minSeqId = 1000;
+ List<Long> seqIds = new ArrayList<>();
+ List<Long> editTimestamps = new ArrayList<>();
+ // All edits except the last have timestamps less than systemTime
+ // Last edit will have a timestamp 5 ms greater than systemTime
+ for (long i = minSeqId; i <= maxSeqId; i += 10) {
+ seqIds.add(i);
+ editTimestamps.add(TimestampType.PHYSICAL
+ .fromEpochTimeMillisToTimestamp(systemTime - (maxSeqId - i) + 5));
+ }
+
+ // Recover edits at sequence id 1030, 1040, and 1050
+ //
+ // The clock's physical time will remain the same for the first two edits since their
+ // timestamps are in the past. Last edit has a timestamp 5 ms ahead of systemTime so the
+ // clock will advance forward.
+ verifyClockUpdatesOnRecoveryEditReplay(systemMonotonicClock, seqIds, editTimestamps,
+ 1030L, systemTime + 5);
+ }
+
@Test
public void testSkipRecoveredEditsReplaySomeIgnored() throws Exception {
byte[] family = Bytes.toBytes("family");