HBASE-18950 Remove Optional parameters in AsyncAdmin interface
This commit is contained in:
parent
bc3f3ee3bc
commit
888f2335c9
|
@ -64,38 +64,49 @@ public interface AsyncAdmin {
|
||||||
/**
|
/**
|
||||||
* List all the userspace tables.
|
* List all the userspace tables.
|
||||||
* @return - returns a list of TableDescriptors wrapped by a {@link CompletableFuture}.
|
* @return - returns a list of TableDescriptors wrapped by a {@link CompletableFuture}.
|
||||||
* @see #listTables(Optional, boolean)
|
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<List<TableDescriptor>> listTables() {
|
default CompletableFuture<List<TableDescriptor>> listTables() {
|
||||||
return listTables(Optional.empty(), false);
|
return listTables(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* List all the tables.
|
||||||
|
* @param includeSysTables False to match only against userspace tables
|
||||||
|
* @return - returns a list of TableDescriptors wrapped by a {@link CompletableFuture}.
|
||||||
|
*/
|
||||||
|
CompletableFuture<List<TableDescriptor>> listTables(boolean includeSysTables);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* List all the tables matching the given pattern.
|
* List all the tables matching the given pattern.
|
||||||
* @param pattern The compiled regular expression to match against
|
* @param pattern The compiled regular expression to match against
|
||||||
* @param includeSysTables False to match only against userspace tables
|
* @param includeSysTables False to match only against userspace tables
|
||||||
* @return - returns a list of TableDescriptors wrapped by a {@link CompletableFuture}.
|
* @return - returns a list of TableDescriptors wrapped by a {@link CompletableFuture}.
|
||||||
*/
|
*/
|
||||||
CompletableFuture<List<TableDescriptor>> listTables(Optional<Pattern> pattern,
|
CompletableFuture<List<TableDescriptor>> listTables(Pattern pattern, boolean includeSysTables);
|
||||||
boolean includeSysTables);
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* List all of the names of userspace tables.
|
* List all of the names of userspace tables.
|
||||||
* @return a list of table names wrapped by a {@link CompletableFuture}.
|
* @return a list of table names wrapped by a {@link CompletableFuture}.
|
||||||
* @see #listTableNames(Optional, boolean)
|
* @see #listTableNames(Pattern, boolean)
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<List<TableName>> listTableNames() {
|
default CompletableFuture<List<TableName>> listTableNames() {
|
||||||
return listTableNames(Optional.empty(), false);
|
return listTableNames(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* List all of the names of tables.
|
||||||
|
* @param includeSysTables False to match only against userspace tables
|
||||||
|
* @return a list of table names wrapped by a {@link CompletableFuture}.
|
||||||
|
*/
|
||||||
|
CompletableFuture<List<TableName>> listTableNames(boolean includeSysTables);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* List all of the names of userspace tables.
|
* List all of the names of userspace tables.
|
||||||
* @param pattern The regular expression to match against
|
* @param pattern The regular expression to match against
|
||||||
* @param includeSysTables False to match only against userspace tables
|
* @param includeSysTables False to match only against userspace tables
|
||||||
* @return a list of table names wrapped by a {@link CompletableFuture}.
|
* @return a list of table names wrapped by a {@link CompletableFuture}.
|
||||||
*/
|
*/
|
||||||
CompletableFuture<List<TableName>> listTableNames(Optional<Pattern> pattern,
|
CompletableFuture<List<TableName>> listTableNames(Pattern pattern, boolean includeSysTables);
|
||||||
boolean includeSysTables);
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Method for getting the tableDescriptor
|
* Method for getting the tableDescriptor
|
||||||
|
@ -108,9 +119,7 @@ public interface AsyncAdmin {
|
||||||
* Creates a new table.
|
* Creates a new table.
|
||||||
* @param desc table descriptor for table
|
* @param desc table descriptor for table
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<Void> createTable(TableDescriptor desc) {
|
CompletableFuture<Void> createTable(TableDescriptor desc);
|
||||||
return createTable(desc, Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Creates a new table with the specified number of regions. The start key specified will become
|
* Creates a new table with the specified number of regions. The start key specified will become
|
||||||
|
@ -133,7 +142,7 @@ public interface AsyncAdmin {
|
||||||
* @param desc table descriptor for table
|
* @param desc table descriptor for table
|
||||||
* @param splitKeys array of split keys for the initial regions of the table
|
* @param splitKeys array of split keys for the initial regions of the table
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> createTable(TableDescriptor desc, Optional<byte[][]> splitKeys);
|
CompletableFuture<Void> createTable(TableDescriptor desc, byte[][] splitKeys);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Deletes a table.
|
* Deletes a table.
|
||||||
|
@ -179,9 +188,7 @@ public interface AsyncAdmin {
|
||||||
* @return true if all regions of the table are available. The return value will be wrapped by a
|
* @return true if all regions of the table are available. The return value will be wrapped by a
|
||||||
* {@link CompletableFuture}.
|
* {@link CompletableFuture}.
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<Boolean> isTableAvailable(TableName tableName) {
|
CompletableFuture<Boolean> isTableAvailable(TableName tableName);
|
||||||
return isTableAvailable(tableName, null);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Use this api to check if the table has been created with the specified number of splitkeys
|
* Use this api to check if the table has been created with the specified number of splitkeys
|
||||||
|
@ -274,9 +281,7 @@ public interface AsyncAdmin {
|
||||||
* was sent to HBase and may need some time to finish the compact operation.
|
* was sent to HBase and may need some time to finish the compact operation.
|
||||||
* @param tableName table to compact
|
* @param tableName table to compact
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<Void> compact(TableName tableName) {
|
CompletableFuture<Void> compact(TableName tableName);
|
||||||
return compact(tableName, Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Compact a column family within a table. When the returned CompletableFuture is done, it only
|
* Compact a column family within a table. When the returned CompletableFuture is done, it only
|
||||||
|
@ -286,16 +291,14 @@ public interface AsyncAdmin {
|
||||||
* @param columnFamily column family within a table. If not present, compact the table's all
|
* @param columnFamily column family within a table. If not present, compact the table's all
|
||||||
* column families.
|
* column families.
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> compact(TableName tableName, Optional<byte[]> columnFamily);
|
CompletableFuture<Void> compact(TableName tableName, byte[] columnFamily);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Compact an individual region. When the returned CompletableFuture is done, it only means the
|
* 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.
|
* compact request was sent to HBase and may need some time to finish the compact operation.
|
||||||
* @param regionName region to compact
|
* @param regionName region to compact
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<Void> compactRegion(byte[] regionName) {
|
CompletableFuture<Void> compactRegion(byte[] regionName);
|
||||||
return compactRegion(regionName, Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Compact a column family within a region. When the returned CompletableFuture is done, it only
|
* Compact a column family within a region. When the returned CompletableFuture is done, it only
|
||||||
|
@ -305,16 +308,14 @@ public interface AsyncAdmin {
|
||||||
* @param columnFamily column family within a region. If not present, compact the region's all
|
* @param columnFamily column family within a region. If not present, compact the region's all
|
||||||
* column families.
|
* column families.
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> compactRegion(byte[] regionName, Optional<byte[]> columnFamily);
|
CompletableFuture<Void> compactRegion(byte[] regionName, byte[] columnFamily);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Major compact a table. When the returned CompletableFuture is done, it only means the compact
|
* 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.
|
* request was sent to HBase and may need some time to finish the compact operation.
|
||||||
* @param tableName table to major compact
|
* @param tableName table to major compact
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<Void> majorCompact(TableName tableName) {
|
CompletableFuture<Void> majorCompact(TableName tableName);
|
||||||
return majorCompact(tableName, Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Major compact a column family within a table. When the returned CompletableFuture is done, it
|
* Major compact a column family within a table. When the returned CompletableFuture is done, it
|
||||||
|
@ -324,16 +325,14 @@ public interface AsyncAdmin {
|
||||||
* @param columnFamily column family within a table. If not present, major compact the table's all
|
* @param columnFamily column family within a table. If not present, major compact the table's all
|
||||||
* column families.
|
* column families.
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> majorCompact(TableName tableName, Optional<byte[]> columnFamily);
|
CompletableFuture<Void> majorCompact(TableName tableName, byte[] columnFamily);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Major compact a region. When the returned CompletableFuture is done, it only means the compact
|
* 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.
|
* request was sent to HBase and may need some time to finish the compact operation.
|
||||||
* @param regionName region to major compact
|
* @param regionName region to major compact
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<Void> majorCompactRegion(byte[] regionName) {
|
CompletableFuture<Void> majorCompactRegion(byte[] regionName);
|
||||||
return majorCompactRegion(regionName, Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Major compact a column family within region. When the returned CompletableFuture is done, it
|
* Major compact a column family within region. When the returned CompletableFuture is done, it
|
||||||
|
@ -343,7 +342,7 @@ public interface AsyncAdmin {
|
||||||
* @param columnFamily column family within a region. If not present, major compact the region's
|
* @param columnFamily column family within a region. If not present, major compact the region's
|
||||||
* all column families.
|
* all column families.
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> majorCompactRegion(byte[] regionName, Optional<byte[]> columnFamily);
|
CompletableFuture<Void> majorCompactRegion(byte[] regionName, byte[] columnFamily);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Compact all regions on the region server.
|
* Compact all regions on the region server.
|
||||||
|
@ -405,9 +404,7 @@ public interface AsyncAdmin {
|
||||||
* Split an individual region.
|
* Split an individual region.
|
||||||
* @param regionName region to split
|
* @param regionName region to split
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<Void> splitRegion(byte[] regionName) {
|
CompletableFuture<Void> splitRegion(byte[] regionName);
|
||||||
return splitRegion(regionName, Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Split a table.
|
* Split a table.
|
||||||
|
@ -422,7 +419,7 @@ public interface AsyncAdmin {
|
||||||
* @param splitPoint the explicit position to split on. If not present, it will decide by region
|
* @param splitPoint the explicit position to split on. If not present, it will decide by region
|
||||||
* server.
|
* server.
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> splitRegion(byte[] regionName, Optional<byte[]> splitPoint);
|
CompletableFuture<Void> splitRegion(byte[] regionName, byte[] splitPoint);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param regionName Encoded or full name of region to assign.
|
* @param regionName Encoded or full name of region to assign.
|
||||||
|
@ -432,7 +429,7 @@ public interface AsyncAdmin {
|
||||||
/**
|
/**
|
||||||
* Unassign a region from current hosting regionserver. Region will then be assigned to a
|
* 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
|
* regionserver chosen at random. Region could be reassigned back to the same server. Use
|
||||||
* {@link #move(byte[], Optional)} if you want to control the region movement.
|
* {@link #move(byte[], ServerName)} if you want to control the region movement.
|
||||||
* @param regionName Encoded or full name of region to unassign. Will clear any existing
|
* @param regionName Encoded or full name of region to unassign. Will clear any existing
|
||||||
* RegionPlan if one found.
|
* RegionPlan if one found.
|
||||||
* @param forcible 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
|
||||||
|
@ -451,6 +448,12 @@ public interface AsyncAdmin {
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> offline(byte[] regionName);
|
CompletableFuture<Void> offline(byte[] regionName);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Move the region <code>r</code> to a random server.
|
||||||
|
* @param regionName Encoded or full name of region to move.
|
||||||
|
*/
|
||||||
|
CompletableFuture<Void> move(byte[] regionName);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Move the region <code>r</code> to <code>dest</code>.
|
* Move the region <code>r</code> to <code>dest</code>.
|
||||||
* @param regionName Encoded or full name of region to move.
|
* @param regionName Encoded or full name of region to move.
|
||||||
|
@ -458,7 +461,7 @@ public interface AsyncAdmin {
|
||||||
* assign to a random server. A server name is made of host, port and startcode. Here is
|
* 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>
|
* an example: <code> host187.example.com,60020,1289493121758</code>
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> move(byte[] regionName, Optional<ServerName> destServerName);
|
CompletableFuture<Void> move(byte[] regionName, ServerName destServerName);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Apply the new quota settings.
|
* Apply the new quota settings.
|
||||||
|
@ -535,9 +538,7 @@ public interface AsyncAdmin {
|
||||||
* @return a list of replication peers description. The return value will be wrapped by a
|
* @return a list of replication peers description. The return value will be wrapped by a
|
||||||
* {@link CompletableFuture}.
|
* {@link CompletableFuture}.
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers() {
|
CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers();
|
||||||
return listReplicationPeers(Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Return a list of replication peers.
|
* Return a list of replication peers.
|
||||||
|
@ -545,8 +546,7 @@ public interface AsyncAdmin {
|
||||||
* @return a list of replication peers description. The return value will be wrapped by a
|
* @return a list of replication peers description. The return value will be wrapped by a
|
||||||
* {@link CompletableFuture}.
|
* {@link CompletableFuture}.
|
||||||
*/
|
*/
|
||||||
CompletableFuture<List<ReplicationPeerDescription>>
|
CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Pattern pattern);
|
||||||
listReplicationPeers(Optional<Pattern> pattern);
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Find all table and column families that are replicated from this cluster
|
* Find all table and column families that are replicated from this cluster
|
||||||
|
@ -652,16 +652,22 @@ public interface AsyncAdmin {
|
||||||
* @return a list of snapshot descriptors for completed snapshots wrapped by a
|
* @return a list of snapshot descriptors for completed snapshots wrapped by a
|
||||||
* {@link CompletableFuture}
|
* {@link CompletableFuture}
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<List<SnapshotDescription>> listSnapshots() {
|
CompletableFuture<List<SnapshotDescription>> listSnapshots();
|
||||||
return listSnapshots(Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* List all the completed snapshots matching the given pattern.
|
* List all the completed snapshots matching the given pattern.
|
||||||
* @param pattern The compiled regular expression to match against
|
* @param pattern The compiled regular expression to match against
|
||||||
* @return - returns a List of SnapshotDescription wrapped by a {@link CompletableFuture}
|
* @return - returns a List of SnapshotDescription wrapped by a {@link CompletableFuture}
|
||||||
*/
|
*/
|
||||||
CompletableFuture<List<SnapshotDescription>> listSnapshots(Optional<Pattern> pattern);
|
CompletableFuture<List<SnapshotDescription>> listSnapshots(Pattern pattern);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* List all the completed snapshots matching the given table name pattern.
|
||||||
|
* @param tableNamePattern The compiled table name regular expression to match against
|
||||||
|
* @return - returns a List of completed SnapshotDescription wrapped by a
|
||||||
|
* {@link CompletableFuture}
|
||||||
|
*/
|
||||||
|
CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* List all the completed snapshots matching the given table name regular expression and snapshot
|
* List all the completed snapshots matching the given table name regular expression and snapshot
|
||||||
|
@ -680,13 +686,22 @@ public interface AsyncAdmin {
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> deleteSnapshot(String snapshotName);
|
CompletableFuture<Void> deleteSnapshot(String snapshotName);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Delete all existing snapshots.
|
||||||
|
*/
|
||||||
|
CompletableFuture<Void> deleteSnapshots();
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Delete existing snapshots whose names match the pattern passed.
|
* Delete existing snapshots whose names match the pattern passed.
|
||||||
* @param pattern pattern for names of the snapshot to match
|
* @param pattern pattern for names of the snapshot to match
|
||||||
*/
|
*/
|
||||||
default CompletableFuture<Void> deleteSnapshots(Pattern pattern) {
|
CompletableFuture<Void> deleteSnapshots(Pattern pattern);
|
||||||
return deleteTableSnapshots(null, pattern);
|
|
||||||
}
|
/**
|
||||||
|
* Delete all existing snapshots matching the given table name pattern.
|
||||||
|
* @param tableNamePattern The compiled table name regular expression to match against
|
||||||
|
*/
|
||||||
|
CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Delete all existing snapshots matching the given table name regular expression and snapshot
|
* Delete all existing snapshots matching the given table name regular expression and snapshot
|
||||||
|
@ -822,15 +837,6 @@ public interface AsyncAdmin {
|
||||||
ClusterStatus::getMasterInfoPort);
|
ClusterStatus::getMasterInfoPort);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Get a list of {@link RegionLoad} of all regions hosted on a region seerver.
|
|
||||||
* @param serverName
|
|
||||||
* @return a list of {@link RegionLoad} wrapped by {@link CompletableFuture}
|
|
||||||
*/
|
|
||||||
default CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName) {
|
|
||||||
return getRegionLoads(serverName, Optional.empty());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Shuts down the HBase cluster.
|
* Shuts down the HBase cluster.
|
||||||
*/
|
*/
|
||||||
|
@ -877,14 +883,20 @@ public interface AsyncAdmin {
|
||||||
*/
|
*/
|
||||||
CompletableFuture<Void> clearCompactionQueues(ServerName serverName, Set<String> queues);
|
CompletableFuture<Void> clearCompactionQueues(ServerName serverName, Set<String> queues);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get a list of {@link RegionLoad} of all regions hosted on a region seerver.
|
||||||
|
* @param serverName
|
||||||
|
* @return a list of {@link RegionLoad} wrapped by {@link CompletableFuture}
|
||||||
|
*/
|
||||||
|
CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Get a list of {@link RegionLoad} of all regions hosted on a region seerver for a table.
|
* Get a list of {@link RegionLoad} of all regions hosted on a region seerver for a table.
|
||||||
* @param serverName
|
* @param serverName
|
||||||
* @param tableName
|
* @param tableName
|
||||||
* @return a list of {@link RegionLoad} wrapped by {@link CompletableFuture}
|
* @return a list of {@link RegionLoad} wrapped by {@link CompletableFuture}
|
||||||
*/
|
*/
|
||||||
CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName,
|
CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName, TableName tableName);
|
||||||
Optional<TableName> tableName);
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Check whether master is in maintenance mode
|
* Check whether master is in maintenance mode
|
||||||
|
|
|
@ -43,7 +43,6 @@ import org.apache.hadoop.hbase.quotas.QuotaFilter;
|
||||||
import org.apache.hadoop.hbase.quotas.QuotaSettings;
|
import org.apache.hadoop.hbase.quotas.QuotaSettings;
|
||||||
import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
|
import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
|
||||||
import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
|
import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
|
||||||
import org.apache.hadoop.hbase.util.Pair;
|
|
||||||
import org.apache.yetus.audience.InterfaceAudience;
|
import org.apache.yetus.audience.InterfaceAudience;
|
||||||
|
|
||||||
import com.google.protobuf.RpcChannel;
|
import com.google.protobuf.RpcChannel;
|
||||||
|
@ -84,13 +83,23 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<TableDescriptor>> listTables(Optional<Pattern> pattern,
|
public CompletableFuture<List<TableDescriptor>> listTables(boolean includeSysTables) {
|
||||||
|
return wrap(rawAdmin.listTables(includeSysTables));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<TableDescriptor>> listTables(Pattern pattern,
|
||||||
boolean includeSysTables) {
|
boolean includeSysTables) {
|
||||||
return wrap(rawAdmin.listTables(pattern, includeSysTables));
|
return wrap(rawAdmin.listTables(pattern, includeSysTables));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<TableName>> listTableNames(Optional<Pattern> pattern,
|
public CompletableFuture<List<TableName>> listTableNames(boolean includeSysTables) {
|
||||||
|
return wrap(rawAdmin.listTableNames(includeSysTables));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<TableName>> listTableNames(Pattern pattern,
|
||||||
boolean includeSysTables) {
|
boolean includeSysTables) {
|
||||||
return wrap(rawAdmin.listTableNames(pattern, includeSysTables));
|
return wrap(rawAdmin.listTableNames(pattern, includeSysTables));
|
||||||
}
|
}
|
||||||
|
@ -100,6 +109,11 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return wrap(rawAdmin.getTableDescriptor(tableName));
|
return wrap(rawAdmin.getTableDescriptor(tableName));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> createTable(TableDescriptor desc) {
|
||||||
|
return wrap(rawAdmin.createTable(desc));
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> createTable(TableDescriptor desc, byte[] startKey, byte[] endKey,
|
public CompletableFuture<Void> createTable(TableDescriptor desc, byte[] startKey, byte[] endKey,
|
||||||
int numRegions) {
|
int numRegions) {
|
||||||
|
@ -107,7 +121,7 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> createTable(TableDescriptor desc, Optional<byte[][]> splitKeys) {
|
public CompletableFuture<Void> createTable(TableDescriptor desc, byte[][] splitKeys) {
|
||||||
return wrap(rawAdmin.createTable(desc, splitKeys));
|
return wrap(rawAdmin.createTable(desc, splitKeys));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -141,6 +155,11 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return wrap(rawAdmin.isTableDisabled(tableName));
|
return wrap(rawAdmin.isTableDisabled(tableName));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Boolean> isTableAvailable(TableName tableName) {
|
||||||
|
return wrap(rawAdmin.isTableAvailable(tableName));
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Boolean> isTableAvailable(TableName tableName, byte[][] splitKeys) {
|
public CompletableFuture<Boolean> isTableAvailable(TableName tableName, byte[][] splitKeys) {
|
||||||
return wrap(rawAdmin.isTableAvailable(tableName, splitKeys));
|
return wrap(rawAdmin.isTableAvailable(tableName, splitKeys));
|
||||||
|
@ -209,23 +228,42 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> compact(TableName tableName, Optional<byte[]> columnFamily) {
|
public CompletableFuture<Void> compact(TableName tableName) {
|
||||||
|
return wrap(rawAdmin.compact(tableName));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> compact(TableName tableName, byte[] columnFamily) {
|
||||||
return wrap(rawAdmin.compact(tableName, columnFamily));
|
return wrap(rawAdmin.compact(tableName, columnFamily));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> compactRegion(byte[] regionName, Optional<byte[]> columnFamily) {
|
public CompletableFuture<Void> compactRegion(byte[] regionName) {
|
||||||
|
return wrap(rawAdmin.compactRegion(regionName));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> compactRegion(byte[] regionName, byte[] columnFamily) {
|
||||||
return wrap(rawAdmin.compactRegion(regionName, columnFamily));
|
return wrap(rawAdmin.compactRegion(regionName, columnFamily));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> majorCompact(TableName tableName, Optional<byte[]> columnFamily) {
|
public CompletableFuture<Void> majorCompact(TableName tableName) {
|
||||||
|
return wrap(rawAdmin.majorCompact(tableName));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> majorCompact(TableName tableName, byte[] columnFamily) {
|
||||||
return wrap(rawAdmin.majorCompact(tableName, columnFamily));
|
return wrap(rawAdmin.majorCompact(tableName, columnFamily));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void>
|
public CompletableFuture<Void> majorCompactRegion(byte[] regionName) {
|
||||||
majorCompactRegion(byte[] regionName, Optional<byte[]> columnFamily) {
|
return wrap(rawAdmin.majorCompactRegion(regionName));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> majorCompactRegion(byte[] regionName, byte[] columnFamily) {
|
||||||
return wrap(rawAdmin.majorCompactRegion(regionName, columnFamily));
|
return wrap(rawAdmin.majorCompactRegion(regionName, columnFamily));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -276,7 +314,12 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> splitRegion(byte[] regionName, Optional<byte[]> splitPoint) {
|
public CompletableFuture<Void> splitRegion(byte[] regionName) {
|
||||||
|
return wrap(rawAdmin.splitRegion(regionName));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> splitRegion(byte[] regionName, byte[] splitPoint) {
|
||||||
return wrap(rawAdmin.splitRegion(regionName, splitPoint));
|
return wrap(rawAdmin.splitRegion(regionName, splitPoint));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -296,7 +339,12 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> move(byte[] regionName, Optional<ServerName> destServerName) {
|
public CompletableFuture<Void> move(byte[] regionName) {
|
||||||
|
return wrap(rawAdmin.move(regionName));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> move(byte[] regionName, ServerName destServerName) {
|
||||||
return wrap(rawAdmin.move(regionName, destServerName));
|
return wrap(rawAdmin.move(regionName, destServerName));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -355,8 +403,12 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(
|
public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers() {
|
||||||
Optional<Pattern> pattern) {
|
return wrap(rawAdmin.listReplicationPeers());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Pattern pattern) {
|
||||||
return wrap(rawAdmin.listReplicationPeers(pattern));
|
return wrap(rawAdmin.listReplicationPeers(pattern));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -391,10 +443,20 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<SnapshotDescription>> listSnapshots(Optional<Pattern> pattern) {
|
public CompletableFuture<List<SnapshotDescription>> listSnapshots() {
|
||||||
|
return wrap(rawAdmin.listSnapshots());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<SnapshotDescription>> listSnapshots(Pattern pattern) {
|
||||||
return wrap(rawAdmin.listSnapshots(pattern));
|
return wrap(rawAdmin.listSnapshots(pattern));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern) {
|
||||||
|
return wrap(rawAdmin.listTableSnapshots(tableNamePattern));
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern,
|
public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern,
|
||||||
Pattern snapshotNamePattern) {
|
Pattern snapshotNamePattern) {
|
||||||
|
@ -406,6 +468,21 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return wrap(rawAdmin.deleteSnapshot(snapshotName));
|
return wrap(rawAdmin.deleteSnapshot(snapshotName));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> deleteSnapshots() {
|
||||||
|
return wrap(rawAdmin.deleteSnapshots());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> deleteSnapshots(Pattern pattern) {
|
||||||
|
return wrap(rawAdmin.deleteSnapshots(pattern));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern) {
|
||||||
|
return wrap(rawAdmin.deleteTableSnapshots(tableNamePattern));
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern,
|
public CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern,
|
||||||
Pattern snapshotNamePattern) {
|
Pattern snapshotNamePattern) {
|
||||||
|
@ -512,9 +589,14 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return wrap(rawAdmin.getSecurityCapabilities());
|
return wrap(rawAdmin.getSecurityCapabilities());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName) {
|
||||||
|
return wrap(rawAdmin.getRegionLoads(serverName));
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName,
|
public CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName,
|
||||||
Optional<TableName> tableName) {
|
TableName tableName) {
|
||||||
return wrap(rawAdmin.getRegionLoads(serverName, tableName));
|
return wrap(rawAdmin.getRegionLoads(serverName, tableName));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -427,8 +427,8 @@ class AsyncRpcRetryingCallerFactory {
|
||||||
|
|
||||||
public AsyncAdminRequestRetryingCaller<T> build() {
|
public AsyncAdminRequestRetryingCaller<T> build() {
|
||||||
return new AsyncAdminRequestRetryingCaller<T>(retryTimer, conn, pauseNs, maxAttempts,
|
return new AsyncAdminRequestRetryingCaller<T>(retryTimer, conn, pauseNs, maxAttempts,
|
||||||
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, serverName, checkNotNull(callable,
|
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, checkNotNull(serverName,
|
||||||
"action is null"));
|
"serverName is null"), checkNotNull(callable, "action is null"));
|
||||||
}
|
}
|
||||||
|
|
||||||
public CompletableFuture<T> call() {
|
public CompletableFuture<T> call() {
|
||||||
|
@ -488,8 +488,8 @@ class AsyncRpcRetryingCallerFactory {
|
||||||
|
|
||||||
public AsyncServerRequestRpcRetryingCaller<T> build() {
|
public AsyncServerRequestRpcRetryingCaller<T> build() {
|
||||||
return new AsyncServerRequestRpcRetryingCaller<T>(retryTimer, conn, pauseNs, maxAttempts,
|
return new AsyncServerRequestRpcRetryingCaller<T>(retryTimer, conn, pauseNs, maxAttempts,
|
||||||
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, serverName, checkNotNull(callable,
|
operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt, checkNotNull(serverName,
|
||||||
"action is null"));
|
"serverName is null"), checkNotNull(callable, "action is null"));
|
||||||
}
|
}
|
||||||
|
|
||||||
public CompletableFuture<T> call() {
|
public CompletableFuture<T> call() {
|
||||||
|
|
|
@ -1371,14 +1371,14 @@ public class HBaseAdmin implements Admin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void move(final byte [] encodedRegionName, final byte [] destServerName)
|
public void move(final byte[] encodedRegionName, final byte[] destServerName) throws IOException {
|
||||||
throws IOException {
|
|
||||||
executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
|
executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
|
||||||
@Override
|
@Override
|
||||||
protected Void rpcCall() throws Exception {
|
protected Void rpcCall() throws Exception {
|
||||||
setPriority(encodedRegionName);
|
setPriority(encodedRegionName);
|
||||||
MoveRegionRequest request =
|
MoveRegionRequest request =
|
||||||
RequestConverter.buildMoveRegionRequest(encodedRegionName, destServerName);
|
RequestConverter.buildMoveRegionRequest(encodedRegionName,
|
||||||
|
destServerName != null ? ServerName.valueOf(Bytes.toString(destServerName)) : null);
|
||||||
master.moveRegion(getRpcController(), request);
|
master.moveRegion(getRpcController(), request);
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
|
@ -26,7 +26,6 @@ import java.util.Collection;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
import java.util.EnumSet;
|
import java.util.EnumSet;
|
||||||
import java.util.HashMap;
|
import java.util.HashMap;
|
||||||
import java.util.LinkedList;
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
|
@ -82,9 +81,7 @@ import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
|
||||||
import org.apache.hadoop.hbase.util.Bytes;
|
import org.apache.hadoop.hbase.util.Bytes;
|
||||||
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
|
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
|
||||||
import org.apache.hadoop.hbase.util.ForeignExceptionUtil;
|
import org.apache.hadoop.hbase.util.ForeignExceptionUtil;
|
||||||
import org.apache.hadoop.hbase.util.Pair;
|
|
||||||
import org.apache.yetus.audience.InterfaceAudience;
|
import org.apache.yetus.audience.InterfaceAudience;
|
||||||
|
|
||||||
import org.apache.hadoop.hbase.shaded.com.google.protobuf.RpcCallback;
|
import org.apache.hadoop.hbase.shaded.com.google.protobuf.RpcCallback;
|
||||||
import org.apache.hadoop.hbase.shaded.io.netty.util.Timeout;
|
import org.apache.hadoop.hbase.shaded.io.netty.util.Timeout;
|
||||||
import org.apache.hadoop.hbase.shaded.io.netty.util.TimerTask;
|
import org.apache.hadoop.hbase.shaded.io.netty.util.TimerTask;
|
||||||
|
@ -245,6 +242,7 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.Updat
|
||||||
import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos;
|
import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos;
|
||||||
|
|
||||||
import com.google.common.annotations.VisibleForTesting;
|
import com.google.common.annotations.VisibleForTesting;
|
||||||
|
import com.google.common.base.Preconditions;
|
||||||
import com.google.protobuf.Message;
|
import com.google.protobuf.Message;
|
||||||
import com.google.protobuf.RpcChannel;
|
import com.google.protobuf.RpcChannel;
|
||||||
|
|
||||||
|
@ -385,28 +383,54 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<TableDescriptor>> listTables(Optional<Pattern> pattern,
|
public CompletableFuture<List<TableDescriptor>> listTables(boolean includeSysTables) {
|
||||||
|
return getTableDescriptors(RequestConverter.buildGetTableDescriptorsRequest(null,
|
||||||
|
includeSysTables));
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* {@link #listTables(boolean)}
|
||||||
|
*/
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<TableDescriptor>> listTables(Pattern pattern,
|
||||||
boolean includeSysTables) {
|
boolean includeSysTables) {
|
||||||
|
Preconditions.checkNotNull(pattern,
|
||||||
|
"pattern is null. If you don't specify a pattern, use listTables(boolean) instead");
|
||||||
|
return getTableDescriptors(RequestConverter.buildGetTableDescriptorsRequest(pattern,
|
||||||
|
includeSysTables));
|
||||||
|
}
|
||||||
|
|
||||||
|
private CompletableFuture<List<TableDescriptor>>
|
||||||
|
getTableDescriptors(GetTableDescriptorsRequest request) {
|
||||||
return this.<List<TableDescriptor>> newMasterCaller()
|
return this.<List<TableDescriptor>> newMasterCaller()
|
||||||
.action((controller, stub) -> this
|
.action((controller, stub) -> this
|
||||||
.<GetTableDescriptorsRequest, GetTableDescriptorsResponse, List<TableDescriptor>> call(
|
.<GetTableDescriptorsRequest, GetTableDescriptorsResponse, List<TableDescriptor>> call(
|
||||||
controller, stub,
|
controller, stub, request, (s, c, req, done) -> s.getTableDescriptors(c, req, done),
|
||||||
RequestConverter.buildGetTableDescriptorsRequest(pattern, includeSysTables),
|
|
||||||
(s, c, req, done) -> s.getTableDescriptors(c, req, done),
|
|
||||||
(resp) -> ProtobufUtil.toTableDescriptorList(resp)))
|
(resp) -> ProtobufUtil.toTableDescriptorList(resp)))
|
||||||
.call();
|
.call();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<TableName>> listTableNames(Optional<Pattern> pattern,
|
public CompletableFuture<List<TableName>> listTableNames(boolean includeSysTables) {
|
||||||
boolean includeSysTables) {
|
return getTableNames(RequestConverter.buildGetTableNamesRequest(null, includeSysTables));
|
||||||
return this.<List<TableName>> newMasterCaller()
|
}
|
||||||
.action((controller, stub) -> this
|
|
||||||
.<GetTableNamesRequest, GetTableNamesResponse, List<TableName>> call(controller, stub,
|
@Override
|
||||||
RequestConverter.buildGetTableNamesRequest(pattern, includeSysTables),
|
public CompletableFuture<List<TableName>>
|
||||||
(s, c, req, done) -> s.getTableNames(c, req, done),
|
listTableNames(Pattern pattern, boolean includeSysTables) {
|
||||||
(resp) -> ProtobufUtil.toTableNameList(resp.getTableNamesList())))
|
Preconditions.checkNotNull(pattern,
|
||||||
.call();
|
"pattern is null. If you don't specify a pattern, use listTableNames(boolean) instead");
|
||||||
|
return getTableNames(RequestConverter.buildGetTableNamesRequest(pattern, includeSysTables));
|
||||||
|
}
|
||||||
|
|
||||||
|
private CompletableFuture<List<TableName>> getTableNames(GetTableNamesRequest request) {
|
||||||
|
return this
|
||||||
|
.<List<TableName>> newMasterCaller()
|
||||||
|
.action(
|
||||||
|
(controller, stub) -> this
|
||||||
|
.<GetTableNamesRequest, GetTableNamesResponse, List<TableName>> call(controller,
|
||||||
|
stub, request, (s, c, req, done) -> s.getTableNames(c, req, done),
|
||||||
|
(resp) -> ProtobufUtil.toTableNameList(resp.getTableNamesList()))).call();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -432,32 +456,42 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return future;
|
return future;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> createTable(TableDescriptor desc) {
|
||||||
|
return createTable(desc.getTableName(),
|
||||||
|
RequestConverter.buildCreateTableRequest(desc, null, ng.getNonceGroup(), ng.newNonce()));
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> createTable(TableDescriptor desc, byte[] startKey, byte[] endKey,
|
public CompletableFuture<Void> createTable(TableDescriptor desc, byte[] startKey, byte[] endKey,
|
||||||
int numRegions) {
|
int numRegions) {
|
||||||
try {
|
try {
|
||||||
return createTable(desc, Optional.of(getSplitKeys(startKey, endKey, numRegions)));
|
return createTable(desc, getSplitKeys(startKey, endKey, numRegions));
|
||||||
} catch (IllegalArgumentException e) {
|
} catch (IllegalArgumentException e) {
|
||||||
return failedFuture(e);
|
return failedFuture(e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> createTable(TableDescriptor desc, Optional<byte[][]> splitKeys) {
|
public CompletableFuture<Void> createTable(TableDescriptor desc, byte[][] splitKeys) {
|
||||||
if (desc.getTableName() == null) {
|
Preconditions.checkNotNull(splitKeys, "splitKeys is null. If you don't specify splitKeys,"
|
||||||
return failedFuture(new IllegalArgumentException("TableName cannot be null"));
|
+ " use createTable(TableDescriptor) instead");
|
||||||
}
|
|
||||||
try {
|
try {
|
||||||
splitKeys.ifPresent(keys -> verifySplitKeys(keys));
|
verifySplitKeys(splitKeys);
|
||||||
return this.<CreateTableRequest, CreateTableResponse> procedureCall(RequestConverter
|
return createTable(desc.getTableName(), RequestConverter.buildCreateTableRequest(desc,
|
||||||
.buildCreateTableRequest(desc, splitKeys, ng.getNonceGroup(), ng.newNonce()), (s, c, req,
|
splitKeys, ng.getNonceGroup(), ng.newNonce()));
|
||||||
done) -> s.createTable(c, req, done), (resp) -> resp.getProcId(),
|
|
||||||
new CreateTableProcedureBiConsumer(this, desc.getTableName()));
|
|
||||||
} catch (IllegalArgumentException e) {
|
} catch (IllegalArgumentException e) {
|
||||||
return failedFuture(e);
|
return failedFuture(e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private CompletableFuture<Void> createTable(TableName tableName, CreateTableRequest request) {
|
||||||
|
Preconditions.checkNotNull(tableName, "table name is null");
|
||||||
|
return this.<CreateTableRequest, CreateTableResponse> procedureCall(request,
|
||||||
|
(s, c, req, done) -> s.createTable(c, req, done), (resp) -> resp.getProcId(),
|
||||||
|
new CreateTableProcedureBiConsumer(this, tableName));
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> deleteTable(TableName tableName) {
|
public CompletableFuture<Void> deleteTable(TableName tableName) {
|
||||||
return this.<DeleteTableRequest, DeleteTableResponse> procedureCall(RequestConverter
|
return this.<DeleteTableRequest, DeleteTableResponse> procedureCall(RequestConverter
|
||||||
|
@ -526,11 +560,18 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Boolean> isTableAvailable(TableName tableName) {
|
public CompletableFuture<Boolean> isTableAvailable(TableName tableName) {
|
||||||
return isTableAvailable(tableName, null);
|
return isTableAvailable(tableName, Optional.empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Boolean> isTableAvailable(TableName tableName, byte[][] splitKeys) {
|
public CompletableFuture<Boolean> isTableAvailable(TableName tableName, byte[][] splitKeys) {
|
||||||
|
Preconditions.checkNotNull(splitKeys, "splitKeys is null. If you don't specify splitKeys,"
|
||||||
|
+ " use isTableAvailable(TableName) instead");
|
||||||
|
return isTableAvailable(tableName, Optional.of(splitKeys));
|
||||||
|
}
|
||||||
|
|
||||||
|
private CompletableFuture<Boolean> isTableAvailable(TableName tableName,
|
||||||
|
Optional<byte[][]> splitKeys) {
|
||||||
CompletableFuture<Boolean> future = new CompletableFuture<>();
|
CompletableFuture<Boolean> future = new CompletableFuture<>();
|
||||||
isTableEnabled(tableName).whenComplete(
|
isTableEnabled(tableName).whenComplete(
|
||||||
(enabled, error) -> {
|
(enabled, error) -> {
|
||||||
|
@ -548,54 +589,46 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
future.completeExceptionally(error1);
|
future.completeExceptionally(error1);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
int notDeployed = 0;
|
List<HRegionLocation> notDeployedRegions =
|
||||||
int regionCount = 0;
|
locations.stream().filter(loc -> loc.getServerName() == null)
|
||||||
for (HRegionLocation location : locations) {
|
.collect(Collectors.toList());
|
||||||
RegionInfo info = location.getRegionInfo();
|
if (notDeployedRegions.size() > 0) {
|
||||||
if (location.getServerName() == null) {
|
|
||||||
if (LOG.isDebugEnabled()) {
|
|
||||||
LOG.debug("Table " + tableName + " has not deployed region "
|
|
||||||
+ info.getEncodedName());
|
|
||||||
}
|
|
||||||
notDeployed++;
|
|
||||||
} else if (splitKeys != null
|
|
||||||
&& !Bytes.equals(info.getStartKey(), HConstants.EMPTY_BYTE_ARRAY)) {
|
|
||||||
for (byte[] splitKey : splitKeys) {
|
|
||||||
// Just check if the splitkey is available
|
|
||||||
if (Bytes.equals(info.getStartKey(), splitKey)) {
|
|
||||||
regionCount++;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
// Always empty start row should be counted
|
|
||||||
regionCount++;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (notDeployed > 0) {
|
|
||||||
if (LOG.isDebugEnabled()) {
|
if (LOG.isDebugEnabled()) {
|
||||||
LOG.debug("Table " + tableName + " has " + notDeployed + " regions");
|
LOG.debug("Table " + tableName + " has " + notDeployedRegions.size()
|
||||||
|
+ " regions");
|
||||||
}
|
}
|
||||||
future.complete(false);
|
future.complete(false);
|
||||||
} else if (splitKeys != null && regionCount != splitKeys.length + 1) {
|
return;
|
||||||
if (LOG.isDebugEnabled()) {
|
|
||||||
LOG.debug("Table " + tableName + " expected to have "
|
|
||||||
+ (splitKeys.length + 1) + " regions, but only " + regionCount
|
|
||||||
+ " available");
|
|
||||||
}
|
|
||||||
future.complete(false);
|
|
||||||
} else {
|
|
||||||
if (LOG.isDebugEnabled()) {
|
|
||||||
LOG.debug("Table " + tableName + " should be available");
|
|
||||||
}
|
|
||||||
future.complete(true);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Optional<Boolean> available =
|
||||||
|
splitKeys.map(keys -> compareRegionsWithSplitKeys(locations, keys));
|
||||||
|
future.complete(available.orElse(true));
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
return future;
|
return future;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private boolean compareRegionsWithSplitKeys(List<HRegionLocation> locations, byte[][] splitKeys) {
|
||||||
|
int regionCount = 0;
|
||||||
|
for (HRegionLocation location : locations) {
|
||||||
|
RegionInfo info = location.getRegion();
|
||||||
|
if (Bytes.equals(info.getStartKey(), HConstants.EMPTY_BYTE_ARRAY)) {
|
||||||
|
regionCount++;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
for (byte[] splitKey : splitKeys) {
|
||||||
|
// Just check if the splitkey is available
|
||||||
|
if (Bytes.equals(info.getStartKey(), splitKey)) {
|
||||||
|
regionCount++;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return regionCount == splitKeys.length + 1;
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> addColumnFamily(TableName tableName, ColumnFamilyDescriptor columnFamily) {
|
public CompletableFuture<Void> addColumnFamily(TableName tableName, ColumnFamilyDescriptor columnFamily) {
|
||||||
return this.<AddColumnRequest, AddColumnResponse> procedureCall(
|
return this.<AddColumnRequest, AddColumnResponse> procedureCall(
|
||||||
|
@ -757,22 +790,50 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> compact(TableName tableName, Optional<byte[]> columnFamily) {
|
public CompletableFuture<Void> compact(TableName tableName) {
|
||||||
|
return compact(tableName, null, false, CompactType.NORMAL);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> compact(TableName tableName, byte[] columnFamily) {
|
||||||
|
Preconditions.checkNotNull(columnFamily,
|
||||||
|
"columnFamily is null. If you don't specify a columnFamily, use compact(TableName) instead");
|
||||||
return compact(tableName, columnFamily, false, CompactType.NORMAL);
|
return compact(tableName, columnFamily, false, CompactType.NORMAL);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> compactRegion(byte[] regionName, Optional<byte[]> columnFamily) {
|
public CompletableFuture<Void> compactRegion(byte[] regionName) {
|
||||||
|
return compactRegion(regionName, null, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> compactRegion(byte[] regionName, byte[] columnFamily) {
|
||||||
|
Preconditions.checkNotNull(columnFamily, "columnFamily is null."
|
||||||
|
+ " If you don't specify a columnFamily, use compactRegion(regionName) instead");
|
||||||
return compactRegion(regionName, columnFamily, false);
|
return compactRegion(regionName, columnFamily, false);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> majorCompact(TableName tableName, Optional<byte[]> columnFamily) {
|
public CompletableFuture<Void> majorCompact(TableName tableName) {
|
||||||
|
return compact(tableName, null, true, CompactType.NORMAL);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> majorCompact(TableName tableName, byte[] columnFamily) {
|
||||||
|
Preconditions.checkNotNull(columnFamily, "columnFamily is null."
|
||||||
|
+ " If you don't specify a columnFamily, use majorCompact(TableName) instead");
|
||||||
return compact(tableName, columnFamily, true, CompactType.NORMAL);
|
return compact(tableName, columnFamily, true, CompactType.NORMAL);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> majorCompactRegion(byte[] regionName, Optional<byte[]> columnFamily) {
|
public CompletableFuture<Void> majorCompactRegion(byte[] regionName) {
|
||||||
|
return compactRegion(regionName, null, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> majorCompactRegion(byte[] regionName, byte[] columnFamily) {
|
||||||
|
Preconditions.checkNotNull(columnFamily, "columnFamily is null."
|
||||||
|
+ " If you don't specify a columnFamily, use majorCompactRegion(regionName) instead");
|
||||||
return compactRegion(regionName, columnFamily, true);
|
return compactRegion(regionName, columnFamily, true);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -795,7 +856,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
List<CompletableFuture<Void>> compactFutures = new ArrayList<>();
|
List<CompletableFuture<Void>> compactFutures = new ArrayList<>();
|
||||||
if (hRegionInfos != null) {
|
if (hRegionInfos != null) {
|
||||||
hRegionInfos.forEach(region -> compactFutures.add(compact(sn, region, major, Optional.empty())));
|
hRegionInfos.forEach(region -> compactFutures.add(compact(sn, region, major, null)));
|
||||||
}
|
}
|
||||||
CompletableFuture
|
CompletableFuture
|
||||||
.allOf(compactFutures.toArray(new CompletableFuture<?>[compactFutures.size()]))
|
.allOf(compactFutures.toArray(new CompletableFuture<?>[compactFutures.size()]))
|
||||||
|
@ -810,7 +871,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return future;
|
return future;
|
||||||
}
|
}
|
||||||
|
|
||||||
private CompletableFuture<Void> compactRegion(byte[] regionName, Optional<byte[]> columnFamily,
|
private CompletableFuture<Void> compactRegion(byte[] regionName, byte[] columnFamily,
|
||||||
boolean major) {
|
boolean major) {
|
||||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||||
getRegionLocation(regionName).whenComplete(
|
getRegionLocation(regionName).whenComplete(
|
||||||
|
@ -868,7 +929,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
/**
|
/**
|
||||||
* Compact column family of a table, Asynchronous operation even if CompletableFuture.get()
|
* Compact column family of a table, Asynchronous operation even if CompletableFuture.get()
|
||||||
*/
|
*/
|
||||||
private CompletableFuture<Void> compact(final TableName tableName, Optional<byte[]> columnFamily,
|
private CompletableFuture<Void> compact(final TableName tableName, byte[] columnFamily,
|
||||||
final boolean major, CompactType compactType) {
|
final boolean major, CompactType compactType) {
|
||||||
if (CompactType.MOB.equals(compactType)) {
|
if (CompactType.MOB.equals(compactType)) {
|
||||||
// TODO support MOB compact.
|
// TODO support MOB compact.
|
||||||
|
@ -905,7 +966,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
* Compact the region at specific region server.
|
* Compact the region at specific region server.
|
||||||
*/
|
*/
|
||||||
private CompletableFuture<Void> compact(final ServerName sn, final RegionInfo hri,
|
private CompletableFuture<Void> compact(final ServerName sn, final RegionInfo hri,
|
||||||
final boolean major, Optional<byte[]> columnFamily) {
|
final boolean major, byte[] columnFamily) {
|
||||||
return this
|
return this
|
||||||
.<Void> newAdminCaller()
|
.<Void> newAdminCaller()
|
||||||
.serverName(sn)
|
.serverName(sn)
|
||||||
|
@ -1078,7 +1139,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
if (hri == null || hri.isSplitParent()
|
if (hri == null || hri.isSplitParent()
|
||||||
|| hri.getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID)
|
|| hri.getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID)
|
||||||
continue;
|
continue;
|
||||||
splitFutures.add(split(hri, Optional.empty()));
|
splitFutures.add(split(hri, null));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -1114,7 +1175,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
result.completeExceptionally(new IllegalArgumentException(
|
result.completeExceptionally(new IllegalArgumentException(
|
||||||
"Region does not found: rowKey=" + Bytes.toStringBinary(splitPoint)));
|
"Region does not found: rowKey=" + Bytes.toStringBinary(splitPoint)));
|
||||||
} else {
|
} else {
|
||||||
splitRegion(loc.getRegionInfo().getRegionName(), Optional.of(splitPoint))
|
splitRegion(loc.getRegionInfo().getRegionName(), splitPoint)
|
||||||
.whenComplete((ret, err2) -> {
|
.whenComplete((ret, err2) -> {
|
||||||
if (err2 != null) {
|
if (err2 != null) {
|
||||||
result.completeExceptionally(err2);
|
result.completeExceptionally(err2);
|
||||||
|
@ -1129,11 +1190,11 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> splitRegion(byte[] regionName, Optional<byte[]> splitPoint) {
|
public CompletableFuture<Void> splitRegion(byte[] regionName) {
|
||||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||||
getRegionLocation(regionName).whenComplete(
|
getRegionLocation(regionName).whenComplete(
|
||||||
(location, err) -> {
|
(location, err) -> {
|
||||||
RegionInfo regionInfo = location.getRegionInfo();
|
RegionInfo regionInfo = location.getRegion();
|
||||||
if (regionInfo.getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID) {
|
if (regionInfo.getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID) {
|
||||||
future.completeExceptionally(new IllegalArgumentException(
|
future.completeExceptionally(new IllegalArgumentException(
|
||||||
"Can't split replicas directly. "
|
"Can't split replicas directly. "
|
||||||
|
@ -1146,6 +1207,43 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
.toStringBinary(regionName)));
|
.toStringBinary(regionName)));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
split(regionInfo, null).whenComplete((ret, err2) -> {
|
||||||
|
if (err2 != null) {
|
||||||
|
future.completeExceptionally(err2);
|
||||||
|
} else {
|
||||||
|
future.complete(ret);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
|
return future;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> splitRegion(byte[] regionName, byte[] splitPoint) {
|
||||||
|
Preconditions.checkNotNull(splitPoint,
|
||||||
|
"splitPoint is null. If you don't specify a splitPoint, use splitRegion(byte[]) instead");
|
||||||
|
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||||
|
getRegionLocation(regionName).whenComplete(
|
||||||
|
(location, err) -> {
|
||||||
|
RegionInfo regionInfo = location.getRegion();
|
||||||
|
if (regionInfo.getReplicaId() != RegionInfo.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;
|
||||||
|
}
|
||||||
|
if (regionInfo.getStartKey() != null
|
||||||
|
&& Bytes.compareTo(regionInfo.getStartKey(), splitPoint) == 0) {
|
||||||
|
future.completeExceptionally(new IllegalArgumentException(
|
||||||
|
"should not give a splitkey which equals to startkey!"));
|
||||||
|
return;
|
||||||
|
}
|
||||||
split(regionInfo, splitPoint).whenComplete((ret, err2) -> {
|
split(regionInfo, splitPoint).whenComplete((ret, err2) -> {
|
||||||
if (err2 != null) {
|
if (err2 != null) {
|
||||||
future.completeExceptionally(err2);
|
future.completeExceptionally(err2);
|
||||||
|
@ -1157,20 +1255,13 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return future;
|
return future;
|
||||||
}
|
}
|
||||||
|
|
||||||
private CompletableFuture<Void> split(final RegionInfo hri,
|
private CompletableFuture<Void> split(final RegionInfo hri, byte[] splitPoint) {
|
||||||
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!"));
|
|
||||||
}
|
|
||||||
|
|
||||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||||
TableName tableName = hri.getTable();
|
TableName tableName = hri.getTable();
|
||||||
SplitTableRegionRequest request = null;
|
SplitTableRegionRequest request = null;
|
||||||
try {
|
try {
|
||||||
request = RequestConverter
|
request = RequestConverter
|
||||||
.buildSplitTableRegionRequest(hri, splitPoint.isPresent() ? splitPoint.get() : null,
|
.buildSplitTableRegionRequest(hri, splitPoint,
|
||||||
ng.getNonceGroup(), ng.newNonce());
|
ng.getNonceGroup(), ng.newNonce());
|
||||||
} catch (DeserializationException e) {
|
} catch (DeserializationException e) {
|
||||||
future.completeExceptionally(e);
|
future.completeExceptionally(e);
|
||||||
|
@ -1266,7 +1357,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> move(byte[] regionName, Optional<ServerName> destServerName) {
|
public CompletableFuture<Void> move(byte[] regionName) {
|
||||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||||
getRegionInfo(regionName).whenComplete(
|
getRegionInfo(regionName).whenComplete(
|
||||||
(regionInfo, err) -> {
|
(regionInfo, err) -> {
|
||||||
|
@ -1274,12 +1365,9 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
future.completeExceptionally(err);
|
future.completeExceptionally(err);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
this.<Void> newMasterCaller()
|
moveRegion(
|
||||||
.action(
|
RequestConverter.buildMoveRegionRequest(regionInfo.getEncodedNameAsBytes(), null))
|
||||||
(controller, stub) -> this.<MoveRegionRequest, MoveRegionResponse, Void> call(
|
.whenComplete((ret, err2) -> {
|
||||||
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) {
|
if (err2 != null) {
|
||||||
future.completeExceptionally(err2);
|
future.completeExceptionally(err2);
|
||||||
} else {
|
} else {
|
||||||
|
@ -1290,6 +1378,37 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return future;
|
return future;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> move(byte[] regionName, ServerName destServerName) {
|
||||||
|
Preconditions.checkNotNull(destServerName,
|
||||||
|
"destServerName is null. If you don't specify a destServerName, use move(byte[]) instead");
|
||||||
|
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||||
|
getRegionInfo(regionName).whenComplete((regionInfo, err) -> {
|
||||||
|
if (err != null) {
|
||||||
|
future.completeExceptionally(err);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
moveRegion(
|
||||||
|
RequestConverter.buildMoveRegionRequest(regionInfo.getEncodedNameAsBytes(), destServerName))
|
||||||
|
.whenComplete((ret, err2) -> {
|
||||||
|
if (err2 != null) {
|
||||||
|
future.completeExceptionally(err2);
|
||||||
|
} else {
|
||||||
|
future.complete(ret);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
|
return future;
|
||||||
|
}
|
||||||
|
|
||||||
|
private CompletableFuture<Void> moveRegion(MoveRegionRequest request) {
|
||||||
|
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();
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> setQuota(QuotaSettings quota) {
|
public CompletableFuture<Void> setQuota(QuotaSettings quota) {
|
||||||
return this
|
return this
|
||||||
|
@ -1452,7 +1571,19 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Optional<Pattern> pattern) {
|
public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers() {
|
||||||
|
return listReplicationPeers(RequestConverter.buildListReplicationPeersRequest(null));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Pattern pattern) {
|
||||||
|
Preconditions.checkNotNull(pattern,
|
||||||
|
"pattern is null. If you don't specify a pattern, use listReplicationPeers() instead");
|
||||||
|
return listReplicationPeers(RequestConverter.buildListReplicationPeersRequest(pattern));
|
||||||
|
}
|
||||||
|
|
||||||
|
private CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(
|
||||||
|
ListReplicationPeersRequest request) {
|
||||||
return this
|
return this
|
||||||
.<List<ReplicationPeerDescription>> newMasterCaller()
|
.<List<ReplicationPeerDescription>> newMasterCaller()
|
||||||
.action(
|
.action(
|
||||||
|
@ -1460,7 +1591,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
.<ListReplicationPeersRequest, ListReplicationPeersResponse, List<ReplicationPeerDescription>> call(
|
.<ListReplicationPeersRequest, ListReplicationPeersResponse, List<ReplicationPeerDescription>> call(
|
||||||
controller,
|
controller,
|
||||||
stub,
|
stub,
|
||||||
RequestConverter.buildListReplicationPeersRequest(pattern),
|
request,
|
||||||
(s, c, req, done) -> s.listReplicationPeers(c, req, done),
|
(s, c, req, done) -> s.listReplicationPeers(c, req, done),
|
||||||
(resp) -> resp.getPeerDescList().stream()
|
(resp) -> resp.getPeerDescList().stream()
|
||||||
.map(ReplicationSerDeHelper::toReplicationPeerDescription)
|
.map(ReplicationSerDeHelper::toReplicationPeerDescription)
|
||||||
|
@ -1570,7 +1701,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> restoreSnapshot(String snapshotName, boolean takeFailSafeSnapshot) {
|
public CompletableFuture<Void> restoreSnapshot(String snapshotName, boolean takeFailSafeSnapshot) {
|
||||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||||
listSnapshots(Optional.of(Pattern.compile(snapshotName))).whenComplete(
|
listSnapshots(Pattern.compile(snapshotName)).whenComplete(
|
||||||
(snapshotDescriptions, err) -> {
|
(snapshotDescriptions, err) -> {
|
||||||
if (err != null) {
|
if (err != null) {
|
||||||
future.completeExceptionally(err);
|
future.completeExceptionally(err);
|
||||||
|
@ -1715,37 +1846,47 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<SnapshotDescription>> listSnapshots(Optional<Pattern> pattern) {
|
public CompletableFuture<List<SnapshotDescription>> listSnapshots() {
|
||||||
CompletableFuture<List<SnapshotDescription>> future = new CompletableFuture<>();
|
return getCompletedSnapshots(null);
|
||||||
this.<GetCompletedSnapshotsResponse> newMasterCaller()
|
}
|
||||||
.action(
|
|
||||||
(controller, stub) -> this
|
@Override
|
||||||
.<GetCompletedSnapshotsRequest, GetCompletedSnapshotsResponse, GetCompletedSnapshotsResponse> call(
|
public CompletableFuture<List<SnapshotDescription>> listSnapshots(Pattern pattern) {
|
||||||
controller, stub, GetCompletedSnapshotsRequest.newBuilder().build(), (s, c, req,
|
Preconditions.checkNotNull(pattern,
|
||||||
done) -> s.getCompletedSnapshots(c, req, done), resp -> resp))
|
"pattern is null. If you don't specify a pattern, use listSnapshots() instead");
|
||||||
.call()
|
return getCompletedSnapshots(pattern);
|
||||||
.whenComplete(
|
}
|
||||||
(resp, err) -> {
|
|
||||||
if (err != null) {
|
private CompletableFuture<List<SnapshotDescription>> getCompletedSnapshots(Pattern pattern) {
|
||||||
future.completeExceptionally(err);
|
return this.<List<SnapshotDescription>> newMasterCaller().action((controller, stub) -> this
|
||||||
return;
|
.<GetCompletedSnapshotsRequest, GetCompletedSnapshotsResponse, List<SnapshotDescription>>
|
||||||
}
|
call(controller, stub, GetCompletedSnapshotsRequest.newBuilder().build(),
|
||||||
future.complete(resp
|
(s, c, req, done) -> s.getCompletedSnapshots(c, req, done),
|
||||||
.getSnapshotsList()
|
resp -> ProtobufUtil.toSnapshotDescriptionList(resp, pattern)))
|
||||||
.stream()
|
.call();
|
||||||
.map(ProtobufUtil::createSnapshotDesc)
|
}
|
||||||
.filter(
|
|
||||||
snap -> pattern.isPresent() ? pattern.get().matcher(snap.getName()).matches()
|
@Override
|
||||||
: true).collect(Collectors.toList()));
|
public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern) {
|
||||||
});
|
Preconditions.checkNotNull(tableNamePattern, "tableNamePattern is null."
|
||||||
return future;
|
+ " If you don't specify a tableNamePattern, use listSnapshots() instead");
|
||||||
|
return getCompletedSnapshots(tableNamePattern, null);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern,
|
public CompletableFuture<List<SnapshotDescription>> listTableSnapshots(Pattern tableNamePattern,
|
||||||
Pattern snapshotNamePattern) {
|
Pattern snapshotNamePattern) {
|
||||||
|
Preconditions.checkNotNull(tableNamePattern, "tableNamePattern is null."
|
||||||
|
+ " If you don't specify a tableNamePattern, use listSnapshots(Pattern) instead");
|
||||||
|
Preconditions.checkNotNull(snapshotNamePattern, "snapshotNamePattern is null."
|
||||||
|
+ " If you don't specify a snapshotNamePattern, use listTableSnapshots(Pattern) instead");
|
||||||
|
return getCompletedSnapshots(tableNamePattern, snapshotNamePattern);
|
||||||
|
}
|
||||||
|
|
||||||
|
private CompletableFuture<List<SnapshotDescription>> getCompletedSnapshots(
|
||||||
|
Pattern tableNamePattern, Pattern snapshotNamePattern) {
|
||||||
CompletableFuture<List<SnapshotDescription>> future = new CompletableFuture<>();
|
CompletableFuture<List<SnapshotDescription>> future = new CompletableFuture<>();
|
||||||
listTableNames(Optional.ofNullable(tableNamePattern), false).whenComplete(
|
listTableNames(tableNamePattern, false).whenComplete(
|
||||||
(tableNames, err) -> {
|
(tableNames, err) -> {
|
||||||
if (err != null) {
|
if (err != null) {
|
||||||
future.completeExceptionally(err);
|
future.completeExceptionally(err);
|
||||||
|
@ -1755,7 +1896,7 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
future.complete(Collections.emptyList());
|
future.complete(Collections.emptyList());
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
listSnapshots(Optional.ofNullable(snapshotNamePattern)).whenComplete(
|
getCompletedSnapshots(snapshotNamePattern).whenComplete(
|
||||||
(snapshotDescList, err2) -> {
|
(snapshotDescList, err2) -> {
|
||||||
if (err2 != null) {
|
if (err2 != null) {
|
||||||
future.completeExceptionally(err2);
|
future.completeExceptionally(err2);
|
||||||
|
@ -1778,32 +1919,60 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
return internalDeleteSnapshot(new SnapshotDescription(snapshotName));
|
return internalDeleteSnapshot(new SnapshotDescription(snapshotName));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> deleteSnapshots() {
|
||||||
|
return internalDeleteSnapshots(null, null);
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> deleteSnapshots(Pattern snapshotNamePattern) {
|
public CompletableFuture<Void> deleteSnapshots(Pattern snapshotNamePattern) {
|
||||||
return deleteTableSnapshots(null, snapshotNamePattern);
|
Preconditions.checkNotNull(snapshotNamePattern, "snapshotNamePattern is null."
|
||||||
|
+ " If you don't specify a snapshotNamePattern, use deleteSnapshots() instead");
|
||||||
|
return internalDeleteSnapshots(null, snapshotNamePattern);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern) {
|
||||||
|
Preconditions.checkNotNull(tableNamePattern, "tableNamePattern is null."
|
||||||
|
+ " If you don't specify a tableNamePattern, use deleteSnapshots() instead");
|
||||||
|
return internalDeleteSnapshots(tableNamePattern, null);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern,
|
public CompletableFuture<Void> deleteTableSnapshots(Pattern tableNamePattern,
|
||||||
Pattern snapshotNamePattern) {
|
Pattern snapshotNamePattern) {
|
||||||
|
Preconditions.checkNotNull(tableNamePattern, "tableNamePattern is null."
|
||||||
|
+ " If you don't specify a tableNamePattern, use deleteSnapshots(Pattern) instead");
|
||||||
|
Preconditions.checkNotNull(snapshotNamePattern, "snapshotNamePattern is null."
|
||||||
|
+ " If you don't specify a snapshotNamePattern, use deleteSnapshots(Pattern) instead");
|
||||||
|
return internalDeleteSnapshots(tableNamePattern, snapshotNamePattern);
|
||||||
|
}
|
||||||
|
|
||||||
|
private CompletableFuture<Void> internalDeleteSnapshots(Pattern tableNamePattern,
|
||||||
|
Pattern snapshotNamePattern) {
|
||||||
|
CompletableFuture<List<SnapshotDescription>> listSnapshotsFuture;
|
||||||
|
if (tableNamePattern == null) {
|
||||||
|
listSnapshotsFuture = getCompletedSnapshots(snapshotNamePattern);
|
||||||
|
} else {
|
||||||
|
listSnapshotsFuture = getCompletedSnapshots(tableNamePattern, snapshotNamePattern);
|
||||||
|
}
|
||||||
CompletableFuture<Void> future = new CompletableFuture<>();
|
CompletableFuture<Void> future = new CompletableFuture<>();
|
||||||
listTableSnapshots(tableNamePattern, snapshotNamePattern).whenComplete(
|
listSnapshotsFuture.whenComplete(((snapshotDescriptions, err) -> {
|
||||||
((snapshotDescriptions, err) -> {
|
if (err != null) {
|
||||||
if (err != null) {
|
future.completeExceptionally(err);
|
||||||
future.completeExceptionally(err);
|
return;
|
||||||
return;
|
}
|
||||||
}
|
if (snapshotDescriptions == null || snapshotDescriptions.isEmpty()) {
|
||||||
if (snapshotDescriptions == null || snapshotDescriptions.isEmpty()) {
|
future.complete(null);
|
||||||
future.complete(null);
|
return;
|
||||||
return;
|
}
|
||||||
}
|
List<CompletableFuture<Void>> deleteSnapshotFutures = new ArrayList<>();
|
||||||
List<CompletableFuture<Void>> deleteSnapshotFutures = new ArrayList<>();
|
snapshotDescriptions.forEach(snapDesc -> deleteSnapshotFutures
|
||||||
snapshotDescriptions.forEach(snapDesc -> deleteSnapshotFutures
|
.add(internalDeleteSnapshot(snapDesc)));
|
||||||
.add(internalDeleteSnapshot(snapDesc)));
|
CompletableFuture.allOf(
|
||||||
CompletableFuture.allOf(
|
deleteSnapshotFutures.toArray(new CompletableFuture<?>[deleteSnapshotFutures.size()]))
|
||||||
deleteSnapshotFutures.toArray(new CompletableFuture<?>[deleteSnapshotFutures.size()]))
|
.thenAccept(v -> future.complete(v));
|
||||||
.thenAccept(v -> future.complete(v));
|
}));
|
||||||
}));
|
|
||||||
return future;
|
return future;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -2484,17 +2653,27 @@ public class RawAsyncHBaseAdmin implements AsyncAdmin {
|
||||||
.toSecurityCapabilityList(resp.getCapabilitiesList()))).call();
|
.toSecurityCapabilityList(resp.getCapabilitiesList()))).call();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName) {
|
||||||
|
return getRegionLoad(GetRegionLoadRequest.newBuilder().build(), serverName);
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName,
|
public CompletableFuture<List<RegionLoad>> getRegionLoads(ServerName serverName,
|
||||||
Optional<TableName> tableName) {
|
TableName tableName) {
|
||||||
return this
|
Preconditions.checkNotNull(tableName,
|
||||||
.<List<RegionLoad>> newAdminCaller()
|
"tableName is null. If you don't specify a tableName, use getRegionLoads() instead");
|
||||||
.action(
|
return getRegionLoad(RequestConverter.buildGetRegionLoadRequest(tableName), serverName);
|
||||||
(controller, stub) -> this
|
}
|
||||||
.<GetRegionLoadRequest, GetRegionLoadResponse, List<RegionLoad>> adminCall(
|
|
||||||
controller, stub, RequestConverter.buildGetRegionLoadRequest(tableName), (s, c,
|
private CompletableFuture<List<RegionLoad>> getRegionLoad(GetRegionLoadRequest request,
|
||||||
req, done) -> s.getRegionLoad(controller, req, done),
|
ServerName serverName) {
|
||||||
ProtobufUtil::getRegionLoadInfo)).serverName(serverName).call();
|
return this.<List<RegionLoad>> newAdminCaller()
|
||||||
|
.action((controller, stub) -> this
|
||||||
|
.<GetRegionLoadRequest, GetRegionLoadResponse, List<RegionLoad>> adminCall(controller,
|
||||||
|
stub, request, (s, c, req, done) -> s.getRegionLoad(controller, req, done),
|
||||||
|
ProtobufUtil::getRegionLoadInfo))
|
||||||
|
.serverName(serverName).call();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|
|
@ -36,6 +36,7 @@ import java.util.Optional;
|
||||||
import java.util.concurrent.Callable;
|
import java.util.concurrent.Callable;
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
import java.util.function.Function;
|
import java.util.function.Function;
|
||||||
|
import java.util.regex.Pattern;
|
||||||
import java.util.stream.Collectors;
|
import java.util.stream.Collectors;
|
||||||
import org.apache.hadoop.conf.Configuration;
|
import org.apache.hadoop.conf.Configuration;
|
||||||
import org.apache.hadoop.fs.Path;
|
import org.apache.hadoop.fs.Path;
|
||||||
|
@ -158,6 +159,7 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.LockServiceProtos;
|
||||||
import org.apache.hadoop.hbase.shaded.protobuf.generated.MapReduceProtos;
|
import org.apache.hadoop.hbase.shaded.protobuf.generated.MapReduceProtos;
|
||||||
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos;
|
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos;
|
||||||
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.CreateTableRequest;
|
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.CreateTableRequest;
|
||||||
|
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetCompletedSnapshotsResponse;
|
||||||
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetTableDescriptorsResponse;
|
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetTableDescriptorsResponse;
|
||||||
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListNamespaceDescriptorsResponse;
|
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListNamespaceDescriptorsResponse;
|
||||||
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MajorCompactionTimestampResponse;
|
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MajorCompactionTimestampResponse;
|
||||||
|
@ -1733,7 +1735,7 @@ public final class ProtobufUtil {
|
||||||
final RpcController controller, final AdminService.BlockingInterface admin,
|
final RpcController controller, final AdminService.BlockingInterface admin,
|
||||||
final TableName tableName) throws IOException {
|
final TableName tableName) throws IOException {
|
||||||
GetRegionLoadRequest request =
|
GetRegionLoadRequest request =
|
||||||
RequestConverter.buildGetRegionLoadRequest(Optional.ofNullable(tableName));
|
RequestConverter.buildGetRegionLoadRequest(tableName);
|
||||||
GetRegionLoadResponse response;
|
GetRegionLoadResponse response;
|
||||||
try {
|
try {
|
||||||
response = admin.getRegionLoad(controller, request);
|
response = admin.getRegionLoad(controller, request);
|
||||||
|
@ -3376,4 +3378,11 @@ public final class ProtobufUtil {
|
||||||
}
|
}
|
||||||
return rib.build();
|
return rib.build();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public static List<SnapshotDescription> toSnapshotDescriptionList(
|
||||||
|
GetCompletedSnapshotsResponse response, Pattern pattern) {
|
||||||
|
return response.getSnapshotsList().stream().map(ProtobufUtil::createSnapshotDesc)
|
||||||
|
.filter(snap -> pattern != null ? pattern.matcher(snap.getName()).matches() : true)
|
||||||
|
.collect(Collectors.toList());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -19,11 +19,9 @@ package org.apache.hadoop.hbase.shaded.protobuf;
|
||||||
|
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Arrays;
|
|
||||||
import java.util.Collection;
|
import java.util.Collection;
|
||||||
import java.util.EnumSet;
|
import java.util.EnumSet;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Optional;
|
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
import java.util.regex.Pattern;
|
import java.util.regex.Pattern;
|
||||||
|
|
||||||
|
@ -824,132 +822,112 @@ public final class RequestConverter {
|
||||||
* Create a protocol buffer GetRegionLoadRequest for all regions/regions of a table.
|
* Create a protocol buffer GetRegionLoadRequest for all regions/regions of a table.
|
||||||
* @param tableName the table for which regionLoad should be obtained from RS
|
* @param tableName the table for which regionLoad should be obtained from RS
|
||||||
* @return a protocol buffer GetRegionLoadRequest
|
* @return a protocol buffer GetRegionLoadRequest
|
||||||
* @deprecated use {@link #buildGetRegionLoadRequest(Optional)} instead.
|
|
||||||
*/
|
*/
|
||||||
@Deprecated
|
|
||||||
public static GetRegionLoadRequest buildGetRegionLoadRequest(final TableName tableName) {
|
public static GetRegionLoadRequest buildGetRegionLoadRequest(final TableName tableName) {
|
||||||
return buildGetRegionLoadRequest(Optional.ofNullable(tableName));
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Create a protocol buffer GetRegionLoadRequest for all regions/regions of a table.
|
|
||||||
* @param tableName the table for which regionLoad should be obtained from RS
|
|
||||||
* @return a protocol buffer GetRegionLoadRequest
|
|
||||||
*/
|
|
||||||
public static GetRegionLoadRequest buildGetRegionLoadRequest(Optional<TableName> tableName) {
|
|
||||||
GetRegionLoadRequest.Builder builder = GetRegionLoadRequest.newBuilder();
|
GetRegionLoadRequest.Builder builder = GetRegionLoadRequest.newBuilder();
|
||||||
tableName.ifPresent(table -> builder.setTableName(ProtobufUtil.toProtoTableName(table)));
|
if (tableName != null) {
|
||||||
|
builder.setTableName(ProtobufUtil.toProtoTableName(tableName));
|
||||||
|
}
|
||||||
return builder.build();
|
return builder.build();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Create a protocol buffer GetOnlineRegionRequest
|
* Create a protocol buffer GetOnlineRegionRequest
|
||||||
*
|
* @return a protocol buffer GetOnlineRegionRequest
|
||||||
* @return a protocol buffer GetOnlineRegionRequest
|
*/
|
||||||
*/
|
public static GetOnlineRegionRequest buildGetOnlineRegionRequest() {
|
||||||
public static GetOnlineRegionRequest buildGetOnlineRegionRequest() {
|
return GetOnlineRegionRequest.newBuilder().build();
|
||||||
return GetOnlineRegionRequest.newBuilder().build();
|
}
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Create a protocol buffer FlushRegionRequest for a given region name
|
|
||||||
*
|
|
||||||
* @param regionName the name of the region to get info
|
|
||||||
* @return a protocol buffer FlushRegionRequest
|
|
||||||
*/
|
|
||||||
public static FlushRegionRequest
|
|
||||||
buildFlushRegionRequest(final byte[] regionName) {
|
|
||||||
return buildFlushRegionRequest(regionName, false);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Create a protocol buffer FlushRegionRequest for a given region name
|
|
||||||
*
|
|
||||||
* @param regionName the name of the region to get info
|
|
||||||
* @return a protocol buffer FlushRegionRequest
|
|
||||||
*/
|
|
||||||
public static FlushRegionRequest
|
|
||||||
buildFlushRegionRequest(final byte[] regionName, boolean writeFlushWALMarker) {
|
|
||||||
FlushRegionRequest.Builder builder = FlushRegionRequest.newBuilder();
|
|
||||||
RegionSpecifier region = buildRegionSpecifier(
|
|
||||||
RegionSpecifierType.REGION_NAME, regionName);
|
|
||||||
builder.setRegion(region);
|
|
||||||
builder.setWriteFlushWalMarker(writeFlushWALMarker);
|
|
||||||
return builder.build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Create a protocol buffer OpenRegionRequest to open a list of regions
|
|
||||||
*
|
|
||||||
* @param server the serverName for the RPC
|
|
||||||
* @param regionOpenInfos info of a list of regions to open
|
|
||||||
* @param openForReplay
|
|
||||||
* @return a protocol buffer OpenRegionRequest
|
|
||||||
*/
|
|
||||||
public static OpenRegionRequest
|
|
||||||
buildOpenRegionRequest(ServerName server, final List<Pair<RegionInfo,
|
|
||||||
List<ServerName>>> regionOpenInfos, Boolean openForReplay) {
|
|
||||||
OpenRegionRequest.Builder builder = OpenRegionRequest.newBuilder();
|
|
||||||
for (Pair<RegionInfo, List<ServerName>> regionOpenInfo: regionOpenInfos) {
|
|
||||||
builder.addOpenInfo(buildRegionOpenInfo(regionOpenInfo.getFirst(),
|
|
||||||
regionOpenInfo.getSecond(), openForReplay));
|
|
||||||
}
|
|
||||||
if (server != null) {
|
|
||||||
builder.setServerStartCode(server.getStartcode());
|
|
||||||
}
|
|
||||||
// send the master's wall clock time as well, so that the RS can refer to it
|
|
||||||
builder.setMasterSystemTime(EnvironmentEdgeManager.currentTime());
|
|
||||||
return builder.build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Create a protocol buffer OpenRegionRequest for a given region
|
|
||||||
*
|
|
||||||
* @param server the serverName for the RPC
|
|
||||||
* @param region the region to open
|
|
||||||
* @param favoredNodes
|
|
||||||
* @param openForReplay
|
|
||||||
* @return a protocol buffer OpenRegionRequest
|
|
||||||
*/
|
|
||||||
public static OpenRegionRequest buildOpenRegionRequest(ServerName server,
|
|
||||||
final RegionInfo region, List<ServerName> favoredNodes,
|
|
||||||
Boolean openForReplay) {
|
|
||||||
OpenRegionRequest.Builder builder = OpenRegionRequest.newBuilder();
|
|
||||||
builder.addOpenInfo(buildRegionOpenInfo(region, favoredNodes,
|
|
||||||
openForReplay));
|
|
||||||
if (server != null) {
|
|
||||||
builder.setServerStartCode(server.getStartcode());
|
|
||||||
}
|
|
||||||
builder.setMasterSystemTime(EnvironmentEdgeManager.currentTime());
|
|
||||||
return builder.build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Create a protocol buffer UpdateFavoredNodesRequest to update a list of favorednode mappings
|
|
||||||
* @param updateRegionInfos
|
|
||||||
* @return a protocol buffer UpdateFavoredNodesRequest
|
|
||||||
*/
|
|
||||||
public static UpdateFavoredNodesRequest buildUpdateFavoredNodesRequest(
|
|
||||||
final List<Pair<RegionInfo, List<ServerName>>> updateRegionInfos) {
|
|
||||||
UpdateFavoredNodesRequest.Builder ubuilder = UpdateFavoredNodesRequest.newBuilder();
|
|
||||||
if (updateRegionInfos != null && !updateRegionInfos.isEmpty()) {
|
|
||||||
RegionUpdateInfo.Builder builder = RegionUpdateInfo.newBuilder();
|
|
||||||
for (Pair<RegionInfo, List<ServerName>> pair : updateRegionInfos) {
|
|
||||||
builder.setRegion(ProtobufUtil.toRegionInfo(pair.getFirst()));
|
|
||||||
for (ServerName server : pair.getSecond()) {
|
|
||||||
builder.addFavoredNodes(ProtobufUtil.toServerName(server));
|
|
||||||
}
|
|
||||||
ubuilder.addUpdateInfo(builder.build());
|
|
||||||
builder.clear();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return ubuilder.build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Create a WarmupRegionRequest for a given region name
|
* Create a protocol buffer FlushRegionRequest for a given region name
|
||||||
*
|
* @param regionName the name of the region to get info
|
||||||
* @param regionInfo Region we are warming up
|
* @return a protocol buffer FlushRegionRequest
|
||||||
|
*/
|
||||||
|
public static FlushRegionRequest buildFlushRegionRequest(final byte[] regionName) {
|
||||||
|
return buildFlushRegionRequest(regionName, false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Create a protocol buffer FlushRegionRequest for a given region name
|
||||||
|
* @param regionName the name of the region to get info
|
||||||
|
* @return a protocol buffer FlushRegionRequest
|
||||||
|
*/
|
||||||
|
public static FlushRegionRequest buildFlushRegionRequest(final byte[] regionName,
|
||||||
|
boolean writeFlushWALMarker) {
|
||||||
|
FlushRegionRequest.Builder builder = FlushRegionRequest.newBuilder();
|
||||||
|
RegionSpecifier region = buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName);
|
||||||
|
builder.setRegion(region);
|
||||||
|
builder.setWriteFlushWalMarker(writeFlushWALMarker);
|
||||||
|
return builder.build();
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Create a protocol buffer OpenRegionRequest to open a list of regions
|
||||||
|
* @param server the serverName for the RPC
|
||||||
|
* @param regionOpenInfos info of a list of regions to open
|
||||||
|
* @param openForReplay whether open for replay
|
||||||
|
* @return a protocol buffer OpenRegionRequest
|
||||||
|
*/
|
||||||
|
public static OpenRegionRequest buildOpenRegionRequest(ServerName server,
|
||||||
|
final List<Pair<RegionInfo, List<ServerName>>> regionOpenInfos, Boolean openForReplay) {
|
||||||
|
OpenRegionRequest.Builder builder = OpenRegionRequest.newBuilder();
|
||||||
|
for (Pair<RegionInfo, List<ServerName>> regionOpenInfo : regionOpenInfos) {
|
||||||
|
builder.addOpenInfo(buildRegionOpenInfo(regionOpenInfo.getFirst(),
|
||||||
|
regionOpenInfo.getSecond(), openForReplay));
|
||||||
|
}
|
||||||
|
if (server != null) {
|
||||||
|
builder.setServerStartCode(server.getStartcode());
|
||||||
|
}
|
||||||
|
// send the master's wall clock time as well, so that the RS can refer to it
|
||||||
|
builder.setMasterSystemTime(EnvironmentEdgeManager.currentTime());
|
||||||
|
return builder.build();
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Create a protocol buffer OpenRegionRequest for a given region
|
||||||
|
* @param server the serverName for the RPC
|
||||||
|
* @param region the region to open
|
||||||
|
* @param favoredNodes a list of favored nodes
|
||||||
|
* @param openForReplay whether open for replay
|
||||||
|
* @return a protocol buffer OpenRegionRequest
|
||||||
|
*/
|
||||||
|
public static OpenRegionRequest buildOpenRegionRequest(ServerName server,
|
||||||
|
final RegionInfo region, List<ServerName> favoredNodes, Boolean openForReplay) {
|
||||||
|
OpenRegionRequest.Builder builder = OpenRegionRequest.newBuilder();
|
||||||
|
builder.addOpenInfo(buildRegionOpenInfo(region, favoredNodes, openForReplay));
|
||||||
|
if (server != null) {
|
||||||
|
builder.setServerStartCode(server.getStartcode());
|
||||||
|
}
|
||||||
|
builder.setMasterSystemTime(EnvironmentEdgeManager.currentTime());
|
||||||
|
return builder.build();
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Create a protocol buffer UpdateFavoredNodesRequest to update a list of favorednode mappings
|
||||||
|
* @param updateRegionInfos a list of favored node mappings
|
||||||
|
* @return a protocol buffer UpdateFavoredNodesRequest
|
||||||
|
*/
|
||||||
|
public static UpdateFavoredNodesRequest buildUpdateFavoredNodesRequest(
|
||||||
|
final List<Pair<RegionInfo, List<ServerName>>> updateRegionInfos) {
|
||||||
|
UpdateFavoredNodesRequest.Builder ubuilder = UpdateFavoredNodesRequest.newBuilder();
|
||||||
|
if (updateRegionInfos != null && !updateRegionInfos.isEmpty()) {
|
||||||
|
RegionUpdateInfo.Builder builder = RegionUpdateInfo.newBuilder();
|
||||||
|
for (Pair<RegionInfo, List<ServerName>> pair : updateRegionInfos) {
|
||||||
|
builder.setRegion(ProtobufUtil.toRegionInfo(pair.getFirst()));
|
||||||
|
for (ServerName server : pair.getSecond()) {
|
||||||
|
builder.addFavoredNodes(ProtobufUtil.toServerName(server));
|
||||||
|
}
|
||||||
|
ubuilder.addUpdateInfo(builder.build());
|
||||||
|
builder.clear();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return ubuilder.build();
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Create a WarmupRegionRequest for a given region name
|
||||||
|
* @param regionInfo Region we are warming up
|
||||||
*/
|
*/
|
||||||
public static WarmupRegionRequest buildWarmupRegionRequest(final RegionInfo regionInfo) {
|
public static WarmupRegionRequest buildWarmupRegionRequest(final RegionInfo regionInfo) {
|
||||||
WarmupRegionRequest.Builder builder = WarmupRegionRequest.newBuilder();
|
WarmupRegionRequest.Builder builder = WarmupRegionRequest.newBuilder();
|
||||||
|
@ -963,72 +941,57 @@ public final class RequestConverter {
|
||||||
* @param major indicator if it is a major compaction
|
* @param major indicator if it is a major compaction
|
||||||
* @param columnFamily
|
* @param columnFamily
|
||||||
* @return a CompactRegionRequest
|
* @return a CompactRegionRequest
|
||||||
* @deprecated Use {@link #buildCompactRegionRequest(byte[], boolean, Optional)} instead.
|
|
||||||
*/
|
*/
|
||||||
@Deprecated
|
|
||||||
public static CompactRegionRequest buildCompactRegionRequest(byte[] regionName, boolean major,
|
public static CompactRegionRequest buildCompactRegionRequest(byte[] regionName, boolean major,
|
||||||
byte[] columnFamily) {
|
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();
|
CompactRegionRequest.Builder builder = CompactRegionRequest.newBuilder();
|
||||||
RegionSpecifier region = buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName);
|
RegionSpecifier region = buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName);
|
||||||
builder.setRegion(region);
|
builder.setRegion(region);
|
||||||
builder.setMajor(major);
|
builder.setMajor(major);
|
||||||
columnFamily.ifPresent(family -> builder.setFamily(UnsafeByteOperations.unsafeWrap(family)));
|
if (columnFamily != null) {
|
||||||
|
builder.setFamily(UnsafeByteOperations.unsafeWrap(columnFamily));
|
||||||
|
}
|
||||||
return builder.build();
|
return builder.build();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @see {@link #buildRollWALWriterRequest()}
|
* @see {@link #buildRollWALWriterRequest()}
|
||||||
*/
|
*/
|
||||||
private static RollWALWriterRequest ROLL_WAL_WRITER_REQUEST =
|
private static RollWALWriterRequest ROLL_WAL_WRITER_REQUEST = RollWALWriterRequest.newBuilder()
|
||||||
RollWALWriterRequest.newBuilder().build();
|
.build();
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Create a new RollWALWriterRequest
|
* Create a new RollWALWriterRequest
|
||||||
*
|
* @return a ReplicateWALEntryRequest
|
||||||
* @return a ReplicateWALEntryRequest
|
*/
|
||||||
*/
|
public static RollWALWriterRequest buildRollWALWriterRequest() {
|
||||||
public static RollWALWriterRequest buildRollWALWriterRequest() {
|
return ROLL_WAL_WRITER_REQUEST;
|
||||||
return ROLL_WAL_WRITER_REQUEST;
|
}
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @see {@link #buildGetServerInfoRequest()}
|
* @see {@link #buildGetServerInfoRequest()}
|
||||||
*/
|
*/
|
||||||
private static GetServerInfoRequest GET_SERVER_INFO_REQUEST =
|
private static GetServerInfoRequest GET_SERVER_INFO_REQUEST = GetServerInfoRequest.newBuilder()
|
||||||
GetServerInfoRequest.newBuilder().build();
|
.build();
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Create a new GetServerInfoRequest
|
* Create a new GetServerInfoRequest
|
||||||
*
|
* @return a GetServerInfoRequest
|
||||||
* @return a GetServerInfoRequest
|
*/
|
||||||
*/
|
public static GetServerInfoRequest buildGetServerInfoRequest() {
|
||||||
public static GetServerInfoRequest buildGetServerInfoRequest() {
|
return GET_SERVER_INFO_REQUEST;
|
||||||
return GET_SERVER_INFO_REQUEST;
|
}
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Create a new StopServerRequest
|
* Create a new StopServerRequest
|
||||||
*
|
* @param reason the reason to stop the server
|
||||||
* @param reason the reason to stop the server
|
* @return a StopServerRequest
|
||||||
* @return a StopServerRequest
|
*/
|
||||||
*/
|
public static StopServerRequest buildStopServerRequest(final String reason) {
|
||||||
public static StopServerRequest buildStopServerRequest(final String reason) {
|
StopServerRequest.Builder builder = StopServerRequest.newBuilder();
|
||||||
StopServerRequest.Builder builder = StopServerRequest.newBuilder();
|
builder.setReason(reason);
|
||||||
builder.setReason(reason);
|
return builder.build();
|
||||||
return builder.build();
|
}
|
||||||
}
|
|
||||||
|
|
||||||
//End utilities for Admin
|
//End utilities for Admin
|
||||||
|
|
||||||
|
@ -1131,28 +1094,6 @@ public final class RequestConverter {
|
||||||
return builder.build();
|
return builder.build();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* 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 {
|
|
||||||
MoveRegionRequest.Builder builder = MoveRegionRequest.newBuilder();
|
|
||||||
builder.setRegion(
|
|
||||||
buildRegionSpecifier(RegionSpecifierType.ENCODED_REGION_NAME,encodedRegionName));
|
|
||||||
if (destServerName != null) {
|
|
||||||
builder.setDestServerName(
|
|
||||||
ProtobufUtil.toServerName(ServerName.valueOf(Bytes.toString(destServerName))));
|
|
||||||
}
|
|
||||||
return builder.build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Create a protocol buffer MoveRegionRequest
|
* Create a protocol buffer MoveRegionRequest
|
||||||
* @param encodedRegionName
|
* @param encodedRegionName
|
||||||
|
@ -1160,12 +1101,13 @@ public final class RequestConverter {
|
||||||
* @return A MoveRegionRequest
|
* @return A MoveRegionRequest
|
||||||
*/
|
*/
|
||||||
public static MoveRegionRequest buildMoveRegionRequest(byte[] encodedRegionName,
|
public static MoveRegionRequest buildMoveRegionRequest(byte[] encodedRegionName,
|
||||||
Optional<ServerName> destServerName) {
|
ServerName destServerName) {
|
||||||
MoveRegionRequest.Builder builder = MoveRegionRequest.newBuilder();
|
MoveRegionRequest.Builder builder = MoveRegionRequest.newBuilder();
|
||||||
builder.setRegion(buildRegionSpecifier(RegionSpecifierType.ENCODED_REGION_NAME,
|
builder.setRegion(buildRegionSpecifier(RegionSpecifierType.ENCODED_REGION_NAME,
|
||||||
encodedRegionName));
|
encodedRegionName));
|
||||||
destServerName.ifPresent(serverName -> builder.setDestServerName(ProtobufUtil
|
if (destServerName != null) {
|
||||||
.toServerName(serverName)));
|
builder.setDestServerName(ProtobufUtil.toServerName(destServerName));
|
||||||
|
}
|
||||||
return builder.build();
|
return builder.build();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -1320,21 +1262,13 @@ public final class RequestConverter {
|
||||||
final byte [][] splitKeys,
|
final byte [][] splitKeys,
|
||||||
final long nonceGroup,
|
final long nonceGroup,
|
||||||
final long nonce) {
|
final long nonce) {
|
||||||
return buildCreateTableRequest(tableDescriptor, Optional.ofNullable(splitKeys), nonceGroup, nonce);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Creates a protocol buffer CreateTableRequest
|
|
||||||
* @param tableDescriptor
|
|
||||||
* @param splitKeys
|
|
||||||
* @return a CreateTableRequest
|
|
||||||
*/
|
|
||||||
public static CreateTableRequest buildCreateTableRequest(TableDescriptor tableDescriptor,
|
|
||||||
Optional<byte[][]> splitKeys, long nonceGroup, long nonce) {
|
|
||||||
CreateTableRequest.Builder builder = CreateTableRequest.newBuilder();
|
CreateTableRequest.Builder builder = CreateTableRequest.newBuilder();
|
||||||
builder.setTableSchema(ProtobufUtil.toTableSchema(tableDescriptor));
|
builder.setTableSchema(ProtobufUtil.toTableSchema(tableDescriptor));
|
||||||
splitKeys.ifPresent(keys -> Arrays.stream(keys).forEach(
|
if (splitKeys != null) {
|
||||||
key -> builder.addSplitKeys(UnsafeByteOperations.unsafeWrap(key))));
|
for(byte[] key : splitKeys) {
|
||||||
|
builder.addSplitKeys(UnsafeByteOperations.unsafeWrap(key));
|
||||||
|
}
|
||||||
|
}
|
||||||
builder.setNonceGroup(nonceGroup);
|
builder.setNonceGroup(nonceGroup);
|
||||||
builder.setNonce(nonce);
|
builder.setNonce(nonce);
|
||||||
return builder.build();
|
return builder.build();
|
||||||
|
@ -1396,25 +1330,13 @@ public final class RequestConverter {
|
||||||
* @param pattern The compiled regular expression to match against
|
* @param pattern The compiled regular expression to match against
|
||||||
* @param includeSysTables False to match only against userspace tables
|
* @param includeSysTables False to match only against userspace tables
|
||||||
* @return a GetTableDescriptorsRequest
|
* @return a GetTableDescriptorsRequest
|
||||||
* @deprecated Use {@link #buildGetTableDescriptorsRequest(Optional, boolean)} instead.
|
|
||||||
*/
|
*/
|
||||||
@Deprecated
|
|
||||||
public static GetTableDescriptorsRequest buildGetTableDescriptorsRequest(final Pattern pattern,
|
public static GetTableDescriptorsRequest buildGetTableDescriptorsRequest(final Pattern pattern,
|
||||||
boolean includeSysTables) {
|
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();
|
GetTableDescriptorsRequest.Builder builder = GetTableDescriptorsRequest.newBuilder();
|
||||||
pattern.ifPresent(p -> builder.setRegex(p.toString()));
|
if (pattern != null) {
|
||||||
|
builder.setRegex(pattern.toString());
|
||||||
|
}
|
||||||
builder.setIncludeSysTables(includeSysTables);
|
builder.setIncludeSysTables(includeSysTables);
|
||||||
return builder.build();
|
return builder.build();
|
||||||
}
|
}
|
||||||
|
@ -1425,25 +1347,13 @@ public final class RequestConverter {
|
||||||
* @param pattern The compiled regular expression to match against
|
* @param pattern The compiled regular expression to match against
|
||||||
* @param includeSysTables False to match only against userspace tables
|
* @param includeSysTables False to match only against userspace tables
|
||||||
* @return a GetTableNamesRequest
|
* @return a GetTableNamesRequest
|
||||||
* @deprecated Use {@link #buildGetTableNamesRequest(Optional, boolean)} instead.
|
|
||||||
*/
|
*/
|
||||||
@Deprecated
|
|
||||||
public static GetTableNamesRequest buildGetTableNamesRequest(final Pattern pattern,
|
public static GetTableNamesRequest buildGetTableNamesRequest(final Pattern pattern,
|
||||||
boolean includeSysTables) {
|
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();
|
GetTableNamesRequest.Builder builder = GetTableNamesRequest.newBuilder();
|
||||||
pattern.ifPresent(p -> builder.setRegex(p.toString()));
|
if (pattern != null) {
|
||||||
|
builder.setRegex(pattern.toString());
|
||||||
|
}
|
||||||
builder.setIncludeSysTables(includeSysTables);
|
builder.setIncludeSysTables(includeSysTables);
|
||||||
return builder.build();
|
return builder.build();
|
||||||
}
|
}
|
||||||
|
@ -1745,18 +1655,11 @@ public final class RequestConverter {
|
||||||
return builder.build();
|
return builder.build();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* @deprecated Use {@link #buildListReplicationPeersRequest(Optional)} instead.
|
|
||||||
*/
|
|
||||||
@Deprecated
|
|
||||||
public static ListReplicationPeersRequest buildListReplicationPeersRequest(Pattern pattern) {
|
public static ListReplicationPeersRequest buildListReplicationPeersRequest(Pattern pattern) {
|
||||||
return buildListReplicationPeersRequest(Optional.ofNullable(pattern));
|
|
||||||
}
|
|
||||||
|
|
||||||
public static ListReplicationPeersRequest
|
|
||||||
buildListReplicationPeersRequest(Optional<Pattern> pattern) {
|
|
||||||
ListReplicationPeersRequest.Builder builder = ListReplicationPeersRequest.newBuilder();
|
ListReplicationPeersRequest.Builder builder = ListReplicationPeersRequest.newBuilder();
|
||||||
pattern.ifPresent(p -> builder.setRegex(p.toString()));
|
if (pattern != null) {
|
||||||
|
builder.setRegex(pattern.toString());
|
||||||
|
}
|
||||||
return builder.build();
|
return builder.build();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -20,7 +20,6 @@ package org.apache.hadoop.hbase.regionserver;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Optional;
|
|
||||||
import java.util.concurrent.atomic.AtomicLong;
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
|
||||||
import org.apache.commons.logging.Log;
|
import org.apache.commons.logging.Log;
|
||||||
|
@ -138,7 +137,7 @@ public class TestHRegionServerBulkLoadWithOldSecureEndpoint extends TestHRegionS
|
||||||
conn.getAdmin(getLocation().getServerName());
|
conn.getAdmin(getLocation().getServerName());
|
||||||
CompactRegionRequest request =
|
CompactRegionRequest request =
|
||||||
RequestConverter.buildCompactRegionRequest(
|
RequestConverter.buildCompactRegionRequest(
|
||||||
getLocation().getRegionInfo().getRegionName(), true, Optional.empty());
|
getLocation().getRegionInfo().getRegionName(), true, null);
|
||||||
server.compactRegion(null, request);
|
server.compactRegion(null, request);
|
||||||
numCompactions.incrementAndGet();
|
numCompactions.incrementAndGet();
|
||||||
return null;
|
return null;
|
||||||
|
|
|
@ -21,10 +21,8 @@ import static org.apache.hadoop.hbase.client.AsyncProcess.START_LOG_ERRORS_AFTER
|
||||||
|
|
||||||
import java.util.Arrays;
|
import java.util.Arrays;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Optional;
|
import java.util.concurrent.CompletableFuture;
|
||||||
import java.util.concurrent.ExecutionException;
|
|
||||||
import java.util.concurrent.ForkJoinPool;
|
import java.util.concurrent.ForkJoinPool;
|
||||||
import java.util.concurrent.TimeUnit;
|
|
||||||
import java.util.function.Supplier;
|
import java.util.function.Supplier;
|
||||||
import java.util.regex.Pattern;
|
import java.util.regex.Pattern;
|
||||||
|
|
||||||
|
@ -41,8 +39,6 @@ import org.junit.Before;
|
||||||
import org.junit.BeforeClass;
|
import org.junit.BeforeClass;
|
||||||
import org.junit.Rule;
|
import org.junit.Rule;
|
||||||
import org.junit.rules.TestName;
|
import org.junit.rules.TestName;
|
||||||
import org.junit.runner.RunWith;
|
|
||||||
import org.junit.runners.Parameterized;
|
|
||||||
import org.junit.runners.Parameterized.Parameter;
|
import org.junit.runners.Parameterized.Parameter;
|
||||||
import org.junit.runners.Parameterized.Parameters;
|
import org.junit.runners.Parameterized.Parameters;
|
||||||
|
|
||||||
|
@ -106,7 +102,7 @@ public abstract class TestAsyncAdminBase {
|
||||||
|
|
||||||
@After
|
@After
|
||||||
public void tearDown() throws Exception {
|
public void tearDown() throws Exception {
|
||||||
admin.listTableNames(Optional.of(Pattern.compile(tableName.getNameAsString() + ".*")), false)
|
admin.listTableNames(Pattern.compile(tableName.getNameAsString() + ".*"), false)
|
||||||
.whenCompleteAsync((tables, err) -> {
|
.whenCompleteAsync((tables, err) -> {
|
||||||
if (tables != null) {
|
if (tables != null) {
|
||||||
tables.forEach(table -> {
|
tables.forEach(table -> {
|
||||||
|
@ -122,19 +118,21 @@ public abstract class TestAsyncAdminBase {
|
||||||
}
|
}
|
||||||
|
|
||||||
protected void createTableWithDefaultConf(TableName tableName) {
|
protected void createTableWithDefaultConf(TableName tableName) {
|
||||||
createTableWithDefaultConf(tableName, Optional.empty());
|
createTableWithDefaultConf(tableName, null);
|
||||||
}
|
}
|
||||||
|
|
||||||
protected void createTableWithDefaultConf(TableName tableName, Optional<byte[][]> splitKeys) {
|
protected void createTableWithDefaultConf(TableName tableName, byte[][] splitKeys) {
|
||||||
createTableWithDefaultConf(tableName, splitKeys, FAMILY);
|
createTableWithDefaultConf(tableName, splitKeys, FAMILY);
|
||||||
}
|
}
|
||||||
|
|
||||||
protected void createTableWithDefaultConf(TableName tableName, Optional<byte[][]> splitKeys,
|
protected void createTableWithDefaultConf(TableName tableName, byte[][] splitKeys,
|
||||||
byte[]... families) {
|
byte[]... families) {
|
||||||
TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
|
TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
|
||||||
for (byte[] family : families) {
|
for (byte[] family : families) {
|
||||||
builder.addColumnFamily(ColumnFamilyDescriptorBuilder.of(family));
|
builder.addColumnFamily(ColumnFamilyDescriptorBuilder.of(family));
|
||||||
}
|
}
|
||||||
admin.createTable(builder.build(), splitKeys).join();
|
CompletableFuture<Void> future = splitKeys == null ? admin.createTable(builder.build())
|
||||||
|
: admin.createTable(builder.build(), splitKeys);
|
||||||
|
future.join();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -31,7 +31,6 @@ import java.util.Collection;
|
||||||
import java.util.EnumSet;
|
import java.util.EnumSet;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.Optional;
|
|
||||||
|
|
||||||
import org.apache.hadoop.conf.Configuration;
|
import org.apache.hadoop.conf.Configuration;
|
||||||
import org.apache.hadoop.hbase.ClusterStatus;
|
import org.apache.hadoop.hbase.ClusterStatus;
|
||||||
|
@ -254,7 +253,7 @@ public class TestAsyncClusterAdminApi extends TestAsyncAdminBase {
|
||||||
List<RegionInfo> tableRegions = admin.getTableRegions(table).get();
|
List<RegionInfo> tableRegions = admin.getTableRegions(table).get();
|
||||||
List<RegionLoad> regionLoads = Lists.newArrayList();
|
List<RegionLoad> regionLoads = Lists.newArrayList();
|
||||||
for (ServerName serverName : servers) {
|
for (ServerName serverName : servers) {
|
||||||
regionLoads.addAll(admin.getRegionLoads(serverName, Optional.of(table)).get());
|
regionLoads.addAll(admin.getRegionLoads(serverName, table).get());
|
||||||
}
|
}
|
||||||
checkRegionsAndRegionLoads(tableRegions, regionLoads);
|
checkRegionsAndRegionLoads(tableRegions, regionLoads);
|
||||||
}
|
}
|
||||||
|
|
|
@ -200,7 +200,7 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
|
||||||
}
|
}
|
||||||
|
|
||||||
assertTrue(destServerName != null && !destServerName.equals(serverName));
|
assertTrue(destServerName != null && !destServerName.equals(serverName));
|
||||||
admin.move(hri.getRegionName(), Optional.of(destServerName)).get();
|
admin.move(hri.getRegionName(), destServerName).get();
|
||||||
|
|
||||||
long timeoutTime = System.currentTimeMillis() + 30000;
|
long timeoutTime = System.currentTimeMillis() + 30000;
|
||||||
while (true) {
|
while (true) {
|
||||||
|
@ -362,7 +362,7 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
|
||||||
@Test
|
@Test
|
||||||
public void testMergeRegions() throws Exception {
|
public void testMergeRegions() throws Exception {
|
||||||
byte[][] splitRows = new byte[][] { Bytes.toBytes("3"), Bytes.toBytes("6") };
|
byte[][] splitRows = new byte[][] { Bytes.toBytes("3"), Bytes.toBytes("6") };
|
||||||
createTableWithDefaultConf(tableName, Optional.of(splitRows));
|
createTableWithDefaultConf(tableName, splitRows);
|
||||||
|
|
||||||
RawAsyncTable metaTable = ASYNC_CONN.getRawTable(META_TABLE_NAME);
|
RawAsyncTable metaTable = ASYNC_CONN.getRawTable(META_TABLE_NAME);
|
||||||
List<HRegionLocation> regionLocations =
|
List<HRegionLocation> regionLocations =
|
||||||
|
@ -419,8 +419,11 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
|
||||||
table.putAll(puts).join();
|
table.putAll(puts).join();
|
||||||
|
|
||||||
if (isSplitRegion) {
|
if (isSplitRegion) {
|
||||||
admin.splitRegion(regionLocations.get(0).getRegionInfo().getRegionName(),
|
if (splitPoint == null) {
|
||||||
Optional.ofNullable(splitPoint)).get();
|
admin.splitRegion(regionLocations.get(0).getRegionInfo().getRegionName()).get();
|
||||||
|
} else {
|
||||||
|
admin.splitRegion(regionLocations.get(0).getRegionInfo().getRegionName(), splitPoint).get();
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
if (splitPoint == null) {
|
if (splitPoint == null) {
|
||||||
admin.split(tableName).get();
|
admin.split(tableName).get();
|
||||||
|
@ -450,7 +453,7 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
|
||||||
@Test
|
@Test
|
||||||
public void testCompactRegionServer() throws Exception {
|
public void testCompactRegionServer() throws Exception {
|
||||||
byte[][] families = { Bytes.toBytes("f1"), Bytes.toBytes("f2"), Bytes.toBytes("f3") };
|
byte[][] families = { Bytes.toBytes("f1"), Bytes.toBytes("f2"), Bytes.toBytes("f3") };
|
||||||
createTableWithDefaultConf(tableName, Optional.empty(), families);
|
createTableWithDefaultConf(tableName, null, families);
|
||||||
loadData(tableName, families, 3000, 8);
|
loadData(tableName, families, 3000, 8);
|
||||||
|
|
||||||
List<HRegionServer> rsList =
|
List<HRegionServer> rsList =
|
||||||
|
@ -491,7 +494,7 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
|
||||||
byte[] family = Bytes.toBytes("family");
|
byte[] family = Bytes.toBytes("family");
|
||||||
byte[][] families =
|
byte[][] families =
|
||||||
{ family, Bytes.add(family, Bytes.toBytes("2")), Bytes.add(family, Bytes.toBytes("3")) };
|
{ family, Bytes.add(family, Bytes.toBytes("2")), Bytes.add(family, Bytes.toBytes("3")) };
|
||||||
createTableWithDefaultConf(tableName, Optional.empty(), families);
|
createTableWithDefaultConf(tableName, null, families);
|
||||||
loadData(tableName, families, 3000, flushes);
|
loadData(tableName, families, 3000, flushes);
|
||||||
|
|
||||||
List<Region> regions = new ArrayList<>();
|
List<Region> regions = new ArrayList<>();
|
||||||
|
@ -506,15 +509,15 @@ public class TestAsyncRegionAdminApi extends TestAsyncAdminBase {
|
||||||
assertTrue(countBefore > 0); // there should be some data files
|
assertTrue(countBefore > 0); // there should be some data files
|
||||||
if (expectedState == CompactionState.MINOR) {
|
if (expectedState == CompactionState.MINOR) {
|
||||||
if (singleFamily) {
|
if (singleFamily) {
|
||||||
admin.compact(tableName, Optional.of(family)).get();
|
admin.compact(tableName, family).get();
|
||||||
} else {
|
} else {
|
||||||
admin.compact(tableName, Optional.empty()).get();
|
admin.compact(tableName).get();
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
if (singleFamily) {
|
if (singleFamily) {
|
||||||
admin.majorCompact(tableName, Optional.of(family)).get();
|
admin.majorCompact(tableName, family).get();
|
||||||
} else {
|
} else {
|
||||||
admin.majorCompact(tableName, Optional.empty()).get();
|
admin.majorCompact(tableName).get();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -32,7 +32,6 @@ import org.junit.runners.Parameterized;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Optional;
|
|
||||||
import java.util.regex.Pattern;
|
import java.util.regex.Pattern;
|
||||||
|
|
||||||
@RunWith(Parameterized.class)
|
@RunWith(Parameterized.class)
|
||||||
|
@ -169,13 +168,10 @@ public class TestAsyncSnapshotAdminApi extends TestAsyncAdminBase {
|
||||||
admin.snapshot(snapshotName3, tableName).get();
|
admin.snapshot(snapshotName3, tableName).get();
|
||||||
Assert.assertEquals(admin.listSnapshots().get().size(), 3);
|
Assert.assertEquals(admin.listSnapshots().get().size(), 3);
|
||||||
|
|
||||||
Assert.assertEquals(admin.listSnapshots(Optional.of(Pattern.compile("(.*)"))).get().size(), 3);
|
Assert.assertEquals(admin.listSnapshots(Pattern.compile("(.*)")).get().size(), 3);
|
||||||
Assert.assertEquals(admin.listSnapshots(Optional.of(Pattern.compile("snapshotName(\\d+)")))
|
Assert.assertEquals(admin.listSnapshots(Pattern.compile("snapshotName(\\d+)")).get().size(), 3);
|
||||||
.get().size(), 3);
|
Assert.assertEquals(admin.listSnapshots(Pattern.compile("snapshotName[1|3]")).get().size(), 2);
|
||||||
Assert.assertEquals(admin.listSnapshots(Optional.of(Pattern.compile("snapshotName[1|3]")))
|
Assert.assertEquals(admin.listSnapshots(Pattern.compile("snapshot(.*)")).get().size(), 3);
|
||||||
.get().size(), 2);
|
|
||||||
Assert.assertEquals(admin.listSnapshots(Optional.of(Pattern.compile("snapshot(.*)"))).get()
|
|
||||||
.size(), 3);
|
|
||||||
Assert.assertEquals(
|
Assert.assertEquals(
|
||||||
admin.listTableSnapshots(Pattern.compile("testListSnapshots"), Pattern.compile("s(.*)")).get()
|
admin.listTableSnapshots(Pattern.compile("testListSnapshots"), Pattern.compile("s(.*)")).get()
|
||||||
.size(),
|
.size(),
|
||||||
|
|
|
@ -120,9 +120,9 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
admin.deleteTable(tables[i]).join();
|
admin.deleteTable(tables[i]).join();
|
||||||
}
|
}
|
||||||
|
|
||||||
tableDescs = admin.listTables(Optional.empty(), true).get();
|
tableDescs = admin.listTables(true).get();
|
||||||
assertTrue("Not found system tables", tableDescs.size() > 0);
|
assertTrue("Not found system tables", tableDescs.size() > 0);
|
||||||
tableNames = admin.listTableNames(Optional.empty(), true).get();
|
tableNames = admin.listTableNames(true).get();
|
||||||
assertTrue("Not found system tables", tableNames.size() > 0);
|
assertTrue("Not found system tables", tableNames.size() > 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -169,7 +169,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
assertEquals("Table should have only 1 region", 1, regionLocations.size());
|
assertEquals("Table should have only 1 region", 1, regionLocations.size());
|
||||||
|
|
||||||
final TableName tableName2 = TableName.valueOf(tableName.getNameAsString() + "_2");
|
final TableName tableName2 = TableName.valueOf(tableName.getNameAsString() + "_2");
|
||||||
createTableWithDefaultConf(tableName2, Optional.of(new byte[][] { new byte[] { 42 } }));
|
createTableWithDefaultConf(tableName2, new byte[][] { new byte[] { 42 } });
|
||||||
regionLocations =
|
regionLocations =
|
||||||
AsyncMetaTableAccessor.getTableHRegionLocations(metaTable, Optional.of(tableName2)).get();
|
AsyncMetaTableAccessor.getTableHRegionLocations(metaTable, Optional.of(tableName2)).get();
|
||||||
assertEquals("Table should have only 2 region", 2, regionLocations.size());
|
assertEquals("Table should have only 2 region", 2, regionLocations.size());
|
||||||
|
@ -208,7 +208,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
new byte[] { 7, 7, 7 }, new byte[] { 8, 8, 8 }, new byte[] { 9, 9, 9 }, };
|
new byte[] { 7, 7, 7 }, new byte[] { 8, 8, 8 }, new byte[] { 9, 9, 9 }, };
|
||||||
int expectedRegions = splitKeys.length + 1;
|
int expectedRegions = splitKeys.length + 1;
|
||||||
boolean tablesOnMaster = LoadBalancer.isTablesOnMaster(TEST_UTIL.getConfiguration());
|
boolean tablesOnMaster = LoadBalancer.isTablesOnMaster(TEST_UTIL.getConfiguration());
|
||||||
createTableWithDefaultConf(tableName, Optional.of(splitKeys));
|
createTableWithDefaultConf(tableName, splitKeys);
|
||||||
|
|
||||||
boolean tableAvailable = admin.isTableAvailable(tableName, splitKeys).get();
|
boolean tableAvailable = admin.isTableAvailable(tableName, splitKeys).get();
|
||||||
assertTrue("Table should be created with splitKyes + 1 rows in META", tableAvailable);
|
assertTrue("Table should be created with splitKyes + 1 rows in META", tableAvailable);
|
||||||
|
@ -342,7 +342,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
new byte[] { 3, 3, 3 }, new byte[] { 2, 2, 2 } };
|
new byte[] { 3, 3, 3 }, new byte[] { 2, 2, 2 } };
|
||||||
final TableName tableName4 = TableName.valueOf(tableName.getNameAsString() + "_4");;
|
final TableName tableName4 = TableName.valueOf(tableName.getNameAsString() + "_4");;
|
||||||
try {
|
try {
|
||||||
createTableWithDefaultConf(tableName4, Optional.of(splitKeys));
|
createTableWithDefaultConf(tableName4, splitKeys);
|
||||||
fail("Should not be able to create this table because of " + "duplicate split keys");
|
fail("Should not be able to create this table because of " + "duplicate split keys");
|
||||||
} catch (CompletionException e) {
|
} catch (CompletionException e) {
|
||||||
assertTrue(e.getCause() instanceof IllegalArgumentException);
|
assertTrue(e.getCause() instanceof IllegalArgumentException);
|
||||||
|
@ -376,7 +376,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
byte[][] splitKeys = new byte[1][];
|
byte[][] splitKeys = new byte[1][];
|
||||||
splitKeys[0] = HConstants.EMPTY_BYTE_ARRAY;
|
splitKeys[0] = HConstants.EMPTY_BYTE_ARRAY;
|
||||||
try {
|
try {
|
||||||
createTableWithDefaultConf(tableName, Optional.of(splitKeys));
|
createTableWithDefaultConf(tableName, splitKeys);
|
||||||
fail("Test case should fail as empty split key is passed.");
|
fail("Test case should fail as empty split key is passed.");
|
||||||
} catch (CompletionException e) {
|
} catch (CompletionException e) {
|
||||||
assertTrue(e.getCause() instanceof IllegalArgumentException);
|
assertTrue(e.getCause() instanceof IllegalArgumentException);
|
||||||
|
@ -390,7 +390,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
splitKeys[1] = HConstants.EMPTY_BYTE_ARRAY;
|
splitKeys[1] = HConstants.EMPTY_BYTE_ARRAY;
|
||||||
splitKeys[2] = "region2".getBytes();
|
splitKeys[2] = "region2".getBytes();
|
||||||
try {
|
try {
|
||||||
createTableWithDefaultConf(tableName, Optional.of(splitKeys));
|
createTableWithDefaultConf(tableName, splitKeys);
|
||||||
fail("Test case should fail as empty split key is passed.");
|
fail("Test case should fail as empty split key is passed.");
|
||||||
} catch (CompletionException e) {
|
} catch (CompletionException e) {
|
||||||
assertTrue(e.getCause() instanceof IllegalArgumentException);
|
assertTrue(e.getCause() instanceof IllegalArgumentException);
|
||||||
|
@ -423,7 +423,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
splitKeys[1] = Bytes.toBytes(8);
|
splitKeys[1] = Bytes.toBytes(8);
|
||||||
|
|
||||||
// Create & Fill the table
|
// Create & Fill the table
|
||||||
createTableWithDefaultConf(tableName, Optional.of(splitKeys));
|
createTableWithDefaultConf(tableName, splitKeys);
|
||||||
RawAsyncTable table = ASYNC_CONN.getRawTable(tableName);
|
RawAsyncTable table = ASYNC_CONN.getRawTable(tableName);
|
||||||
int expectedRows = 10;
|
int expectedRows = 10;
|
||||||
for (int i = 0; i < expectedRows; i++) {
|
for (int i = 0; i < expectedRows; i++) {
|
||||||
|
@ -517,7 +517,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
table1.get(get).get();
|
table1.get(get).get();
|
||||||
table2.get(get).get();
|
table2.get(get).get();
|
||||||
|
|
||||||
admin.listTableNames(Optional.of(Pattern.compile(tableName.getNameAsString() + ".*")), false)
|
admin.listTableNames(Pattern.compile(tableName.getNameAsString() + ".*"), false)
|
||||||
.get().forEach(t -> admin.disableTable(t).join());
|
.get().forEach(t -> admin.disableTable(t).join());
|
||||||
|
|
||||||
// Test that tables are disabled
|
// Test that tables are disabled
|
||||||
|
@ -541,7 +541,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
assertEquals(TableState.State.DISABLED, getStateFromMeta(tableName1));
|
assertEquals(TableState.State.DISABLED, getStateFromMeta(tableName1));
|
||||||
assertEquals(TableState.State.DISABLED, getStateFromMeta(tableName2));
|
assertEquals(TableState.State.DISABLED, getStateFromMeta(tableName2));
|
||||||
|
|
||||||
admin.listTableNames(Optional.of(Pattern.compile(tableName.getNameAsString() + ".*")), false)
|
admin.listTableNames(Pattern.compile(tableName.getNameAsString() + ".*"), false)
|
||||||
.get().forEach(t -> admin.enableTable(t).join());
|
.get().forEach(t -> admin.enableTable(t).join());
|
||||||
|
|
||||||
// Test that tables are enabled
|
// Test that tables are enabled
|
||||||
|
@ -567,7 +567,7 @@ public class TestAsyncTableAdminApi extends TestAsyncAdminBase {
|
||||||
new byte[] { 4, 4, 4 }, new byte[] { 5, 5, 5 }, new byte[] { 6, 6, 6 },
|
new byte[] { 4, 4, 4 }, new byte[] { 5, 5, 5 }, new byte[] { 6, 6, 6 },
|
||||||
new byte[] { 7, 7, 7 }, new byte[] { 8, 8, 8 }, new byte[] { 9, 9, 9 } };
|
new byte[] { 7, 7, 7 }, new byte[] { 8, 8, 8 }, new byte[] { 9, 9, 9 } };
|
||||||
int expectedRegions = splitKeys.length + 1;
|
int expectedRegions = splitKeys.length + 1;
|
||||||
createTableWithDefaultConf(tableName, Optional.of(splitKeys));
|
createTableWithDefaultConf(tableName, splitKeys);
|
||||||
|
|
||||||
RawAsyncTable metaTable = ASYNC_CONN.getRawTable(META_TABLE_NAME);
|
RawAsyncTable metaTable = ASYNC_CONN.getRawTable(META_TABLE_NAME);
|
||||||
List<HRegionLocation> regions =
|
List<HRegionLocation> regions =
|
||||||
|
|
|
@ -1543,7 +1543,7 @@ public class TestMasterObserver {
|
||||||
assertTrue("Found server", found);
|
assertTrue("Found server", found);
|
||||||
LOG.info("Found " + destName);
|
LOG.info("Found " + destName);
|
||||||
master.getMasterRpcServices().moveRegion(null, RequestConverter.buildMoveRegionRequest(
|
master.getMasterRpcServices().moveRegion(null, RequestConverter.buildMoveRegionRequest(
|
||||||
firstGoodPair.getRegionInfo().getEncodedNameAsBytes(),Bytes.toBytes(destName)));
|
firstGoodPair.getRegionInfo().getEncodedNameAsBytes(), ServerName.valueOf(destName)));
|
||||||
assertTrue("Coprocessor should have been called on region move",
|
assertTrue("Coprocessor should have been called on region move",
|
||||||
cp.wasMoveCalled());
|
cp.wasMoveCalled());
|
||||||
|
|
||||||
|
@ -1565,11 +1565,12 @@ public class TestMasterObserver {
|
||||||
UTIL.waitUntilNoRegionsInTransition();
|
UTIL.waitUntilNoRegionsInTransition();
|
||||||
List<RegionInfo> openRegions = ProtobufUtil.getOnlineRegions(rs.getRSRpcServices());
|
List<RegionInfo> openRegions = ProtobufUtil.getOnlineRegions(rs.getRSRpcServices());
|
||||||
int moveCnt = openRegions.size()/2;
|
int moveCnt = openRegions.size()/2;
|
||||||
for (int i=0; i<moveCnt; i++) {
|
for (int i = 0; i < moveCnt; i++) {
|
||||||
RegionInfo info = openRegions.get(i);
|
RegionInfo info = openRegions.get(i);
|
||||||
if (!info.isMetaRegion()) {
|
if (!info.isMetaRegion()) {
|
||||||
master.getMasterRpcServices().moveRegion(null, RequestConverter.buildMoveRegionRequest(
|
master.getMasterRpcServices().moveRegion(null,
|
||||||
openRegions.get(i).getEncodedNameAsBytes(), destRS));
|
RequestConverter.buildMoveRegionRequest(openRegions.get(i).getEncodedNameAsBytes(),
|
||||||
|
ServerName.valueOf(Bytes.toString(destRS))));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
//Make sure no regions are in transition now
|
//Make sure no regions are in transition now
|
||||||
|
|
|
@ -242,7 +242,7 @@ public class TestHRegionServerBulkLoad {
|
||||||
AdminProtos.AdminService.BlockingInterface server =
|
AdminProtos.AdminService.BlockingInterface server =
|
||||||
conn.getAdmin(getLocation().getServerName());
|
conn.getAdmin(getLocation().getServerName());
|
||||||
CompactRegionRequest request = RequestConverter.buildCompactRegionRequest(
|
CompactRegionRequest request = RequestConverter.buildCompactRegionRequest(
|
||||||
getLocation().getRegionInfo().getRegionName(), true, Optional.empty());
|
getLocation().getRegionInfo().getRegionName(), true, null);
|
||||||
server.compactRegion(null, request);
|
server.compactRegion(null, request);
|
||||||
numCompactions.incrementAndGet();
|
numCompactions.incrementAndGet();
|
||||||
return null;
|
return null;
|
||||||
|
|
|
@ -20,7 +20,6 @@ package org.apache.hadoop.hbase.regionserver;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Optional;
|
|
||||||
import java.util.concurrent.atomic.AtomicLong;
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
|
|
||||||
import org.apache.commons.logging.Log;
|
import org.apache.commons.logging.Log;
|
||||||
|
@ -124,7 +123,7 @@ public class TestHRegionServerBulkLoadWithOldClient extends TestHRegionServerBul
|
||||||
conn.getAdmin(getLocation().getServerName());
|
conn.getAdmin(getLocation().getServerName());
|
||||||
CompactRegionRequest request =
|
CompactRegionRequest request =
|
||||||
RequestConverter.buildCompactRegionRequest(
|
RequestConverter.buildCompactRegionRequest(
|
||||||
getLocation().getRegionInfo().getRegionName(), true, Optional.empty());
|
getLocation().getRegionInfo().getRegionName(), true, null);
|
||||||
server.compactRegion(null, request);
|
server.compactRegion(null, request);
|
||||||
numCompactions.incrementAndGet();
|
numCompactions.incrementAndGet();
|
||||||
return null;
|
return null;
|
||||||
|
|
Loading…
Reference in New Issue