openinx commented on a change in pull request #126: HBASE-21718 Implement Admin based on AsyncAdmin URL: https://github.com/apache/hbase/pull/126#discussion_r273767968
########## File path: hbase-client/src/main/java/org/apache/hadoop/hbase/client/AdminOverAsyncAdmin.java ########## @@ -0,0 +1,929 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hbase.client; + +import static org.apache.hadoop.hbase.client.ConnectionUtils.setCoprocessorError; +import static org.apache.hadoop.hbase.util.FutureUtils.get; + +import com.google.protobuf.Descriptors.MethodDescriptor; +import com.google.protobuf.Message; +import com.google.protobuf.RpcCallback; +import com.google.protobuf.RpcChannel; +import com.google.protobuf.RpcController; +import com.google.protobuf.ServiceException; +import java.io.IOException; +import java.util.Arrays; +import java.util.EnumSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Future; +import java.util.regex.Pattern; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.CacheEvictionStats; +import org.apache.hadoop.hbase.ClusterMetrics; +import org.apache.hadoop.hbase.ClusterMetrics.Option; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.NamespaceDescriptor; +import org.apache.hadoop.hbase.NamespaceNotFoundException; +import org.apache.hadoop.hbase.RegionMetrics; +import org.apache.hadoop.hbase.ServerName; +import org.apache.hadoop.hbase.TableExistsException; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.TableNotFoundException; +import org.apache.hadoop.hbase.client.replication.TableCFs; +import org.apache.hadoop.hbase.client.security.SecurityCapability; +import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel; +import org.apache.hadoop.hbase.quotas.QuotaFilter; +import org.apache.hadoop.hbase.quotas.QuotaSettings; +import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshotView; +import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException; +import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; +import org.apache.hadoop.hbase.replication.ReplicationPeerDescription; +import org.apache.hadoop.hbase.replication.SyncReplicationState; +import org.apache.hadoop.hbase.security.access.GetUserPermissionsRequest; +import org.apache.hadoop.hbase.security.access.UserPermission; +import org.apache.hadoop.hbase.snapshot.HBaseSnapshotException; +import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException; +import org.apache.hadoop.hbase.snapshot.SnapshotCreationException; +import org.apache.hadoop.hbase.snapshot.UnknownSnapshotException; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.yetus.audience.InterfaceAudience; + +/** + * The {@link Admin} implementation which is based on an {@link AsyncAdmin}. + */ [email protected] +class AdminOverAsyncAdmin implements Admin { + + private volatile boolean aborted = false; + + private final Connection conn; + + private final RawAsyncHBaseAdmin admin; + + private final int operationTimeout; + + private final int syncWaitTimeout; + + public AdminOverAsyncAdmin(Connection conn, RawAsyncHBaseAdmin admin) { + this.conn = conn; + this.admin = admin; + this.operationTimeout = conn.getConfiguration().getInt( + HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT); + this.syncWaitTimeout = + conn.getConfiguration().getInt("hbase.client.sync.wait.timeout.msec", 10 * 60000); // 10min + } + + @Override + public int getOperationTimeout() { + return operationTimeout; + } + + @Override + public int getSyncWaitTimeout() { + return syncWaitTimeout; + } + + @Override + public void abort(String why, Throwable e) { + this.aborted = true; + } + + @Override + public boolean isAborted() { + return aborted; + } + + @Override + public Connection getConnection() { + return conn; + } + + @Override + public boolean tableExists(TableName tableName) throws IOException { + return get(admin.tableExists(tableName)); + } + + @Override + public List<TableDescriptor> listTableDescriptors() throws IOException { + return get(admin.listTableDescriptors()); + } + + @Override + public List<TableDescriptor> listTableDescriptors(Pattern pattern, boolean includeSysTables) + throws IOException { + return get(admin.listTableDescriptors(pattern, includeSysTables)); + } + + @Override + public TableName[] listTableNames() throws IOException { + return get(admin.listTableNames()).toArray(new TableName[0]); + } + + @Override + public TableName[] listTableNames(Pattern pattern, boolean includeSysTables) throws IOException { + return get(admin.listTableNames(pattern, includeSysTables)).toArray(new TableName[0]); + } + + @Override + public TableDescriptor getDescriptor(TableName tableName) + throws TableNotFoundException, IOException { + return get(admin.getDescriptor(tableName)); + } + + @Override + public void createTable(TableDescriptor desc, byte[] startKey, byte[] endKey, int numRegions) + throws IOException { + get(admin.createTable(desc, startKey, endKey, numRegions)); + } + + @Override + public Future<Void> createTableAsync(TableDescriptor desc) + throws IOException { + return admin.createTable(desc); + } + + @Override + public Future<Void> createTableAsync(TableDescriptor desc, byte[][] splitKeys) + throws IOException { + return admin.createTable(desc, splitKeys); + } + + @Override + public Future<Void> deleteTableAsync(TableName tableName) throws IOException { + return admin.deleteTable(tableName); + } + + @Override + public Future<Void> truncateTableAsync(TableName tableName, boolean preserveSplits) + throws IOException { + return admin.truncateTable(tableName, preserveSplits); + } + + @Override + public Future<Void> enableTableAsync(TableName tableName) throws IOException { + return admin.enableTable(tableName); + } + + @Override + public Future<Void> disableTableAsync(TableName tableName) throws IOException { + return admin.disableTable(tableName); + } + + @Override + public boolean isTableEnabled(TableName tableName) throws IOException { + return get(admin.isTableEnabled(tableName)); + } + + @Override + public boolean isTableDisabled(TableName tableName) throws IOException { + return get(admin.isTableDisabled(tableName)); + } + + @Override + public boolean isTableAvailable(TableName tableName) throws IOException { + return get(admin.isTableAvailable(tableName)); + } + + @Override + public Future<Void> addColumnFamilyAsync(TableName tableName, ColumnFamilyDescriptor columnFamily) + throws IOException { + return admin.addColumnFamily(tableName, columnFamily); + } + + @Override + public Future<Void> deleteColumnFamilyAsync(TableName tableName, byte[] columnFamily) + throws IOException { + return admin.deleteColumnFamily(tableName, columnFamily); + } + + @Override + public Future<Void> modifyColumnFamilyAsync(TableName tableName, + ColumnFamilyDescriptor columnFamily) throws IOException { + return admin.modifyColumnFamily(tableName, columnFamily); + } + + @Override + public List<RegionInfo> getRegions(ServerName serverName) throws IOException { + return get(admin.getRegions(serverName)); + } + + @Override + public void flush(TableName tableName) throws IOException { + get(admin.flush(tableName)); + } + + @Override + public void flushRegion(byte[] regionName) throws IOException { + get(admin.flushRegion(regionName)); + } + + @Override + public void flushRegionServer(ServerName serverName) throws IOException { + get(admin.flushRegionServer(serverName)); + } + + @Override + public void compact(TableName tableName) throws IOException { + get(admin.compact(tableName)); + } + + @Override + public void compactRegion(byte[] regionName) throws IOException { + get(admin.compactRegion(regionName)); + } + + @Override + public void compact(TableName tableName, byte[] columnFamily) throws IOException { + get(admin.compact(tableName, columnFamily)); + } + + @Override + public void compactRegion(byte[] regionName, byte[] columnFamily) throws IOException { + get(admin.compactRegion(regionName, columnFamily)); + } + + @Override + public void compact(TableName tableName, CompactType compactType) + throws IOException, InterruptedException { + get(admin.compact(tableName, compactType)); + } + + @Override + public void compact(TableName tableName, byte[] columnFamily, CompactType compactType) + throws IOException, InterruptedException { + get(admin.compact(tableName, columnFamily, compactType)); + } + + @Override + public void majorCompact(TableName tableName) throws IOException { + get(admin.majorCompact(tableName)); + } + + @Override + public void majorCompactRegion(byte[] regionName) throws IOException { + get(admin.majorCompactRegion(regionName)); + } + + @Override + public void majorCompact(TableName tableName, byte[] columnFamily) throws IOException { + get(admin.majorCompact(tableName, columnFamily)); + } + + @Override + public void majorCompactRegion(byte[] regionName, byte[] columnFamily) throws IOException { + get(admin.majorCompactRegion(regionName, columnFamily)); + } + + @Override + public void majorCompact(TableName tableName, CompactType compactType) + throws IOException, InterruptedException { + get(admin.majorCompact(tableName, compactType)); + } + + @Override + public void majorCompact(TableName tableName, byte[] columnFamily, CompactType compactType) + throws IOException, InterruptedException { + get(admin.majorCompact(tableName, columnFamily, compactType)); + } + + @Override + public Map<ServerName, Boolean> compactionSwitch(boolean switchState, + List<String> serverNamesList) throws IOException { + return get(admin.compactionSwitch(switchState, serverNamesList)); + } + + @Override + public void compactRegionServer(ServerName serverName) throws IOException { + get(admin.compactRegionServer(serverName)); + } + + @Override + public void majorCompactRegionServer(ServerName serverName) throws IOException { + get(admin.majorCompactRegionServer(serverName)); + } + + @Override + public void move(byte[] encodedRegionName) throws IOException { + get(admin.move(encodedRegionName)); + } + + @Override + public void move(byte[] encodedRegionName, ServerName destServerName) throws IOException { + get(admin.move(encodedRegionName, destServerName)); + } + + @Override + public void assign(byte[] regionName) throws IOException { + get(admin.assign(regionName)); + } + + @Override + public void unassign(byte[] regionName, boolean force) throws IOException { + get(admin.unassign(regionName, force)); + } + + @Override + public void offline(byte[] regionName) throws IOException { + get(admin.offline(regionName)); + } + + @Override + public boolean balancerSwitch(boolean onOrOff, boolean synchronous) throws IOException { + return get(admin.balancerSwitch(onOrOff)); + } + + @Override + public boolean balance() throws IOException { + return get(admin.balance()); + } + + @Override + public boolean balance(boolean force) throws IOException { + return get(admin.balance(force)); + } + + @Override + public boolean isBalancerEnabled() throws IOException { + return get(admin.isBalancerEnabled()); + } + + @Override + public CacheEvictionStats clearBlockCache(TableName tableName) throws IOException { + return get(admin.clearBlockCache(tableName)); + } + + @Override + public boolean normalize() throws IOException { + return get(admin.normalize()); + } + + @Override + public boolean isNormalizerEnabled() throws IOException { + return get(admin.isNormalizerEnabled()); + } + + @Override + public boolean normalizerSwitch(boolean on) throws IOException { + return get(admin.normalizerSwitch(on)); + } + + @Override + public boolean catalogJanitorSwitch(boolean onOrOff) throws IOException { + return get(admin.catalogJanitorSwitch(onOrOff)); + } + + @Override + public int runCatalogJanitor() throws IOException { + return get(admin.runCatalogJanitor()); + } + + @Override + public boolean isCatalogJanitorEnabled() throws IOException { + return get(admin.isCatalogJanitorEnabled()); + } + + @Override + public boolean cleanerChoreSwitch(boolean onOrOff) throws IOException { + return get(admin.cleanerChoreSwitch(onOrOff)); + } + + @Override + public boolean runCleanerChore() throws IOException { + return get(admin.runCleanerChore()); + } + + @Override + public boolean isCleanerChoreEnabled() throws IOException { + return get(admin.isCleanerChoreEnabled()); + } + + @Override + public Future<Void> mergeRegionsAsync(byte[][] nameOfRegionsToMerge, boolean forcible) + throws IOException { + return admin.mergeRegions(Arrays.asList(nameOfRegionsToMerge), forcible); + } + + @Override + public void split(TableName tableName) throws IOException { + get(admin.split(tableName)); + } + + @Override + public void split(TableName tableName, byte[] splitPoint) throws IOException { + get(admin.split(tableName, splitPoint)); + } + + @Override + public Future<Void> splitRegionAsync(byte[] regionName) throws IOException { + return admin.splitRegion(regionName); + } + + @Override + public Future<Void> splitRegionAsync(byte[] regionName, byte[] splitPoint) throws IOException { + return admin.splitRegion(regionName, splitPoint); + } + + @Override + public Future<Void> modifyTableAsync(TableDescriptor td) throws IOException { + return admin.modifyTable(td); + } + + @Override + public void shutdown() throws IOException { + get(admin.shutdown()); + } + + @Override + public void stopMaster() throws IOException { + get(admin.stopMaster()); + } + + @Override + public boolean isMasterInMaintenanceMode() throws IOException { + return get(admin.isMasterInMaintenanceMode()); + } + + @Override + public void stopRegionServer(String hostnamePort) throws IOException { + get(admin.stopRegionServer(ServerName.valueOf(hostnamePort, 0))); + } + + @Override + public ClusterMetrics getClusterMetrics(EnumSet<Option> options) throws IOException { + return get(admin.getClusterMetrics(options)); + } + + @Override + public List<RegionMetrics> getRegionMetrics(ServerName serverName) throws IOException { + return get(admin.getRegionMetrics(serverName)); + } + + @Override + public List<RegionMetrics> getRegionMetrics(ServerName serverName, TableName tableName) + throws IOException { + return get(admin.getRegionMetrics(serverName, tableName)); + } + + @Override + public Configuration getConfiguration() { + return conn.getConfiguration(); + } + + @Override + public Future<Void> createNamespaceAsync(NamespaceDescriptor descriptor) throws IOException { + return admin.createNamespace(descriptor); + } + + @Override + public Future<Void> modifyNamespaceAsync(NamespaceDescriptor descriptor) throws IOException { + return admin.modifyNamespace(descriptor); + } + + @Override + public Future<Void> deleteNamespaceAsync(String name) throws IOException { + return admin.deleteNamespace(name); + } + + @Override + public NamespaceDescriptor getNamespaceDescriptor(String name) + throws NamespaceNotFoundException, IOException { + return get(admin.getNamespaceDescriptor(name)); + } + + @Override + public NamespaceDescriptor[] listNamespaceDescriptors() throws IOException { + return get(admin.listNamespaceDescriptors()).toArray(new NamespaceDescriptor[0]); + } + + @Override + public List<TableDescriptor> listTableDescriptorsByNamespace(byte[] name) throws IOException { + return get(admin.listTableDescriptorsByNamespace(Bytes.toString(name))); + } + + @Override + public TableName[] listTableNamesByNamespace(String name) throws IOException { + return get(admin.listTableNamesByNamespace(name)).toArray(new TableName[0]); + } + + @Override + public List<RegionInfo> getRegions(TableName tableName) throws IOException { + return get(admin.getRegions(tableName)); + } + + @Override + public void close() { + // do nothing, AsyncAdmin is not a Closeable. + } + + @Override + public List<TableDescriptor> listTableDescriptors(List<TableName> tableNames) throws IOException { + return get(admin.listTableDescriptors(tableNames)); + } + + @Override + public Future<Boolean> abortProcedureAsync(long procId, boolean mayInterruptIfRunning) + throws IOException { + return admin.abortProcedure(procId, mayInterruptIfRunning); + } + + @Override + public String getProcedures() throws IOException { + return get(admin.getProcedures()); + } + + @Override + public String getLocks() throws IOException { + return get(admin.getLocks()); + } + + @Override + public void rollWALWriter(ServerName serverName) throws IOException, FailedLogCloseException { + get(admin.rollWALWriter(serverName)); + } + + @Override + public CompactionState getCompactionState(TableName tableName) throws IOException { + return get(admin.getCompactionState(tableName)); + } + + @Override + public CompactionState getCompactionState(TableName tableName, CompactType compactType) + throws IOException { + return get(admin.getCompactionState(tableName, compactType)); + } + + @Override + public CompactionState getCompactionStateForRegion(byte[] regionName) throws IOException { + return get(admin.getCompactionStateForRegion(regionName)); + } + + @Override + public long getLastMajorCompactionTimestamp(TableName tableName) throws IOException { + return get(admin.getLastMajorCompactionTimestamp(tableName)).orElse(0L); + } + + @Override + public long getLastMajorCompactionTimestampForRegion(byte[] regionName) throws IOException { + return get(admin.getLastMajorCompactionTimestampForRegion(regionName)).orElse(0L); + } + + @Override + public void snapshot(SnapshotDescription snapshot) + throws IOException, SnapshotCreationException, IllegalArgumentException { + get(admin.snapshot(snapshot)); + } + + @Override + public Future<Void> snapshotAsync(SnapshotDescription snapshot) + throws IOException, SnapshotCreationException { + return admin.snapshot(snapshot); + } + + @Override + public boolean isSnapshotFinished(SnapshotDescription snapshot) + throws IOException, HBaseSnapshotException, UnknownSnapshotException { + return get(admin.isSnapshotFinished(snapshot)); + } + + @Override + public void restoreSnapshot(String snapshotName) throws IOException, RestoreSnapshotException { + get(admin.restoreSnapshot(snapshotName)); + } + + @Override + public void restoreSnapshot(String snapshotName, boolean takeFailSafeSnapshot, boolean restoreAcl) + throws IOException, RestoreSnapshotException { + get(admin.restoreSnapshot(snapshotName, takeFailSafeSnapshot, restoreAcl)); + } + + @Override + public Future<Void> cloneSnapshotAsync(String snapshotName, TableName tableName, + boolean restoreAcl) throws IOException, TableExistsException, RestoreSnapshotException { + return admin.cloneSnapshot(snapshotName, tableName, restoreAcl); + } + + @Override + public void execProcedure(String signature, String instance, Map<String, String> props) + throws IOException { + get(admin.execProcedure(signature, instance, props)); + } + + @Override + public byte[] execProcedureWithReturn(String signature, String instance, + Map<String, String> props) throws IOException { + return get(admin.execProcedureWithReturn(signature, instance, props)); + } + + @Override + public boolean isProcedureFinished(String signature, String instance, Map<String, String> props) + throws IOException { + return get(admin.isProcedureFinished(signature, instance, props)); + } + + @Override + public List<SnapshotDescription> listSnapshots() throws IOException { + return get(admin.listSnapshots()); + } + + @Override + public List<SnapshotDescription> listSnapshots(Pattern pattern) throws IOException { + return get(admin.listSnapshots(pattern)); + } + + @Override + public List<SnapshotDescription> listTableSnapshots(Pattern tableNamePattern, + Pattern snapshotNamePattern) throws IOException { + return get(admin.listTableSnapshots(tableNamePattern, snapshotNamePattern)); + } + + @Override + public void deleteSnapshot(String snapshotName) throws IOException { + get(admin.deleteSnapshot(snapshotName)); + } + + @Override + public void deleteSnapshots(Pattern pattern) throws IOException { + get(admin.deleteSnapshots(pattern)); + } + + @Override + public void deleteTableSnapshots(Pattern tableNamePattern, Pattern snapshotNamePattern) + throws IOException { + get(admin.deleteTableSnapshots(tableNamePattern, snapshotNamePattern)); + } + + @Override + public void setQuota(QuotaSettings quota) throws IOException { + get(admin.setQuota(quota)); + } + + @Override + public List<QuotaSettings> getQuota(QuotaFilter filter) throws IOException { + return get(admin.getQuota(filter)); + } + + @SuppressWarnings("deprecation") + private static final class SyncCoprocessorRpcChannelOverAsync implements CoprocessorRpcChannel { + + private final RpcChannel delegate; + + public SyncCoprocessorRpcChannelOverAsync(RpcChannel delegate) { + this.delegate = delegate; + } + + @Override + public void callMethod(MethodDescriptor method, RpcController controller, Message request, + Message responsePrototype, RpcCallback<Message> done) { + ClientCoprocessorRpcController c = new ClientCoprocessorRpcController(); + CoprocessorBlockingRpcCallback<Message> callback = new CoprocessorBlockingRpcCallback<>(); + delegate.callMethod(method, c, request, responsePrototype, callback); + Message ret; + try { + ret = callback.get(); + } catch (IOException e) { + setCoprocessorError(controller, e); + return; + } + if (c.failed()) { + setCoprocessorError(controller, c.getFailed()); + } + done.run(ret); + } + + @Override + public Message callBlockingMethod(MethodDescriptor method, RpcController controller, + Message request, Message responsePrototype) throws ServiceException { + ClientCoprocessorRpcController c = new ClientCoprocessorRpcController(); + CoprocessorBlockingRpcCallback<Message> done = new CoprocessorBlockingRpcCallback<>(); + callMethod(method, c, request, responsePrototype, done); + Message ret; + try { + ret = done.get(); + } catch (IOException e) { + throw new ServiceException(e); + } + if (c.failed()) { + setCoprocessorError(controller, c.getFailed()); + throw new ServiceException(c.getFailed()); + } + return ret; + } + } + + @SuppressWarnings("deprecation") + @Override + public CoprocessorRpcChannel coprocessorService() { + return new SyncCoprocessorRpcChannelOverAsync( + new MasterCoprocessorRpcChannelImpl(admin.<Message> newMasterCaller())); + } + + @SuppressWarnings("deprecation") + @Override + public CoprocessorRpcChannel coprocessorService(ServerName serverName) { + return new SyncCoprocessorRpcChannelOverAsync(new RegionServerCoprocessorRpcChannelImpl( + admin.<Message> newServerCaller().serverName(serverName))); + } + + @Override + public void updateConfiguration(ServerName server) throws IOException { Review comment: May we need an deprecated annotation here ? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
