You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2018/06/21 02:49:35 UTC
[02/50] [abbrv] hbase git commit: HBASE-20727 Persist
FlushedSequenceId to speed up WAL split after cluster restart
HBASE-20727 Persist FlushedSequenceId to speed up WAL split after cluster restart
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/b336da92
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/b336da92
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/b336da92
Branch: refs/heads/HBASE-19064
Commit: b336da925ac5c5ee3565112de4b808fe2eed08a2
Parents: 78da0e3
Author: Allan Yang <al...@apache.org>
Authored: Tue Jun 19 09:45:47 2018 +0800
Committer: Allan Yang <al...@163.com>
Committed: Tue Jun 19 09:45:47 2018 +0800
----------------------------------------------------------------------
.../src/main/protobuf/HBase.proto | 15 ++
.../org/apache/hadoop/hbase/master/HMaster.java | 8 +
.../hadoop/hbase/master/ServerManager.java | 213 ++++++++++++++++++-
.../apache/hadoop/hbase/master/TestMaster.java | 33 +++
4 files changed, 268 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/b336da92/hbase-protocol-shaded/src/main/protobuf/HBase.proto
----------------------------------------------------------------------
diff --git a/hbase-protocol-shaded/src/main/protobuf/HBase.proto b/hbase-protocol-shaded/src/main/protobuf/HBase.proto
index 29067f1..0af2ffd 100644
--- a/hbase-protocol-shaded/src/main/protobuf/HBase.proto
+++ b/hbase-protocol-shaded/src/main/protobuf/HBase.proto
@@ -252,4 +252,19 @@ message CacheEvictionStats {
optional int64 bytes_evicted = 2;
optional int64 max_cache_size = 3;
repeated RegionExceptionMessage exception = 4;
+}
+
+message FlushedStoreSequenceId {
+ required bytes family = 1;
+ required uint64 seqId = 2;
+}
+
+message FlushedRegionSequenceId {
+ required bytes regionEncodedName = 1;
+ required uint64 seqId = 2;
+ repeated FlushedStoreSequenceId stores = 3;
+}
+
+message FlushedSequenceId {
+ repeated FlushedRegionSequenceId regionSequenceId = 1;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/b336da92/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java
index 883bb4f..38aac50 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java
@@ -862,6 +862,13 @@ public class HMaster extends HRegionServer implements MasterServices {
status.setStatus("Initializing ZK system trackers");
initializeZKBasedSystemTrackers();
+ status.setStatus("Loading last flushed sequence id of regions");
+ try {
+ this.serverManager.loadLastFlushedSequenceIds();
+ } catch (IOException e) {
+ LOG.debug("Failed to load last flushed sequence id of regions"
+ + " from file system", e);
+ }
// Set ourselves as active Master now our claim has succeeded up in zk.
this.activeMaster = true;
@@ -946,6 +953,7 @@ public class HMaster extends HRegionServer implements MasterServices {
getChoreService().scheduleChore(normalizerChore);
this.catalogJanitorChore = new CatalogJanitor(this);
getChoreService().scheduleChore(catalogJanitorChore);
+ this.serverManager.startChore();
status.setStatus("Starting cluster schema service");
initClusterSchemaService();
http://git-wip-us.apache.org/repos/asf/hbase/blob/b336da92/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java
index c746502..cfbd52f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java
@@ -38,10 +38,15 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.ClockOutOfSyncException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.RegionMetrics;
+import org.apache.hadoop.hbase.ScheduledChore;
import org.apache.hadoop.hbase.ServerMetrics;
import org.apache.hadoop.hbase.ServerMetricsBuilder;
import org.apache.hadoop.hbase.ServerName;
@@ -51,9 +56,11 @@ import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RetriesExhaustedException;
import org.apache.hadoop.hbase.ipc.HBaseRpcController;
import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
+import org.apache.hadoop.hbase.master.assignment.RegionStates;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
import org.apache.yetus.audience.InterfaceAudience;
@@ -62,12 +69,16 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
+import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClusterStatusProtos.RegionStoreSequenceIds;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClusterStatusProtos.StoreSequenceId;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.FlushedRegionSequenceId;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.FlushedSequenceId;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.FlushedStoreSequenceId;
import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.RegionServerStartupRequest;
/**
@@ -106,6 +117,22 @@ public class ServerManager {
public static final String WAIT_ON_REGIONSERVERS_INTERVAL =
"hbase.master.wait.on.regionservers.interval";
+ /**
+ * see HBASE-20727
+ * if set to true, flushedSequenceIdByRegion and storeFlushedSequenceIdsByRegion
+ * will be persisted to HDFS and loaded when master restart to speed up log split
+ */
+ public static final String PERSIST_FLUSHEDSEQUENCEID =
+ "hbase.master.persist.flushedsequenceid.enabled";
+
+ public static final boolean PERSIST_FLUSHEDSEQUENCEID_DEFAULT = true;
+
+ public static final String FLUSHEDSEQUENCEID_FLUSHER_INTERVAL =
+ "hbase.master.flushedsequenceid.flusher.interval";
+
+ public static final int FLUSHEDSEQUENCEID_FLUSHER_INTERVAL_DEFAULT =
+ 3 * 60 * 60 * 1000; // 3 hours
+
private static final Logger LOG = LoggerFactory.getLogger(ServerManager.class);
// Set if we are to shutdown the cluster.
@@ -117,6 +144,13 @@ public class ServerManager {
private final ConcurrentNavigableMap<byte[], Long> flushedSequenceIdByRegion =
new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR);
+ private boolean persistFlushedSequenceId = true;
+ private volatile boolean isFlushSeqIdPersistInProgress = false;
+ /** File on hdfs to store last flushed sequence id of regions */
+ private static final String LAST_FLUSHED_SEQ_ID_FILE = ".lastflushedseqids";
+ private FlushedSequenceIdFlusher flushedSeqIdFlusher;
+
+
/**
* The last flushed sequence id for a store in a region.
*/
@@ -194,6 +228,8 @@ public class ServerManager {
warningSkew = c.getLong("hbase.master.warningclockskew", 10000);
this.connection = master.getClusterConnection();
this.rpcControllerFactory = this.connection == null? null: connection.getRpcControllerFactory();
+ persistFlushedSequenceId = c.getBoolean(PERSIST_FLUSHEDSEQUENCEID,
+ PERSIST_FLUSHEDSEQUENCEID_DEFAULT);
}
/**
@@ -424,6 +460,11 @@ public class ServerManager {
this.rsAdmins.remove(serverName);
}
+ @VisibleForTesting
+ public ConcurrentNavigableMap<byte[], Long> getFlushedSequenceIdByRegion() {
+ return flushedSequenceIdByRegion;
+ }
+
public RegionStoreSequenceIds getLastFlushedSequenceId(byte[] encodedRegionName) {
RegionStoreSequenceIds.Builder builder = RegionStoreSequenceIds.newBuilder();
Long seqId = flushedSequenceIdByRegion.get(encodedRegionName);
@@ -601,6 +642,10 @@ public class ServerManager {
listener.serverRemoved(serverName);
}
}
+ // trigger a persist of flushedSeqId
+ if (flushedSeqIdFlusher != null) {
+ flushedSeqIdFlusher.triggerNow();
+ }
return true;
}
@@ -968,10 +1013,36 @@ public class ServerManager {
}
/**
+ * start chore in ServerManager
+ */
+ public void startChore() {
+ Configuration c = master.getConfiguration();
+ if (persistFlushedSequenceId) {
+ // when reach here, RegionStates should loaded, firstly, we call remove deleted regions
+ removeDeletedRegionFromLoadedFlushedSequenceIds();
+ int flushPeriod = c.getInt(FLUSHEDSEQUENCEID_FLUSHER_INTERVAL,
+ FLUSHEDSEQUENCEID_FLUSHER_INTERVAL_DEFAULT);
+ flushedSeqIdFlusher = new FlushedSequenceIdFlusher(
+ "FlushedSequenceIdFlusher", flushPeriod);
+ master.getChoreService().scheduleChore(flushedSeqIdFlusher);
+ }
+ }
+
+ /**
* Stop the ServerManager.
*/
public void stop() {
- // Nothing to do.
+ if (flushedSeqIdFlusher != null) {
+ flushedSeqIdFlusher.cancel();
+ }
+ if (persistFlushedSequenceId) {
+ try {
+ persistRegionLastFlushedSequenceIds();
+ } catch (IOException e) {
+ LOG.warn("Failed to persist last flushed sequence id of regions"
+ + " to file system", e);
+ }
+ }
}
/**
@@ -1065,4 +1136,144 @@ public class ServerManager {
ServerMetrics serverMetrics = onlineServers.get(serverName);
return serverMetrics != null ? serverMetrics.getInfoServerPort() : 0;
}
+
+ /**
+ * Persist last flushed sequence id of each region to HDFS
+ * @throws IOException if persit to HDFS fails
+ */
+ private void persistRegionLastFlushedSequenceIds() throws IOException {
+ if (isFlushSeqIdPersistInProgress) {
+ return;
+ }
+ isFlushSeqIdPersistInProgress = true;
+ try {
+ Configuration conf = master.getConfiguration();
+ Path rootDir = FSUtils.getRootDir(conf);
+ Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);
+ FileSystem fs = FileSystem.get(conf);
+ if (fs.exists(lastFlushedSeqIdPath)) {
+ LOG.info("Rewriting .lastflushedseqids file at: "
+ + lastFlushedSeqIdPath);
+ if (!fs.delete(lastFlushedSeqIdPath, false)) {
+ throw new IOException("Unable to remove existing "
+ + lastFlushedSeqIdPath);
+ }
+ } else {
+ LOG.info("Writing .lastflushedseqids file at: " + lastFlushedSeqIdPath);
+ }
+ FSDataOutputStream out = fs.create(lastFlushedSeqIdPath);
+ FlushedSequenceId.Builder flushedSequenceIdBuilder =
+ FlushedSequenceId.newBuilder();
+ try {
+ for (Entry<byte[], Long> entry : flushedSequenceIdByRegion.entrySet()) {
+ FlushedRegionSequenceId.Builder flushedRegionSequenceIdBuilder =
+ FlushedRegionSequenceId.newBuilder();
+ flushedRegionSequenceIdBuilder.setRegionEncodedName(
+ ByteString.copyFrom(entry.getKey()));
+ flushedRegionSequenceIdBuilder.setSeqId(entry.getValue());
+ ConcurrentNavigableMap<byte[], Long> storeSeqIds =
+ storeFlushedSequenceIdsByRegion.get(entry.getKey());
+ if (storeSeqIds != null) {
+ for (Entry<byte[], Long> store : storeSeqIds.entrySet()) {
+ FlushedStoreSequenceId.Builder flushedStoreSequenceIdBuilder =
+ FlushedStoreSequenceId.newBuilder();
+ flushedStoreSequenceIdBuilder.setFamily(ByteString.copyFrom(store.getKey()));
+ flushedStoreSequenceIdBuilder.setSeqId(store.getValue());
+ flushedRegionSequenceIdBuilder.addStores(flushedStoreSequenceIdBuilder);
+ }
+ }
+ flushedSequenceIdBuilder.addRegionSequenceId(flushedRegionSequenceIdBuilder);
+ }
+ flushedSequenceIdBuilder.build().writeDelimitedTo(out);
+ } finally {
+ if (out != null) {
+ out.close();
+ }
+ }
+ } finally {
+ isFlushSeqIdPersistInProgress = false;
+ }
+ }
+
+ /**
+ * Load last flushed sequence id of each region from HDFS, if persisted
+ */
+ public void loadLastFlushedSequenceIds() throws IOException {
+ if (!persistFlushedSequenceId) {
+ return;
+ }
+ Configuration conf = master.getConfiguration();
+ Path rootDir = FSUtils.getRootDir(conf);
+ Path lastFlushedSeqIdPath = new Path(rootDir, LAST_FLUSHED_SEQ_ID_FILE);
+ FileSystem fs = FileSystem.get(conf);
+ if (!fs.exists(lastFlushedSeqIdPath)) {
+ LOG.info("No .lastflushedseqids found at" + lastFlushedSeqIdPath
+ + " will record last flushed sequence id"
+ + " for regions by regionserver report all over again");
+ return;
+ } else {
+ LOG.info("begin to load .lastflushedseqids at " + lastFlushedSeqIdPath);
+ }
+ FSDataInputStream in = fs.open(lastFlushedSeqIdPath);
+ try {
+ FlushedSequenceId flushedSequenceId =
+ FlushedSequenceId.parseDelimitedFrom(in);
+ for (FlushedRegionSequenceId flushedRegionSequenceId : flushedSequenceId
+ .getRegionSequenceIdList()) {
+ byte[] encodedRegionName = flushedRegionSequenceId
+ .getRegionEncodedName().toByteArray();
+ flushedSequenceIdByRegion
+ .putIfAbsent(encodedRegionName, flushedRegionSequenceId.getSeqId());
+ if (flushedRegionSequenceId.getStoresList() != null
+ && flushedRegionSequenceId.getStoresList().size() != 0) {
+ ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
+ computeIfAbsent(storeFlushedSequenceIdsByRegion, encodedRegionName,
+ () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));
+ for (FlushedStoreSequenceId flushedStoreSequenceId : flushedRegionSequenceId
+ .getStoresList()) {
+ storeFlushedSequenceId
+ .put(flushedStoreSequenceId.getFamily().toByteArray(),
+ flushedStoreSequenceId.getSeqId());
+ }
+ }
+ }
+ } finally {
+ in.close();
+ }
+ }
+
+ /**
+ * Regions may have been removed between latest persist of FlushedSequenceIds
+ * and master abort. So after loading FlushedSequenceIds from file, and after
+ * meta loaded, we need to remove the deleted region according to RegionStates.
+ */
+ public void removeDeletedRegionFromLoadedFlushedSequenceIds() {
+ RegionStates regionStates = master.getAssignmentManager().getRegionStates();
+ Iterator<byte[]> it = flushedSequenceIdByRegion.keySet().iterator();
+ while(it.hasNext()) {
+ byte[] regionEncodedName = it.next();
+ if (regionStates.getRegionState(Bytes.toStringBinary(regionEncodedName)) == null) {
+ it.remove();
+ storeFlushedSequenceIdsByRegion.remove(regionEncodedName);
+ }
+ }
+ }
+
+
+ private class FlushedSequenceIdFlusher extends ScheduledChore {
+
+ public FlushedSequenceIdFlusher(String name, int p) {
+ super(name, master, p, 60 * 1000); //delay one minute before first execute
+ }
+
+ @Override
+ protected void chore() {
+ try {
+ persistRegionLastFlushedSequenceIds();
+ } catch (IOException e) {
+ LOG.debug("Failed to persist last flushed sequence id of regions"
+ + " to file system", e);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/b336da92/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMaster.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMaster.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMaster.java
index 11df313..8abaf60 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMaster.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMaster.java
@@ -24,6 +24,8 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HColumnDescriptor;
@@ -45,6 +47,7 @@ import org.apache.hadoop.hbase.testclassification.MasterTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.util.StringUtils;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -192,5 +195,35 @@ public class TestMaster {
TEST_UTIL.deleteTable(tableName);
}
}
+
+ @Test
+ public void testFlushedSequenceIdPersistLoad() throws Exception {
+ Configuration conf = TEST_UTIL.getConfiguration();
+ int msgInterval = conf.getInt("hbase.regionserver.msginterval", 100);
+ // insert some data into META
+ TableName tableName = TableName.valueOf("testFlushSeqId");
+ HTableDescriptor desc = new HTableDescriptor(tableName);
+ desc.addFamily(new HColumnDescriptor(Bytes.toBytes("cf")));
+ Table table = TEST_UTIL.createTable(desc, null);
+ // flush META region
+ TEST_UTIL.flush(TableName.META_TABLE_NAME);
+ // wait for regionserver report
+ Threads.sleep(msgInterval * 2);
+ // record flush seqid before cluster shutdown
+ Map<byte[], Long> regionMapBefore =
+ TEST_UTIL.getHBaseCluster().getMaster().getServerManager()
+ .getFlushedSequenceIdByRegion();
+ // restart hbase cluster which will cause flushed sequence id persist and reload
+ TEST_UTIL.getMiniHBaseCluster().shutdown();
+ TEST_UTIL.restartHBaseCluster(2);
+ TEST_UTIL.waitUntilNoRegionsInTransition();
+ // check equality after reloading flushed sequence id map
+ Map<byte[], Long> regionMapAfter =
+ TEST_UTIL.getHBaseCluster().getMaster().getServerManager()
+ .getFlushedSequenceIdByRegion();
+ assertTrue(regionMapBefore.equals(regionMapAfter));
+
+
+ }
}