diff --git hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionConfiguration.java hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionConfiguration.java index bea91da..2883111 100644 --- hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionConfiguration.java +++ hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionConfiguration.java @@ -40,6 +40,7 @@ public class ConnectionConfiguration { private final long scannerMaxResultSize; private final int primaryCallTimeoutMicroSecond; private final int replicaCallTimeoutMicroSecondScan; + private final int metaReplicaCallTimeoutMicroSecondScan; private final int retries; private final int maxKeyValueSize; private final int rpcTimeout; @@ -55,9 +56,8 @@ public class ConnectionConfiguration { ConnectionConfiguration(Configuration conf) { this.writeBufferSize = conf.getLong(WRITE_BUFFER_SIZE_KEY, WRITE_BUFFER_SIZE_DEFAULT); - this.metaOperationTimeout = conf.getInt( - HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, - HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT); + this.metaOperationTimeout = conf.getInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, + HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT); this.operationTimeout = conf.getInt( HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT); @@ -67,7 +67,7 @@ public class ConnectionConfiguration { this.scannerMaxResultSize = conf.getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, - HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE); + HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE); this.primaryCallTimeoutMicroSecond = conf.getInt("hbase.client.primaryCallTimeout.get", 10000); // 10ms @@ -75,6 +75,10 @@ public class ConnectionConfiguration { this.replicaCallTimeoutMicroSecondScan = conf.getInt("hbase.client.replicaCallTimeout.scan", 1000000); // 1000 ms + this.metaReplicaCallTimeoutMicroSecondScan = + conf.getInt(HConstants.HBASE_CLIENT_MEAT_REPLICA_CALL_TIMEOUT_SCAN, + HConstants.HBASE_CLIENT_MEAT_REPLICA_CALL_TIMEOUT_SCAN_DEFAULT); + this.retries = conf.getInt( HConstants.HBASE_CLIENT_RETRIES_NUMBER, HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER); @@ -107,6 +111,8 @@ public class ConnectionConfiguration { this.scannerMaxResultSize = HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE; this.primaryCallTimeoutMicroSecond = 10000; this.replicaCallTimeoutMicroSecondScan = 1000000; + this.metaReplicaCallTimeoutMicroSecondScan = + HConstants.HBASE_CLIENT_MEAT_REPLICA_CALL_TIMEOUT_SCAN_DEFAULT; this.retries = HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER; this.clientScannerAsyncPrefetch = Scan.DEFAULT_HBASE_CLIENT_SCANNER_ASYNC_PREFETCH; this.maxKeyValueSize = MAX_KEYVALUE_SIZE_DEFAULT; @@ -147,6 +153,10 @@ public class ConnectionConfiguration { return replicaCallTimeoutMicroSecondScan; } + public int getMetaReplicaCallTimeoutMicroSecondScan() { + return metaReplicaCallTimeoutMicroSecondScan; + } + public int getRetriesNumber() { return retries; } diff --git hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java index 6859cb3..564cfb4 100644 --- hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java +++ hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java @@ -139,7 +139,8 @@ class ConnectionImplementation implements ClusterConnection, Closeable { private final boolean hostnamesCanChange; private final long pause; private final long pauseForCQTBE;// pause for CallQueueTooBigException, if specified - private final boolean useMetaReplicas; + private boolean useMetaReplicas; + private final int metaReplicaCallTimeoutScanInMicroSecond; private final int numTries; final int rpcTimeout; @@ -235,6 +236,9 @@ class ConnectionImplementation implements ClusterConnection, Closeable { } this.useMetaReplicas = conf.getBoolean(HConstants.USE_META_REPLICAS, HConstants.DEFAULT_USE_META_REPLICAS); + this.metaReplicaCallTimeoutScanInMicroSecond = + connectionConfig.getMetaReplicaCallTimeoutMicroSecondScan(); + // how many times to try, one more than max *retry* time this.numTries = retries2Attempts(connectionConfig.getRetriesNumber()); this.rpcTimeout = conf.getInt( @@ -306,6 +310,14 @@ class ConnectionImplementation implements ClusterConnection, Closeable { } /** + * @param useMetaReplicas + */ + @VisibleForTesting + void setUseMetaReplicas(final boolean useMetaReplicas) { + this.useMetaReplicas = useMetaReplicas; + } + + /** * @param conn The connection for which to replace the generator. * @param cnm Replaces the nonce generator used, for testing. * @return old nonce generator. @@ -820,7 +832,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable { s.resetMvccReadPoint(); try (ReversedClientScanner rcs = new ReversedClientScanner(conf, s, TableName.META_TABLE_NAME, this, rpcCallerFactory, - rpcControllerFactory, getMetaLookupPool(), 0)) { + rpcControllerFactory, getMetaLookupPool(), metaReplicaCallTimeoutScanInMicroSecond)) { regionInfoRow = rcs.next(); } diff --git hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java index 3c1b021..c9aff60 100644 --- hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java +++ hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java @@ -307,6 +307,13 @@ public final class HConstants { /** Default HBase client operation timeout, which is tantamount to a blocking call */ public static final int DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT = 1200000; + /** Parameter name for HBase client meta replica scan call timeout. */ + public static final String HBASE_CLIENT_MEAT_REPLICA_CALL_TIMEOUT_SCAN = + "hbase.client.metaReplicaCallTimeout.scan"; + + /** Default HBase client meta replica scan call timeout, 1 second */ + public static final int HBASE_CLIENT_MEAT_REPLICA_CALL_TIMEOUT_SCAN_DEFAULT = 1000000; + /** Used to construct the name of the log directory for a region server */ public static final String HREGION_LOGDIR_NAME = "WALs"; diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java index 2c77541..ad93a68 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java @@ -39,6 +39,9 @@ import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.RegionLocations; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.Waiter; import org.apache.hadoop.hbase.client.replication.ReplicationAdmin; @@ -75,6 +78,7 @@ public class TestReplicaWithCluster { private static final byte[] f = HConstants.CATALOG_FAMILY; private final static int REFRESH_PERIOD = 1000; + private final static int META_SCAN_TIMEOUT_IN_MILLISEC = 200; /** * This copro is used to synchronize the tests. @@ -157,6 +161,34 @@ public class TestReplicaWithCluster { } } + /** + * This copro is used to slow down the primary meta region scan a bit + */ + public static class RegionServerHostingPrimayMetaRegionSlowCopro implements RegionObserver { + static boolean slowDownPrimaryMetaScan = false; + + @Override + public RegionScanner preScannerOpen(final ObserverContext e, + final Scan scan, final RegionScanner s) throws IOException { + + int replicaId = e.getEnvironment().getRegion().getRegionInfo().getReplicaId(); + + // Slow down with the primary meta region scan + if (slowDownPrimaryMetaScan && (e.getEnvironment().getRegion().getRegionInfo().isMetaRegion() + && (replicaId == 0))) { + LOG.info("Scan with primary meta region, slow down a bit"); + try { + Thread.sleep(META_SCAN_TIMEOUT_IN_MILLISEC - 50); + } catch (InterruptedException ie) { + // Ingore + } + + } + return null; + } + } + + @BeforeClass public static void beforeClass() throws Exception { // enable store file refreshing @@ -178,6 +210,19 @@ public class TestReplicaWithCluster { // Retry less so it can fail faster HTU.getConfiguration().setInt("hbase.client.retries.number", 1); + // Enable meta replica at server side + HTU.getConfiguration().setInt("hbase.meta.replica.count", 2); + + // Make sure master does not host system tables. + HTU.getConfiguration().set("hbase.balancer.tablesOnMaster", "none"); + + // Set system coprocessor so it can be applied to meta regions + HTU.getConfiguration().set("hbase.coprocessor.region.classes", + RegionServerHostingPrimayMetaRegionSlowCopro.class.getName()); + + HTU.getConfiguration().setInt(HConstants.HBASE_CLIENT_MEAT_REPLICA_CALL_TIMEOUT_SCAN, + META_SCAN_TIMEOUT_IN_MILLISEC * 1000); + HTU.startMiniCluster(NB_SERVERS); HTU.getHBaseCluster().startMaster(); } @@ -570,4 +615,74 @@ public class TestReplicaWithCluster { HTU.deleteTable(hdt.getTableName()); } } + + // This test is to test when hbase.client.metaReplicaCallTimeout.scan is configured, meta table + // scan will always get the result from primary meta region as long as the result is returned + // within configured hbase.client.metaReplicaCallTimeout.scan from primary meta region. + @Test + public void testGetRegionLocationFromPrimaryMetaRegion() throws IOException, InterruptedException { + HTU.getAdmin().setBalancerRunning(false, true); + + ((ConnectionImplementation) HTU.getAdmin().getConnection()).setUseMetaReplicas(true); + + // Create table then get the single region for our new table. + HTableDescriptor hdt = HTU.createTableDescriptor("testGetRegionLocationFromPrimaryMetaRegion"); + hdt.setRegionReplication(2); + try { + + Table table = HTU.createTable(hdt, new byte[][] { f }, null); + + // Get Meta location + RegionLocations mrl = ((ClusterConnection) HTU.getConnection()) + .locateRegion(TableName.META_TABLE_NAME, HConstants.EMPTY_START_ROW, false, false); + + // Get user table location + RegionLocations url = ((ClusterConnection) HTU.getConnection()) + .locateRegion(hdt.getTableName(), row, false, false); + + // Make sure that user primary region is co-hosted with the meta region + if (!url.getDefaultRegionLocation().getServerName() + .equals(mrl.getDefaultRegionLocation().getServerName())) { + HTU.moveRegionAndWait(url.getDefaultRegionLocation().getRegionInfo(), + mrl.getDefaultRegionLocation().getServerName()); + } + + // Make sure that the user replica region is not hosted by the same region server with + // primary + if (url.getRegionLocation(1).getServerName() + .equals(mrl.getDefaultRegionLocation().getServerName())) { + HTU.moveRegionAndWait(url.getRegionLocation(1).getRegionInfo(), + url.getDefaultRegionLocation().getServerName()); + } + + RegionServerHostingPrimayMetaRegionSlowCopro.slowDownPrimaryMetaScan = true; + + // Wait until the meta table is updated with new location info + while (true) { + mrl = ((ClusterConnection) HTU.getConnection()) + .locateRegion(TableName.META_TABLE_NAME, HConstants.EMPTY_START_ROW, false, false); + + // Get user table location + url = ((ClusterConnection) HTU.getConnection()) + .locateRegion(hdt.getTableName(), row, false, true); + + LOG.info("meta locations " + mrl); + LOG.info("table locations " + url); + ServerName a = url.getDefaultRegionLocation().getServerName(); + ServerName b = mrl.getDefaultRegionLocation().getServerName(); + if (a.equals(b)) { + break; + } else { + LOG.info("Waiting for new region info to be updated in meta table"); + Thread.sleep(100); + } + } + } finally { + RegionServerHostingPrimayMetaRegionSlowCopro.slowDownPrimaryMetaScan = false; + ((ConnectionImplementation) HTU.getAdmin().getConnection()).setUseMetaReplicas(false); + HTU.getAdmin().setBalancerRunning(true, true); + HTU.getAdmin().disableTable(hdt.getTableName()); + HTU.deleteTable(hdt.getTableName()); + } + } }