You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zg...@apache.org on 2017/06/26 09:27:47 UTC
[1/3] hbase git commit: HBASE-18234 Revisit the async admin api
Repository: hbase
Updated Branches:
refs/heads/master 0e8e176eb -> 2d781aa15
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java
----------------------------------------------------------------------
diff --git a/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java
index 10a4d19..dc895f6 100644
--- a/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java
+++ b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.logging.Log;
@@ -136,7 +137,7 @@ public class TestHRegionServerBulkLoadWithOldSecureEndpoint extends TestHRegionS
conn.getAdmin(getLocation().getServerName());
CompactRegionRequest request =
RequestConverter.buildCompactRegionRequest(
- getLocation().getRegionInfo().getRegionName(), true, null);
+ getLocation().getRegionInfo().getRegionName(), true, Optional.empty());
server.compactRegion(null, request);
numCompactions.incrementAndGet();
return null;
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncBalancerAdminApi.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncBalancerAdminApi.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncBalancerAdminApi.java
index b1df40e..00303e2 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncBalancerAdminApi.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncBalancerAdminApi.java
@@ -29,23 +29,23 @@ public class TestAsyncBalancerAdminApi extends TestAsyncAdminBase {
@Test
public void testBalancer() throws Exception {
- boolean initialState = admin.isBalancerEnabled().get();
+ boolean initialState = admin.isBalancerOn().get();
// Start the balancer, wait for it.
- boolean prevState = admin.setBalancerRunning(!initialState).get();
+ boolean prevState = admin.setBalancerOn(!initialState).get();
// The previous state should be the original state we observed
assertEquals(initialState, prevState);
// Current state should be opposite of the original
- assertEquals(!initialState, admin.isBalancerEnabled().get());
+ assertEquals(!initialState, admin.isBalancerOn().get());
// Reset it back to what it was
- prevState = admin.setBalancerRunning(initialState).get();
+ prevState = admin.setBalancerOn(initialState).get();
// The previous state should be the opposite of the initial state
assertEquals(!initialState, prevState);
// Current state should be the original state again
- assertEquals(initialState, admin.isBalancerEnabled().get());
+ assertEquals(initialState, admin.isBalancerOn().get());
}
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncNamespaceAdminApi.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncNamespaceAdminApi.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncNamespaceAdminApi.java
index ed2e246..eccff3f 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncNamespaceAdminApi.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncNamespaceAdminApi.java
@@ -74,7 +74,7 @@ public class TestAsyncNamespaceAdminApi extends TestAsyncAdminBase {
// create namespace and verify
admin.createNamespace(NamespaceDescriptor.create(nsName).build()).join();
- assertEquals(3, admin.listNamespaceDescriptors().get().length);
+ assertEquals(3, admin.listNamespaceDescriptors().get().size());
TEST_UTIL.waitFor(60000, new Waiter.Predicate<Exception>() {
@Override
public boolean evaluate() throws Exception {
@@ -84,7 +84,7 @@ public class TestAsyncNamespaceAdminApi extends TestAsyncAdminBase {
assertNotNull(zkNamespaceManager.get(nsName));
// delete namespace and verify
admin.deleteNamespace(nsName).join();
- assertEquals(2, admin.listNamespaceDescriptors().get().length);
+ assertEquals(2, admin.listNamespaceDescriptors().get().size());
assertEquals(2, zkNamespaceManager.list().size());
assertNull(zkNamespaceManager.get(nsName));
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcedureAdminApi.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcedureAdminApi.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcedureAdminApi.java
index 82f627c..832bcbe 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcedureAdminApi.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcedureAdminApi.java
@@ -34,6 +34,7 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Random;
@@ -94,8 +95,8 @@ public class TestAsyncProcedureAdminApi extends TestAsyncAdminBase {
@Test
public void listProcedure() throws Exception {
- ProcedureInfo[] procList = admin.listProcedures().get();
- assertTrue(procList.length >= 0);
+ List<ProcedureInfo> procList = admin.listProcedures().get();
+ assertTrue(procList.size() >= 0);
}
@Test
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncRegionAdminApi.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncRegionAdminApi.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncRegionAdminApi.java
index 3bdd5fd..a3afabc 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncRegionAdminApi.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncRegionAdminApi.java
@@ -25,6 +25,7 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Optional;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
@@ -84,8 +85,8 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
for (HRegionInfo regionInfo : onlineRegions) {
if (!regionInfo.getTable().isSystemTable()) {
info = regionInfo;
- boolean closed = admin.closeRegionWithEncodedRegionName(regionInfo.getEncodedName(),
- rs.getServerName().getServerName()).get();
+ boolean closed = admin.closeRegion(regionInfo.getRegionName(),
+ Optional.of(rs.getServerName())).get();
assertTrue(closed);
}
}
@@ -114,7 +115,7 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
info = regionInfo;
boolean catchNotServingException = false;
try {
- admin.closeRegionWithEncodedRegionName("sample", rs.getServerName().getServerName())
+ admin.closeRegion(Bytes.toBytes("sample"), Optional.of(rs.getServerName()))
.get();
} catch (Exception e) {
catchNotServingException = true;
@@ -130,76 +131,19 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
}
@Test
- public void testCloseRegionWhenServerNameIsNull() throws Exception {
- byte[] TABLENAME = Bytes.toBytes("TestHBACloseRegion3");
- createTableWithDefaultConf(TableName.valueOf(TABLENAME));
-
- HRegionServer rs = TEST_UTIL.getRSForFirstRegionInTable(TableName.valueOf(TABLENAME));
-
- try {
- List<HRegionInfo> onlineRegions = ProtobufUtil.getOnlineRegions(rs.getRSRpcServices());
- for (HRegionInfo regionInfo : onlineRegions) {
- if (!regionInfo.isMetaTable()) {
- if (regionInfo.getRegionNameAsString().contains("TestHBACloseRegion3")) {
- admin.closeRegionWithEncodedRegionName(regionInfo.getEncodedName(), null).get();
- }
- }
- }
- fail("The test should throw exception if the servername passed is null.");
- } catch (IllegalArgumentException e) {
- }
- }
-
- @Test
public void testCloseRegionWhenServerNameIsEmpty() throws Exception {
byte[] TABLENAME = Bytes.toBytes("TestHBACloseRegionWhenServerNameIsEmpty");
createTableWithDefaultConf(TableName.valueOf(TABLENAME));
HRegionServer rs = TEST_UTIL.getRSForFirstRegionInTable(TableName.valueOf(TABLENAME));
-
- try {
- List<HRegionInfo> onlineRegions = ProtobufUtil.getOnlineRegions(rs.getRSRpcServices());
- for (HRegionInfo regionInfo : onlineRegions) {
- if (!regionInfo.isMetaTable()) {
- if (regionInfo.getRegionNameAsString()
- .contains("TestHBACloseRegionWhenServerNameIsEmpty")) {
- admin.closeRegionWithEncodedRegionName(regionInfo.getEncodedName(), " ").get();
- }
- }
- }
- fail("The test should throw exception if the servername passed is empty.");
- } catch (IllegalArgumentException e) {
- }
- }
-
- @Test
- public void testCloseRegionWhenEncodedRegionNameIsNotGiven() throws Exception {
- byte[] TABLENAME = Bytes.toBytes("TestHBACloseRegion4");
- createTableWithDefaultConf(TableName.valueOf(TABLENAME));
-
- HRegionInfo info = null;
- HRegionServer rs = TEST_UTIL.getRSForFirstRegionInTable(TableName.valueOf(TABLENAME));
-
List<HRegionInfo> onlineRegions = ProtobufUtil.getOnlineRegions(rs.getRSRpcServices());
for (HRegionInfo regionInfo : onlineRegions) {
if (!regionInfo.isMetaTable()) {
- if (regionInfo.getRegionNameAsString().contains("TestHBACloseRegion4")) {
- info = regionInfo;
- boolean catchNotServingException = false;
- try {
- admin.closeRegionWithEncodedRegionName(regionInfo.getRegionNameAsString(),
- rs.getServerName().getServerName()).get();
- } catch (Exception e) {
- // expected, ignore it.
- catchNotServingException = true;
- }
- assertTrue(catchNotServingException);
+ if (regionInfo.getRegionNameAsString().contains("TestHBACloseRegionWhenServerNameIsEmpty")) {
+ admin.closeRegion(regionInfo.getRegionName(), Optional.empty()).get();
}
}
}
- onlineRegions = ProtobufUtil.getOnlineRegions(rs.getRSRpcServices());
- assertTrue("The region should be present in online regions list.",
- onlineRegions.contains(info));
}
@Test
@@ -214,10 +158,10 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
HRegionLocation regionLocation = locator.getRegionLocation(Bytes.toBytes("mmm"));
HRegionInfo region = regionLocation.getRegionInfo();
byte[] regionName = region.getRegionName();
- Pair<HRegionInfo, ServerName> pair = rawAdmin.getRegion(regionName).get();
- assertTrue(Bytes.equals(regionName, pair.getFirst().getRegionName()));
- pair = rawAdmin.getRegion(region.getEncodedNameAsBytes()).get();
- assertTrue(Bytes.equals(regionName, pair.getFirst().getRegionName()));
+ HRegionLocation location = rawAdmin.getRegionLocation(regionName).get();
+ assertTrue(Bytes.equals(regionName, location.getRegionInfo().getRegionName()));
+ location = rawAdmin.getRegionLocation(region.getEncodedNameAsBytes()).get();
+ assertTrue(Bytes.equals(regionName, location.getRegionInfo().getRegionName()));
}
}
@@ -289,7 +233,7 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
table.put(puts);
if (isSplitRegion) {
- admin.splitRegion(regions.get(0).getRegionName(), splitPoint).get();
+ admin.splitRegion(regions.get(0).getRegionName(), Optional.ofNullable(splitPoint)).get();
} else {
if (splitPoint == null) {
admin.split(tableName).get();
@@ -456,7 +400,7 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
}
}
assertTrue(destServerName != null && !destServerName.equals(serverName));
- admin.move(hri.getEncodedNameAsBytes(), Bytes.toBytes(destServerName.getServerName())).get();
+ admin.move(hri.getRegionName(), Optional.of(destServerName)).get();
long timeoutTime = System.currentTimeMillis() + 30000;
while (true) {
@@ -607,15 +551,15 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
assertTrue(countBefore > 0); // there should be some data files
if (expectedState == CompactionState.MINOR) {
if (singleFamily) {
- admin.compact(table, family).get();
+ admin.compact(table, Optional.of(family)).get();
} else {
- admin.compact(table).get();
+ admin.compact(table, Optional.empty()).get();
}
} else {
if (singleFamily) {
- admin.majorCompact(table, family).get();
+ admin.majorCompact(table, Optional.of(family)).get();
} else {
- admin.majorCompact(table).get();
+ admin.majorCompact(table, Optional.empty()).get();
}
}
long curt = System.currentTimeMillis();
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncSnapshotAdminApi.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncSnapshotAdminApi.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncSnapshotAdminApi.java
index f71f311..3e0c261 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncSnapshotAdminApi.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncSnapshotAdminApi.java
@@ -53,9 +53,9 @@ public class TestAsyncSnapshotAdminApi extends TestAsyncAdminBase {
@After
public void cleanup() throws Exception {
- admin.deleteSnapshots(".*").get();
- admin.disableTables(".*").get();
- admin.deleteTables(".*").get();
+ admin.deleteSnapshots(Pattern.compile(".*")).get();
+ admin.disableTables(Pattern.compile(".*")).get();
+ admin.deleteTables(Pattern.compile(".*")).get();
}
@Test
@@ -175,13 +175,22 @@ public class TestAsyncSnapshotAdminApi extends TestAsyncAdminBase {
admin.snapshot(snapshotName3, tableName).get();
Assert.assertEquals(admin.listSnapshots().get().size(), 3);
- Assert.assertEquals(admin.listSnapshots("(.*)").get().size(), 3);
- Assert.assertEquals(admin.listSnapshots("snapshotName(\\d+)").get().size(), 3);
- Assert.assertEquals(admin.listSnapshots("snapshotName[1|3]").get().size(), 2);
+ Assert.assertEquals(admin.listSnapshots(Pattern.compile("(.*)")).get().size(), 3);
+ Assert.assertEquals(admin.listSnapshots(Pattern.compile("snapshotName(\\d+)")).get().size(), 3);
+ Assert.assertEquals(admin.listSnapshots(Pattern.compile("snapshotName[1|3]")).get().size(), 2);
Assert.assertEquals(admin.listSnapshots(Pattern.compile("snapshot(.*)")).get().size(), 3);
- Assert.assertEquals(admin.listTableSnapshots("testListSnapshots", "s(.*)").get().size(), 3);
- Assert.assertEquals(admin.listTableSnapshots("fakeTableName", "snap(.*)").get().size(), 0);
- Assert.assertEquals(admin.listTableSnapshots("test(.*)", "snap(.*)[1|3]").get().size(), 2);
+ Assert.assertEquals(
+ admin.listTableSnapshots(Pattern.compile("testListSnapshots"), Pattern.compile("s(.*)")).get()
+ .size(),
+ 3);
+ Assert.assertEquals(
+ admin.listTableSnapshots(Pattern.compile("fakeTableName"), Pattern.compile("snap(.*)")).get()
+ .size(),
+ 0);
+ Assert.assertEquals(
+ admin.listTableSnapshots(Pattern.compile("test(.*)"), Pattern.compile("snap(.*)[1|3]")).get()
+ .size(),
+ 2);
}
@Test
@@ -201,19 +210,19 @@ public class TestAsyncSnapshotAdminApi extends TestAsyncAdminBase {
admin.deleteSnapshot(snapshotName1).get();
Assert.assertEquals(admin.listSnapshots().get().size(), 2);
- admin.deleteSnapshots("(.*)abc").get();
+ admin.deleteSnapshots(Pattern.compile("(.*)abc")).get();
Assert.assertEquals(admin.listSnapshots().get().size(), 2);
- admin.deleteSnapshots("(.*)1").get();
+ admin.deleteSnapshots(Pattern.compile("(.*)1")).get();
Assert.assertEquals(admin.listSnapshots().get().size(), 2);
- admin.deleteTableSnapshots("(.*)", "(.*)1").get();
+ admin.deleteTableSnapshots(Pattern.compile("(.*)"), Pattern.compile("(.*)1")).get();
Assert.assertEquals(admin.listSnapshots().get().size(), 2);
- admin.deleteTableSnapshots("(.*)", "(.*)2").get();
+ admin.deleteTableSnapshots(Pattern.compile("(.*)"), Pattern.compile("(.*)2")).get();
Assert.assertEquals(admin.listSnapshots().get().size(), 1);
- admin.deleteTableSnapshots("(.*)", "(.*)3").get();
+ admin.deleteTableSnapshots(Pattern.compile("(.*)"), Pattern.compile("(.*)3")).get();
Assert.assertEquals(admin.listSnapshots().get().size(), 0);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java
index c0ccd5e..6586a03 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableAdminApi.java
@@ -80,7 +80,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
@Test
public void testListTables() throws Exception {
- int numTables = admin.listTables().get().length;
+ int numTables = admin.listTables().get().size();
final TableName tableName1 = TableName.valueOf(name.getMethodName() + "1");
final TableName tableName2 = TableName.valueOf(name.getMethodName() + "2");
final TableName tableName3 = TableName.valueOf(name.getMethodName() + "3");
@@ -89,13 +89,13 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
TEST_UTIL.createTable(tables[i], FAMILY);
}
- TableDescriptor[] tableDescs = admin.listTables().get();
- int size = tableDescs.length;
+ List<TableDescriptor> tableDescs = admin.listTables().get();
+ int size = tableDescs.size();
assertTrue(size >= tables.length);
for (int i = 0; i < tables.length && i < size; i++) {
boolean found = false;
- for (int j = 0; j < tableDescs.length; j++) {
- if (tableDescs[j].getTableName().equals(tables[i])) {
+ for (int j = 0; j < size; j++) {
+ if (tableDescs.get(j).getTableName().equals(tables[i])) {
found = true;
break;
}
@@ -103,13 +103,13 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
assertTrue("Not found: " + tables[i], found);
}
- TableName[] tableNames = admin.listTableNames().get();
- size = tableNames.length;
+ List<TableName> tableNames = admin.listTableNames().get();
+ size = tableNames.size();
assertTrue(size == (numTables + tables.length));
for (int i = 0; i < tables.length && i < size; i++) {
boolean found = false;
- for (int j = 0; j < tableNames.length; j++) {
- if (tableNames[j].equals(tables[i])) {
+ for (int j = 0; j < size; j++) {
+ if (tableNames.get(j).equals(tables[i])) {
found = true;
break;
}
@@ -121,10 +121,10 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
TEST_UTIL.deleteTable(tables[i]);
}
- tableDescs = admin.listTables((Pattern) null, true).get();
- assertTrue("Not found system tables", tableDescs.length > 0);
- tableNames = admin.listTableNames((Pattern) null, true).get();
- assertTrue("Not found system tables", tableNames.length > 0);
+ tableDescs = admin.listTables(Optional.empty(), true).get();
+ assertTrue("Not found system tables", tableDescs.size() > 0);
+ tableNames = admin.listTableNames(Optional.empty(), true).get();
+ assertTrue("Not found system tables", tableNames.size() > 0);
}
@Test(timeout = 300000)
@@ -143,13 +143,13 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
@Test(timeout = 300000)
public void testCreateTable() throws Exception {
- TableDescriptor[] tables = admin.listTables().get();
- int numTables = tables.length;
+ List<TableDescriptor> tables = admin.listTables().get();
+ int numTables = tables.size();
final TableName tableName = TableName.valueOf(name.getMethodName());
admin.createTable(new HTableDescriptor(tableName).addFamily(new HColumnDescriptor(FAMILY)))
.join();
tables = admin.listTables().get();
- assertEquals(numTables + 1, tables.length);
+ assertEquals(numTables + 1, tables.size());
assertTrue("Table must be enabled.", TEST_UTIL.getHBaseCluster().getMaster()
.getTableStateManager().isTableState(tableName, TableState.State.ENABLED));
assertEquals(TableState.State.ENABLED, getStateFromMeta(tableName));
@@ -449,8 +449,8 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
} catch (Exception e) {
}
});
- TableDescriptor[] failed = admin.deleteTables(Pattern.compile("testDeleteTables.*")).get();
- assertEquals(0, failed.length);
+ List<TableDescriptor> failed = admin.deleteTables(Pattern.compile("testDeleteTables.*")).get();
+ assertEquals(0, failed.size());
Arrays.stream(tables).forEach((table) -> {
admin.tableExists(table).thenAccept((exist) -> assertFalse(exist)).join();
});
@@ -572,7 +572,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
ht1.get(get);
ht2.get(get);
- this.admin.disableTables("testDisableAndEnableTable.*").join();
+ this.admin.disableTables(Pattern.compile("testDisableAndEnableTable.*")).join();
// Test that tables are disabled
get = new Get(row);
@@ -589,7 +589,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
assertEquals(TableState.State.DISABLED, getStateFromMeta(tableName2));
assertTrue(ok);
- this.admin.enableTables("testDisableAndEnableTable.*").join();
+ this.admin.enableTables(Pattern.compile("testDisableAndEnableTable.*")).join();
// Test that tables are enabled
try {
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java
index 61fe2cc..d640cbf 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java
@@ -28,6 +28,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.logging.Log;
@@ -236,7 +237,7 @@ public class TestHRegionServerBulkLoad {
AdminProtos.AdminService.BlockingInterface server =
conn.getAdmin(getLocation().getServerName());
CompactRegionRequest request = RequestConverter.buildCompactRegionRequest(
- getLocation().getRegionInfo().getRegionName(), true, null);
+ getLocation().getRegionInfo().getRegionName(), true, Optional.empty());
server.compactRegion(null, request);
numCompactions.incrementAndGet();
return null;
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java
index 7aa1b31..a28f8f1 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.regionserver;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.logging.Log;
@@ -122,7 +123,7 @@ public class TestHRegionServerBulkLoadWithOldClient extends TestHRegionServerBul
conn.getAdmin(getLocation().getServerName());
CompactRegionRequest request =
RequestConverter.buildCompactRegionRequest(
- getLocation().getRegionInfo().getRegionName(), true, null);
+ getLocation().getRegionInfo().getRegionName(), true, Optional.empty());
server.compactRegion(null, request);
numCompactions.incrementAndGet();
return null;
[2/3] hbase git commit: HBASE-18234 Revisit the async admin api
Posted by zg...@apache.org.
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java
index c972b4c..8505241 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java
@@ -40,10 +40,13 @@ import com.google.common.annotations.VisibleForTesting;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
+
import java.util.stream.Stream;
+
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.directory.api.util.OptionalComponentsMonitor;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.MetaTableAccessor;
@@ -190,7 +193,6 @@ import org.apache.hadoop.hbase.util.Pair;
* The implementation of AsyncAdmin.
*/
@InterfaceAudience.Private
-@InterfaceStability.Evolving
public class AsyncHBaseAdmin implements AsyncAdmin {
public static final String FLUSH_TABLE_PROCEDURE_SIGNATURE = "flush-table-proc";
@@ -278,7 +280,6 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
return future;
}
- //TODO abstract call and adminCall into a single method.
private <PREQ, PRESP, RESP> CompletableFuture<RESP> adminCall(HBaseRpcController controller,
AdminService.Interface stub, PREQ preq, AdminRpcCall<PRESP, PREQ> rpcCall,
Converter<RESP, PRESP> respConverter) {
@@ -318,25 +319,26 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
CompletableFuture<Void> operate(TableName table);
}
- private CompletableFuture<TableDescriptor[]> batchTableOperations(Pattern pattern,
+ private CompletableFuture<List<TableDescriptor>> batchTableOperations(Pattern pattern,
TableOperator operator, String operationType) {
- CompletableFuture<TableDescriptor[]> future = new CompletableFuture<>();
+ CompletableFuture<List<TableDescriptor>> future = new CompletableFuture<>();
List<TableDescriptor> failed = new LinkedList<>();
- listTables(pattern, false).whenComplete(
+ listTables(Optional.ofNullable(pattern), false).whenComplete(
(tables, error) -> {
if (error != null) {
future.completeExceptionally(error);
return;
}
- CompletableFuture[] futures = Arrays.stream(tables)
- .map((table) -> operator.operate(table.getTableName()).whenComplete((v, ex) -> {
- if (ex != null) {
- LOG.info("Failed to " + operationType + " table " + table.getTableName(), ex);
- failed.add(table);
- }
- })).<CompletableFuture> toArray(size -> new CompletableFuture[size]);
+ CompletableFuture[] futures =
+ tables.stream()
+ .map((table) -> operator.operate(table.getTableName()).whenComplete((v, ex) -> {
+ if (ex != null) {
+ LOG.info("Failed to " + operationType + " table " + table.getTableName(), ex);
+ failed.add(table);
+ }
+ })).<CompletableFuture> toArray(size -> new CompletableFuture[size]);
CompletableFuture.allOf(futures).thenAccept((v) -> {
- future.complete(failed.toArray(new TableDescriptor[failed.size()]));
+ future.complete(failed);
});
});
return future;
@@ -353,47 +355,28 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<TableDescriptor[]> listTables() {
- return listTables((Pattern) null, false);
- }
-
- @Override
- public CompletableFuture<TableDescriptor[]> listTables(String regex, boolean includeSysTables) {
- return listTables(Pattern.compile(regex), false);
- }
-
- @Override
- public CompletableFuture<TableDescriptor[]> listTables(Pattern pattern, boolean includeSysTables) {
- return this
- .<TableDescriptor[]>newMasterCaller()
- .action(
- (controller, stub) -> this
- .<GetTableDescriptorsRequest, GetTableDescriptorsResponse, TableDescriptor[]> call(
- controller, stub, RequestConverter.buildGetTableDescriptorsRequest(pattern,
- includeSysTables), (s, c, req, done) -> s.getTableDescriptors(c, req, done), (
- resp) -> ProtobufUtil.getTableDescriptorArray(resp))).call();
- }
-
- @Override
- public CompletableFuture<TableName[]> listTableNames() {
- return listTableNames((Pattern) null, false);
- }
-
- @Override
- public CompletableFuture<TableName[]> listTableNames(String regex, boolean includeSysTables) {
- return listTableNames(Pattern.compile(regex), false);
+ public CompletableFuture<List<TableDescriptor>> listTables(Optional<Pattern> pattern,
+ boolean includeSysTables) {
+ return this.<List<TableDescriptor>> newMasterCaller()
+ .action((controller, stub) -> this
+ .<GetTableDescriptorsRequest, GetTableDescriptorsResponse, List<TableDescriptor>> call(
+ controller, stub,
+ RequestConverter.buildGetTableDescriptorsRequest(pattern, includeSysTables),
+ (s, c, req, done) -> s.getTableDescriptors(c, req, done),
+ (resp) -> ProtobufUtil.toTableDescriptorList(resp)))
+ .call();
}
@Override
- public CompletableFuture<TableName[]> listTableNames(Pattern pattern, boolean includeSysTables) {
- return this
- .<TableName[]>newMasterCaller()
- .action(
- (controller, stub) -> this
- .<GetTableNamesRequest, GetTableNamesResponse, TableName[]> call(controller, stub,
- RequestConverter.buildGetTableNamesRequest(pattern, includeSysTables), (s, c, req,
- done) -> s.getTableNames(c, req, done), (resp) -> ProtobufUtil
- .getTableNameArray(resp.getTableNamesList()))).call();
+ public CompletableFuture<List<TableName>> listTableNames(Optional<Pattern> pattern,
+ boolean includeSysTables) {
+ return this.<List<TableName>> newMasterCaller()
+ .action((controller, stub) -> this
+ .<GetTableNamesRequest, GetTableNamesResponse, List<TableName>> call(controller, stub,
+ RequestConverter.buildGetTableNamesRequest(pattern, includeSysTables),
+ (s, c, req, done) -> s.getTableNames(c, req, done),
+ (resp) -> ProtobufUtil.toTableNameList(resp.getTableNamesList())))
+ .call();
}
@Override
@@ -472,12 +455,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<TableDescriptor[]> deleteTables(String regex) {
- return deleteTables(Pattern.compile(regex));
- }
-
- @Override
- public CompletableFuture<TableDescriptor[]> deleteTables(Pattern pattern) {
+ public CompletableFuture<List<TableDescriptor>> deleteTables(Pattern pattern) {
return batchTableOperations(pattern, (table) -> deleteTable(table), "DELETE");
}
@@ -498,12 +476,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<TableDescriptor[]> enableTables(String regex) {
- return enableTables(Pattern.compile(regex));
- }
-
- @Override
- public CompletableFuture<TableDescriptor[]> enableTables(Pattern pattern) {
+ public CompletableFuture<List<TableDescriptor>> enableTables(Pattern pattern) {
return batchTableOperations(pattern, (table) -> enableTable(table), "ENABLE");
}
@@ -516,16 +489,10 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<TableDescriptor[]> disableTables(String regex) {
- return disableTables(Pattern.compile(regex));
- }
-
- @Override
- public CompletableFuture<TableDescriptor[]> disableTables(Pattern pattern) {
+ public CompletableFuture<List<TableDescriptor>> disableTables(Pattern pattern) {
return batchTableOperations(pattern, (table) -> disableTable(table), "DISABLE");
}
-
@Override
public CompletableFuture<Boolean> isTableEnabled(TableName tableName) {
CompletableFuture<Boolean> future = new CompletableFuture<>();
@@ -577,7 +544,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
if (!enabled) {
future.complete(false);
} else {
- AsyncMetaTableAccessor.getTableRegionsAndLocations(metaTable, Optional.of(tableName))
+ AsyncMetaTableAccessor.getTableHRegionLocations(metaTable, Optional.of(tableName))
.whenComplete(
(locations, error1) -> {
if (error1 != null) {
@@ -586,12 +553,12 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
int notDeployed = 0;
int regionCount = 0;
- for (Pair<HRegionInfo, ServerName> pair : locations) {
- HRegionInfo info = pair.getFirst();
- if (pair.getSecond() == null) {
+ for (HRegionLocation location : locations) {
+ HRegionInfo info = location.getRegionInfo();
+ if (location.getServerName() == null) {
if (LOG.isDebugEnabled()) {
LOG.debug("Table " + tableName + " has not deployed region "
- + pair.getFirst().getEncodedName());
+ + info.getEncodedName());
}
notDeployed++;
} else if (splitKeys != null
@@ -706,21 +673,21 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<NamespaceDescriptor[]> listNamespaceDescriptors() {
+ public CompletableFuture<List<NamespaceDescriptor>> listNamespaceDescriptors() {
return this
- .<NamespaceDescriptor[]> newMasterCaller()
+ .<List<NamespaceDescriptor>> newMasterCaller()
.action(
(controller, stub) -> this
- .<ListNamespaceDescriptorsRequest, ListNamespaceDescriptorsResponse, NamespaceDescriptor[]> call(
+ .<ListNamespaceDescriptorsRequest, ListNamespaceDescriptorsResponse, List<NamespaceDescriptor>> call(
controller, stub, ListNamespaceDescriptorsRequest.newBuilder().build(), (s, c, req,
done) -> s.listNamespaceDescriptors(c, req, done), (resp) -> ProtobufUtil
- .getNamespaceDescriptorArray(resp))).call();
+ .toNamespaceDescriptorList(resp))).call();
}
@Override
- public CompletableFuture<Boolean> setBalancerRunning(final boolean on) {
+ public CompletableFuture<Boolean> setBalancerOn(final boolean on) {
return this
- .<Boolean>newMasterCaller()
+ .<Boolean> newMasterCaller()
.action(
(controller, stub) -> this
.<SetBalancerRunningRequest, SetBalancerRunningResponse, Boolean> call(controller,
@@ -730,24 +697,19 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<Boolean> balancer() {
- return balancer(false);
- }
-
- @Override
- public CompletableFuture<Boolean> balancer(boolean force) {
+ public CompletableFuture<Boolean> balance(boolean forcible) {
return this
- .<Boolean>newMasterCaller()
+ .<Boolean> newMasterCaller()
.action(
(controller, stub) -> this.<BalanceRequest, BalanceResponse, Boolean> call(controller,
- stub, RequestConverter.buildBalanceRequest(force),
+ stub, RequestConverter.buildBalanceRequest(forcible),
(s, c, req, done) -> s.balance(c, req, done), (resp) -> resp.getBalancerRan())).call();
}
@Override
- public CompletableFuture<Boolean> isBalancerEnabled() {
+ public CompletableFuture<Boolean> isBalancerOn() {
return this
- .<Boolean>newMasterCaller()
+ .<Boolean> newMasterCaller()
.action(
(controller, stub) -> this.<IsBalancerEnabledRequest, IsBalancerEnabledResponse, Boolean> call(
controller, stub, RequestConverter.buildIsBalancerEnabledRequest(),
@@ -756,109 +718,38 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<Void> closeRegion(String regionname, String serverName) {
- return closeRegion(Bytes.toBytes(regionname), serverName);
- }
-
- @Override
- public CompletableFuture<Void> closeRegion(byte[] regionName, String serverName) {
- CompletableFuture<Void> future = new CompletableFuture<>();
- getRegion(regionName).whenComplete((p, err) -> {
+ public CompletableFuture<Boolean> closeRegion(byte[] regionName, Optional<ServerName> serverName) {
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ getRegionLocation(regionName).whenComplete((location, err) -> {
if (err != null) {
future.completeExceptionally(err);
return;
}
- if (p == null || p.getFirst() == null) {
- future.completeExceptionally(new UnknownRegionException(Bytes.toStringBinary(regionName)));
- return;
- }
- if (serverName != null) {
- closeRegion(ServerName.valueOf(serverName), p.getFirst()).whenComplete((p2, err2) -> {
+ ServerName server = serverName.isPresent() ? serverName.get() : location.getServerName();
+ if (server == null) {
+ future.completeExceptionally(new NotServingRegionException(regionName));
+ } else {
+ closeRegion(location.getRegionInfo(), server).whenComplete((result, err2) -> {
if (err2 != null) {
future.completeExceptionally(err2);
- }else{
- future.complete(null);
+ } else {
+ future.complete(result);
}
});
- } else {
- if (p.getSecond() == null) {
- future.completeExceptionally(new NotServingRegionException(regionName));
- } else {
- closeRegion(p.getSecond(), p.getFirst()).whenComplete((p2, err2) -> {
- if (err2 != null) {
- future.completeExceptionally(err2);
- }else{
- future.complete(null);
- }
- });
- }
}
});
return future;
}
- CompletableFuture<Pair<HRegionInfo, ServerName>> getRegion(byte[] regionName) {
- if (regionName == null) {
- return failedFuture(new IllegalArgumentException("Pass region name"));
- }
- CompletableFuture<Pair<HRegionInfo, ServerName>> future = new CompletableFuture<>();
- AsyncMetaTableAccessor.getRegion(metaTable, regionName).whenComplete(
- (p, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- } else if (p != null) {
- future.complete(p);
- } else {
- metaTable.scanAll(
- new Scan().setReadType(ReadType.PREAD).addFamily(HConstants.CATALOG_FAMILY))
- .whenComplete((results, err2) -> {
- if (err2 != null) {
- future.completeExceptionally(err2);
- return;
- }
- String encodedName = Bytes.toString(regionName);
- if (results != null && !results.isEmpty()) {
- for (Result r : results) {
- if (r.isEmpty() || MetaTableAccessor.getHRegionInfo(r) == null) continue;
- RegionLocations rl = MetaTableAccessor.getRegionLocations(r);
- if (rl != null) {
- for (HRegionLocation h : rl.getRegionLocations()) {
- if (h != null && encodedName.equals(h.getRegionInfo().getEncodedName())) {
- future.complete(new Pair<>(h.getRegionInfo(), h.getServerName()));
- return;
- }
- }
- }
- }
- }
- future.complete(null);
- });
- }
- });
- return future;
- }
-
- @Override
- public CompletableFuture<Boolean> closeRegionWithEncodedRegionName(String encodedRegionName,
- String serverName) {
+ private CompletableFuture<Boolean> closeRegion(HRegionInfo hri, ServerName serverName) {
return this
.<Boolean> newAdminCaller()
.action(
(controller, stub) -> this.<CloseRegionRequest, CloseRegionResponse, Boolean> adminCall(
controller, stub,
- ProtobufUtil.buildCloseRegionRequest(ServerName.valueOf(serverName), encodedRegionName),
- (s, c, req, done) -> s.closeRegion(controller, req, done), (resp) -> resp.getClosed()))
- .serverName(ServerName.valueOf(serverName)).call();
- }
-
- @Override
- public CompletableFuture<Void> closeRegion(ServerName sn, HRegionInfo hri) {
- return this.<Void> newAdminCaller()
- .action(
- (controller, stub) -> this.<CloseRegionRequest, CloseRegionResponse, Void> adminCall(
- controller, stub, ProtobufUtil.buildCloseRegionRequest(sn, hri.getRegionName()),
- (s, c, req, done) -> s.closeRegion(controller, req, done), resp -> null))
- .serverName(sn).call();
+ ProtobufUtil.buildCloseRegionRequest(serverName, hri.getRegionName()),
+ (s, c, req, done) -> s.closeRegion(controller, req, done), resp -> resp.getClosed()))
+ .serverName(serverName).call();
}
@Override
@@ -905,75 +796,54 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
@Override
public CompletableFuture<Void> flushRegion(byte[] regionName) {
CompletableFuture<Void> future = new CompletableFuture<>();
- getRegion(regionName).whenComplete((p, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- return;
- }
- if (p == null || p.getFirst() == null) {
- future.completeExceptionally(
- new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName)));
- return;
- }
- if (p.getSecond() == null) {
- future.completeExceptionally(
- new NoServerForRegionException(Bytes.toStringBinary(regionName)));
- return;
- }
+ getRegionLocation(regionName).whenComplete(
+ (location, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
+ ServerName serverName = location.getServerName();
+ if (serverName == null) {
+ future.completeExceptionally(new NoServerForRegionException(Bytes
+ .toStringBinary(regionName)));
+ return;
+ }
- this.<Void> newAdminCaller().serverName(p.getSecond())
- .action((controller, stub) -> this
- .<FlushRegionRequest, FlushRegionResponse, Void> adminCall(controller, stub,
- RequestConverter.buildFlushRegionRequest(p.getFirst().getRegionName()),
- (s, c, req, done) -> s.flushRegion(c, req, done), resp -> null))
- .call().whenComplete((ret, err2) -> {
- if (err2 != null) {
- future.completeExceptionally(err2);
- } else {
- future.complete(ret);
- }
- });
- });
+ HRegionInfo regionInfo = location.getRegionInfo();
+ this.<Void> newAdminCaller()
+ .serverName(serverName)
+ .action(
+ (controller, stub) -> this.<FlushRegionRequest, FlushRegionResponse, Void> adminCall(
+ controller, stub, RequestConverter.buildFlushRegionRequest(regionInfo
+ .getRegionName()), (s, c, req, done) -> s.flushRegion(c, req, done),
+ resp -> null)).call().whenComplete((ret, err2) -> {
+ if (err2 != null) {
+ future.completeExceptionally(err2);
+ } else {
+ future.complete(ret);
+ }
+ });
+ });
return future;
}
@Override
- public CompletableFuture<Void> compact(TableName tableName) {
- return compact(tableName, null, false, CompactType.NORMAL);
- }
-
- @Override
- public CompletableFuture<Void> compact(TableName tableName, byte[] columnFamily) {
+ public CompletableFuture<Void> compact(TableName tableName, Optional<byte[]> columnFamily) {
return compact(tableName, columnFamily, false, CompactType.NORMAL);
}
@Override
- public CompletableFuture<Void> compactRegion(byte[] regionName) {
- return compactRegion(regionName, null, false);
- }
-
- @Override
- public CompletableFuture<Void> compactRegion(byte[] regionName, byte[] columnFamily) {
+ public CompletableFuture<Void> compactRegion(byte[] regionName, Optional<byte[]> columnFamily) {
return compactRegion(regionName, columnFamily, false);
}
@Override
- public CompletableFuture<Void> majorCompact(TableName tableName) {
- return compact(tableName, null, true, CompactType.NORMAL);
- }
-
- @Override
- public CompletableFuture<Void> majorCompact(TableName tableName, byte[] columnFamily) {
+ public CompletableFuture<Void> majorCompact(TableName tableName, Optional<byte[]> columnFamily) {
return compact(tableName, columnFamily, true, CompactType.NORMAL);
}
@Override
- public CompletableFuture<Void> majorCompactRegion(byte[] regionName) {
- return compactRegion(regionName, null, true);
- }
-
- @Override
- public CompletableFuture<Void> majorCompactRegion(byte[] regionName, byte[] columnFamily) {
+ public CompletableFuture<Void> majorCompactRegion(byte[] regionName, Optional<byte[]> columnFamily) {
return compactRegion(regionName, columnFamily, true);
}
@@ -996,7 +866,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
List<CompletableFuture<Void>> compactFutures = new ArrayList<>();
if (hRegionInfos != null) {
- hRegionInfos.forEach(region -> compactFutures.add(compact(sn, region, major, null)));
+ hRegionInfos.forEach(region -> compactFutures.add(compact(sn, region, major, Optional.empty())));
}
CompletableFuture
.allOf(compactFutures.toArray(new CompletableFuture<?>[compactFutures.size()]))
@@ -1011,33 +881,30 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
return future;
}
- private CompletableFuture<Void> compactRegion(final byte[] regionName, final byte[] columnFamily,
- final boolean major) {
+ private CompletableFuture<Void> compactRegion(byte[] regionName, Optional<byte[]> columnFamily,
+ boolean major) {
CompletableFuture<Void> future = new CompletableFuture<>();
- getRegion(regionName).whenComplete((p, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- return;
- }
- if (p == null || p.getFirst() == null) {
- future.completeExceptionally(
- new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName)));
- return;
- }
- if (p.getSecond() == null) {
- // found a region without region server assigned.
- future.completeExceptionally(
- new NoServerForRegionException(Bytes.toStringBinary(regionName)));
- return;
- }
- compact(p.getSecond(), p.getFirst(), major, columnFamily).whenComplete((ret, err2) -> {
- if (err2 != null) {
- future.completeExceptionally(err2);
- } else {
- future.complete(ret);
+ getRegionLocation(regionName).whenComplete(
+ (location, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
+ ServerName serverName = location.getServerName();
+ if (serverName == null) {
+ future.completeExceptionally(new NoServerForRegionException(Bytes
+ .toStringBinary(regionName)));
+ return;
}
+ compact(location.getServerName(), location.getRegionInfo(), major, columnFamily)
+ .whenComplete((ret, err2) -> {
+ if (err2 != null) {
+ future.completeExceptionally(err2);
+ } else {
+ future.complete(ret);
+ }
+ });
});
- });
return future;
}
@@ -1045,45 +912,34 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
* List all region locations for the specific table.
*/
private CompletableFuture<List<HRegionLocation>> getTableHRegionLocations(TableName tableName) {
- CompletableFuture<List<HRegionLocation>> future = new CompletableFuture<>();
if (TableName.META_TABLE_NAME.equals(tableName)) {
+ CompletableFuture<List<HRegionLocation>> future = new CompletableFuture<>();
// For meta table, we use zk to fetch all locations.
AsyncRegistry registry = AsyncRegistryFactory.getRegistry(connection.getConfiguration());
- registry.getMetaRegionLocation().whenComplete((metaRegions, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- } else if (metaRegions == null || metaRegions.isEmpty()
- || metaRegions.getDefaultRegionLocation() == null) {
- future.completeExceptionally(new IOException("meta region does not found"));
- } else {
- future.complete(Collections.singletonList(metaRegions.getDefaultRegionLocation()));
- }
- // close the registry.
- IOUtils.closeQuietly(registry);
- });
+ registry.getMetaRegionLocation().whenComplete(
+ (metaRegions, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ } else if (metaRegions == null || metaRegions.isEmpty()
+ || metaRegions.getDefaultRegionLocation() == null) {
+ future.completeExceptionally(new IOException("meta region does not found"));
+ } else {
+ future.complete(Collections.singletonList(metaRegions.getDefaultRegionLocation()));
+ }
+ // close the registry.
+ IOUtils.closeQuietly(registry);
+ });
+ return future;
} else {
// For non-meta table, we fetch all locations by scanning hbase:meta table
- AsyncMetaTableAccessor.getTableRegionsAndLocations(metaTable, Optional.of(tableName))
- .whenComplete((locations, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- } else if (locations == null || locations.isEmpty()) {
- future.complete(Collections.emptyList());
- } else {
- List<HRegionLocation> regionLocations = locations.stream()
- .map(loc -> new HRegionLocation(loc.getFirst(), loc.getSecond()))
- .collect(Collectors.toList());
- future.complete(regionLocations);
- }
- });
+ return AsyncMetaTableAccessor.getTableHRegionLocations(metaTable, Optional.of(tableName));
}
- return future;
}
/**
* Compact column family of a table, Asynchronous operation even if CompletableFuture.get()
*/
- private CompletableFuture<Void> compact(final TableName tableName, final byte[] columnFamily,
+ private CompletableFuture<Void> compact(final TableName tableName, Optional<byte[]> columnFamily,
final boolean major, CompactType compactType) {
if (CompactType.MOB.equals(compactType)) {
// TODO support MOB compact.
@@ -1120,13 +976,15 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
* Compact the region at specific region server.
*/
private CompletableFuture<Void> compact(final ServerName sn, final HRegionInfo hri,
- final boolean major, final byte[] family) {
- return this.<Void> newAdminCaller().serverName(sn)
- .action((controller, stub) -> this
- .<CompactRegionRequest, CompactRegionResponse, Void> adminCall(controller, stub,
- RequestConverter.buildCompactRegionRequest(hri.getRegionName(), major, family),
- (s, c, req, done) -> s.compactRegion(c, req, done), resp -> null))
- .call();
+ final boolean major, Optional<byte[]> columnFamily) {
+ return this
+ .<Void> newAdminCaller()
+ .serverName(sn)
+ .action(
+ (controller, stub) -> this.<CompactRegionRequest, CompactRegionResponse, Void> adminCall(
+ controller, stub, RequestConverter.buildCompactRegionRequest(hri.getRegionName(),
+ major, columnFamily), (s, c, req, done) -> s.compactRegion(c, req, done),
+ resp -> null)).call();
}
private byte[] toEncodeRegionName(byte[] regionName) {
@@ -1140,32 +998,29 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
private void checkAndGetTableName(byte[] encodeRegionName, AtomicReference<TableName> tableName,
CompletableFuture<TableName> result) {
- getRegion(encodeRegionName).whenComplete((p, err) -> {
- if (err != null) {
- result.completeExceptionally(err);
- return;
- }
- if (p == null) {
- result.completeExceptionally(new UnknownRegionException(
- "Can't invoke merge on unknown region " + Bytes.toStringBinary(encodeRegionName)));
- return;
- }
- if (p.getFirst().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
- result.completeExceptionally(
- new IllegalArgumentException("Can't invoke merge on non-default regions directly"));
- return;
- }
- if (!tableName.compareAndSet(null, p.getFirst().getTable())) {
- if (!tableName.get().equals(p.getFirst().getTable())) {
- // tables of this two region should be same.
- result.completeExceptionally(
- new IllegalArgumentException("Cannot merge regions from two different tables "
- + tableName.get() + " and " + p.getFirst().getTable()));
- } else {
- result.complete(tableName.get());
+ getRegionLocation(encodeRegionName).whenComplete(
+ (location, err) -> {
+ if (err != null) {
+ result.completeExceptionally(err);
+ return;
}
- }
- });
+ HRegionInfo regionInfo = location.getRegionInfo();
+ if (regionInfo.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
+ result.completeExceptionally(new IllegalArgumentException(
+ "Can't invoke merge on non-default regions directly"));
+ return;
+ }
+ if (!tableName.compareAndSet(null, regionInfo.getTable())) {
+ if (!tableName.get().equals(regionInfo.getTable())) {
+ // tables of this two region should be same.
+ result.completeExceptionally(new IllegalArgumentException(
+ "Cannot merge regions from two different tables " + tableName.get() + " and "
+ + regionInfo.getTable()));
+ } else {
+ result.complete(tableName.get());
+ }
+ }
+ });
}
private CompletableFuture<TableName> checkRegionsAndGetTableName(byte[] encodeRegionNameA,
@@ -1249,7 +1104,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
if (hri == null || hri.isSplitParent()
|| hri.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID)
continue;
- splitFutures.add(split(h.getServerName(), hri, null));
+ splitFutures.add(split(h.getServerName(), hri, Optional.empty()));
}
}
}
@@ -1272,11 +1127,6 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<Void> splitRegion(byte[] regionName) {
- return splitRegion(regionName, null);
- }
-
- @Override
public CompletableFuture<Void> split(TableName tableName, byte[] splitPoint) {
CompletableFuture<Void> result = new CompletableFuture<>();
if (splitPoint == null) {
@@ -1290,7 +1140,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
result.completeExceptionally(new IllegalArgumentException(
"Region does not found: rowKey=" + Bytes.toStringBinary(splitPoint)));
} else {
- splitRegion(loc.getRegionInfo().getRegionName(), splitPoint)
+ splitRegion(loc.getRegionInfo().getRegionName(), Optional.of(splitPoint))
.whenComplete((ret, err2) -> {
if (err2 != null) {
result.completeExceptionally(err2);
@@ -1305,182 +1155,149 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<Void> splitRegion(byte[] regionName, byte[] splitPoint) {
+ public CompletableFuture<Void> splitRegion(byte[] regionName, Optional<byte[]> splitPoint) {
CompletableFuture<Void> future = new CompletableFuture<>();
- getRegion(regionName).whenComplete((p, err) -> {
- if (p == null) {
- future.completeExceptionally(
- new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName)));
- return;
- }
- if (p.getFirst() != null && p.getFirst().getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
- future.completeExceptionally(new IllegalArgumentException("Can't split replicas directly. "
- + "Replicas are auto-split when their primary is split."));
- return;
- }
- if (p.getSecond() == null) {
- future.completeExceptionally(
- new NoServerForRegionException(Bytes.toStringBinary(regionName)));
- return;
- }
- split(p.getSecond(), p.getFirst(), splitPoint).whenComplete((ret, err2) -> {
- if (err2 != null) {
- future.completeExceptionally(err2);
- } else {
- future.complete(ret);
+ getRegionLocation(regionName).whenComplete(
+ (location, err) -> {
+ HRegionInfo regionInfo = location.getRegionInfo();
+ if (regionInfo.getReplicaId() != HRegionInfo.DEFAULT_REPLICA_ID) {
+ future.completeExceptionally(new IllegalArgumentException(
+ "Can't split replicas directly. "
+ + "Replicas are auto-split when their primary is split."));
+ return;
}
+ ServerName serverName = location.getServerName();
+ if (serverName == null) {
+ future.completeExceptionally(new NoServerForRegionException(Bytes
+ .toStringBinary(regionName)));
+ return;
+ }
+ split(serverName, regionInfo, splitPoint).whenComplete((ret, err2) -> {
+ if (err2 != null) {
+ future.completeExceptionally(err2);
+ } else {
+ future.complete(ret);
+ }
+ });
});
- });
return future;
}
- @VisibleForTesting
- public CompletableFuture<Void> split(final ServerName sn, final HRegionInfo hri,
- byte[] splitPoint) {
- if (hri.getStartKey() != null && splitPoint != null
- && Bytes.compareTo(hri.getStartKey(), splitPoint) == 0) {
- return failedFuture(
- new IllegalArgumentException("should not give a splitkey which equals to startkey!"));
+ private CompletableFuture<Void> split(final ServerName sn, final HRegionInfo hri,
+ Optional<byte[]> splitPoint) {
+ if (hri.getStartKey() != null && splitPoint.isPresent()
+ && Bytes.compareTo(hri.getStartKey(), splitPoint.get()) == 0) {
+ return failedFuture(new IllegalArgumentException(
+ "should not give a splitkey which equals to startkey!"));
}
- return this.<Void> newAdminCaller()
+ return this
+ .<Void> newAdminCaller()
.action(
(controller, stub) -> this.<SplitRegionRequest, SplitRegionResponse, Void> adminCall(
- controller, stub, ProtobufUtil.buildSplitRegionRequest(hri.getRegionName(), splitPoint),
+ controller, stub,
+ ProtobufUtil.buildSplitRegionRequest(hri.getRegionName(), splitPoint),
(s, c, req, done) -> s.splitRegion(controller, req, done), resp -> null))
.serverName(sn).call();
}
- /**
- * Turn regionNameOrEncodedRegionName into regionName, if region does not found, then it'll throw
- * an IllegalArgumentException wrapped by a {@link CompletableFuture}
- * @param regionNameOrEncodedRegionName
- * @return
- */
- CompletableFuture<byte[]> getRegionName(byte[] regionNameOrEncodedRegionName) {
- CompletableFuture<byte[]> future = new CompletableFuture<>();
- if (Bytes
- .equals(regionNameOrEncodedRegionName, HRegionInfo.FIRST_META_REGIONINFO.getRegionName())
- || Bytes.equals(regionNameOrEncodedRegionName,
- HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes())) {
- future.complete(HRegionInfo.FIRST_META_REGIONINFO.getRegionName());
- return future;
- }
-
- getRegion(regionNameOrEncodedRegionName).whenComplete((p, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- }
- if (p != null && p.getFirst() != null) {
- future.complete(p.getFirst().getRegionName());
- } else {
- future.completeExceptionally(
- new IllegalArgumentException("Invalid region name or encoded region name: "
- + Bytes.toStringBinary(regionNameOrEncodedRegionName)));
- }
- });
- return future;
- }
-
@Override
public CompletableFuture<Void> assign(byte[] regionName) {
CompletableFuture<Void> future = new CompletableFuture<>();
- getRegionName(regionName).whenComplete((fullRegionName, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- } else {
+ getRegionInfo(regionName).whenComplete(
+ (regionInfo, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
this.<Void> newMasterCaller()
.action(
((controller, stub) -> this.<AssignRegionRequest, AssignRegionResponse, Void> call(
- controller, stub, RequestConverter.buildAssignRegionRequest(fullRegionName),
- (s, c, req, done) -> s.assignRegion(c, req, done), resp -> null)))
- .call().whenComplete((ret, err2) -> {
+ controller, stub, RequestConverter.buildAssignRegionRequest(regionInfo
+ .getRegionName()), (s, c, req, done) -> s.assignRegion(c, req, done),
+ resp -> null))).call().whenComplete((ret, err2) -> {
if (err2 != null) {
future.completeExceptionally(err2);
} else {
future.complete(ret);
}
});
- }
- });
+ });
return future;
}
@Override
- public CompletableFuture<Void> unassign(byte[] regionName, boolean force) {
+ public CompletableFuture<Void> unassign(byte[] regionName, boolean forcible) {
CompletableFuture<Void> future = new CompletableFuture<>();
- getRegionName(regionName).whenComplete((fullRegionName, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- } else {
+ getRegionInfo(regionName).whenComplete(
+ (regionInfo, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
this.<Void> newMasterCaller()
- .action(((controller, stub) -> this
- .<UnassignRegionRequest, UnassignRegionResponse, Void> call(controller, stub,
- RequestConverter.buildUnassignRegionRequest(fullRegionName, force),
- (s, c, req, done) -> s.unassignRegion(c, req, done), resp -> null)))
- .call().whenComplete((ret, err2) -> {
+ .action(
+ ((controller, stub) -> this
+ .<UnassignRegionRequest, UnassignRegionResponse, Void> call(controller, stub,
+ RequestConverter.buildUnassignRegionRequest(regionInfo.getRegionName(), forcible),
+ (s, c, req, done) -> s.unassignRegion(c, req, done), resp -> null))).call()
+ .whenComplete((ret, err2) -> {
if (err2 != null) {
future.completeExceptionally(err2);
} else {
future.complete(ret);
}
});
- }
- });
+ });
return future;
}
@Override
public CompletableFuture<Void> offline(byte[] regionName) {
CompletableFuture<Void> future = new CompletableFuture<>();
- getRegionName(regionName).whenComplete((fullRegionName, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- } else {
+ getRegionInfo(regionName).whenComplete(
+ (regionInfo, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
this.<Void> newMasterCaller()
.action(
((controller, stub) -> this.<OfflineRegionRequest, OfflineRegionResponse, Void> call(
- controller, stub, RequestConverter.buildOfflineRegionRequest(fullRegionName),
- (s, c, req, done) -> s.offlineRegion(c, req, done), resp -> null)))
- .call().whenComplete((ret, err2) -> {
+ controller, stub, RequestConverter.buildOfflineRegionRequest(regionInfo
+ .getRegionName()), (s, c, req, done) -> s.offlineRegion(c, req, done),
+ resp -> null))).call().whenComplete((ret, err2) -> {
if (err2 != null) {
future.completeExceptionally(err2);
} else {
future.complete(ret);
}
});
- }
- });
+ });
return future;
}
@Override
- public CompletableFuture<Void> move(byte[] regionName, byte[] destServerName) {
+ public CompletableFuture<Void> move(byte[] regionName, Optional<ServerName> destServerName) {
CompletableFuture<Void> future = new CompletableFuture<>();
- getRegionName(regionName).whenComplete((fullRegionName, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- } else {
- final MoveRegionRequest request;
- try {
- request = RequestConverter.buildMoveRegionRequest(
- Bytes.toBytes(HRegionInfo.encodeRegionName(fullRegionName)), destServerName);
- } catch (DeserializationException e) {
- future.completeExceptionally(e);
+ getRegionInfo(regionName).whenComplete(
+ (regionInfo, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
return;
}
this.<Void> newMasterCaller()
- .action((controller, stub) -> this.<MoveRegionRequest, MoveRegionResponse, Void> call(
- controller, stub, request, (s, c, req, done) -> s.moveRegion(c, req, done),
- resp -> null))
- .call().whenComplete((ret, err2) -> {
+ .action(
+ (controller, stub) -> this.<MoveRegionRequest, MoveRegionResponse, Void> call(
+ controller, stub, RequestConverter.buildMoveRegionRequest(
+ regionInfo.getEncodedNameAsBytes(), destServerName), (s, c, req, done) -> s
+ .moveRegion(c, req, done), resp -> null)).call().whenComplete((ret, err2) -> {
if (err2 != null) {
future.completeExceptionally(err2);
} else {
future.complete(ret);
}
});
- }
- });
+ });
return future;
}
@@ -1644,17 +1461,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers() {
- return listReplicationPeers((Pattern) null);
- }
-
- @Override
- public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(String regex) {
- return listReplicationPeers(Pattern.compile(regex));
- }
-
- @Override
- public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Pattern pattern) {
+ public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Optional<Pattern> pattern) {
return this
.<List<ReplicationPeerDescription>> newMasterCaller()
.action(
@@ -1676,18 +1483,17 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
(tables, error) -> {
if (!completeExceptionally(future, error)) {
List<TableCFs> replicatedTableCFs = new ArrayList<>();
- Arrays.asList(tables).forEach(
- table -> {
- Map<String, Integer> cfs = new HashMap<>();
- Stream.of(table.getColumnFamilies())
- .filter(column -> column.getScope() != HConstants.REPLICATION_SCOPE_LOCAL)
- .forEach(column -> {
- cfs.put(column.getNameAsString(), column.getScope());
- });
- if (!cfs.isEmpty()) {
- replicatedTableCFs.add(new TableCFs(table.getTableName(), cfs));
- }
- });
+ tables.forEach(table -> {
+ Map<String, Integer> cfs = new HashMap<>();
+ Stream.of(table.getColumnFamilies())
+ .filter(column -> column.getScope() != HConstants.REPLICATION_SCOPE_LOCAL)
+ .forEach(column -> {
+ cfs.put(column.getNameAsString(), column.getScope());
+ });
+ if (!cfs.isEmpty()) {
+ replicatedTableCFs.add(new TableCFs(table.getTableName(), cfs));
+ }
+ });
future.complete(replicatedTableCFs);
}
});
@@ -1707,8 +1513,8 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
@Override
public CompletableFuture<Void> snapshot(SnapshotDescription snapshotDesc) {
- SnapshotProtos.SnapshotDescription snapshot =
- ProtobufUtil.createHBaseProtosSnapshotDesc(snapshotDesc);
+ SnapshotProtos.SnapshotDescription snapshot = ProtobufUtil
+ .createHBaseProtosSnapshotDesc(snapshotDesc);
try {
ClientSnapshotDescriptionUtils.assertSnapshotRequestIsValid(snapshot);
} catch (IllegalArgumentException e) {
@@ -1717,10 +1523,10 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
CompletableFuture<Void> future = new CompletableFuture<>();
final SnapshotRequest request = SnapshotRequest.newBuilder().setSnapshot(snapshot).build();
this.<Long> newMasterCaller()
- .action((controller, stub) -> this.<SnapshotRequest, SnapshotResponse, Long> call(
- controller, stub, request, (s, c, req, done) -> s.snapshot(c, req, done),
- resp -> resp.getExpectedTimeout()))
- .call().whenComplete((expectedTimeout, err) -> {
+ .action(
+ (controller, stub) -> this.<SnapshotRequest, SnapshotResponse, Long> call(controller,
+ stub, request, (s, c, req, done) -> s.snapshot(c, req, done),
+ resp -> resp.getExpectedTimeout())).call().whenComplete((expectedTimeout, err) -> {
if (err != null) {
future.completeExceptionally(err);
return;
@@ -1734,25 +1540,24 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
@Override
public void run(Timeout timeout) throws Exception {
if (EnvironmentEdgeManager.currentTime() < endTime) {
- isSnapshotFinished(snapshotDesc).whenComplete((done, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
+ isSnapshotFinished(snapshotDesc).whenComplete((done, err2) -> {
+ if (err2 != null) {
+ future.completeExceptionally(err2);
} else if (done) {
future.complete(null);
} else {
// retry again after pauseTime.
- long pauseTime = ConnectionUtils
- .getPauseTime(TimeUnit.NANOSECONDS.toMillis(pauseNs), ++tries);
- pauseTime = Math.min(pauseTime, maxPauseTime);
- AsyncConnectionImpl.RETRY_TIMER.newTimeout(this, pauseTime,
- TimeUnit.MILLISECONDS);
- }
- });
+ long pauseTime = ConnectionUtils.getPauseTime(
+ TimeUnit.NANOSECONDS.toMillis(pauseNs), ++tries);
+ pauseTime = Math.min(pauseTime, maxPauseTime);
+ AsyncConnectionImpl.RETRY_TIMER
+ .newTimeout(this, pauseTime, TimeUnit.MILLISECONDS);
+ }
+ } );
} else {
- future.completeExceptionally(new SnapshotCreationException(
- "Snapshot '" + snapshot.getName() + "' wasn't completed in expectedTime:"
- + expectedTimeout + " ms",
- snapshotDesc));
+ future.completeExceptionally(new SnapshotCreationException("Snapshot '"
+ + snapshot.getName() + "' wasn't completed in expectedTime:" + expectedTimeout
+ + " ms", snapshotDesc));
}
}
};
@@ -1763,13 +1568,15 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
@Override
public CompletableFuture<Boolean> isSnapshotFinished(SnapshotDescription snapshot) {
- return this.<Boolean> newMasterCaller()
- .action((controller, stub) -> this
- .<IsSnapshotDoneRequest, IsSnapshotDoneResponse, Boolean> call(controller, stub,
- IsSnapshotDoneRequest.newBuilder()
- .setSnapshot(ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot)).build(),
- (s, c, req, done) -> s.isSnapshotDone(c, req, done), resp -> resp.getDone()))
- .call();
+ return this
+ .<Boolean> newMasterCaller()
+ .action(
+ (controller, stub) -> this.<IsSnapshotDoneRequest, IsSnapshotDoneResponse, Boolean> call(
+ controller,
+ stub,
+ IsSnapshotDoneRequest.newBuilder()
+ .setSnapshot(ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot)).build(), (s, c,
+ req, done) -> s.isSnapshotDone(c, req, done), resp -> resp.getDone())).call();
}
@Override
@@ -1780,109 +1587,110 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
return restoreSnapshot(snapshotName, takeFailSafeSnapshot);
}
- private CompletableFuture<Void> restoreSnapshotWithFailSafe(String snapshotName,
- TableName tableName, boolean takeFailSafeSnapshot) {
+ @Override
+ public CompletableFuture<Void> restoreSnapshot(String snapshotName, boolean takeFailSafeSnapshot) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ listSnapshots(Pattern.compile(snapshotName)).whenComplete(
+ (snapshotDescriptions, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
+ TableName tableName = null;
+ if (snapshotDescriptions != null && !snapshotDescriptions.isEmpty()) {
+ for (SnapshotDescription snap : snapshotDescriptions) {
+ if (snap.getName().equals(snapshotName)) {
+ tableName = snap.getTableName();
+ break;
+ }
+ }
+ }
+ if (tableName == null) {
+ future.completeExceptionally(new RestoreSnapshotException(
+ "Unable to find the table name for snapshot=" + snapshotName));
+ return;
+ }
+ final TableName finalTableName = tableName;
+ tableExists(finalTableName)
+ .whenComplete((exists, err2) -> {
+ if (err2 != null) {
+ future.completeExceptionally(err2);
+ } else if (!exists) {
+ // if table does not exist, then just clone snapshot into new table.
+ completeConditionalOnFuture(future,
+ internalRestoreSnapshot(snapshotName, finalTableName));
+ } else {
+ isTableDisabled(finalTableName).whenComplete(
+ (disabled, err4) -> {
+ if (err4 != null) {
+ future.completeExceptionally(err4);
+ } else if (!disabled) {
+ future.completeExceptionally(new TableNotDisabledException(finalTableName));
+ } else {
+ completeConditionalOnFuture(future,
+ restoreSnapshot(snapshotName, finalTableName, takeFailSafeSnapshot));
+ }
+ });
+ }
+ } );
+ });
+ return future;
+ }
+
+ private CompletableFuture<Void> restoreSnapshot(String snapshotName, TableName tableName,
+ boolean takeFailSafeSnapshot) {
if (takeFailSafeSnapshot) {
CompletableFuture<Void> future = new CompletableFuture<>();
// Step.1 Take a snapshot of the current state
- String failSafeSnapshotSnapshotNameFormat =
- this.connection.getConfiguration().get(HConstants.SNAPSHOT_RESTORE_FAILSAFE_NAME,
- HConstants.DEFAULT_SNAPSHOT_RESTORE_FAILSAFE_NAME);
- final String failSafeSnapshotSnapshotName =
- failSafeSnapshotSnapshotNameFormat.replace("{snapshot.name}", snapshotName)
- .replace("{table.name}", tableName.toString().replace(TableName.NAMESPACE_DELIM, '.'))
- .replace("{restore.timestamp}", String.valueOf(EnvironmentEdgeManager.currentTime()));
+ String failSafeSnapshotSnapshotNameFormat = this.connection.getConfiguration().get(
+ HConstants.SNAPSHOT_RESTORE_FAILSAFE_NAME,
+ HConstants.DEFAULT_SNAPSHOT_RESTORE_FAILSAFE_NAME);
+ final String failSafeSnapshotSnapshotName = failSafeSnapshotSnapshotNameFormat
+ .replace("{snapshot.name}", snapshotName)
+ .replace("{table.name}", tableName.toString().replace(TableName.NAMESPACE_DELIM, '.'))
+ .replace("{restore.timestamp}", String.valueOf(EnvironmentEdgeManager.currentTime()));
LOG.info("Taking restore-failsafe snapshot: " + failSafeSnapshotSnapshotName);
snapshot(failSafeSnapshotSnapshotName, tableName).whenComplete((ret, err) -> {
if (err != null) {
future.completeExceptionally(err);
} else {
// Step.2 Restore snapshot
- internalRestoreSnapshot(snapshotName, tableName).whenComplete((void2, err2) -> {
- if (err2 != null) {
- // Step.3.a Something went wrong during the restore and try to rollback.
- internalRestoreSnapshot(failSafeSnapshotSnapshotName, tableName)
- .whenComplete((void3, err3) -> {
- if (err3 != null) {
- future.completeExceptionally(err3);
- } else {
- String msg =
- "Restore snapshot=" + snapshotName + " failed. Rollback to snapshot="
- + failSafeSnapshotSnapshotName + " succeeded.";
- future.completeExceptionally(new RestoreSnapshotException(msg));
- }
- });
- } else {
- // Step.3.b If the restore is succeeded, delete the pre-restore snapshot.
- LOG.info("Deleting restore-failsafe snapshot: " + failSafeSnapshotSnapshotName);
- deleteSnapshot(failSafeSnapshotSnapshotName).whenComplete((ret3, err3) -> {
- if (err3 != null) {
- LOG.error(
- "Unable to remove the failsafe snapshot: " + failSafeSnapshotSnapshotName,
- err3);
- future.completeExceptionally(err3);
- } else {
- future.complete(ret3);
- }
- });
- }
- });
+ internalRestoreSnapshot(snapshotName, tableName).whenComplete((void2, err2) -> {
+ if (err2 != null) {
+ // Step.3.a Something went wrong during the restore and try to rollback.
+ internalRestoreSnapshot(failSafeSnapshotSnapshotName, tableName).whenComplete(
+ (void3, err3) -> {
+ if (err3 != null) {
+ future.completeExceptionally(err3);
+ } else {
+ String msg = "Restore snapshot=" + snapshotName + " failed. Rollback to snapshot="
+ + failSafeSnapshotSnapshotName + " succeeded.";
+ future.completeExceptionally(new RestoreSnapshotException(msg));
+ }
+ });
+ } else {
+ // Step.3.b If the restore is succeeded, delete the pre-restore snapshot.
+ LOG.info("Deleting restore-failsafe snapshot: " + failSafeSnapshotSnapshotName);
+ deleteSnapshot(failSafeSnapshotSnapshotName).whenComplete(
+ (ret3, err3) -> {
+ if (err3 != null) {
+ LOG.error(
+ "Unable to remove the failsafe snapshot: " + failSafeSnapshotSnapshotName, err3);
+ future.completeExceptionally(err3);
+ } else {
+ future.complete(ret3);
+ }
+ });
}
- });
+ } );
+ }
+ } );
return future;
} else {
return internalRestoreSnapshot(snapshotName, tableName);
}
}
- @Override
- public CompletableFuture<Void> restoreSnapshot(String snapshotName,
- boolean takeFailSafeSnapshot) {
- CompletableFuture<Void> future = new CompletableFuture<>();
- listSnapshots(snapshotName).whenComplete((snapshotDescriptions, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- return;
- }
- TableName tableName = null;
- if (snapshotDescriptions != null && !snapshotDescriptions.isEmpty()) {
- for (SnapshotDescription snap : snapshotDescriptions) {
- if (snap.getName().equals(snapshotName)) {
- tableName = snap.getTableName();
- break;
- }
- }
- }
- if (tableName == null) {
- future.completeExceptionally(new RestoreSnapshotException(
- "Unable to find the table name for snapshot=" + snapshotName));
- return;
- }
- final TableName finalTableName = tableName;
- tableExists(finalTableName).whenComplete((exists, err2) -> {
- if (err2 != null) {
- future.completeExceptionally(err2);
- } else if (!exists) {
- // if table does not exist, then just clone snapshot into new table.
- completeConditionalOnFuture(future,
- internalRestoreSnapshot(snapshotName, finalTableName));
- } else {
- isTableDisabled(finalTableName).whenComplete((disabled, err4) -> {
- if (err4 != null) {
- future.completeExceptionally(err4);
- } else if (!disabled) {
- future.completeExceptionally(new TableNotDisabledException(finalTableName));
- } else {
- completeConditionalOnFuture(future,
- restoreSnapshotWithFailSafe(snapshotName, finalTableName, takeFailSafeSnapshot));
- }
- });
- }
- });
- });
- return future;
- }
-
private <T> void completeConditionalOnFuture(CompletableFuture<T> dependentFuture,
CompletableFuture<T> parentFuture) {
parentFuture.whenComplete((res, err) -> {
@@ -1909,8 +1717,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
return future;
}
- private CompletableFuture<Void> internalRestoreSnapshot(String snapshotName,
- TableName tableName) {
+ private CompletableFuture<Void> internalRestoreSnapshot(String snapshotName, TableName tableName) {
SnapshotProtos.SnapshotDescription snapshot = SnapshotProtos.SnapshotDescription.newBuilder()
.setName(snapshotName).setTable(tableName.getNameAsString()).build();
try {
@@ -1918,86 +1725,78 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
} catch (IllegalArgumentException e) {
return failedFuture(e);
}
- return waitProcedureResult(
- this.<Long> newMasterCaller()
- .action((controller, stub) -> this
- .<RestoreSnapshotRequest, RestoreSnapshotResponse, Long> call(controller, stub,
- RestoreSnapshotRequest.newBuilder().setSnapshot(snapshot)
- .setNonceGroup(ng.getNonceGroup()).setNonce(ng.newNonce()).build(),
- (s, c, req, done) -> s.restoreSnapshot(c, req, done),
- (resp) -> resp.getProcId()))
- .call());
+ return waitProcedureResult(this
+ .<Long> newMasterCaller()
+ .action(
+ (controller, stub) -> this.<RestoreSnapshotRequest, RestoreSnapshotResponse, Long> call(
+ controller, stub, RestoreSnapshotRequest.newBuilder().setSnapshot(snapshot)
+ .setNonceGroup(ng.getNonceGroup()).setNonce(ng.newNonce()).build(), (s, c, req,
+ done) -> s.restoreSnapshot(c, req, done), (resp) -> resp.getProcId())).call());
}
@Override
public CompletableFuture<List<SnapshotDescription>> listSnapshots() {
- return this.<List<SnapshotDescription>> newMasterCaller()
- .action((controller, stub) -> this
- .<GetCompletedSnapshotsRequest, GetCompletedSnapshotsResponse, List<SnapshotDescription>> call(
- controller, stub, GetCompletedSnapshotsRequest.newBuilder().build(),
- (s, c, req, done) -> s.getCompletedSnapshots(c, req, done),
- resp -> resp.getSnapshotsList().stream().map(ProtobufUtil::createSnapshotDesc)
- .collect(Collectors.toList())))
+ return this
+ .<List<SnapshotDescription>> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<GetCompletedSnapshotsRequest, GetCompletedSnapshotsResponse, List<SnapshotDescription>> call(
+ controller, stub, GetCompletedSnapshotsRequest.newBuilder().build(), (s, c, req,
+ done) -> s.getCompletedSnapshots(c, req, done), resp -> resp.getSnapshotsList()
+ .stream().map(ProtobufUtil::createSnapshotDesc).collect(Collectors.toList())))
.call();
}
@Override
- public CompletableFuture<List<SnapshotDescription>> listSnapshots(String regex) {
- return listSnapshots(Pattern.compile(regex));
- }
-
- @Override
public CompletableFuture<List<SnapshotDescription>> listSnapshots(Pattern pattern) {
CompletableFuture<List<SnapshotDescription>> future = new CompletableFuture<>();
- listSnapshots().whenComplete((snapshotDescList, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- return;
- }
- if (snapshotDescList == null || snapshotDescList.isEmpty()) {
- future.complete(Collections.emptyList());
- return;
- }
- future.complete(snapshotDescList.stream()
- .filter(snap -> pattern.matcher(snap.getName()).matches()).collect(Collectors.toList()));
- });
+ listSnapshots()
+ .whenComplete(
+ (snapshotDescList, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
+ if (snapshotDescList == null || snapshotDescList.isEmpty()) {
+ future.complete(Collections.emptyList());
+ return;
+ }
+ future.complete(snapshotDescList.stream()
+ .filter(snap -> pattern.matcher(snap.getName()).matches())
+ .collect(Collectors.toList()));
+ });
return future;
}
@Override
- public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(String tableNameRegex,
- String snapshotNameRegex) {
- return listTableSnapshots(Pattern.compile(tableNameRegex), Pattern.compile(snapshotNameRegex));
- }
-
- @Override
public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern,
Pattern snapshotNamePattern) {
CompletableFuture<List<SnapshotDescription>> future = new CompletableFuture<>();
- listTableNames(tableNamePattern, false).whenComplete((tableNames, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- return;
- }
- if (tableNames == null || tableNames.length <= 0) {
- future.complete(Collections.emptyList());
- return;
- }
- List<TableName> tableNameList = Arrays.asList(tableNames);
- listSnapshots(snapshotNamePattern).whenComplete((snapshotDescList, err2) -> {
- if (err2 != null) {
- future.completeExceptionally(err2);
+ listTableNames(Optional.ofNullable(tableNamePattern), false).whenComplete(
+ (tableNames, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
return;
}
- if (snapshotDescList == null || snapshotDescList.isEmpty()) {
+ if (tableNames == null || tableNames.size() <= 0) {
future.complete(Collections.emptyList());
return;
}
- future.complete(snapshotDescList.stream()
- .filter(snap -> (snap != null && tableNameList.contains(snap.getTableName())))
- .collect(Collectors.toList()));
+ listSnapshots(snapshotNamePattern).whenComplete(
+ (snapshotDescList, err2) -> {
+ if (err2 != null) {
+ future.completeExceptionally(err2);
+ return;
+ }
+ if (snapshotDescList == null || snapshotDescList.isEmpty()) {
+ future.complete(Collections.emptyList());
+ return;
+ }
+ future.complete(snapshotDescList.stream()
+ .filter(snap -> (snap != null && tableNames.contains(snap.getTableName())))
+ .collect(Collectors.toList()));
+ });
});
- });
return future;
}
@@ -2007,47 +1806,46 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<Void> deleteSnapshots(String regex) {
- return deleteSnapshots(Pattern.compile(regex));
- }
-
- @Override
public CompletableFuture<Void> deleteSnapshots(Pattern snapshotNamePattern) {
return deleteTableSnapshots(null, snapshotNamePattern);
}
@Override
- public CompletableFuture<Void> deleteTableSnapshots(String tableNameRegex,
- String snapshotNameRegex) {
- return deleteTableSnapshots(Pattern.compile(tableNameRegex),
- Pattern.compile(snapshotNameRegex));
- }
-
- @Override
public CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern,
Pattern snapshotNamePattern) {
CompletableFuture<Void> future = new CompletableFuture<>();
- listTableSnapshots(tableNamePattern, snapshotNamePattern)
- .whenComplete(((snapshotDescriptions, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- return;
- }
- if (snapshotDescriptions == null || snapshotDescriptions.isEmpty()) {
- future.complete(null);
- return;
- }
- List<CompletableFuture<Void>> deleteSnapshotFutures = new ArrayList<>();
- snapshotDescriptions
- .forEach(snapDesc -> deleteSnapshotFutures.add(internalDeleteSnapshot(snapDesc)));
- CompletableFuture
- .allOf(deleteSnapshotFutures
- .toArray(new CompletableFuture<?>[deleteSnapshotFutures.size()]))
- .thenAccept(v -> future.complete(v));
- }));
+ listTableSnapshots(tableNamePattern, snapshotNamePattern).whenComplete(
+ ((snapshotDescriptions, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
+ if (snapshotDescriptions == null || snapshotDescriptions.isEmpty()) {
+ future.complete(null);
+ return;
+ }
+ List<CompletableFuture<Void>> deleteSnapshotFutures = new ArrayList<>();
+ snapshotDescriptions.forEach(snapDesc -> deleteSnapshotFutures
+ .add(internalDeleteSnapshot(snapDesc)));
+ CompletableFuture.allOf(
+ deleteSnapshotFutures.toArray(new CompletableFuture<?>[deleteSnapshotFutures.size()]))
+ .thenAccept(v -> future.complete(v));
+ }));
return future;
}
+ private CompletableFuture<Void> internalDeleteSnapshot(SnapshotDescription snapshot) {
+ return this
+ .<Void> newMasterCaller()
+ .action(
+ (controller, stub) -> this.<DeleteSnapshotRequest, DeleteSnapshotResponse, Void> call(
+ controller,
+ stub,
+ DeleteSnapshotRequest.newBuilder()
+ .setSnapshot(ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot)).build(), (s, c,
+ req, done) -> s.deleteSnapshot(c, req, done), resp -> null)).call();
+ }
+
@Override
public CompletableFuture<Void> execProcedure(String signature, String instance,
Map<String, String> props) {
@@ -2072,9 +1870,9 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
@Override
public void run(Timeout timeout) throws Exception {
if (EnvironmentEdgeManager.currentTime() < endTime) {
- isProcedureFinished(signature, instance, props).whenComplete((done, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
+ isProcedureFinished(signature, instance, props).whenComplete((done, err2) -> {
+ if (err2 != null) {
+ future.completeExceptionally(err2);
return;
}
if (done) {
@@ -2137,24 +1935,87 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<ProcedureInfo[]> listProcedures() {
- return this.<ProcedureInfo[]> newMasterCaller()
- .action((controller, stub) -> this
- .<ListProceduresRequest, ListProceduresResponse, ProcedureInfo[]> call(controller, stub,
- ListProceduresRequest.newBuilder().build(),
- (s, c, req, done) -> s.listProcedures(c, req, done), resp -> resp.getProcedureList()
- .stream().map(ProtobufUtil::toProcedureInfo).toArray(ProcedureInfo[]::new)))
- .call();
+ public CompletableFuture<List<ProcedureInfo>> listProcedures() {
+ return this
+ .<List<ProcedureInfo>> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<ListProceduresRequest, ListProceduresResponse, List<ProcedureInfo>> call(
+ controller, stub, ListProceduresRequest.newBuilder().build(),
+ (s, c, req, done) -> s.listProcedures(c, req, done),
+ resp -> resp.getProcedureList().stream().map(ProtobufUtil::toProcedureInfo)
+ .collect(Collectors.toList()))).call();
}
- private CompletableFuture<Void> internalDeleteSnapshot(SnapshotDescription snapshot) {
- return this.<Void> newMasterCaller()
- .action((controller, stub) -> this
- .<DeleteSnapshotRequest, DeleteSnapshotResponse, Void> call(controller, stub,
- DeleteSnapshotRequest.newBuilder()
- .setSnapshot(ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot)).build(),
- (s, c, req, done) -> s.deleteSnapshot(c, req, done), resp -> null))
- .call();
+ /**
+ * Get the region location for the passed region name. The region name may be a full region name
+ * or encoded region name. If the region does not found, then it'll throw an
+ * UnknownRegionException wrapped by a {@link CompletableFuture}
+ * @param regionNameOrEncodedRegionName
+ * @return region location, wrapped by a {@link CompletableFuture}
+ */
+ @VisibleForTesting
+ CompletableFuture<HRegionLocation> getRegionLocation(byte[] regionNameOrEncodedRegionName) {
+ if (regionNameOrEncodedRegionName == null) {
+ return failedFuture(new IllegalArgumentException("Passed region name can't be null"));
+ }
+ try {
+ CompletableFuture<Optional<HRegionLocation>> future;
+ if (HRegionInfo.isEncodedRegionName(regionNameOrEncodedRegionName)) {
+ future = AsyncMetaTableAccessor.getRegionLocationWithEncodedName(metaTable,
+ regionNameOrEncodedRegionName);
+ } else {
+ future = AsyncMetaTableAccessor.getRegionLocation(metaTable, regionNameOrEncodedRegionName);
+ }
+
+ CompletableFuture<HRegionLocation> returnedFuture = new CompletableFuture<>();
+ future.whenComplete((location, err) -> {
+ if (err != null) {
+ returnedFuture.completeExceptionally(err);
+ return;
+ }
+ if (!location.isPresent() || location.get().getRegionInfo() == null) {
+ returnedFuture.completeExceptionally(new UnknownRegionException(
+ "Invalid region name or encoded region name: "
+ + Bytes.toStringBinary(regionNameOrEncodedRegionName)));
+ } else {
+ returnedFuture.complete(location.get());
+ }
+ });
+ return returnedFuture;
+ } catch (IOException e) {
+ return failedFuture(e);
+ }
+ }
+
+ /**
+ * Get the region info for the passed region name. The region name may be a full region name or
+ * encoded region name. If the region does not found, then it'll throw an UnknownRegionException
+ * wrapped by a {@link CompletableFuture}
+ * @param regionNameOrEncodedRegionName
+ * @return region info, wrapped by a {@link CompletableFuture}
+ */
+ private CompletableFuture<HRegionInfo> getRegionInfo(byte[] regionNameOrEncodedRegionName) {
+ if (regionNameOrEncodedRegionName == null) {
+ return failedFuture(new IllegalArgumentException("Passed region name can't be null"));
+ }
+
+ if (Bytes.equals(regionNameOrEncodedRegionName,
+ HRegionInfo.FIRST_META_REGIONINFO.getRegionName())
+ || Bytes.equals(regionNameOrEncodedRegionName,
+ HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes())) {
+ return CompletableFuture.completedFuture(HRegionInfo.FIRST_META_REGIONINFO);
+ }
+
+ CompletableFuture<HRegionInfo> future = new CompletableFuture<>();
+ getRegionLocation(regionNameOrEncodedRegionName).whenComplete((location, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ } else {
+ future.complete(location.getRegionInfo());
+ }
+ });
+ return future;
}
private byte[][] getSplitKeys(byte[] startKey, byte[] endKey, int numRegions) {
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
index b196911..5f8924f 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ProtobufUtil.java
@@ -33,8 +33,10 @@ import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.NavigableSet;
+import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
@@ -414,18 +416,14 @@ public final class ProtobufUtil {
}
/**
- * Get NamespaceDescriptor[] from ListNamespaceDescriptorsResponse protobuf
+ * Get a list of NamespaceDescriptor from ListNamespaceDescriptorsResponse protobuf
* @param proto the ListNamespaceDescriptorsResponse
- * @return NamespaceDescriptor[]
+ * @return a list of NamespaceDescriptor
*/
- public static NamespaceDescriptor[] getNamespaceDescriptorArray(
+ public static List<NamespaceDescriptor> toNamespaceDescriptorList(
ListNamespaceDescriptorsResponse proto) {
- List<HBaseProtos.NamespaceDescriptor> list = proto.getNamespaceDescriptorList();
- NamespaceDescriptor[] res = new NamespaceDescriptor[list.size()];
- for (int i = 0; i < list.size(); i++) {
- res[i] = ProtobufUtil.toNamespaceDescriptor(list.get(i));
- }
- return res;
+ return proto.getNamespaceDescriptorList().stream().map(ProtobufUtil::toNamespaceDescriptor)
+ .collect(Collectors.toList());
}
/**
@@ -433,7 +431,7 @@ public final class ProtobufUtil {
*
* @param proto the GetTableDescriptorsResponse
* @return a immutable HTableDescriptor array
- * @deprecated Use {@link #getTableDescriptorArray} after removing the HTableDescriptor
+ * @deprecated Use {@link #toTableDescriptorList} after removing the HTableDescriptor
*/
@Deprecated
public static HTableDescriptor[] getHTableDescriptorArray(GetTableDescriptorsResponse proto) {
@@ -447,18 +445,17 @@ public final class ProtobufUtil {
}
/**
- * Get TableDescriptor[] from GetTableDescriptorsResponse protobuf
+ * Get a list of TableDescriptor from GetTableDescriptorsResponse protobuf
*
* @param proto the GetTableDescriptorsResponse
- * @return TableDescriptor[]
+ * @return a list of TableDescriptor
*/
- public static TableDescriptor[] getTableDescriptorArray(GetTableDescriptorsResponse proto) {
- if (proto == null) return new TableDescriptor[0];
- return proto.getTableSchemaList()
- .stream()
- .map(ProtobufUtil::convertToTableDesc)
- .toArray(size -> new TableDescriptor[size]);
+ public static List<TableDescriptor> toTableDescriptorList(GetTableDescriptorsResponse proto) {
+ if (proto == null) return new ArrayList<>();
+ return proto.getTableSchemaList().stream().map(ProtobufUtil::convertToTableDesc)
+ .collect(Collectors.toList());
}
+
/**
* get the split keys in form "byte [][]" from a CreateTableRequest proto
*
@@ -2398,6 +2395,13 @@ public final class ProtobufUtil {
.setQualifier(UnsafeByteOperations.unsafeWrap(tableName.getQualifier())).build();
}
+ public static List<TableName> toTableNameList(List<HBaseProtos.TableName> tableNamesList) {
+ if (tableNamesList == null) {
+ return new ArrayList<>();
+ }
+ return tableNamesList.stream().map(ProtobufUtil::toTableName).collect(Collectors.toList());
+ }
+
public static TableName[] getTableNameArray(List<HBaseProtos.TableName> tableNamesList) {
if (tableNamesList == null) {
return new TableName[0];
@@ -3345,23 +3349,33 @@ public final class ProtobufUtil {
}
/**
- * Create a SplitRegionRequest for a given region name
- *
- * @param regionName the name of the region to split
- * @param splitPoint the split point
- * @return a SplitRegionRequest
- */
- public static SplitRegionRequest buildSplitRegionRequest(
- final byte[] regionName, final byte[] splitPoint) {
- SplitRegionRequest.Builder builder = SplitRegionRequest.newBuilder();
- RegionSpecifier region = RequestConverter.buildRegionSpecifier(
- RegionSpecifierType.REGION_NAME, regionName);
- builder.setRegion(region);
- if (splitPoint != null) {
- builder.setSplitPoint(UnsafeByteOperations.unsafeWrap(splitPoint));
- }
- return builder.build();
- }
+ * Create a SplitRegionRequest for a given region name
+ * @param regionName the name of the region to split
+ * @param splitPoint the split point
+ * @return a SplitRegionRequest
+ * @deprecated Use {@link #buildSplitRegionRequest(byte[], Optional)} instead.
+ */
+ @Deprecated
+ public static SplitRegionRequest buildSplitRegionRequest(final byte[] regionName,
+ final byte[] splitPoint) {
+ return buildSplitRegionRequest(regionName, Optional.ofNullable(splitPoint));
+ }
+
+ /**
+ * Create a SplitRegionRequest for a given region name
+ * @param regionName the name of the region to split
+ * @param splitPoint the split point
+ * @return a SplitRegionRequest
+ */
+ public static SplitRegionRequest buildSplitRegionRequest(byte[] regionName,
+ Optional<byte[]> splitPoint) {
+ SplitRegionRequest.Builder builder = SplitRegionRequest.newBuilder();
+ RegionSpecifier region =
+ RequestConverter.buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName);
+ builder.setRegion(region);
+ splitPoint.ifPresent(sp -> builder.setSplitPoint(UnsafeByteOperations.unsafeWrap(sp)));
+ return builder.build();
+ }
public static ProcedureDescription buildProcedureDescription(String signature, String instance,
Map<String, String> props) {
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/RequestConverter.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/RequestConverter.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/RequestConverter.java
index 67f7d0a..39ae6a5 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/RequestConverter.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/RequestConverter.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.shaded.protobuf;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
import java.util.regex.Pattern;
@@ -919,25 +920,37 @@ public final class RequestConverter {
builder.setRegionInfo(HRegionInfo.convert(regionInfo));
return builder.build();
}
- /**
- * Create a CompactRegionRequest for a given region name
- *
- * @param regionName the name of the region to get info
- * @param major indicator if it is a major compaction
- * @return a CompactRegionRequest
- */
- public static CompactRegionRequest buildCompactRegionRequest(
- final byte[] regionName, final boolean major, final byte [] family) {
- CompactRegionRequest.Builder builder = CompactRegionRequest.newBuilder();
- RegionSpecifier region = buildRegionSpecifier(
- RegionSpecifierType.REGION_NAME, regionName);
- builder.setRegion(region);
- builder.setMajor(major);
- if (family != null) {
- builder.setFamily(UnsafeByteOperations.unsafeWrap(family));
- }
- return builder.build();
- }
+
+ /**
+ * Create a CompactRegionRequest for a given region name
+ * @param regionName the name of the region to get info
+ * @param major indicator if it is a major compaction
+ * @param columnFamily
+ * @return a CompactRegionRequest
+ * @deprecated Use {@link #buildCompactRegionRequest(byte[], boolean, Optional)} instead.
+ */
+ @Deprecated
+ public static CompactRegionRequest buildCompactRegionRequest(byte[] regionName, boolean major,
+ byte[] columnFamily) {
+ return buildCompactRegionRequest(regionName, major, Optional.ofNullable(columnFamily));
+ }
+
+ /**
+ * Create a CompactRegionRequest for a given region name
+ * @param regionName the name of the region to get info
+ * @param major indicator if it is a major compaction
+ * @param columnFamily
+ * @return a CompactRegionRequest
+ */
+ public static CompactRegionRequest buildCompactRegionRequest(byte[] regionName, boolean major,
+ Optional<byte[]> columnFamily) {
+ CompactRegionRequest.Builder builder = CompactRegionRequest.newBuilder();
+ RegionSpecifier region = buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName);
+ builder.setRegion(region);
+ builder.setMajor(major);
+ columnFamily.ifPresent(family -> builder.setFamily(UnsafeByteOperations.unsafeWrap(family)));
+ return builder.build();
+ }
/**
* @see {@link #buildRollWALWriterRequest()}
@@ -1084,12 +1097,13 @@ public final class RequestConverter {
/**
* Create a protocol buffer MoveRegionRequest
- *
* @param encodedRegionName
* @param destServerName
* @return A MoveRegionRequest
* @throws DeserializationException
+ * @deprecated Use {@link #buildMoveRegionRequest(byte[], Optional)} instead.
*/
+ @Deprecated
public static MoveRegionRequest buildMoveRegionRequest(
final byte [] encodedRegionName, final byte [] destServerName) throws
DeserializationException {
@@ -1103,6 +1117,22 @@ public final class RequestConverter {
return builder.build();
}
+ /**
+ * Create a protocol buffer MoveRegionRequest
+ * @param encodedRegionName
+ * @param destServerName
+ * @return A MoveRegionRequest
+ */
+ public static MoveRegionRequest buildMoveRegionRequest(byte[] encodedRegionName,
+ Optional<ServerName> destServerName) {
+ MoveRegionRequest.Builder builder = MoveRegionRequest.newBuilder();
+ builder.setRegion(buildRegionSpecifier(RegionSpecifierType.ENCODED_REGION_NAME,
+ encodedRegionName));
+ destServerName.ifPresent(serverName -> builder.setDestServerName(ProtobufUtil
+ .toServerName(serverName)));
+ return builder.build();
+ }
+
public static MergeTableRegionsRequest buildMergeTableRegionsRequest(
final byte[][] encodedNameOfdaughaterRegions,
final boolean forcible,
@@ -1310,11 +1340,25 @@ public final class RequestConverter {
* @param pattern The compiled regular expression to match against
* @param includeSysTables False to match only against userspace tables
* @return a GetTableDescriptorsRequest
+ * @deprecated Use {@link #buildGetTableDescriptorsRequest(Optional, boolean)} instead.
*/
+ @Deprecated
public static GetTableDescriptorsRequest buildGetTableDescriptorsRequest(final Pattern pattern,
boolean includeSysTables) {
+ return buildGetTableDescriptorsRequest(Optional.ofNullable(pattern), includeSysTables);
+ }
+
+ /**
+ * Creates a protocol buffer GetTableDescriptorsRequest
+ *
+ * @param pattern The compiled regular expression to match against
+ * @param includeSysTables False to match only against userspace tables
+ * @return a GetTableDescriptorsRequest
+ */
+ public static GetTableDescriptorsRequest
+ buildGetTableDescriptorsRequest(Optional<Pattern> pattern, boolean includeSysTables) {
GetTableDescriptorsRequest.Builder builder = GetTableDescriptorsRequest.newBuilder();
- if (pattern != null) builder.setRegex(pattern.toString());
+ pattern.ifPresent(p -> builder.setRegex(p.toString()));
builder.setIncludeSysTables(includeSysTables);
return builder.build();
}
@@ -1325,11 +1369,25 @@ public final class RequestConverter {
* @param pattern The compiled regular expression to match against
* @param includeSysTables False to match only against userspace tables
* @return a GetTableNamesRequest
+ * @deprecated Use {@link #buildGetTableNamesRequest(Optional, boolean)} instead.
*/
+ @Deprecated
public static GetTableNamesRequest buildGetTableNamesRequest(final Pattern pattern,
boolean includeSysTables) {
+ return buildGetTableNamesRequest(Optional.ofNullable(pattern), includeSysTables);
+ }
+
+ /**
+ * Creates a protocol buffer GetTableNamesRequest
+ *
+ * @param pattern The compiled regular expression to match against
+ * @param includeSysTables False to match only against userspace tables
+ * @return a GetTableNamesRequest
+ */
+ public static GetTableNamesRequest buildGetTableNamesRequest(Optional<Pattern> pattern,
+ boolean includeSysTables) {
GetTableNamesRequest.Builder builder = GetTableNamesRequest.newBuilder();
- if (pattern != null) builder.setRegex(pattern.toString());
+ pattern.ifPresent(p -> builder.setRegex(p.toString()));
builder.setIncludeSysTables(includeSysTables);
return builder.build();
}
@@ -1635,11 +1693,18 @@ public final class RequestConverter {
return builder.build();
}
+ /**
+ * @deprecated Use {@link #buildListReplicationPeersRequest(Optional)} instead.
+ */
+ @Deprecated
public static ListReplicationPeersRequest buildListReplicationPeersRequest(Pattern pattern) {
+ return buildListReplicationPeersRequest(Optional.ofNullable(pattern));
+ }
+
+ public static ListReplicationPeersRequest
+ buildListReplicationPeersRequest(Optional<Pattern> pattern) {
ListReplicationPeersRequest.Builder builder = ListReplicationPeersRequest.newBuilder();
- if (pattern != null) {
- builder.setRegex(pattern.toString());
- }
+ pattern.ifPresent(p -> builder.setRegex(p.toString()));
return builder.build();
}
[3/3] hbase git commit: HBASE-18234 Revisit the async admin api
Posted by zg...@apache.org.
HBASE-18234 Revisit the async admin api
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/2d781aa1
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/2d781aa1
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/2d781aa1
Branch: refs/heads/master
Commit: 2d781aa15c9535c3845692a275c8babca2c45568
Parents: 0e8e176
Author: Guanghao Zhang <zg...@apache.org>
Authored: Mon Jun 26 10:12:38 2017 +0800
Committer: Guanghao Zhang <zg...@apache.org>
Committed: Mon Jun 26 17:27:09 2017 +0800
----------------------------------------------------------------------
.../hadoop/hbase/AsyncMetaTableAccessor.java | 116 +-
.../apache/hadoop/hbase/MetaTableAccessor.java | 9 +-
.../apache/hadoop/hbase/client/AsyncAdmin.java | 373 ++----
.../hadoop/hbase/client/AsyncHBaseAdmin.java | 1211 ++++++++----------
.../hbase/shaded/protobuf/ProtobufUtil.java | 84 +-
.../hbase/shaded/protobuf/RequestConverter.java | 115 +-
...gionServerBulkLoadWithOldSecureEndpoint.java | 3 +-
.../hbase/client/TestAsyncBalancerAdminApi.java | 10 +-
.../client/TestAsyncNamespaceAdminApi.java | 4 +-
.../client/TestAsyncProcedureAdminApi.java | 5 +-
.../hbase/client/TestAsyncRegionAdminApi.java | 88 +-
.../hbase/client/TestAsyncSnapshotAdminApi.java | 37 +-
.../hbase/client/TestAsyncTableAdminApi.java | 40 +-
.../regionserver/TestHRegionServerBulkLoad.java | 3 +-
.../TestHRegionServerBulkLoadWithOldClient.java | 3 +-
15 files changed, 981 insertions(+), 1120 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-client/src/main/java/org/apache/hadoop/hbase/AsyncMetaTableAccessor.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/AsyncMetaTableAccessor.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/AsyncMetaTableAccessor.java
index 6988047..18fca6b 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/AsyncMetaTableAccessor.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/AsyncMetaTableAccessor.java
@@ -22,6 +22,7 @@ import static org.apache.hadoop.hbase.TableName.META_TABLE_NAME;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
@@ -30,6 +31,7 @@ import java.util.SortedMap;
import java.util.concurrent.CompletableFuture;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -37,6 +39,7 @@ import org.apache.hadoop.hbase.MetaTableAccessor.CollectingVisitor;
import org.apache.hadoop.hbase.MetaTableAccessor.QueryType;
import org.apache.hadoop.hbase.MetaTableAccessor.Visitor;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Consistency;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.RawAsyncTable;
@@ -45,6 +48,7 @@ import org.apache.hadoop.hbase.client.RegionReplicaUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.TableState;
+import org.apache.hadoop.hbase.client.Scan.ReadType;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
@@ -98,34 +102,72 @@ public class AsyncMetaTableAccessor {
return future;
}
- public static CompletableFuture<Pair<HRegionInfo, ServerName>> getRegion(RawAsyncTable metaTable,
- byte[] regionName) {
- CompletableFuture<Pair<HRegionInfo, ServerName>> future = new CompletableFuture<>();
- byte[] row = regionName;
- HRegionInfo parsedInfo = null;
+ /**
+ * Returns the HRegionLocation from meta for the given region
+ * @param metaTable
+ * @param regionName region we're looking for
+ * @return HRegionLocation for the given region
+ */
+ public static CompletableFuture<Optional<HRegionLocation>> getRegionLocation(
+ RawAsyncTable metaTable, byte[] regionName) {
+ CompletableFuture<Optional<HRegionLocation>> future = new CompletableFuture<>();
try {
- parsedInfo = MetaTableAccessor.parseRegionInfoFromRegionName(regionName);
- row = MetaTableAccessor.getMetaKeyForRegion(parsedInfo);
- } catch (Exception parseEx) {
- // Ignore if regionName is a encoded region name.
+ HRegionInfo parsedRegionInfo = MetaTableAccessor.parseRegionInfoFromRegionName(regionName);
+ metaTable.get(
+ new Get(MetaTableAccessor.getMetaKeyForRegion(parsedRegionInfo))
+ .addFamily(HConstants.CATALOG_FAMILY)).whenComplete(
+ (r, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
+ future.complete(getRegionLocations(r).map(
+ locations -> locations.getRegionLocation(parsedRegionInfo.getReplicaId())));
+ });
+ } catch (IOException parseEx) {
+ LOG.warn("Failed to parse the passed region name: " + Bytes.toStringBinary(regionName));
+ future.completeExceptionally(parseEx);
}
+ return future;
+ }
- final HRegionInfo finalHRI = parsedInfo;
- metaTable.get(new Get(row).addFamily(HConstants.CATALOG_FAMILY)).whenComplete((r, err) -> {
- if (err != null) {
- future.completeExceptionally(err);
- return;
- }
- RegionLocations locations = MetaTableAccessor.getRegionLocations(r);
- HRegionLocation hrl = locations == null ? null
- : locations.getRegionLocation(finalHRI == null ? 0 : finalHRI.getReplicaId());
- if (hrl == null) {
- future.complete(null);
- } else {
- future.complete(new Pair<>(hrl.getRegionInfo(), hrl.getServerName()));
- }
- });
-
+ /**
+ * Returns the HRegionLocation from meta for the given encoded region name
+ * @param metaTable
+ * @param encodedRegionName region we're looking for
+ * @return HRegionLocation for the given region
+ */
+ public static CompletableFuture<Optional<HRegionLocation>> getRegionLocationWithEncodedName(
+ RawAsyncTable metaTable, byte[] encodedRegionName) {
+ CompletableFuture<Optional<HRegionLocation>> future = new CompletableFuture<>();
+ metaTable.scanAll(new Scan().setReadType(ReadType.PREAD).addFamily(HConstants.CATALOG_FAMILY))
+ .whenComplete(
+ (results, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ return;
+ }
+ String encodedRegionNameStr = Bytes.toString(encodedRegionName);
+ results
+ .stream()
+ .filter(result -> !result.isEmpty())
+ .filter(result -> MetaTableAccessor.getHRegionInfo(result) != null)
+ .forEach(
+ result -> {
+ getRegionLocations(result).ifPresent(
+ locations -> {
+ for (HRegionLocation location : locations.getRegionLocations()) {
+ if (location != null
+ && encodedRegionNameStr.equals(location.getRegionInfo()
+ .getEncodedName())) {
+ future.complete(Optional.of(location));
+ return;
+ }
+ }
+ });
+ });
+ future.complete(Optional.empty());
+ });
return future;
}
@@ -143,15 +185,29 @@ public class AsyncMetaTableAccessor {
}
/**
- * Used to get table regions' info and server.
+ * Used to get all region locations for the specific table.
* @param metaTable
* @param tableName table we're looking for, can be null for getting all regions
- * @return the list of regioninfos and server. The return value will be wrapped by a
+ * @return the list of region locations. The return value will be wrapped by a
* {@link CompletableFuture}.
*/
- public static CompletableFuture<List<Pair<HRegionInfo, ServerName>>> getTableRegionsAndLocations(
+ public static CompletableFuture<List<HRegionLocation>> getTableHRegionLocations(
RawAsyncTable metaTable, final Optional<TableName> tableName) {
- return getTableRegionsAndLocations(metaTable, tableName, true);
+ CompletableFuture<List<HRegionLocation>> future = new CompletableFuture<>();
+ getTableRegionsAndLocations(metaTable, tableName, true).whenComplete(
+ (locations, err) -> {
+ if (err != null) {
+ future.completeExceptionally(err);
+ } else if (locations == null || locations.isEmpty()) {
+ future.complete(Collections.emptyList());
+ } else {
+ List<HRegionLocation> regionLocations = locations.stream()
+ .map(loc -> new HRegionLocation(loc.getFirst(), loc.getSecond()))
+ .collect(Collectors.toList());
+ future.complete(regionLocations);
+ }
+ });
+ return future;
}
/**
@@ -162,7 +218,7 @@ public class AsyncMetaTableAccessor {
* @return the list of regioninfos and server. The return value will be wrapped by a
* {@link CompletableFuture}.
*/
- public static CompletableFuture<List<Pair<HRegionInfo, ServerName>>> getTableRegionsAndLocations(
+ private static CompletableFuture<List<Pair<HRegionInfo, ServerName>>> getTableRegionsAndLocations(
RawAsyncTable metaTable, final Optional<TableName> tableName,
final boolean excludeOfflinedSplitParents) {
CompletableFuture<List<Pair<HRegionInfo, ServerName>>> future = new CompletableFuture<>();
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-client/src/main/java/org/apache/hadoop/hbase/MetaTableAccessor.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/MetaTableAccessor.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/MetaTableAccessor.java
index 9eb5111..3870fd1 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/MetaTableAccessor.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/MetaTableAccessor.java
@@ -364,13 +364,12 @@ public class MetaTableAccessor {
* is stored in the name, so the returned object should only be used for the fields
* in the regionName.
*/
- protected static HRegionInfo parseRegionInfoFromRegionName(byte[] regionName)
- throws IOException {
+ public static HRegionInfo parseRegionInfoFromRegionName(byte[] regionName) throws IOException {
byte[][] fields = HRegionInfo.parseRegionName(regionName);
- long regionId = Long.parseLong(Bytes.toString(fields[2]));
+ long regionId = Long.parseLong(Bytes.toString(fields[2]));
int replicaId = fields.length > 3 ? Integer.parseInt(Bytes.toString(fields[3]), 16) : 0;
- return new HRegionInfo(
- TableName.valueOf(fields[0]), fields[1], fields[1], false, regionId, replicaId);
+ return new HRegionInfo(TableName.valueOf(fields[0]), fields[1], fields[1], false, regionId,
+ replicaId);
}
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/2d781aa1/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
index 1a3cae2..3b022f4 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.client;
import java.util.List;
import java.util.Collection;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.regex.Pattern;
@@ -42,7 +43,7 @@ import org.apache.hadoop.hbase.util.Pair;
* This feature is still under development, so marked as IA.Private. Will change to public when
* done. Use it with caution.
*/
-@InterfaceAudience.Private
+@InterfaceAudience.Public
public interface AsyncAdmin {
/**
@@ -55,63 +56,50 @@ public interface AsyncAdmin {
* @return True if table exists already. The return value will be wrapped by a
* {@link CompletableFuture}.
*/
- CompletableFuture<Boolean> tableExists(final TableName tableName);
+ CompletableFuture<Boolean> tableExists(TableName tableName);
/**
* List all the userspace tables.
- * @return - returns an array of TableDescriptors wrapped by a {@link CompletableFuture}.
- * @see #listTables(Pattern, boolean)
+ * @return - returns a list of TableDescriptors wrapped by a {@link CompletableFuture}.
+ * @see #listTables(Optional, boolean)
*/
- CompletableFuture<TableDescriptor[]> listTables();
-
- /**
- * List all the tables matching the given pattern.
- * @param regex The regular expression to match against
- * @param includeSysTables False to match only against userspace tables
- * @return - returns an array of TableDescriptors wrapped by a {@link CompletableFuture}.
- * @see #listTables(Pattern, boolean)
- */
- CompletableFuture<TableDescriptor[]> listTables(String regex, boolean includeSysTables);
+ default CompletableFuture<List<TableDescriptor>> listTables() {
+ return listTables(Optional.empty(), false);
+ }
/**
* List all the tables matching the given pattern.
* @param pattern The compiled regular expression to match against
* @param includeSysTables False to match only against userspace tables
- * @return - returns an array of TableDescriptors wrapped by a {@link CompletableFuture}.
- */
- CompletableFuture<TableDescriptor[]> listTables(Pattern pattern, boolean includeSysTables);
-
- /**
- * List all of the names of userspace tables.
- * @return TableName[] an array of table names wrapped by a {@link CompletableFuture}.
- * @see #listTableNames(Pattern, boolean)
+ * @return - returns a list of TableDescriptors wrapped by a {@link CompletableFuture}.
*/
- CompletableFuture<TableName[]> listTableNames();
+ CompletableFuture<List<TableDescriptor>> listTables(Optional<Pattern> pattern,
+ boolean includeSysTables);
/**
* List all of the names of userspace tables.
- * @param regex The regular expression to match against
- * @param includeSysTables False to match only against userspace tables
- * @return TableName[] an array of table names wrapped by a {@link CompletableFuture}.
- * @see #listTableNames(Pattern, boolean)
+ * @return a list of table names wrapped by a {@link CompletableFuture}.
+ * @see #listTableNames(Optional, boolean)
*/
- CompletableFuture<TableName[]> listTableNames(final String regex, final boolean includeSysTables);
+ default CompletableFuture<List<TableName>> listTableNames() {
+ return listTableNames(Optional.empty(), false);
+ }
/**
* List all of the names of userspace tables.
* @param pattern The regular expression to match against
* @param includeSysTables False to match only against userspace tables
- * @return TableName[] an array of table names wrapped by a {@link CompletableFuture}.
+ * @return a list of table names wrapped by a {@link CompletableFuture}.
*/
- CompletableFuture<TableName[]> listTableNames(final Pattern pattern,
- final boolean includeSysTables);
+ CompletableFuture<List<TableName>> listTableNames(Optional<Pattern> pattern,
+ boolean includeSysTables);
/**
* Method for getting the tableDescriptor
* @param tableName as a {@link TableName}
* @return the read-only tableDescriptor wrapped by a {@link CompletableFuture}.
*/
- CompletableFuture<TableDescriptor> getTableDescriptor(final TableName tableName);
+ CompletableFuture<TableDescriptor> getTableDescriptor(TableName tableName);
/**
* Creates a new table.
@@ -140,94 +128,63 @@ public interface AsyncAdmin {
* @param desc table descriptor for table
* @param splitKeys array of split keys for the initial regions of the table
*/
- CompletableFuture<Void> createTable(final TableDescriptor desc, byte[][] splitKeys);
+ CompletableFuture<Void> createTable(TableDescriptor desc, byte[][] splitKeys);
/**
* Deletes a table.
* @param tableName name of table to delete
*/
- CompletableFuture<Void> deleteTable(final TableName tableName);
-
- /**
- * Deletes tables matching the passed in pattern and wait on completion. Warning: Use this method
- * carefully, there is no prompting and the effect is immediate. Consider using
- * {@link #listTables(String, boolean)} and
- * {@link #deleteTable(org.apache.hadoop.hbase.TableName)}
- * @param regex The regular expression to match table names against
- * @return Table descriptors for tables that couldn't be deleted. The return value will be wrapped
- * by a {@link CompletableFuture}. The return HTDs are read-only.
- */
- CompletableFuture<TableDescriptor[]> deleteTables(String regex);
+ CompletableFuture<Void> deleteTable(TableName tableName);
/**
* Delete tables matching the passed in pattern and wait on completion. Warning: Use this method
* carefully, there is no prompting and the effect is immediate. Consider using
- * {@link #listTables(Pattern, boolean) } and
+ * {@link #listTableNames(Optional, boolean) } and
* {@link #deleteTable(org.apache.hadoop.hbase.TableName)}
* @param pattern The pattern to match table names against
* @return Table descriptors for tables that couldn't be deleted. The return value will be wrapped
* by a {@link CompletableFuture}. The return HTDs are read-only.
*/
- CompletableFuture<TableDescriptor[]> deleteTables(Pattern pattern);
+ CompletableFuture<List<TableDescriptor>> deleteTables(Pattern pattern);
/**
* Truncate a table.
* @param tableName name of table to truncate
* @param preserveSplits True if the splits should be preserved
*/
- CompletableFuture<Void> truncateTable(final TableName tableName, final boolean preserveSplits);
+ CompletableFuture<Void> truncateTable(TableName tableName, boolean preserveSplits);
/**
* Enable a table. The table has to be in disabled state for it to be enabled.
* @param tableName name of the table
*/
- CompletableFuture<Void> enableTable(final TableName tableName);
+ CompletableFuture<Void> enableTable(TableName tableName);
/**
* Enable tables matching the passed in pattern. Warning: Use this method carefully, there is no
- * prompting and the effect is immediate. Consider using {@link #listTables(Pattern, boolean)} and
- * {@link #enableTable(TableName)}
- * @param regex The regular expression to match table names against
- * @return Table descriptors for tables that couldn't be enabled. The return value will be wrapped
- * by a {@link CompletableFuture}. The return HTDs are read-only.
- */
- CompletableFuture<TableDescriptor[]> enableTables(String regex);
-
- /**
- * Enable tables matching the passed in pattern. Warning: Use this method carefully, there is no
- * prompting and the effect is immediate. Consider using {@link #listTables(Pattern, boolean)} and
+ * prompting and the effect is immediate. Consider using {@link #listTables(Optional, boolean)} and
* {@link #enableTable(TableName)}
* @param pattern The pattern to match table names against
* @return Table descriptors for tables that couldn't be enabled. The return value will be wrapped
* by a {@link CompletableFuture}. The return HTDs are read-only.
*/
- CompletableFuture<TableDescriptor[]> enableTables(Pattern pattern);
+ CompletableFuture<List<TableDescriptor>> enableTables(Pattern pattern);
/**
* Disable a table. The table has to be in enabled state for it to be disabled.
* @param tableName
*/
- CompletableFuture<Void> disableTable(final TableName tableName);
+ CompletableFuture<Void> disableTable(TableName tableName);
/**
* Disable tables matching the passed in pattern. Warning: Use this method carefully, there is no
- * prompting and the effect is immediate. Consider using {@link #listTables(Pattern, boolean)} and
- * {@link #disableTable(TableName)}
- * @param regex The regular expression to match table names against
- * @return Table descriptors for tables that couldn't be disabled. The return value will be wrapped by a
- * {@link CompletableFuture}. The return HTDs are read-only.
- */
- CompletableFuture<TableDescriptor[]> disableTables(String regex);
-
- /**
- * Disable tables matching the passed in pattern. Warning: Use this method carefully, there is no
- * prompting and the effect is immediate. Consider using {@link #listTables(Pattern, boolean)} and
+ * prompting and the effect is immediate. Consider using {@link #listTables(Optional, boolean)} and
* {@link #disableTable(TableName)}
* @param pattern The pattern to match table names against
* @return Table descriptors for tables that couldn't be disabled. The return value will be wrapped by a
* {@link CompletableFuture}. The return HTDs are read-only.
*/
- CompletableFuture<TableDescriptor[]> disableTables(Pattern pattern);
+ CompletableFuture<List<TableDescriptor>> disableTables(Pattern pattern);
/**
* @param tableName name of table to check
@@ -261,61 +218,61 @@ public interface AsyncAdmin {
* yet to be updated Pair.getSecond() is the total number of regions of the table. The
* return value will be wrapped by a {@link CompletableFuture}.
*/
- CompletableFuture<Pair<Integer, Integer>> getAlterStatus(final TableName tableName);
+ CompletableFuture<Pair<Integer, Integer>> getAlterStatus(TableName tableName);
/**
* Add a column family to an existing table.
* @param tableName name of the table to add column family to
* @param columnFamily column family descriptor of column family to be added
*/
- CompletableFuture<Void> addColumnFamily(final TableName tableName,
- final ColumnFamilyDescriptor columnFamily);
+ CompletableFuture<Void> addColumnFamily(TableName tableName,
+ ColumnFamilyDescriptor columnFamily);
/**
* Delete a column family from a table.
* @param tableName name of table
* @param columnFamily name of column family to be deleted
*/
- CompletableFuture<Void> deleteColumnFamily(final TableName tableName, final byte[] columnFamily);
+ CompletableFuture<Void> deleteColumnFamily(TableName tableName, byte[] columnFamily);
/**
* Modify an existing column family on a table.
* @param tableName name of table
* @param columnFamily new column family descriptor to use
*/
- CompletableFuture<Void> modifyColumnFamily(final TableName tableName,
- final ColumnFamilyDescriptor columnFamily);
+ CompletableFuture<Void> modifyColumnFamily(TableName tableName,
+ ColumnFamilyDescriptor columnFamily);
/**
* Create a new namespace.
* @param descriptor descriptor which describes the new namespace
*/
- CompletableFuture<Void> createNamespace(final NamespaceDescriptor descriptor);
+ CompletableFuture<Void> createNamespace(NamespaceDescriptor descriptor);
/**
* Modify an existing namespace.
* @param descriptor descriptor which describes the new namespace
*/
- CompletableFuture<Void> modifyNamespace(final NamespaceDescriptor descriptor);
+ CompletableFuture<Void> modifyNamespace(NamespaceDescriptor descriptor);
/**
* Delete an existing namespace. Only empty namespaces (no tables) can be removed.
* @param name namespace name
*/
- CompletableFuture<Void> deleteNamespace(final String name);
+ CompletableFuture<Void> deleteNamespace(String name);
/**
* Get a namespace descriptor by name
* @param name name of namespace descriptor
* @return A descriptor wrapped by a {@link CompletableFuture}.
*/
- CompletableFuture<NamespaceDescriptor> getNamespaceDescriptor(final String name);
+ CompletableFuture<NamespaceDescriptor> getNamespaceDescriptor(String name);
/**
* List available namespace descriptors
* @return List of descriptors wrapped by a {@link CompletableFuture}.
*/
- CompletableFuture<NamespaceDescriptor[]> listNamespaceDescriptors();
+ CompletableFuture<List<NamespaceDescriptor>> listNamespaceDescriptors();
/**
* @param tableName name of table to check
@@ -329,7 +286,7 @@ public interface AsyncAdmin {
* @param on
* @return Previous balancer value wrapped by a {@link CompletableFuture}.
*/
- CompletableFuture<Boolean> setBalancerRunning(final boolean on);
+ CompletableFuture<Boolean> setBalancerOn(boolean on);
/**
* Invoke the balancer. Will run the balancer and if regions to move, it will go ahead and do the
@@ -337,72 +294,38 @@ public interface AsyncAdmin {
* @return True if balancer ran, false otherwise. The return value will be wrapped by a
* {@link CompletableFuture}.
*/
- CompletableFuture<Boolean> balancer();
+ default CompletableFuture<Boolean> balance() {
+ return balance(false);
+ }
/**
* Invoke the balancer. Will run the balancer and if regions to move, it will go ahead and do the
* reassignments. If there is region in transition, force parameter of true would still run
* balancer. Can *not* run for other reasons. Check logs.
- * @param force whether we should force balance even if there is region in transition.
+ * @param forcible whether we should force balance even if there is region in transition.
* @return True if balancer ran, false otherwise. The return value will be wrapped by a
* {@link CompletableFuture}.
*/
- CompletableFuture<Boolean> balancer(boolean force);
+ CompletableFuture<Boolean> balance(boolean forcible);
/**
* Query the current state of the balancer.
- * @return true if the balancer is enabled, false otherwise.
- * The return value will be wrapped by a {@link CompletableFuture}.
- */
- CompletableFuture<Boolean> isBalancerEnabled();
-
- /**
- * Close a region. For expert-admins. Runs close on the regionserver. The master will not be
- * informed of the close.
- *
- * @param regionname region name to close
- * @param serverName If supplied, we'll use this location rather than the one currently in
- * <code>hbase:meta</code>
+ * @return true if the balance switch is on, false otherwise The return value will be wrapped by a
+ * {@link CompletableFuture}.
*/
- CompletableFuture<Void> closeRegion(String regionname, String serverName);
+ CompletableFuture<Boolean> isBalancerOn();
/**
- * Close a region. For expert-admins Runs close on the regionserver. The master will not be
+ * Close a region. For expert-admins Runs close on the regionserver. The master will not be
* informed of the close.
- *
- * @param regionname region name to close
- * @param serverName The servername of the regionserver. If passed null we will use servername
- * found in the hbase:meta table. A server name is made of host, port and startcode. Here is an
- * example: <code> host187.example.com,60020,1289493121758</code>
- */
- CompletableFuture<Void> closeRegion(byte[] regionname, String serverName);
-
- /**
- * For expert-admins. Runs close on the regionserver. Closes a region based on the encoded region
- * name. The region server name is mandatory. If the servername is provided then based on the
- * online regions in the specified regionserver the specified region will be closed. The master
- * will not be informed of the close. Note that the regionname is the encoded regionname.
- *
- * @param encodedRegionName The encoded region name; i.e. the hash that makes up the region name
- * suffix: e.g. if regionname is
- * <code>TestTable,0094429456,1289497600452.527db22f95c8a9e0116f0cc13c680396.</code>,
- * then the encoded region name is: <code>527db22f95c8a9e0116f0cc13c680396</code>.
- * @param serverName The servername of the regionserver. A server name is made of host, port and
- * startcode. This is mandatory. Here is an example:
- * <code> host187.example.com,60020,1289493121758</code>
+ * @param regionName region name to close
+ * @param serverName The servername of the regionserver. If not present, we will use servername
+ * found in the hbase:meta table. A server name is made of host, port and startcode. Here
+ * is an example: <code> host187.example.com,60020,1289493121758</code>
* @return true if the region was closed, false if not. The return value will be wrapped by a
- * {@link CompletableFuture}.
- */
- CompletableFuture<Boolean> closeRegionWithEncodedRegionName(String encodedRegionName, String serverName);
-
- /**
- * Close a region. For expert-admins Runs close on the regionserver. The master will not be
- * informed of the close.
- *
- * @param sn
- * @param hri
+ * {@link CompletableFuture}.
*/
- CompletableFuture<Void> closeRegion(ServerName sn, HRegionInfo hri);
+ CompletableFuture<Boolean> closeRegion(byte[] regionName, Optional<ServerName> serverName);
/**
* Get all the online regions on a region server.
@@ -422,60 +345,80 @@ public interface AsyncAdmin {
CompletableFuture<Void> flushRegion(byte[] regionName);
/**
- * Compact a table. Asynchronous operation even if CompletableFuture.get().
+ * Compact a table. When the returned CompletableFuture is done, it only means the compact request
+ * was sent to HBase and may need some time to finish the compact operation.
* @param tableName table to compact
*/
- CompletableFuture<Void> compact(TableName tableName);
+ default CompletableFuture<Void> compact(TableName tableName) {
+ return compact(tableName, Optional.empty());
+ }
/**
- * Compact a column family within a table. Asynchronous operation even if CompletableFuture.get().
+ * Compact a column family within a table. When the returned CompletableFuture is done, it only
+ * means the compact request was sent to HBase and may need some time to finish the compact
+ * operation.
* @param tableName table to compact
- * @param columnFamily column family within a table
+ * @param columnFamily column family within a table. If not present, compact the table's all
+ * column families.
*/
- CompletableFuture<Void> compact(TableName tableName, byte[] columnFamily);
+ CompletableFuture<Void> compact(TableName tableName, Optional<byte[]> columnFamily);
/**
- * Compact an individual region. Asynchronous operation even if CompletableFuture.get().
+ * Compact an individual region. When the returned CompletableFuture is done, it only means the
+ * compact request was sent to HBase and may need some time to finish the compact operation.
* @param regionName region to compact
*/
- CompletableFuture<Void> compactRegion(byte[] regionName);
+ default CompletableFuture<Void> compactRegion(byte[] regionName) {
+ return compactRegion(regionName, Optional.empty());
+ }
/**
- * Compact a column family within a region. Asynchronous operation even if
- * CompletableFuture.get().
+ * Compact a column family within a region. When the returned CompletableFuture is done, it only
+ * means the compact request was sent to HBase and may need some time to finish the compact
+ * operation.
* @param regionName region to compact
- * @param columnFamily column family within a region
+ * @param columnFamily column family within a region. If not present, compact the region's all
+ * column families.
*/
- CompletableFuture<Void> compactRegion(byte[] regionName, byte[] columnFamily);
+ CompletableFuture<Void> compactRegion(byte[] regionName, Optional<byte[]> columnFamily);
/**
- * Major compact a table. Asynchronous operation even if CompletableFuture.get().
+ * Major compact a table. When the returned CompletableFuture is done, it only means the compact
+ * request was sent to HBase and may need some time to finish the compact operation.
* @param tableName table to major compact
*/
- CompletableFuture<Void> majorCompact(TableName tableName);
+ default CompletableFuture<Void> majorCompact(TableName tableName) {
+ return majorCompact(tableName, Optional.empty());
+ }
/**
- * Major compact a column family within a table. Asynchronous operation even if
- * CompletableFuture.get().
+ * Major compact a column family within a table. When the returned CompletableFuture is done, it
+ * only means the compact request was sent to HBase and may need some time to finish the compact
+ * operation.
* @param tableName table to major compact
- * @param columnFamily column family within a table
+ * @param columnFamily column family within a table. If not present, major compact the table's all
+ * column families.
*/
- CompletableFuture<Void> majorCompact(TableName tableName, byte[] columnFamily);
+ CompletableFuture<Void> majorCompact(TableName tableName, Optional<byte[]> columnFamily);
/**
- * Major compact a table or an individual region. Asynchronous operation even if
- * CompletableFuture.get().
+ * Major compact a region. When the returned CompletableFuture is done, it only means the compact
+ * request was sent to HBase and may need some time to finish the compact operation.
* @param regionName region to major compact
*/
- CompletableFuture<Void> majorCompactRegion(byte[] regionName);
+ default CompletableFuture<Void> majorCompactRegion(byte[] regionName) {
+ return majorCompactRegion(regionName, Optional.empty());
+ }
/**
- * Major compact a column family within region. Asynchronous operation even if
- * CompletableFuture.get().
- * @param regionName egion to major compact
- * @param columnFamily column family within a region
+ * Major compact a column family within region. When the returned CompletableFuture is done, it
+ * only means the compact request was sent to HBase and may need some time to finish the compact
+ * operation.
+ * @param regionName region to major compact
+ * @param columnFamily column family within a region. If not present, major compact the region's
+ * all column families.
*/
- CompletableFuture<Void> majorCompactRegion(byte[] regionName, byte[] columnFamily);
+ CompletableFuture<Void> majorCompactRegion(byte[] regionName, Optional<byte[]> columnFamily);
/**
* Compact all regions on the region server.
@@ -496,51 +439,54 @@ public interface AsyncAdmin {
* @param forcible true if do a compulsory merge, otherwise we will only merge two adjacent
* regions
*/
- CompletableFuture<Void> mergeRegions(final byte[] nameOfRegionA, final byte[] nameOfRegionB,
- final boolean forcible);
+ CompletableFuture<Void> mergeRegions(byte[] nameOfRegionA, byte[] nameOfRegionB,
+ boolean forcible);
/**
* Split a table. The method will execute split action for each region in table.
* @param tableName table to split
*/
- CompletableFuture<Void> split(final TableName tableName);
+ CompletableFuture<Void> split(TableName tableName);
/**
* Split an individual region.
* @param regionName region to split
*/
- CompletableFuture<Void> splitRegion(final byte[] regionName);
+ default CompletableFuture<Void> splitRegion(byte[] regionName) {
+ return splitRegion(regionName, Optional.empty());
+ }
/**
* Split a table.
* @param tableName table to split
* @param splitPoint the explicit position to split on
*/
- CompletableFuture<Void> split(final TableName tableName, final byte[] splitPoint);
+ CompletableFuture<Void> split(TableName tableName, byte[] splitPoint);
/**
* Split an individual region.
* @param regionName region to split
- * @param splitPoint the explicit position to split on
+ * @param splitPoint the explicit position to split on. If not present, it will decide by region
+ * server.
*/
- CompletableFuture<Void> splitRegion(final byte[] regionName, final byte[] splitPoint);
+ CompletableFuture<Void> splitRegion(byte[] regionName, Optional<byte[]> splitPoint);
/**
* @param regionName Encoded or full name of region to assign.
*/
- CompletableFuture<Void> assign(final byte[] regionName);
+ CompletableFuture<Void> assign(byte[] regionName);
/**
* Unassign a region from current hosting regionserver. Region will then be assigned to a
* regionserver chosen at random. Region could be reassigned back to the same server. Use
- * {@link #move(byte[], byte[])} if you want to control the region movement.
+ * {@link #move(byte[], Optional)} if you want to control the region movement.
* @param regionName Encoded or full name of region to unassign. Will clear any existing
* RegionPlan if one found.
- * @param force If true, force unassign (Will remove region from regions-in-transition too if
+ * @param forcible If true, force unassign (Will remove region from regions-in-transition too if
* present. If results in double assignment use hbck -fix to resolve. To be used by
* experts).
*/
- CompletableFuture<Void> unassign(final byte[] regionName, final boolean force);
+ CompletableFuture<Void> unassign(byte[] regionName, boolean forcible);
/**
* Offline specified region from master's in-memory state. It will not attempt to reassign the
@@ -550,22 +496,22 @@ public interface AsyncAdmin {
* experts or hbck.
* @param regionName Encoded or full name of region to offline
*/
- CompletableFuture<Void> offline(final byte[] regionName);
+ CompletableFuture<Void> offline(byte[] regionName);
/**
* Move the region <code>r</code> to <code>dest</code>.
* @param regionName Encoded or full name of region to move.
- * @param destServerName The servername of the destination regionserver. If passed the empty byte
- * array we'll assign to a random server. A server name is made of host, port and
- * startcode. Here is an example: <code> host187.example.com,60020,1289493121758</code>
+ * @param destServerName The servername of the destination regionserver. If not present, we'll
+ * assign to a random server. A server name is made of host, port and startcode. Here is
+ * an example: <code> host187.example.com,60020,1289493121758</code>
*/
- CompletableFuture<Void> move(final byte[] regionName, final byte[] destServerName);
+ CompletableFuture<Void> move(byte[] regionName, Optional<ServerName> destServerName);
/**
* Apply the new quota settings.
* @param quota the quota settings
*/
- CompletableFuture<Void> setQuota(final QuotaSettings quota);
+ CompletableFuture<Void> setQuota(QuotaSettings quota);
/**
* List the quotas based on the filter.
@@ -579,41 +525,41 @@ public interface AsyncAdmin {
* @param peerId a short name that identifies the peer
* @param peerConfig configuration for the replication slave cluster
*/
- CompletableFuture<Void> addReplicationPeer(final String peerId,
- final ReplicationPeerConfig peerConfig);
+ CompletableFuture<Void> addReplicationPeer(String peerId,
+ ReplicationPeerConfig peerConfig);
/**
* Remove a peer and stop the replication
* @param peerId a short name that identifies the peer
*/
- CompletableFuture<Void> removeReplicationPeer(final String peerId);
+ CompletableFuture<Void> removeReplicationPeer(String peerId);
/**
* Restart the replication stream to the specified peer
* @param peerId a short name that identifies the peer
*/
- CompletableFuture<Void> enableReplicationPeer(final String peerId);
+ CompletableFuture<Void> enableReplicationPeer(String peerId);
/**
* Stop the replication stream to the specified peer
* @param peerId a short name that identifies the peer
*/
- CompletableFuture<Void> disableReplicationPeer(final String peerId);
+ CompletableFuture<Void> disableReplicationPeer(String peerId);
/**
* Returns the configured ReplicationPeerConfig for the specified peer
* @param peerId a short name that identifies the peer
* @return ReplicationPeerConfig for the peer wrapped by a {@link CompletableFuture}.
*/
- CompletableFuture<ReplicationPeerConfig> getReplicationPeerConfig(final String peerId);
+ CompletableFuture<ReplicationPeerConfig> getReplicationPeerConfig(String peerId);
/**
* Update the peerConfig for the specified peer
* @param peerId a short name that identifies the peer
* @param peerConfig new config for the peer
*/
- CompletableFuture<Void> updateReplicationPeerConfig(final String peerId,
- final ReplicationPeerConfig peerConfig);
+ CompletableFuture<Void> updateReplicationPeerConfig(String peerId,
+ ReplicationPeerConfig peerConfig);
/**
* Append the replicable table-cf config of the specified peer
@@ -636,15 +582,9 @@ public interface AsyncAdmin {
* @return a list of replication peers description. The return value will be wrapped by a
* {@link CompletableFuture}.
*/
- CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers();
-
- /**
- * Return a list of replication peers.
- * @param regex The regular expression to match peer id
- * @return a list of replication peers description. The return value will be wrapped by a
- * {@link CompletableFuture}.
- */
- CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(String regex);
+ default CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers() {
+ return listReplicationPeers(Optional.empty());
+ }
/**
* Return a list of replication peers.
@@ -652,7 +592,8 @@ public interface AsyncAdmin {
* @return a list of replication peers description. The return value will be wrapped by a
* {@link CompletableFuture}.
*/
- CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Pattern pattern);
+ CompletableFuture<List<ReplicationPeerDescription>>
+ listReplicationPeers(Optional<Pattern> pattern);
/**
* Find all table and column families that are replicated from this cluster
@@ -686,7 +627,7 @@ public interface AsyncAdmin {
* @param tableName name of the table to snapshot
* @param type type of snapshot to take
*/
- CompletableFuture<Void> snapshot(final String snapshotName, final TableName tableName,
+ CompletableFuture<Void> snapshot(String snapshotName, TableName tableName,
SnapshotType type);
/**
@@ -718,7 +659,7 @@ public interface AsyncAdmin {
* @return <tt>true</tt> if the snapshot is completed, <tt>false</tt> if the snapshot is still
* running
*/
- CompletableFuture<Boolean> isSnapshotFinished(final SnapshotDescription snapshot);
+ CompletableFuture<Boolean> isSnapshotFinished(SnapshotDescription snapshot);
/**
* Restore the specified snapshot on the original table. (The table must be disabled) If the
@@ -747,7 +688,7 @@ public interface AsyncAdmin {
* @param snapshotName name of the snapshot to be cloned
* @param tableName name of the table where the snapshot will be restored
*/
- CompletableFuture<Void> cloneSnapshot(final String snapshotName, final TableName tableName);
+ CompletableFuture<Void> cloneSnapshot(String snapshotName, TableName tableName);
/**
* List completed snapshots.
@@ -757,13 +698,6 @@ public interface AsyncAdmin {
CompletableFuture<List<SnapshotDescription>> listSnapshots();
/**
- * List all the completed snapshots matching the given regular expression.
- * @param regex The regular expression to match against
- * @return - returns a List of SnapshotDescription wrapped by a {@link CompletableFuture}
- */
- CompletableFuture<List<SnapshotDescription>> listSnapshots(String regex);
-
- /**
* List all the completed snapshots matching the given pattern.
* @param pattern The compiled regular expression to match against
* @return - returns a List of SnapshotDescription wrapped by a {@link CompletableFuture}
@@ -773,17 +707,6 @@ public interface AsyncAdmin {
/**
* List all the completed snapshots matching the given table name regular expression and snapshot
* name regular expression.
- * @param tableNameRegex The table name regular expression to match against
- * @param snapshotNameRegex The snapshot name regular expression to match against
- * @return - returns a List of completed SnapshotDescription wrapped by a
- * {@link CompletableFuture}
- */
- CompletableFuture<List<SnapshotDescription>> listTableSnapshots(String tableNameRegex,
- String snapshotNameRegex);
-
- /**
- * List all the completed snapshots matching the given table name regular expression and snapshot
- * name regular expression.
* @param tableNamePattern The compiled table name regular expression to match against
* @param snapshotNamePattern The compiled snapshot name regular expression to match against
* @return - returns a List of completed SnapshotDescription wrapped by a
@@ -800,12 +723,6 @@ public interface AsyncAdmin {
/**
* Delete existing snapshots whose names match the pattern passed.
- * @param regex The regular expression to match against
- */
- CompletableFuture<Void> deleteSnapshots(String regex);
-
- /**
- * Delete existing snapshots whose names match the pattern passed.
* @param pattern pattern for names of the snapshot to match
*/
CompletableFuture<Void> deleteSnapshots(Pattern pattern);
@@ -813,14 +730,6 @@ public interface AsyncAdmin {
/**
* Delete all existing snapshots matching the given table name regular expression and snapshot
* name regular expression.
- * @param tableNameRegex The table name regular expression to match against
- * @param snapshotNameRegex The snapshot name regular expression to match against
- */
- CompletableFuture<Void> deleteTableSnapshots(String tableNameRegex, String snapshotNameRegex);
-
- /**
- * Delete all existing snapshots matching the given table name regular expression and snapshot
- * name regular expression.
* @param tableNamePattern The compiled table name regular expression to match against
* @param snapshotNamePattern The compiled snapshot name regular expression to match against
*/
@@ -861,7 +770,7 @@ public interface AsyncAdmin {
* @param instance The instance name of the procedure
* @param props Property/Value pairs of properties passing to the procedure
* @return true if the specified procedure is finished successfully, false if it is still running.
- * The value is vrapped by {@link CompletableFuture}
+ * The value is wrapped by {@link CompletableFuture}
*/
CompletableFuture<Boolean> isProcedureFinished(String signature, String instance,
Map<String, String> props);
@@ -879,5 +788,5 @@ public interface AsyncAdmin {
* List procedures
* @return procedure list wrapped by {@link CompletableFuture}
*/
- CompletableFuture<ProcedureInfo[]> listProcedures();
+ CompletableFuture<List<ProcedureInfo>> listProcedures();
}