hbase-client/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.client;
  19
  20 import com.google.protobuf.Descriptors;
  21 import com.google.protobuf.Message;
  22 import com.google.protobuf.RpcController;
  23 import java.io.Closeable;
  24 import java.io.IOException;
  25 import java.io.InterruptedIOException;
  26 import java.util.ArrayList;
  27 import java.util.Arrays;
  28 import java.util.Collections;
  29 import java.util.EnumSet;
  30 import java.util.HashMap;
  31 import java.util.Iterator;
  32 import java.util.LinkedList;
  33 import java.util.List;
  34 import java.util.Map;
  35 import java.util.Set;
  36 import java.util.concurrent.Callable;
  37 import java.util.concurrent.ExecutionException;
  38 import java.util.concurrent.Future;
  39 import java.util.concurrent.TimeUnit;
  40 import java.util.concurrent.TimeoutException;
  41 import java.util.concurrent.atomic.AtomicInteger;
  42 import java.util.concurrent.atomic.AtomicReference;
  43 import java.util.function.Supplier;
  44 import java.util.regex.Pattern;
  45 import java.util.stream.Collectors;
  46 import java.util.stream.Stream;
  47 import org.apache.hadoop.conf.Configuration;
  48 import org.apache.hadoop.hbase.CacheEvictionStats;
  49 import org.apache.hadoop.hbase.CacheEvictionStatsBuilder;
  50 import org.apache.hadoop.hbase.ClusterMetrics;
  51 import org.apache.hadoop.hbase.ClusterMetrics.Option;
  52 import org.apache.hadoop.hbase.ClusterMetricsBuilder;
  53 import org.apache.hadoop.hbase.DoNotRetryIOException;
  54 import org.apache.hadoop.hbase.HBaseConfiguration;
  55 import org.apache.hadoop.hbase.HConstants;
  56 import org.apache.hadoop.hbase.HRegionLocation;
  57 import org.apache.hadoop.hbase.MasterNotRunningException;
  58 import org.apache.hadoop.hbase.MetaTableAccessor;
  59 import org.apache.hadoop.hbase.NamespaceDescriptor;
  60 import org.apache.hadoop.hbase.NamespaceNotFoundException;
  61 import org.apache.hadoop.hbase.NotServingRegionException;
  62 import org.apache.hadoop.hbase.RegionLocations;
  63 import org.apache.hadoop.hbase.RegionMetrics;
  64 import org.apache.hadoop.hbase.RegionMetricsBuilder;
  65 import org.apache.hadoop.hbase.ServerName;
  66 import org.apache.hadoop.hbase.TableExistsException;
  67 import org.apache.hadoop.hbase.TableName;
  68 import org.apache.hadoop.hbase.TableNotDisabledException;
  69 import org.apache.hadoop.hbase.TableNotFoundException;
  70 import org.apache.hadoop.hbase.UnknownRegionException;
  71 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
  72 import org.apache.hadoop.hbase.client.replication.ReplicationPeerConfigUtil;
  73 import org.apache.hadoop.hbase.client.replication.TableCFs;
  74 import org.apache.hadoop.hbase.client.security.SecurityCapability;
  75 import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
  76 import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
  77 import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils;
  78 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
  79 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
  80 import org.apache.hadoop.hbase.quotas.QuotaFilter;
  81 import org.apache.hadoop.hbase.quotas.QuotaRetriever;
  82 import org.apache.hadoop.hbase.quotas.QuotaSettings;
  83 import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot;
  84 import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException;
  85 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
  86 import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
  87 import org.apache.hadoop.hbase.replication.SyncReplicationState;
  88 import org.apache.hadoop.hbase.security.UserProvider;
  89 import org.apache.hadoop.hbase.security.access.GetUserPermissionsRequest;
  90 import org.apache.hadoop.hbase.security.access.Permission;
  91 import org.apache.hadoop.hbase.security.access.ShadedAccessControlUtil;
  92 import org.apache.hadoop.hbase.security.access.UserPermission;
  93 import org.apache.hadoop.hbase.snapshot.ClientSnapshotDescriptionUtils;
  94 import org.apache.hadoop.hbase.snapshot.HBaseSnapshotException;
  95 import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException;
  96 import org.apache.hadoop.hbase.snapshot.SnapshotCreationException;
  97 import org.apache.hadoop.hbase.snapshot.UnknownSnapshotException;
  98 import org.apache.hadoop.hbase.util.Addressing;
  99 import org.apache.hadoop.hbase.util.Bytes;
 100 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 101 import org.apache.hadoop.hbase.util.ForeignExceptionUtil;
 102 import org.apache.hadoop.hbase.util.Pair;
 103 import org.apache.hadoop.ipc.RemoteException;
 104 import org.apache.hadoop.util.StringUtils;
 105 import org.apache.yetus.audience.InterfaceAudience;
 106 import org.apache.yetus.audience.InterfaceStability;
 107 import org.slf4j.Logger;
 108 import org.slf4j.LoggerFactory;
 109
 110 import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
 111 import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
 112 import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
 113
 114 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
 115 import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
 116 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos;
 117 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.GrantRequest;
 118 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.HasUserPermissionsRequest;
 119 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.RevokeRequest;
 120 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
 121 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
 122 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesRequest;
 123 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheRequest;
 124 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheResponse;
 125 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CompactRegionRequest;
 126 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CompactionSwitchRequest;
 127 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CompactionSwitchResponse;
 128 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.FlushRegionRequest;
 129 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoRequest;
 130 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoResponse;
 131 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RollWALWriterRequest;
 132 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RollWALWriterResponse;
 133 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerRequest;
 134 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
 135 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
 136 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.CoprocessorServiceRequest;
 137 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.CoprocessorServiceResponse;
 138 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
 139 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.ProcedureDescription;
 140 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
 141 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos;
 142 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.AbortProcedureRequest;
 143 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.AbortProcedureResponse;
 144 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.AddColumnRequest;
 145 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.AddColumnResponse;
 146 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.AssignRegionRequest;
 147 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ClearDeadServersRequest;
 148 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.CreateNamespaceRequest;
 149 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.CreateNamespaceResponse;
 150 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.CreateTableRequest;
 151 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.CreateTableResponse;
 152 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DeleteColumnRequest;
 153 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DeleteColumnResponse;
 154 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DeleteNamespaceRequest;
 155 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DeleteNamespaceResponse;
 156 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DeleteSnapshotRequest;
 157 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DeleteTableRequest;
 158 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DeleteTableResponse;
 159 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DisableTableRequest;
 160 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.DisableTableResponse;
 161 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.EnableTableRequest;
 162 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.EnableTableResponse;
 163 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ExecProcedureRequest;
 164 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ExecProcedureResponse;
 165 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetClusterStatusRequest;
 166 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetCompletedSnapshotsRequest;
 167 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetLocksRequest;
 168 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetLocksResponse;
 169 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetNamespaceDescriptorRequest;
 170 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetProcedureResultRequest;
 171 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetProcedureResultResponse;
 172 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetProceduresRequest;
 173 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetProceduresResponse;
 174 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetTableDescriptorsRequest;
 175 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetTableDescriptorsResponse;
 176 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.GetTableNamesRequest;
 177 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsInMaintenanceModeRequest;
 178 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsInMaintenanceModeResponse;
 179 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsProcedureDoneRequest;
 180 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsProcedureDoneResponse;
 181 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsRpcThrottleEnabledRequest;
 182 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsSnapshotDoneRequest;
 183 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.IsSnapshotDoneResponse;
 184 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListDecommissionedRegionServersRequest;
 185 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListNamespaceDescriptorsRequest;
 186 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListNamespacesRequest;
 187 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListTableDescriptorsByNamespaceRequest;
 188 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ListTableNamesByNamespaceRequest;
 189 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MajorCompactionTimestampForRegionRequest;
 190 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MajorCompactionTimestampRequest;
 191 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MergeTableRegionsRequest;
 192 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MergeTableRegionsResponse;
 193 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ModifyColumnRequest;
 194 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ModifyColumnResponse;
 195 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ModifyNamespaceRequest;
 196 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ModifyNamespaceResponse;
 197 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ModifyTableRequest;
 198 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ModifyTableResponse;
 199 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.MoveRegionRequest;
 200 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.RestoreSnapshotRequest;
 201 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.RestoreSnapshotResponse;
 202 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SecurityCapabilitiesRequest;
 203 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SetBalancerRunningRequest;
 204 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SetNormalizerRunningRequest;
 205 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.ShutdownRequest;
 206 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SnapshotRequest;
 207 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SnapshotResponse;
 208 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SplitTableRegionRequest;
 209 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.SplitTableRegionResponse;
 210 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.StopMasterRequest;
 211 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.TruncateTableRequest;
 212 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.TruncateTableResponse;
 213 import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.UnassignRegionRequest;
 214 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetQuotaStatesResponse;
 215 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetSpaceQuotaRegionSizesResponse;
 216 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetSpaceQuotaRegionSizesResponse.RegionSizes;
 217 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetSpaceQuotaSnapshotsResponse;
 218 import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetSpaceQuotaSnapshotsResponse.TableQuotaSnapshot;
 219 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos;
 220 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.AddReplicationPeerResponse;
 221 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.DisableReplicationPeerResponse;
 222 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.EnableReplicationPeerResponse;
 223 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.GetReplicationPeerConfigResponse;
 224 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.RemoveReplicationPeerResponse;
 225 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.TransitReplicationPeerSyncReplicationStateResponse;
 226 import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.UpdateReplicationPeerConfigResponse;
 227 import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos;
 228
 229 /**
 230  * HBaseAdmin is no longer a client API. It is marked InterfaceAudience.Private indicating that
 231  * this is an HBase-internal class as defined in
 232  * https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/InterfaceClassification.html
 233  * There are no guarantees for backwards source / binary compatibility and methods or class can
 234  * change or go away without deprecation.
 235  * Use {@link Connection#getAdmin()} to obtain an instance of {@link Admin} instead of constructing
 236  * an HBaseAdmin directly.
 237  *
 238  * <p>Connection should be an <i>unmanaged</i> connection obtained via
 239  * {@link ConnectionFactory#createConnection(Configuration)}
 240  *
 241  * @see ConnectionFactory
 242  * @see Connection
 243  * @see Admin
 244  */
 245 @InterfaceAudience.Private
 246 public class HBaseAdmin implements Admin {
 247   private static final Logger LOG = LoggerFactory.getLogger(HBaseAdmin.class);
 248
 249   private ConnectionImplementation connection;
 250
 251   private final Configuration conf;
 252   private final long pause;
 253   private final int numRetries;
 254   private final int syncWaitTimeout;
 255   private boolean aborted;
 256   private int operationTimeout;
 257   private int rpcTimeout;
 258   private int getProcedureTimeout;
 259
 260   private RpcRetryingCallerFactory rpcCallerFactory;
 261   private RpcControllerFactory rpcControllerFactory;
 262
 263   private NonceGenerator ng;
 264
 265   @Override
 266   public int getOperationTimeout() {
 267     return operationTimeout;
 268   }
 269
 270   @Override
 271   public int getSyncWaitTimeout() {
 272     return syncWaitTimeout;
 273   }
 274
 275   HBaseAdmin(ConnectionImplementation connection) throws IOException {
 276     this.conf = connection.getConfiguration();
 277     this.connection = connection;
 278
 279     // TODO: receive ConnectionConfiguration here rather than re-parsing these configs every time.
 280     this.pause = this.conf.getLong(HConstants.HBASE_CLIENT_PAUSE,
 281         HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
 282     this.numRetries = this.conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
 283         HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
 284     this.operationTimeout = this.conf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
 285         HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
 286     this.rpcTimeout = this.conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
 287         HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
 288     this.syncWaitTimeout = this.conf.getInt(
 289       "hbase.client.sync.wait.timeout.msec", 10 * 60000); // 10min
 290     this.getProcedureTimeout =
 291         this.conf.getInt("hbase.client.procedure.future.get.timeout.msec", 10 * 60000); // 10min
 292
 293     this.rpcCallerFactory = connection.getRpcRetryingCallerFactory();
 294     this.rpcControllerFactory = connection.getRpcControllerFactory();
 295
 296     this.ng = this.connection.getNonceGenerator();
 297   }
 298
 299   @Override
 300   public void abort(String why, Throwable e) {
 301     // Currently does nothing but throw the passed message and exception
 302     this.aborted = true;
 303     throw new RuntimeException(why, e);
 304   }
 305
 306   @Override
 307   public boolean isAborted() {
 308     return this.aborted;
 309   }
 310
 311   @Override
 312   public boolean abortProcedure(final long procId, final boolean mayInterruptIfRunning)
 313   throws IOException {
 314     return get(abortProcedureAsync(procId, mayInterruptIfRunning), this.syncWaitTimeout,
 315       TimeUnit.MILLISECONDS);
 316   }
 317
 318   @Override
 319   public Future<Boolean> abortProcedureAsync(final long procId, final boolean mayInterruptIfRunning)
 320       throws IOException {
 321     Boolean abortProcResponse =
 322         executeCallable(new MasterCallable<AbortProcedureResponse>(getConnection(),
 323             getRpcControllerFactory()) {
 324       @Override
 325       protected AbortProcedureResponse rpcCall() throws Exception {
 326         AbortProcedureRequest abortProcRequest =
 327             AbortProcedureRequest.newBuilder().setProcId(procId).build();
 328         return master.abortProcedure(getRpcController(), abortProcRequest);
 329       }
 330     }).getIsProcedureAborted();
 331     return new AbortProcedureFuture(this, procId, abortProcResponse);
 332   }
 333
 334   @Override
 335   public List<TableDescriptor> listTableDescriptors() throws IOException {
 336     return listTableDescriptors((Pattern)null, false);
 337   }
 338
 339   @Override
 340   public List<TableDescriptor> listTableDescriptors(Pattern pattern, boolean includeSysTables)
 341       throws IOException {
 342     return executeCallable(new MasterCallable<List<TableDescriptor>>(getConnection(),
 343         getRpcControllerFactory()) {
 344       @Override
 345       protected List<TableDescriptor> rpcCall() throws Exception {
 346         GetTableDescriptorsRequest req =
 347             RequestConverter.buildGetTableDescriptorsRequest(pattern, includeSysTables);
 348         return ProtobufUtil.toTableDescriptorList(master.getTableDescriptors(getRpcController(),
 349             req));
 350       }
 351     });
 352   }
 353
 354   @Override
 355   public TableDescriptor getDescriptor(TableName tableName)
 356       throws TableNotFoundException, IOException {
 357     return getTableDescriptor(tableName, getConnection(), rpcCallerFactory, rpcControllerFactory,
 358        operationTimeout, rpcTimeout);
 359   }
 360
 361   @Override
 362   public Future<Void> modifyTableAsync(TableDescriptor td) throws IOException {
 363     ModifyTableResponse response = executeCallable(
 364       new MasterCallable<ModifyTableResponse>(getConnection(), getRpcControllerFactory()) {
 365         Long nonceGroup = ng.getNonceGroup();
 366         Long nonce = ng.newNonce();
 367         @Override
 368         protected ModifyTableResponse rpcCall() throws Exception {
 369           setPriority(td.getTableName());
 370           ModifyTableRequest request = RequestConverter.buildModifyTableRequest(
 371             td.getTableName(), td, nonceGroup, nonce);
 372           return master.modifyTable(getRpcController(), request);
 373         }
 374       });
 375     return new ModifyTableFuture(this, td.getTableName(), response);
 376   }
 377
 378   @Override
 379   public List<TableDescriptor> listTableDescriptorsByNamespace(byte[] name) throws IOException {
 380     return executeCallable(new MasterCallable<List<TableDescriptor>>(getConnection(),
 381         getRpcControllerFactory()) {
 382       @Override
 383       protected List<TableDescriptor> rpcCall() throws Exception {
 384         return master.listTableDescriptorsByNamespace(getRpcController(),
 385                 ListTableDescriptorsByNamespaceRequest.newBuilder()
 386                   .setNamespaceName(Bytes.toString(name)).build())
 387                 .getTableSchemaList()
 388                 .stream()
 389                 .map(ProtobufUtil::toTableDescriptor)
 390                 .collect(Collectors.toList());
 391       }
 392     });
 393   }
 394
 395   @Override
 396   public List<TableDescriptor> listTableDescriptors(List<TableName> tableNames) throws IOException {
 397     return executeCallable(new MasterCallable<List<TableDescriptor>>(getConnection(),
 398         getRpcControllerFactory()) {
 399       @Override
 400       protected List<TableDescriptor> rpcCall() throws Exception {
 401         GetTableDescriptorsRequest req =
 402             RequestConverter.buildGetTableDescriptorsRequest(tableNames);
 403           return ProtobufUtil.toTableDescriptorList(master.getTableDescriptors(getRpcController(),
 404               req));
 405       }
 406     });
 407   }
 408
 409   @Override
 410   public List<RegionInfo> getRegions(final ServerName sn) throws IOException {
 411     AdminService.BlockingInterface admin = this.connection.getAdmin(sn);
 412     // TODO: There is no timeout on this controller. Set one!
 413     HBaseRpcController controller = rpcControllerFactory.newController();
 414     return ProtobufUtil.getOnlineRegions(controller, admin);
 415   }
 416
 417   @Override
 418   public List<RegionInfo> getRegions(TableName tableName) throws IOException {
 419     if (TableName.isMetaTableName(tableName)) {
 420       return Arrays.asList(RegionInfoBuilder.FIRST_META_REGIONINFO);
 421     } else {
 422       return MetaTableAccessor.getTableRegions(connection, tableName, true);
 423     }
 424   }
 425
 426   private static class AbortProcedureFuture extends ProcedureFuture<Boolean> {
 427     private boolean isAbortInProgress;
 428
 429     public AbortProcedureFuture(
 430         final HBaseAdmin admin,
 431         final Long procId,
 432         final Boolean abortProcResponse) {
 433       super(admin, procId);
 434       this.isAbortInProgress = abortProcResponse;
 435     }
 436
 437     @Override
 438     public Boolean get(long timeout, TimeUnit unit)
 439         throws InterruptedException, ExecutionException, TimeoutException {
 440       if (!this.isAbortInProgress) {
 441         return false;
 442       }
 443       super.get(timeout, unit);
 444       return true;
 445     }
 446   }
 447
 448   /** @return Connection used by this object. */
 449   @Override
 450   public ConnectionImplementation getConnection() {
 451     return connection;
 452   }
 453
 454   @Override
 455   public boolean tableExists(final TableName tableName) throws IOException {
 456     return executeCallable(new RpcRetryingCallable<Boolean>() {
 457       @Override
 458       protected Boolean rpcCall(int callTimeout) throws Exception {
 459         return MetaTableAccessor.tableExists(connection, tableName);
 460       }
 461     });
 462   }
 463
 464   @Override
 465   public TableName[] listTableNames() throws IOException {
 466     return listTableNames((Pattern)null, false);
 467   }
 468
 469   @Override
 470   public TableName[] listTableNames(Pattern pattern) throws IOException {
 471     return listTableNames(pattern, false);
 472   }
 473
 474   @Override
 475   public TableName[] listTableNames(final Pattern pattern, final boolean includeSysTables)
 476       throws IOException {
 477     return executeCallable(new MasterCallable<TableName[]>(getConnection(),
 478         getRpcControllerFactory()) {
 479       @Override
 480       protected TableName[] rpcCall() throws Exception {
 481         GetTableNamesRequest req =
 482             RequestConverter.buildGetTableNamesRequest(pattern, includeSysTables);
 483         return ProtobufUtil.getTableNameArray(master.getTableNames(getRpcController(), req)
 484             .getTableNamesList());
 485       }
 486     });
 487   }
 488
 489   static TableDescriptor getTableDescriptor(final TableName tableName,
 490       ConnectionImplementation connection, RpcRetryingCallerFactory rpcCallerFactory,
 491       final RpcControllerFactory rpcControllerFactory, int operationTimeout, int rpcTimeout)
 492       throws IOException {
 493     if (tableName == null) return null;
 494     TableDescriptor td =
 495       executeCallable(new MasterCallable<TableDescriptor>(connection, rpcControllerFactory) {
 496         @Override
 497         protected TableDescriptor rpcCall() throws Exception {
 498           GetTableDescriptorsRequest req =
 499             RequestConverter.buildGetTableDescriptorsRequest(tableName);
 500           GetTableDescriptorsResponse htds = master.getTableDescriptors(getRpcController(), req);
 501           if (!htds.getTableSchemaList().isEmpty()) {
 502             return ProtobufUtil.toTableDescriptor(htds.getTableSchemaList().get(0));
 503           }
 504           return null;
 505         }
 506       }, rpcCallerFactory, operationTimeout, rpcTimeout);
 507     if (td != null) {
 508       return td;
 509     }
 510     throw new TableNotFoundException(tableName.getNameAsString());
 511   }
 512
 513   private long getPauseTime(int tries) {
 514     int triesCount = tries;
 515     if (triesCount >= HConstants.RETRY_BACKOFF.length) {
 516       triesCount = HConstants.RETRY_BACKOFF.length - 1;
 517     }
 518     return this.pause * HConstants.RETRY_BACKOFF[triesCount];
 519   }
 520
 521   @Override
 522   public void createTable(TableDescriptor desc, byte[] startKey, byte[] endKey, int numRegions)
 523       throws IOException {
 524     if (numRegions < 3) {
 525       throw new IllegalArgumentException("Must create at least three regions");
 526     } else if (Bytes.compareTo(startKey, endKey) >= 0) {
 527       throw new IllegalArgumentException("Start key must be smaller than end key");
 528     }
 529     if (numRegions == 3) {
 530       createTable(desc, new byte[][] { startKey, endKey });
 531       return;
 532     }
 533     byte[][] splitKeys = Bytes.split(startKey, endKey, numRegions - 3);
 534     if (splitKeys == null || splitKeys.length != numRegions - 1) {
 535       throw new IllegalArgumentException("Unable to split key range into enough regions");
 536     }
 537     createTable(desc, splitKeys);
 538   }
 539
 540   @Override
 541   public Future<Void> createTableAsync(final TableDescriptor desc, final byte[][] splitKeys)
 542       throws IOException {
 543     if (desc.getTableName() == null) {
 544       throw new IllegalArgumentException("TableName cannot be null");
 545     }
 546     if (splitKeys != null && splitKeys.length > 0) {
 547       Arrays.sort(splitKeys, Bytes.BYTES_COMPARATOR);
 548       // Verify there are no duplicate split keys
 549       byte[] lastKey = null;
 550       for (byte[] splitKey : splitKeys) {
 551         if (Bytes.compareTo(splitKey, HConstants.EMPTY_BYTE_ARRAY) == 0) {
 552           throw new IllegalArgumentException(
 553               "Empty split key must not be passed in the split keys.");
 554         }
 555         if (lastKey != null && Bytes.equals(splitKey, lastKey)) {
 556           throw new IllegalArgumentException("All split keys must be unique, " +
 557             "found duplicate: " + Bytes.toStringBinary(splitKey) +
 558             ", " + Bytes.toStringBinary(lastKey));
 559         }
 560         lastKey = splitKey;
 561       }
 562     }
 563
 564     CreateTableResponse response = executeCallable(
 565       new MasterCallable<CreateTableResponse>(getConnection(), getRpcControllerFactory()) {
 566         Long nonceGroup = ng.getNonceGroup();
 567         Long nonce = ng.newNonce();
 568         @Override
 569         protected CreateTableResponse rpcCall() throws Exception {
 570           setPriority(desc.getTableName());
 571           CreateTableRequest request = RequestConverter.buildCreateTableRequest(
 572             desc, splitKeys, nonceGroup, nonce);
 573           return master.createTable(getRpcController(), request);
 574         }
 575       });
 576     return new CreateTableFuture(this, desc, splitKeys, response);
 577   }
 578
 579   private static class CreateTableFuture extends TableFuture<Void> {
 580     private final TableDescriptor desc;
 581     private final byte[][] splitKeys;
 582
 583     public CreateTableFuture(final HBaseAdmin admin, final TableDescriptor desc,
 584         final byte[][] splitKeys, final CreateTableResponse response) {
 585       super(admin, desc.getTableName(),
 586               (response != null && response.hasProcId()) ? response.getProcId() : null);
 587       this.splitKeys = splitKeys;
 588       this.desc = desc;
 589     }
 590
 591     @Override
 592     protected TableDescriptor getDescriptor() {
 593       return desc;
 594     }
 595
 596     @Override
 597     public String getOperationType() {
 598       return "CREATE";
 599     }
 600
 601     @Override
 602     protected Void waitOperationResult(final long deadlineTs) throws IOException, TimeoutException {
 603       waitForTableEnabled(deadlineTs);
 604       waitForAllRegionsOnline(deadlineTs, splitKeys);
 605       return null;
 606     }
 607   }
 608
 609   @Override
 610   public Future<Void> deleteTableAsync(final TableName tableName) throws IOException {
 611     DeleteTableResponse response = executeCallable(
 612       new MasterCallable<DeleteTableResponse>(getConnection(), getRpcControllerFactory()) {
 613         Long nonceGroup = ng.getNonceGroup();
 614         Long nonce = ng.newNonce();
 615         @Override
 616         protected DeleteTableResponse rpcCall() throws Exception {
 617           setPriority(tableName);
 618           DeleteTableRequest req =
 619               RequestConverter.buildDeleteTableRequest(tableName, nonceGroup,nonce);
 620           return master.deleteTable(getRpcController(), req);
 621         }
 622       });
 623     return new DeleteTableFuture(this, tableName, response);
 624   }
 625
 626   private static class DeleteTableFuture extends TableFuture<Void> {
 627     public DeleteTableFuture(final HBaseAdmin admin, final TableName tableName,
 628         final DeleteTableResponse response) {
 629       super(admin, tableName,
 630               (response != null && response.hasProcId()) ? response.getProcId() : null);
 631     }
 632
 633     @Override
 634     public String getOperationType() {
 635       return "DELETE";
 636     }
 637
 638     @Override
 639     protected Void waitOperationResult(final long deadlineTs)
 640         throws IOException, TimeoutException {
 641       waitTableNotFound(deadlineTs);
 642       return null;
 643     }
 644
 645     @Override
 646     protected Void postOperationResult(final Void result, final long deadlineTs)
 647         throws IOException, TimeoutException {
 648       // Delete cached information to prevent clients from using old locations
 649       try (RegionLocator locator = getAdmin().getConnection().getRegionLocator(getTableName())) {
 650         locator.clearRegionLocationCache();
 651       }
 652       return super.postOperationResult(result, deadlineTs);
 653     }
 654   }
 655
 656   @Override
 657   public Future<Void> truncateTableAsync(final TableName tableName, final boolean preserveSplits)
 658       throws IOException {
 659     TruncateTableResponse response =
 660         executeCallable(new MasterCallable<TruncateTableResponse>(getConnection(),
 661             getRpcControllerFactory()) {
 662           Long nonceGroup = ng.getNonceGroup();
 663           Long nonce = ng.newNonce();
 664           @Override
 665           protected TruncateTableResponse rpcCall() throws Exception {
 666             setPriority(tableName);
 667             LOG.info("Started truncating " + tableName);
 668             TruncateTableRequest req = RequestConverter.buildTruncateTableRequest(
 669               tableName, preserveSplits, nonceGroup, nonce);
 670             return master.truncateTable(getRpcController(), req);
 671           }
 672         });
 673     return new TruncateTableFuture(this, tableName, preserveSplits, response);
 674   }
 675
 676   private static class TruncateTableFuture extends TableFuture<Void> {
 677     private final boolean preserveSplits;
 678
 679     public TruncateTableFuture(final HBaseAdmin admin, final TableName tableName,
 680         final boolean preserveSplits, final TruncateTableResponse response) {
 681       super(admin, tableName,
 682              (response != null && response.hasProcId()) ? response.getProcId() : null);
 683       this.preserveSplits = preserveSplits;
 684     }
 685
 686     @Override
 687     public String getOperationType() {
 688       return "TRUNCATE";
 689     }
 690
 691     @Override
 692     protected Void waitOperationResult(final long deadlineTs) throws IOException, TimeoutException {
 693       waitForTableEnabled(deadlineTs);
 694       // once the table is enabled, we know the operation is done. so we can fetch the splitKeys
 695       byte[][] splitKeys = preserveSplits ? getAdmin().getTableSplits(getTableName()) : null;
 696       waitForAllRegionsOnline(deadlineTs, splitKeys);
 697       return null;
 698     }
 699   }
 700
 701   private byte[][] getTableSplits(final TableName tableName) throws IOException {
 702     byte[][] splits = null;
 703     try (RegionLocator locator = getConnection().getRegionLocator(tableName)) {
 704       byte[][] startKeys = locator.getStartKeys();
 705       if (startKeys.length == 1) {
 706         return splits;
 707       }
 708       splits = new byte[startKeys.length - 1][];
 709       for (int i = 1; i < startKeys.length; i++) {
 710         splits[i - 1] = startKeys[i];
 711       }
 712     }
 713     return splits;
 714   }
 715
 716   @Override
 717   public Future<Void> enableTableAsync(final TableName tableName) throws IOException {
 718     TableName.isLegalFullyQualifiedTableName(tableName.getName());
 719     EnableTableResponse response = executeCallable(
 720       new MasterCallable<EnableTableResponse>(getConnection(), getRpcControllerFactory()) {
 721         Long nonceGroup = ng.getNonceGroup();
 722         Long nonce = ng.newNonce();
 723         @Override
 724         protected EnableTableResponse rpcCall() throws Exception {
 725           setPriority(tableName);
 726           LOG.info("Started enable of " + tableName);
 727           EnableTableRequest req =
 728               RequestConverter.buildEnableTableRequest(tableName, nonceGroup, nonce);
 729           return master.enableTable(getRpcController(),req);
 730         }
 731       });
 732     return new EnableTableFuture(this, tableName, response);
 733   }
 734
 735   private static class EnableTableFuture extends TableFuture<Void> {
 736     public EnableTableFuture(final HBaseAdmin admin, final TableName tableName,
 737         final EnableTableResponse response) {
 738       super(admin, tableName,
 739               (response != null && response.hasProcId()) ? response.getProcId() : null);
 740     }
 741
 742     @Override
 743     public String getOperationType() {
 744       return "ENABLE";
 745     }
 746
 747     @Override
 748     protected Void waitOperationResult(final long deadlineTs) throws IOException, TimeoutException {
 749       waitForTableEnabled(deadlineTs);
 750       return null;
 751     }
 752   }
 753
 754   @Override
 755   public Future<Void> disableTableAsync(final TableName tableName) throws IOException {
 756     TableName.isLegalFullyQualifiedTableName(tableName.getName());
 757     DisableTableResponse response = executeCallable(
 758       new MasterCallable<DisableTableResponse>(getConnection(), getRpcControllerFactory()) {
 759         Long nonceGroup = ng.getNonceGroup();
 760         Long nonce = ng.newNonce();
 761         @Override
 762         protected DisableTableResponse rpcCall() throws Exception {
 763           setPriority(tableName);
 764           LOG.info("Started disable of " + tableName);
 765           DisableTableRequest req =
 766               RequestConverter.buildDisableTableRequest(
 767                 tableName, nonceGroup, nonce);
 768           return master.disableTable(getRpcController(), req);
 769         }
 770       });
 771     return new DisableTableFuture(this, tableName, response);
 772   }
 773
 774   private static class DisableTableFuture extends TableFuture<Void> {
 775     public DisableTableFuture(final HBaseAdmin admin, final TableName tableName,
 776         final DisableTableResponse response) {
 777       super(admin, tableName,
 778               (response != null && response.hasProcId()) ? response.getProcId() : null);
 779     }
 780
 781     @Override
 782     public String getOperationType() {
 783       return "DISABLE";
 784     }
 785
 786     @Override
 787     protected Void waitOperationResult(long deadlineTs) throws IOException, TimeoutException {
 788       waitForTableDisabled(deadlineTs);
 789       return null;
 790     }
 791   }
 792
 793   @Override
 794   public boolean isTableEnabled(final TableName tableName) throws IOException {
 795     checkTableExists(tableName);
 796     return executeCallable(new RpcRetryingCallable<Boolean>() {
 797       @Override
 798       protected Boolean rpcCall(int callTimeout) throws Exception {
 799         TableState tableState = MetaTableAccessor.getTableState(getConnection(), tableName);
 800         if (tableState == null) {
 801           throw new TableNotFoundException(tableName);
 802         }
 803         return tableState.inStates(TableState.State.ENABLED);
 804       }
 805     });
 806   }
 807
 808   @Override
 809   public boolean isTableDisabled(TableName tableName) throws IOException {
 810     checkTableExists(tableName);
 811     return connection.isTableDisabled(tableName);
 812   }
 813
 814   @Override
 815   public boolean isTableAvailable(TableName tableName) throws IOException {
 816     return connection.isTableAvailable(tableName, null);
 817   }
 818
 819   @Override
 820   public Future<Void> addColumnFamilyAsync(final TableName tableName,
 821       final ColumnFamilyDescriptor columnFamily) throws IOException {
 822     AddColumnResponse response =
 823         executeCallable(new MasterCallable<AddColumnResponse>(getConnection(),
 824             getRpcControllerFactory()) {
 825           Long nonceGroup = ng.getNonceGroup();
 826           Long nonce = ng.newNonce();
 827           @Override
 828           protected AddColumnResponse rpcCall() throws Exception {
 829             setPriority(tableName);
 830             AddColumnRequest req =
 831                 RequestConverter.buildAddColumnRequest(tableName, columnFamily, nonceGroup, nonce);
 832             return master.addColumn(getRpcController(), req);
 833           }
 834         });
 835     return new AddColumnFamilyFuture(this, tableName, response);
 836   }
 837
 838   private static class AddColumnFamilyFuture extends ModifyTableFuture {
 839     public AddColumnFamilyFuture(final HBaseAdmin admin, final TableName tableName,
 840         final AddColumnResponse response) {
 841       super(admin, tableName, (response != null && response.hasProcId()) ? response.getProcId()
 842           : null);
 843     }
 844
 845     @Override
 846     public String getOperationType() {
 847       return "ADD_COLUMN_FAMILY";
 848     }
 849   }
 850
 851   @Override
 852   public Future<Void> deleteColumnFamilyAsync(final TableName tableName, final byte[] columnFamily)
 853       throws IOException {
 854     DeleteColumnResponse response =
 855         executeCallable(new MasterCallable<DeleteColumnResponse>(getConnection(),
 856             getRpcControllerFactory()) {
 857           Long nonceGroup = ng.getNonceGroup();
 858           Long nonce = ng.newNonce();
 859           @Override
 860           protected DeleteColumnResponse rpcCall() throws Exception {
 861             setPriority(tableName);
 862             DeleteColumnRequest req =
 863                 RequestConverter.buildDeleteColumnRequest(tableName, columnFamily,
 864                   nonceGroup, nonce);
 865             return master.deleteColumn(getRpcController(), req);
 866           }
 867         });
 868     return new DeleteColumnFamilyFuture(this, tableName, response);
 869   }
 870
 871   private static class DeleteColumnFamilyFuture extends ModifyTableFuture {
 872     public DeleteColumnFamilyFuture(final HBaseAdmin admin, final TableName tableName,
 873         final DeleteColumnResponse response) {
 874       super(admin, tableName, (response != null && response.hasProcId()) ? response.getProcId()
 875           : null);
 876     }
 877
 878     @Override
 879     public String getOperationType() {
 880       return "DELETE_COLUMN_FAMILY";
 881     }
 882   }
 883
 884   @Override
 885   public Future<Void> modifyColumnFamilyAsync(final TableName tableName,
 886       final ColumnFamilyDescriptor columnFamily) throws IOException {
 887     ModifyColumnResponse response =
 888         executeCallable(new MasterCallable<ModifyColumnResponse>(getConnection(),
 889             getRpcControllerFactory()) {
 890           Long nonceGroup = ng.getNonceGroup();
 891           Long nonce = ng.newNonce();
 892           @Override
 893           protected ModifyColumnResponse rpcCall() throws Exception {
 894             setPriority(tableName);
 895             ModifyColumnRequest req =
 896                 RequestConverter.buildModifyColumnRequest(tableName, columnFamily,
 897                   nonceGroup, nonce);
 898             return master.modifyColumn(getRpcController(), req);
 899           }
 900         });
 901     return new ModifyColumnFamilyFuture(this, tableName, response);
 902   }
 903
 904   private static class ModifyColumnFamilyFuture extends ModifyTableFuture {
 905     public ModifyColumnFamilyFuture(final HBaseAdmin admin, final TableName tableName,
 906         final ModifyColumnResponse response) {
 907       super(admin, tableName, (response != null && response.hasProcId()) ? response.getProcId()
 908           : null);
 909     }
 910
 911     @Override
 912     public String getOperationType() {
 913       return "MODIFY_COLUMN_FAMILY";
 914     }
 915   }
 916
 917   @Override
 918   public void flush(final TableName tableName) throws IOException {
 919     checkTableExists(tableName);
 920     if (isTableDisabled(tableName)) {
 921       LOG.info("Table is disabled: " + tableName.getNameAsString());
 922       return;
 923     }
 924     execProcedure("flush-table-proc", tableName.getNameAsString(), new HashMap<>());
 925   }
 926
 927   @Override
 928   public void flushRegion(final byte[] regionName) throws IOException {
 929     Pair<RegionInfo, ServerName> regionServerPair = getRegion(regionName);
 930     if (regionServerPair == null) {
 931       throw new IllegalArgumentException("Unknown regionname: " + Bytes.toStringBinary(regionName));
 932     }
 933     if (regionServerPair.getSecond() == null) {
 934       throw new NoServerForRegionException(Bytes.toStringBinary(regionName));
 935     }
 936     final RegionInfo regionInfo = regionServerPair.getFirst();
 937     ServerName serverName = regionServerPair.getSecond();
 938     flush(this.connection.getAdmin(serverName), regionInfo);
 939   }
 940
 941   private void flush(AdminService.BlockingInterface admin, final RegionInfo info)
 942     throws IOException {
 943     ProtobufUtil.call(() -> {
 944       // TODO: There is no timeout on this controller. Set one!
 945       HBaseRpcController controller = rpcControllerFactory.newController();
 946       FlushRegionRequest request =
 947         RequestConverter.buildFlushRegionRequest(info.getRegionName());
 948       admin.flushRegion(controller, request);
 949       return null;
 950     });
 951   }
 952
 953   @Override
 954   public void flushRegionServer(ServerName serverName) throws IOException {
 955     for (RegionInfo region : getRegions(serverName)) {
 956       flush(this.connection.getAdmin(serverName), region);
 957     }
 958   }
 959
 960   /**
 961    * {@inheritDoc}
 962    */
 963   @Override
 964   public void compact(final TableName tableName)
 965     throws IOException {
 966     compact(tableName, null, false, CompactType.NORMAL);
 967   }
 968
 969   @Override
 970   public void compactRegion(final byte[] regionName)
 971     throws IOException {
 972     compactRegion(regionName, null, false);
 973   }
 974
 975   /**
 976    * {@inheritDoc}
 977    */
 978   @Override
 979   public void compact(final TableName tableName, final byte[] columnFamily)
 980     throws IOException {
 981     compact(tableName, columnFamily, false, CompactType.NORMAL);
 982   }
 983
 984   /**
 985    * {@inheritDoc}
 986    */
 987   @Override
 988   public void compactRegion(final byte[] regionName, final byte[] columnFamily)
 989     throws IOException {
 990     compactRegion(regionName, columnFamily, false);
 991   }
 992
 993   @Override
 994   public Map<ServerName, Boolean> compactionSwitch(boolean switchState, List<String>
 995       serverNamesList) throws IOException {
 996     List<ServerName> serverList = new ArrayList<>();
 997     if (serverNamesList.isEmpty()) {
 998       ClusterMetrics status = getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS));
 999       serverList.addAll(status.getLiveServerMetrics().keySet());
1000     } else {
1001       for (String regionServerName: serverNamesList) {
1002         ServerName serverName = null;
1003         try {
1004           serverName = ServerName.valueOf(regionServerName);
1005         } catch (Exception e) {
1006           throw new IllegalArgumentException(String.format("Invalid ServerName format: %s",
1007               regionServerName));
1008         }
1009         if (serverName == null) {
1010           throw new IllegalArgumentException(String.format("Null ServerName: %s",
1011               regionServerName));
1012         }
1013         serverList.add(serverName);
1014       }
1015     }
1016     Map<ServerName, Boolean> res = new HashMap<>(serverList.size());
1017     for (ServerName serverName: serverList) {
1018       boolean prev_state = switchCompact(this.connection.getAdmin(serverName), switchState);
1019       res.put(serverName, prev_state);
1020     }
1021     return res;
1022   }
1023
1024   private Boolean switchCompact(AdminService.BlockingInterface admin, boolean onOrOff)
1025       throws IOException {
1026     return executeCallable(new RpcRetryingCallable<Boolean>() {
1027       @Override protected Boolean rpcCall(int callTimeout) throws Exception {
1028         HBaseRpcController controller = rpcControllerFactory.newController();
1029         CompactionSwitchRequest request =
1030             CompactionSwitchRequest.newBuilder().setEnabled(onOrOff).build();
1031         CompactionSwitchResponse compactionSwitchResponse =
1032             admin.compactionSwitch(controller, request);
1033         return compactionSwitchResponse.getPrevState();
1034       }
1035     });
1036   }
1037
1038   @Override
1039   public void compactRegionServer(final ServerName serverName) throws IOException {
1040     for (RegionInfo region : getRegions(serverName)) {
1041       compact(this.connection.getAdmin(serverName), region, false, null);
1042     }
1043   }
1044
1045   @Override
1046   public void majorCompactRegionServer(final ServerName serverName) throws IOException {
1047     for (RegionInfo region : getRegions(serverName)) {
1048       compact(this.connection.getAdmin(serverName), region, true, null);
1049     }
1050   }
1051
1052   @Override
1053   public void majorCompact(final TableName tableName)
1054   throws IOException {
1055     compact(tableName, null, true, CompactType.NORMAL);
1056   }
1057
1058   @Override
1059   public void majorCompactRegion(final byte[] regionName)
1060   throws IOException {
1061     compactRegion(regionName, null, true);
1062   }
1063
1064   /**
1065    * {@inheritDoc}
1066    */
1067   @Override
1068   public void majorCompact(final TableName tableName, final byte[] columnFamily)
1069   throws IOException {
1070     compact(tableName, columnFamily, true, CompactType.NORMAL);
1071   }
1072
1073   @Override
1074   public void majorCompactRegion(final byte[] regionName, final byte[] columnFamily)
1075   throws IOException {
1076     compactRegion(regionName, columnFamily, true);
1077   }
1078
1079   /**
1080    * Compact a table.
1081    * Asynchronous operation.
1082    *
1083    * @param tableName table or region to compact
1084    * @param columnFamily column family within a table or region
1085    * @param major True if we are to do a major compaction.
1086    * @param compactType {@link org.apache.hadoop.hbase.client.CompactType}
1087    * @throws IOException if a remote or network exception occurs
1088    */
1089   private void compact(final TableName tableName, final byte[] columnFamily,final boolean major,
1090                        CompactType compactType) throws IOException {
1091     switch (compactType) {
1092       case MOB:
1093         compact(this.connection.getAdminForMaster(), RegionInfo.createMobRegionInfo(tableName),
1094             major, columnFamily);
1095         break;
1096       case NORMAL:
1097         checkTableExists(tableName);
1098         for (HRegionLocation loc :connection.locateRegions(tableName, false, false)) {
1099           ServerName sn = loc.getServerName();
1100           if (sn == null) {
1101             continue;
1102           }
1103           try {
1104             compact(this.connection.getAdmin(sn), loc.getRegion(), major, columnFamily);
1105           } catch (NotServingRegionException e) {
1106             if (LOG.isDebugEnabled()) {
1107               LOG.debug("Trying to" + (major ? " major" : "") + " compact " + loc.getRegion() +
1108                   ": " + StringUtils.stringifyException(e));
1109             }
1110           }
1111         }
1112         break;
1113       default:
1114         throw new IllegalArgumentException("Unknown compactType: " + compactType);
1115     }
1116   }
1117
1118   /**
1119    * Compact an individual region.
1120    * Asynchronous operation.
1121    *
1122    * @param regionName region to compact
1123    * @param columnFamily column family within a table or region
1124    * @param major True if we are to do a major compaction.
1125    * @throws IOException if a remote or network exception occurs
1126    * @throws InterruptedException
1127    */
1128   private void compactRegion(final byte[] regionName, final byte[] columnFamily,
1129       final boolean major) throws IOException {
1130     Pair<RegionInfo, ServerName> regionServerPair = getRegion(regionName);
1131     if (regionServerPair == null) {
1132       throw new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName));
1133     }
1134     if (regionServerPair.getSecond() == null) {
1135       throw new NoServerForRegionException(Bytes.toStringBinary(regionName));
1136     }
1137     compact(this.connection.getAdmin(regionServerPair.getSecond()), regionServerPair.getFirst(),
1138       major, columnFamily);
1139   }
1140
1141   private void compact(AdminService.BlockingInterface admin, RegionInfo hri, boolean major,
1142       byte[] family) throws IOException {
1143     Callable<Void> callable = new Callable<Void>() {
1144       @Override
1145       public Void call() throws Exception {
1146         // TODO: There is no timeout on this controller. Set one!
1147         HBaseRpcController controller = rpcControllerFactory.newController();
1148         CompactRegionRequest request =
1149             RequestConverter.buildCompactRegionRequest(hri.getRegionName(), major, family);
1150         admin.compactRegion(controller, request);
1151         return null;
1152       }
1153     };
1154     ProtobufUtil.call(callable);
1155   }
1156
1157   @Override
1158   public void move(byte[] encodedRegionName) throws IOException {
1159     move(encodedRegionName, null);
1160   }
1161
1162   public void move(byte[] encodedRegionName, ServerName destServerName) throws IOException {
1163     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
1164       @Override
1165       protected Void rpcCall() throws Exception {
1166         setPriority(encodedRegionName);
1167         MoveRegionRequest request =
1168           RequestConverter.buildMoveRegionRequest(encodedRegionName, destServerName);
1169         master.moveRegion(getRpcController(), request);
1170         return null;
1171       }
1172     });
1173   }
1174
1175   @Override
1176   public void assign(final byte [] regionName) throws MasterNotRunningException,
1177       ZooKeeperConnectionException, IOException {
1178     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
1179       @Override
1180       protected Void rpcCall() throws Exception {
1181         setPriority(regionName);
1182         AssignRegionRequest request =
1183             RequestConverter.buildAssignRegionRequest(getRegionName(regionName));
1184         master.assignRegion(getRpcController(), request);
1185         return null;
1186       }
1187     });
1188   }
1189
1190   @Override
1191   public void unassign(final byte [] regionName, final boolean force) throws IOException {
1192     final byte[] toBeUnassigned = getRegionName(regionName);
1193     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
1194       @Override
1195       protected Void rpcCall() throws Exception {
1196         setPriority(regionName);
1197         UnassignRegionRequest request =
1198             RequestConverter.buildUnassignRegionRequest(toBeUnassigned, force);
1199         master.unassignRegion(getRpcController(), request);
1200         return null;
1201       }
1202     });
1203   }
1204
1205   @Override
1206   public void offline(final byte [] regionName)
1207   throws IOException {
1208     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
1209       @Override
1210       protected Void rpcCall() throws Exception {
1211         setPriority(regionName);
1212         master.offlineRegion(getRpcController(),
1213             RequestConverter.buildOfflineRegionRequest(regionName));
1214         return null;
1215       }
1216     });
1217   }
1218
1219   @Override
1220   public boolean balancerSwitch(final boolean on, final boolean synchronous)
1221   throws IOException {
1222     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1223       @Override
1224       protected Boolean rpcCall() throws Exception {
1225         SetBalancerRunningRequest req =
1226             RequestConverter.buildSetBalancerRunningRequest(on, synchronous);
1227         return master.setBalancerRunning(getRpcController(), req).getPrevBalanceValue();
1228       }
1229     });
1230   }
1231
1232   @Override
1233   public boolean balance() throws IOException {
1234     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1235       @Override
1236       protected Boolean rpcCall() throws Exception {
1237         return master.balance(getRpcController(),
1238             RequestConverter.buildBalanceRequest(false)).getBalancerRan();
1239       }
1240     });
1241   }
1242
1243   @Override
1244   public boolean balance(final boolean force) throws IOException {
1245     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1246       @Override
1247       protected Boolean rpcCall() throws Exception {
1248         return master.balance(getRpcController(),
1249             RequestConverter.buildBalanceRequest(force)).getBalancerRan();
1250       }
1251     });
1252   }
1253
1254   @Override
1255   public boolean isBalancerEnabled() throws IOException {
1256     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1257       @Override
1258       protected Boolean rpcCall() throws Exception {
1259         return master.isBalancerEnabled(getRpcController(),
1260           RequestConverter.buildIsBalancerEnabledRequest()).getEnabled();
1261       }
1262     });
1263   }
1264
1265   /**
1266    * {@inheritDoc}
1267    */
1268   @Override
1269   public CacheEvictionStats clearBlockCache(final TableName tableName) throws IOException {
1270     checkTableExists(tableName);
1271     CacheEvictionStatsBuilder cacheEvictionStats = CacheEvictionStats.builder();
1272     List<Pair<RegionInfo, ServerName>> pairs =
1273       MetaTableAccessor.getTableRegionsAndLocations(connection, tableName);
1274     Map<ServerName, List<RegionInfo>> regionInfoByServerName =
1275         pairs.stream()
1276             .filter(pair -> !(pair.getFirst().isOffline()))
1277             .filter(pair -> pair.getSecond() != null)
1278             .collect(Collectors.groupingBy(pair -> pair.getSecond(),
1279                 Collectors.mapping(pair -> pair.getFirst(), Collectors.toList())));
1280
1281     for (Map.Entry<ServerName, List<RegionInfo>> entry : regionInfoByServerName.entrySet()) {
1282       CacheEvictionStats stats = clearBlockCache(entry.getKey(), entry.getValue());
1283       cacheEvictionStats = cacheEvictionStats.append(stats);
1284       if (stats.getExceptionCount() > 0) {
1285         for (Map.Entry<byte[], Throwable> exception : stats.getExceptions().entrySet()) {
1286           LOG.debug("Failed to clear block cache for "
1287               + Bytes.toStringBinary(exception.getKey())
1288               + " on " + entry.getKey() + ": ", exception.getValue());
1289         }
1290       }
1291     }
1292     return cacheEvictionStats.build();
1293   }
1294
1295   private CacheEvictionStats clearBlockCache(final ServerName sn, final List<RegionInfo> hris)
1296       throws IOException {
1297     HBaseRpcController controller = rpcControllerFactory.newController();
1298     AdminService.BlockingInterface admin = this.connection.getAdmin(sn);
1299     ClearRegionBlockCacheRequest request =
1300       RequestConverter.buildClearRegionBlockCacheRequest(hris);
1301     ClearRegionBlockCacheResponse response;
1302     try {
1303       response = admin.clearRegionBlockCache(controller, request);
1304       return ProtobufUtil.toCacheEvictionStats(response.getStats());
1305     } catch (ServiceException se) {
1306       throw ProtobufUtil.getRemoteException(se);
1307     }
1308   }
1309
1310   /**
1311    * Invoke region normalizer. Can NOT run for various reasons.  Check logs.
1312    *
1313    * @return True if region normalizer ran, false otherwise.
1314    */
1315   @Override
1316   public boolean normalize() throws IOException {
1317     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1318       @Override
1319       protected Boolean rpcCall() throws Exception {
1320         return master.normalize(getRpcController(),
1321             RequestConverter.buildNormalizeRequest()).getNormalizerRan();
1322       }
1323     });
1324   }
1325
1326   @Override
1327   public boolean isNormalizerEnabled() throws IOException {
1328     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1329       @Override
1330       protected Boolean rpcCall() throws Exception {
1331         return master.isNormalizerEnabled(getRpcController(),
1332           RequestConverter.buildIsNormalizerEnabledRequest()).getEnabled();
1333       }
1334     });
1335   }
1336
1337   @Override
1338   public boolean normalizerSwitch(final boolean on) throws IOException {
1339     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1340       @Override
1341       protected Boolean rpcCall() throws Exception {
1342         SetNormalizerRunningRequest req =
1343           RequestConverter.buildSetNormalizerRunningRequest(on);
1344         return master.setNormalizerRunning(getRpcController(), req).getPrevNormalizerValue();
1345       }
1346     });
1347   }
1348
1349   @Override
1350   public boolean catalogJanitorSwitch(final boolean enable) throws IOException {
1351     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1352       @Override
1353       protected Boolean rpcCall() throws Exception {
1354         return master.enableCatalogJanitor(getRpcController(),
1355           RequestConverter.buildEnableCatalogJanitorRequest(enable)).getPrevValue();
1356       }
1357     });
1358   }
1359
1360   @Override
1361   public int runCatalogJanitor() throws IOException {
1362     return executeCallable(new MasterCallable<Integer>(getConnection(), getRpcControllerFactory()) {
1363       @Override
1364       protected Integer rpcCall() throws Exception {
1365         return master.runCatalogScan(getRpcController(),
1366           RequestConverter.buildCatalogScanRequest()).getScanResult();
1367       }
1368     });
1369   }
1370
1371   @Override
1372   public boolean isCatalogJanitorEnabled() throws IOException {
1373     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1374       @Override
1375       protected Boolean rpcCall() throws Exception {
1376         return master.isCatalogJanitorEnabled(getRpcController(),
1377           RequestConverter.buildIsCatalogJanitorEnabledRequest()).getValue();
1378       }
1379     });
1380   }
1381
1382   @Override
1383   public boolean cleanerChoreSwitch(final boolean on) throws IOException {
1384     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1385       @Override public Boolean rpcCall() throws Exception {
1386         return master.setCleanerChoreRunning(getRpcController(),
1387             RequestConverter.buildSetCleanerChoreRunningRequest(on)).getPrevValue();
1388       }
1389     });
1390   }
1391
1392   @Override
1393   public boolean runCleanerChore() throws IOException {
1394     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1395       @Override public Boolean rpcCall() throws Exception {
1396         return master.runCleanerChore(getRpcController(),
1397             RequestConverter.buildRunCleanerChoreRequest()).getCleanerChoreRan();
1398       }
1399     });
1400   }
1401
1402   @Override
1403   public boolean isCleanerChoreEnabled() throws IOException {
1404     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
1405       @Override public Boolean rpcCall() throws Exception {
1406         return master.isCleanerChoreEnabled(getRpcController(),
1407             RequestConverter.buildIsCleanerChoreEnabledRequest()).getValue();
1408       }
1409     });
1410   }
1411
1412   /**
1413    * Merge two regions. Synchronous operation.
1414    * Note: It is not feasible to predict the length of merge.
1415    *   Therefore, this is for internal testing only.
1416    * @param nameOfRegionA encoded or full name of region a
1417    * @param nameOfRegionB encoded or full name of region b
1418    * @param forcible true if do a compulsory merge, otherwise we will only merge
1419    *          two adjacent regions
1420    * @throws IOException if a remote or network exception occurs
1421    */
1422   @VisibleForTesting
1423   public void mergeRegionsSync(
1424       final byte[] nameOfRegionA,
1425       final byte[] nameOfRegionB,
1426       final boolean forcible) throws IOException {
1427     get(
1428       mergeRegionsAsync(nameOfRegionA, nameOfRegionB, forcible),
1429       syncWaitTimeout,
1430       TimeUnit.MILLISECONDS);
1431   }
1432
1433   /**
1434    * Merge two regions. Asynchronous operation.
1435    * @param nameofRegionsToMerge encoded or full name of daughter regions
1436    * @param forcible true if do a compulsory merge, otherwise we will only merge
1437    *          adjacent regions
1438    */
1439   @Override
1440   public Future<Void> mergeRegionsAsync(final byte[][] nameofRegionsToMerge, final boolean forcible)
1441       throws IOException {
1442     Preconditions.checkArgument(nameofRegionsToMerge.length >= 2, "Can not merge only %s region",
1443       nameofRegionsToMerge.length);
1444     byte[][] encodedNameofRegionsToMerge = new byte[nameofRegionsToMerge.length][];
1445     for (int i = 0; i < nameofRegionsToMerge.length; i++) {
1446       encodedNameofRegionsToMerge[i] =
1447         RegionInfo.isEncodedRegionName(nameofRegionsToMerge[i]) ? nameofRegionsToMerge[i]
1448           : Bytes.toBytes(RegionInfo.encodeRegionName(nameofRegionsToMerge[i]));
1449     }
1450
1451     TableName tableName = null;
1452     Pair<RegionInfo, ServerName> pair;
1453
1454     for(int i = 0; i < nameofRegionsToMerge.length; i++) {
1455       pair = getRegion(nameofRegionsToMerge[i]);
1456
1457       if (pair != null) {
1458         if (pair.getFirst().getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID) {
1459           throw new IllegalArgumentException ("Can't invoke merge on non-default regions directly");
1460         }
1461         if (tableName == null) {
1462           tableName = pair.getFirst().getTable();
1463         } else  if (!tableName.equals(pair.getFirst().getTable())) {
1464           throw new IllegalArgumentException ("Cannot merge regions from two different tables " +
1465               tableName + " and " + pair.getFirst().getTable());
1466         }
1467       } else {
1468         throw new UnknownRegionException (
1469           "Can't invoke merge on unknown region "
1470           + Bytes.toStringBinary(encodedNameofRegionsToMerge[i]));
1471       }
1472     }
1473
1474     MergeTableRegionsResponse response =
1475         executeCallable(new MasterCallable<MergeTableRegionsResponse>(getConnection(),
1476             getRpcControllerFactory()) {
1477           Long nonceGroup = ng.getNonceGroup();
1478           Long nonce = ng.newNonce();
1479       @Override
1480       protected MergeTableRegionsResponse rpcCall() throws Exception {
1481         MergeTableRegionsRequest request = RequestConverter
1482             .buildMergeTableRegionsRequest(
1483                 encodedNameofRegionsToMerge,
1484                 forcible,
1485                 nonceGroup,
1486                 nonce);
1487         return master.mergeTableRegions(getRpcController(), request);
1488       }
1489     });
1490     return new MergeTableRegionsFuture(this, tableName, response);
1491   }
1492
1493   private static class MergeTableRegionsFuture extends TableFuture<Void> {
1494     public MergeTableRegionsFuture(
1495         final HBaseAdmin admin,
1496         final TableName tableName,
1497         final MergeTableRegionsResponse response) {
1498       super(admin, tableName,
1499           (response != null && response.hasProcId()) ? response.getProcId() : null);
1500     }
1501
1502     public MergeTableRegionsFuture(
1503         final HBaseAdmin admin,
1504         final TableName tableName,
1505         final Long procId) {
1506       super(admin, tableName, procId);
1507     }
1508
1509     @Override
1510     public String getOperationType() {
1511       return "MERGE_REGIONS";
1512     }
1513   }
1514   /**
1515    * Split one region. Synchronous operation.
1516    * Note: It is not feasible to predict the length of split.
1517    *   Therefore, this is for internal testing only.
1518    * @param regionName encoded or full name of region
1519    * @param splitPoint key where region splits
1520    * @throws IOException if a remote or network exception occurs
1521    */
1522   @VisibleForTesting
1523   public void splitRegionSync(byte[] regionName, byte[] splitPoint) throws IOException {
1524     splitRegionSync(regionName, splitPoint, syncWaitTimeout, TimeUnit.MILLISECONDS);
1525   }
1526
1527
1528   /**
1529    * Split one region. Synchronous operation.
1530    * @param regionName region to be split
1531    * @param splitPoint split point
1532    * @param timeout how long to wait on split
1533    * @param units time units
1534    * @throws IOException if a remote or network exception occurs
1535    */
1536   public void splitRegionSync(byte[] regionName, byte[] splitPoint, final long timeout,
1537       final TimeUnit units) throws IOException {
1538     get(splitRegionAsync(regionName, splitPoint), timeout, units);
1539   }
1540
1541   @Override
1542   public Future<Void> splitRegionAsync(byte[] regionName, byte[] splitPoint)
1543       throws IOException {
1544     byte[] encodedNameofRegionToSplit = RegionInfo.isEncodedRegionName(regionName) ?
1545         regionName : Bytes.toBytes(RegionInfo.encodeRegionName(regionName));
1546     Pair<RegionInfo, ServerName> pair = getRegion(regionName);
1547     if (pair != null) {
1548       if (pair.getFirst() != null &&
1549           pair.getFirst().getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID) {
1550         throw new IllegalArgumentException ("Can't invoke split on non-default regions directly");
1551       }
1552     } else {
1553       throw new UnknownRegionException(
1554         "Can't invoke split on unknown region " + Bytes.toStringBinary(encodedNameofRegionToSplit));
1555     }
1556
1557     return splitRegionAsync(pair.getFirst(), splitPoint);
1558   }
1559
1560   Future<Void> splitRegionAsync(RegionInfo hri, byte[] splitPoint) throws IOException {
1561     TableName tableName = hri.getTable();
1562     if (hri.getStartKey() != null && splitPoint != null &&
1563         Bytes.compareTo(hri.getStartKey(), splitPoint) == 0) {
1564       throw new IOException("should not give a splitkey which equals to startkey!");
1565     }
1566
1567     SplitTableRegionResponse response = executeCallable(
1568         new MasterCallable<SplitTableRegionResponse>(getConnection(), getRpcControllerFactory()) {
1569           Long nonceGroup = ng.getNonceGroup();
1570           Long nonce = ng.newNonce();
1571           @Override
1572           protected SplitTableRegionResponse rpcCall() throws Exception {
1573             setPriority(tableName);
1574             SplitTableRegionRequest request = RequestConverter
1575                 .buildSplitTableRegionRequest(hri, splitPoint, nonceGroup, nonce);
1576             return master.splitRegion(getRpcController(), request);
1577           }
1578         });
1579     return new SplitTableRegionFuture(this, tableName, response);
1580   }
1581
1582   private static class SplitTableRegionFuture extends TableFuture<Void> {
1583     public SplitTableRegionFuture(final HBaseAdmin admin,
1584         final TableName tableName,
1585         final SplitTableRegionResponse response) {
1586       super(admin, tableName,
1587           (response != null && response.hasProcId()) ? response.getProcId() : null);
1588     }
1589
1590     public SplitTableRegionFuture(
1591         final HBaseAdmin admin,
1592         final TableName tableName,
1593         final Long procId) {
1594       super(admin, tableName, procId);
1595     }
1596
1597     @Override
1598     public String getOperationType() {
1599       return "SPLIT_REGION";
1600     }
1601   }
1602
1603   @Override
1604   public void split(final TableName tableName) throws IOException {
1605     split(tableName, null);
1606   }
1607
1608   @Override
1609   public void split(final TableName tableName, final byte[] splitPoint) throws IOException {
1610     checkTableExists(tableName);
1611     for (HRegionLocation loc : connection.locateRegions(tableName, false, false)) {
1612       ServerName sn = loc.getServerName();
1613       if (sn == null) {
1614         continue;
1615       }
1616       RegionInfo r = loc.getRegion();
1617       // check for parents
1618       if (r.isSplitParent()) {
1619         continue;
1620       }
1621       // if a split point given, only split that particular region
1622       if (r.getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID ||
1623           (splitPoint != null && !r.containsRow(splitPoint))) {
1624         continue;
1625       }
1626       // call out to master to do split now
1627       splitRegionAsync(r, splitPoint);
1628     }
1629   }
1630
1631   private static class ModifyTableFuture extends TableFuture<Void> {
1632     public ModifyTableFuture(final HBaseAdmin admin, final TableName tableName,
1633         final ModifyTableResponse response) {
1634       super(admin, tableName,
1635           (response != null && response.hasProcId()) ? response.getProcId() : null);
1636     }
1637
1638     public ModifyTableFuture(final HBaseAdmin admin, final TableName tableName, final Long procId) {
1639       super(admin, tableName, procId);
1640     }
1641
1642     @Override
1643     public String getOperationType() {
1644       return "MODIFY";
1645     }
1646   }
1647
1648   /**
1649    * @param regionName Name of a region.
1650    * @return a pair of HRegionInfo and ServerName if <code>regionName</code> is
1651    *  a verified region name (we call {@link
1652    *  MetaTableAccessor#getRegionLocation(Connection, byte[])}
1653    *  else null.
1654    * Throw IllegalArgumentException if <code>regionName</code> is null.
1655    * @throws IOException if a remote or network exception occurs
1656    */
1657   Pair<RegionInfo, ServerName> getRegion(final byte[] regionName) throws IOException {
1658     if (regionName == null) {
1659       throw new IllegalArgumentException("Pass a table name or region name");
1660     }
1661     Pair<RegionInfo, ServerName> pair = MetaTableAccessor.getRegion(connection, regionName);
1662     if (pair == null) {
1663       final AtomicReference<Pair<RegionInfo, ServerName>> result = new AtomicReference<>(null);
1664       final String encodedName = Bytes.toString(regionName);
1665       MetaTableAccessor.Visitor visitor = new MetaTableAccessor.Visitor() {
1666         @Override
1667         public boolean visit(Result data) throws IOException {
1668           RegionInfo info = MetaTableAccessor.getRegionInfo(data);
1669           if (info == null) {
1670             LOG.warn("No serialized HRegionInfo in " + data);
1671             return true;
1672           }
1673           RegionLocations rl = MetaTableAccessor.getRegionLocations(data);
1674           boolean matched = false;
1675           ServerName sn = null;
1676           if (rl != null) {
1677             for (HRegionLocation h : rl.getRegionLocations()) {
1678               if (h != null && encodedName.equals(h.getRegion().getEncodedName())) {
1679                 sn = h.getServerName();
1680                 info = h.getRegion();
1681                 matched = true;
1682               }
1683             }
1684           }
1685           if (!matched) return true;
1686           result.set(new Pair<>(info, sn));
1687           return false; // found the region, stop
1688         }
1689       };
1690
1691       MetaTableAccessor.fullScanRegions(connection, visitor);
1692       pair = result.get();
1693     }
1694     return pair;
1695   }
1696
1697   /**
1698    * If the input is a region name, it is returned as is. If it's an
1699    * encoded region name, the corresponding region is found from meta
1700    * and its region name is returned. If we can't find any region in
1701    * meta matching the input as either region name or encoded region
1702    * name, the input is returned as is. We don't throw unknown
1703    * region exception.
1704    */
1705   private byte[] getRegionName(final byte[] regionNameOrEncodedRegionName) throws IOException {
1706     if (Bytes.equals(regionNameOrEncodedRegionName,
1707       RegionInfoBuilder.FIRST_META_REGIONINFO.getRegionName()) ||
1708       Bytes.equals(regionNameOrEncodedRegionName,
1709         RegionInfoBuilder.FIRST_META_REGIONINFO.getEncodedNameAsBytes())) {
1710       return RegionInfoBuilder.FIRST_META_REGIONINFO.getRegionName();
1711     }
1712     byte[] tmp = regionNameOrEncodedRegionName;
1713     Pair<RegionInfo, ServerName> regionServerPair = getRegion(regionNameOrEncodedRegionName);
1714     if (regionServerPair != null && regionServerPair.getFirst() != null) {
1715       tmp = regionServerPair.getFirst().getRegionName();
1716     }
1717     return tmp;
1718   }
1719
1720   /**
1721    * Check if table exists or not
1722    * @param tableName Name of a table.
1723    * @return tableName instance
1724    * @throws IOException if a remote or network exception occurs.
1725    * @throws TableNotFoundException if table does not exist.
1726    */
1727   private TableName checkTableExists(final TableName tableName)
1728       throws IOException {
1729     return executeCallable(new RpcRetryingCallable<TableName>() {
1730       @Override
1731       protected TableName rpcCall(int callTimeout) throws Exception {
1732         if (!MetaTableAccessor.tableExists(connection, tableName)) {
1733           throw new TableNotFoundException(tableName);
1734         }
1735         return tableName;
1736       }
1737     });
1738   }
1739
1740   @Override
1741   public synchronized void shutdown() throws IOException {
1742     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
1743       @Override
1744       protected Void rpcCall() throws Exception {
1745         setPriority(HConstants.HIGH_QOS);
1746         master.shutdown(getRpcController(), ShutdownRequest.newBuilder().build());
1747         return null;
1748       }
1749     });
1750   }
1751
1752   @Override
1753   public synchronized void stopMaster() throws IOException {
1754     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
1755       @Override
1756       protected Void rpcCall() throws Exception {
1757         setPriority(HConstants.HIGH_QOS);
1758         master.stopMaster(getRpcController(), StopMasterRequest.newBuilder().build());
1759         return null;
1760       }
1761     });
1762   }
1763
1764   @Override
1765   public synchronized void stopRegionServer(final String hostnamePort)
1766   throws IOException {
1767     String hostname = Addressing.parseHostname(hostnamePort);
1768     int port = Addressing.parsePort(hostnamePort);
1769     final AdminService.BlockingInterface admin =
1770       this.connection.getAdmin(ServerName.valueOf(hostname, port, 0));
1771     // TODO: There is no timeout on this controller. Set one!
1772     HBaseRpcController controller = rpcControllerFactory.newController();
1773     controller.setPriority(HConstants.HIGH_QOS);
1774     StopServerRequest request = RequestConverter.buildStopServerRequest(
1775         "Called by admin client " + this.connection.toString());
1776     try {
1777       admin.stopServer(controller, request);
1778     } catch (Exception e) {
1779       throw ProtobufUtil.handleRemoteException(e);
1780     }
1781   }
1782
1783   @Override
1784   public boolean isMasterInMaintenanceMode() throws IOException {
1785     return executeCallable(new MasterCallable<IsInMaintenanceModeResponse>(getConnection(),
1786         this.rpcControllerFactory) {
1787       @Override
1788       protected IsInMaintenanceModeResponse rpcCall() throws Exception {
1789         return master.isMasterInMaintenanceMode(getRpcController(),
1790             IsInMaintenanceModeRequest.newBuilder().build());
1791       }
1792     }).getInMaintenanceMode();
1793   }
1794
1795   @Override
1796   public ClusterMetrics getClusterMetrics(EnumSet<Option> options) throws IOException {
1797     return executeCallable(new MasterCallable<ClusterMetrics>(getConnection(),
1798         this.rpcControllerFactory) {
1799       @Override
1800       protected ClusterMetrics rpcCall() throws Exception {
1801         GetClusterStatusRequest req = RequestConverter.buildGetClusterStatusRequest(options);
1802         return ClusterMetricsBuilder.toClusterMetrics(
1803           master.getClusterStatus(getRpcController(), req).getClusterStatus());
1804       }
1805     });
1806   }
1807
1808   @Override
1809   public List<RegionMetrics> getRegionMetrics(ServerName serverName, TableName tableName)
1810       throws IOException {
1811     AdminService.BlockingInterface admin = this.connection.getAdmin(serverName);
1812     HBaseRpcController controller = rpcControllerFactory.newController();
1813     AdminProtos.GetRegionLoadRequest request =
1814       RequestConverter.buildGetRegionLoadRequest(tableName);
1815     try {
1816       return admin.getRegionLoad(controller, request).getRegionLoadsList().stream()
1817         .map(RegionMetricsBuilder::toRegionMetrics).collect(Collectors.toList());
1818     } catch (ServiceException se) {
1819       throw ProtobufUtil.getRemoteException(se);
1820     }
1821   }
1822
1823   @Override
1824   public Configuration getConfiguration() {
1825     return this.conf;
1826   }
1827
1828   /**
1829    * Do a get with a timeout against the passed in <code>future</code>.
1830    */
1831   private static <T> T get(final Future<T> future, final long timeout, final TimeUnit units)
1832   throws IOException {
1833     try {
1834       // TODO: how long should we wait? Spin forever?
1835       return future.get(timeout, units);
1836     } catch (InterruptedException e) {
1837       IOException ioe = new InterruptedIOException("Interrupt while waiting on " + future);
1838       ioe.initCause(e);
1839       throw ioe;
1840     } catch (TimeoutException e) {
1841       throw new TimeoutIOException(e);
1842     } catch (ExecutionException e) {
1843       if (e.getCause() instanceof IOException) {
1844         throw (IOException)e.getCause();
1845       } else {
1846         throw new IOException(e.getCause());
1847       }
1848     }
1849   }
1850
1851   @Override
1852   public Future<Void> createNamespaceAsync(final NamespaceDescriptor descriptor)
1853       throws IOException {
1854     CreateNamespaceResponse response =
1855         executeCallable(new MasterCallable<CreateNamespaceResponse>(getConnection(),
1856             getRpcControllerFactory()) {
1857       @Override
1858       protected CreateNamespaceResponse rpcCall() throws Exception {
1859         return master.createNamespace(getRpcController(),
1860           CreateNamespaceRequest.newBuilder().setNamespaceDescriptor(ProtobufUtil.
1861               toProtoNamespaceDescriptor(descriptor)).build());
1862       }
1863     });
1864     return new NamespaceFuture(this, descriptor.getName(), response.getProcId()) {
1865       @Override
1866       public String getOperationType() {
1867         return "CREATE_NAMESPACE";
1868       }
1869     };
1870   }
1871
1872   @Override
1873   public Future<Void> modifyNamespaceAsync(final NamespaceDescriptor descriptor)
1874       throws IOException {
1875     ModifyNamespaceResponse response =
1876         executeCallable(new MasterCallable<ModifyNamespaceResponse>(getConnection(),
1877             getRpcControllerFactory()) {
1878       @Override
1879       protected ModifyNamespaceResponse rpcCall() throws Exception {
1880         // TODO: set priority based on NS?
1881         return master.modifyNamespace(getRpcController(), ModifyNamespaceRequest.newBuilder().
1882           setNamespaceDescriptor(ProtobufUtil.toProtoNamespaceDescriptor(descriptor)).build());
1883        }
1884     });
1885     return new NamespaceFuture(this, descriptor.getName(), response.getProcId()) {
1886       @Override
1887       public String getOperationType() {
1888         return "MODIFY_NAMESPACE";
1889       }
1890     };
1891   }
1892
1893   @Override
1894   public Future<Void> deleteNamespaceAsync(final String name)
1895       throws IOException {
1896     DeleteNamespaceResponse response =
1897         executeCallable(new MasterCallable<DeleteNamespaceResponse>(getConnection(),
1898             getRpcControllerFactory()) {
1899       @Override
1900       protected DeleteNamespaceResponse rpcCall() throws Exception {
1901         // TODO: set priority based on NS?
1902         return master.deleteNamespace(getRpcController(), DeleteNamespaceRequest.newBuilder().
1903           setNamespaceName(name).build());
1904         }
1905       });
1906     return new NamespaceFuture(this, name, response.getProcId()) {
1907       @Override
1908       public String getOperationType() {
1909         return "DELETE_NAMESPACE";
1910       }
1911     };
1912   }
1913
1914   @Override
1915   public NamespaceDescriptor getNamespaceDescriptor(final String name)
1916       throws NamespaceNotFoundException, IOException {
1917     return executeCallable(new MasterCallable<NamespaceDescriptor>(getConnection(),
1918         getRpcControllerFactory()) {
1919       @Override
1920       protected NamespaceDescriptor rpcCall() throws Exception {
1921         return ProtobufUtil.toNamespaceDescriptor(
1922             master.getNamespaceDescriptor(getRpcController(),
1923                 GetNamespaceDescriptorRequest.newBuilder().
1924                   setNamespaceName(name).build()).getNamespaceDescriptor());
1925       }
1926     });
1927   }
1928
1929   /**
1930    * List available namespaces
1931    * @return List of namespace names
1932    * @throws IOException if a remote or network exception occurs
1933    */
1934   @Override
1935   public String[] listNamespaces() throws IOException {
1936     return executeCallable(new MasterCallable<String[]>(getConnection(),
1937         getRpcControllerFactory()) {
1938       @Override
1939       protected String[] rpcCall() throws Exception {
1940         List<String> list = master.listNamespaces(getRpcController(),
1941           ListNamespacesRequest.newBuilder().build()).getNamespaceNameList();
1942         return list.toArray(new String[list.size()]);
1943       }
1944     });
1945   }
1946
1947   /**
1948    * List available namespace descriptors
1949    * @return List of descriptors
1950    * @throws IOException if a remote or network exception occurs
1951    */
1952   @Override
1953   public NamespaceDescriptor[] listNamespaceDescriptors() throws IOException {
1954     return executeCallable(new MasterCallable<NamespaceDescriptor[]>(getConnection(),
1955         getRpcControllerFactory()) {
1956       @Override
1957       protected NamespaceDescriptor[] rpcCall() throws Exception {
1958         List<HBaseProtos.NamespaceDescriptor> list =
1959             master.listNamespaceDescriptors(getRpcController(),
1960               ListNamespaceDescriptorsRequest.newBuilder().build()).getNamespaceDescriptorList();
1961         NamespaceDescriptor[] res = new NamespaceDescriptor[list.size()];
1962         for(int i = 0; i < list.size(); i++) {
1963           res[i] = ProtobufUtil.toNamespaceDescriptor(list.get(i));
1964         }
1965         return res;
1966       }
1967     });
1968   }
1969
1970   @Override
1971   public String getProcedures() throws IOException {
1972     return executeCallable(new MasterCallable<String>(getConnection(),
1973         getRpcControllerFactory()) {
1974       @Override
1975       protected String rpcCall() throws Exception {
1976         GetProceduresRequest request = GetProceduresRequest.newBuilder().build();
1977         GetProceduresResponse response = master.getProcedures(getRpcController(), request);
1978         return ProtobufUtil.toProcedureJson(response.getProcedureList());
1979       }
1980     });
1981   }
1982
1983   @Override
1984   public String getLocks() throws IOException {
1985     return executeCallable(new MasterCallable<String>(getConnection(),
1986         getRpcControllerFactory()) {
1987       @Override
1988       protected String rpcCall() throws Exception {
1989         GetLocksRequest request = GetLocksRequest.newBuilder().build();
1990         GetLocksResponse response = master.getLocks(getRpcController(), request);
1991         return ProtobufUtil.toLockJson(response.getLockList());
1992       }
1993     });
1994   }
1995
1996   @Override
1997   public TableName[] listTableNamesByNamespace(final String name) throws IOException {
1998     return executeCallable(new MasterCallable<TableName[]>(getConnection(),
1999         getRpcControllerFactory()) {
2000       @Override
2001       protected TableName[] rpcCall() throws Exception {
2002         List<HBaseProtos.TableName> tableNames =
2003             master.listTableNamesByNamespace(getRpcController(), ListTableNamesByNamespaceRequest.
2004                 newBuilder().setNamespaceName(name).build())
2005             .getTableNameList();
2006         TableName[] result = new TableName[tableNames.size()];
2007         for (int i = 0; i < tableNames.size(); i++) {
2008           result[i] = ProtobufUtil.toTableName(tableNames.get(i));
2009         }
2010         return result;
2011       }
2012     });
2013   }
2014
2015   /**
2016    * Is HBase available? Throw an exception if not.
2017    * <p/>
2018    * TODO: do not expose ZKConnectionException.
2019    * @param conf system configuration
2020    * @throws MasterNotRunningException if the master is not running.
2021    * @throws ZooKeeperConnectionException if unable to connect to zookeeper.
2022    */
2023   public static void available(final Configuration conf)
2024       throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
2025     Configuration copyOfConf = HBaseConfiguration.create(conf);
2026     // We set it to make it fail as soon as possible if HBase is not available
2027     copyOfConf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);
2028     copyOfConf.setInt("zookeeper.recovery.retry", 0);
2029
2030     // Check ZK first.
2031     // If the connection exists, we may have a connection to ZK that does not work anymore
2032     try (ConnectionImplementation connection = ConnectionFactory.createConnectionImpl(copyOfConf,
2033       null, UserProvider.instantiate(copyOfConf).getCurrent())) {
2034       // can throw MasterNotRunningException
2035       connection.isMasterRunning();
2036     }
2037   }
2038
2039   private RollWALWriterResponse rollWALWriterImpl(final ServerName sn) throws IOException,
2040       FailedLogCloseException {
2041     final AdminService.BlockingInterface admin = this.connection.getAdmin(sn);
2042     RollWALWriterRequest request = RequestConverter.buildRollWALWriterRequest();
2043     // TODO: There is no timeout on this controller. Set one!
2044     HBaseRpcController controller = rpcControllerFactory.newController();
2045     try {
2046       return admin.rollWALWriter(controller, request);
2047     } catch (ServiceException e) {
2048       throw ProtobufUtil.handleRemoteException(e);
2049     }
2050   }
2051
2052   @Override
2053   public synchronized void rollWALWriter(ServerName serverName)
2054       throws IOException, FailedLogCloseException {
2055     rollWALWriterImpl(serverName);
2056   }
2057
2058   @Override
2059   public CompactionState getCompactionState(final TableName tableName)
2060   throws IOException {
2061     return getCompactionState(tableName, CompactType.NORMAL);
2062   }
2063
2064   @Override
2065   public CompactionState getCompactionStateForRegion(final byte[] regionName)
2066   throws IOException {
2067     final Pair<RegionInfo, ServerName> regionServerPair = getRegion(regionName);
2068     if (regionServerPair == null) {
2069       throw new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName));
2070     }
2071     if (regionServerPair.getSecond() == null) {
2072       throw new NoServerForRegionException(Bytes.toStringBinary(regionName));
2073     }
2074     ServerName sn = regionServerPair.getSecond();
2075     final AdminService.BlockingInterface admin = this.connection.getAdmin(sn);
2076     // TODO: There is no timeout on this controller. Set one!
2077     HBaseRpcController controller = rpcControllerFactory.newController();
2078     GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest(
2079       regionServerPair.getFirst().getRegionName(), true);
2080     GetRegionInfoResponse response;
2081     try {
2082       response = admin.getRegionInfo(controller, request);
2083     } catch (ServiceException e) {
2084       throw ProtobufUtil.handleRemoteException(e);
2085     }
2086     if (response.getCompactionState() != null) {
2087       return ProtobufUtil.createCompactionState(response.getCompactionState());
2088     }
2089     return null;
2090   }
2091
2092   @Override
2093   public void snapshot(SnapshotDescription snapshotDesc)
2094       throws IOException, SnapshotCreationException, IllegalArgumentException {
2095     // actually take the snapshot
2096     SnapshotProtos.SnapshotDescription snapshot =
2097       ProtobufUtil.createHBaseProtosSnapshotDesc(snapshotDesc);
2098     SnapshotResponse response = asyncSnapshot(snapshot);
2099     final IsSnapshotDoneRequest request =
2100         IsSnapshotDoneRequest.newBuilder().setSnapshot(snapshot).build();
2101     IsSnapshotDoneResponse done = null;
2102     long start = EnvironmentEdgeManager.currentTime();
2103     long max = response.getExpectedTimeout();
2104     long maxPauseTime = max / this.numRetries;
2105     int tries = 0;
2106     LOG.debug("Waiting a max of " + max + " ms for snapshot '" +
2107         ClientSnapshotDescriptionUtils.toString(snapshot) + "'' to complete. (max " +
2108         maxPauseTime + " ms per retry)");
2109     while (tries == 0
2110         || ((EnvironmentEdgeManager.currentTime() - start) < max && !done.getDone())) {
2111       try {
2112         // sleep a backoff <= pauseTime amount
2113         long sleep = getPauseTime(tries++);
2114         sleep = sleep > maxPauseTime ? maxPauseTime : sleep;
2115         LOG.debug("(#" + tries + ") Sleeping: " + sleep +
2116           "ms while waiting for snapshot completion.");
2117         Thread.sleep(sleep);
2118       } catch (InterruptedException e) {
2119         throw (InterruptedIOException)new InterruptedIOException("Interrupted").initCause(e);
2120       }
2121       LOG.debug("Getting current status of snapshot from master...");
2122       done = executeCallable(new MasterCallable<IsSnapshotDoneResponse>(getConnection(),
2123           getRpcControllerFactory()) {
2124         @Override
2125         protected IsSnapshotDoneResponse rpcCall() throws Exception {
2126           return master.isSnapshotDone(getRpcController(), request);
2127         }
2128       });
2129     }
2130     if (!done.getDone()) {
2131       throw new SnapshotCreationException("Snapshot '" + snapshot.getName()
2132           + "' wasn't completed in expectedTime:" + max + " ms", snapshotDesc);
2133     }
2134   }
2135
2136   @Override
2137   public Future<Void> snapshotAsync(SnapshotDescription snapshotDesc)
2138       throws IOException, SnapshotCreationException {
2139     asyncSnapshot(ProtobufUtil.createHBaseProtosSnapshotDesc(snapshotDesc));
2140     return new ProcedureFuture<Void>(this, null) {
2141
2142       @Override
2143       protected Void waitOperationResult(long deadlineTs) throws IOException, TimeoutException {
2144         waitForState(deadlineTs, new WaitForStateCallable() {
2145
2146           @Override
2147           public void throwInterruptedException() throws InterruptedIOException {
2148             throw new InterruptedIOException(
2149               "Interrupted while waiting for taking snapshot" + snapshotDesc);
2150           }
2151
2152           @Override
2153           public void throwTimeoutException(long elapsedTime) throws TimeoutException {
2154             throw new TimeoutException("Snapshot '" + snapshotDesc.getName() +
2155               "' wasn't completed in expectedTime:" + elapsedTime + " ms");
2156           }
2157
2158           @Override
2159           public boolean checkState(int tries) throws IOException {
2160             return isSnapshotFinished(snapshotDesc);
2161           }
2162         });
2163         return null;
2164       }
2165     };
2166   }
2167
2168   private SnapshotResponse asyncSnapshot(SnapshotProtos.SnapshotDescription snapshot)
2169       throws IOException {
2170     ClientSnapshotDescriptionUtils.assertSnapshotRequestIsValid(snapshot);
2171     final SnapshotRequest request = SnapshotRequest.newBuilder().setSnapshot(snapshot)
2172         .build();
2173     // run the snapshot on the master
2174     return executeCallable(new MasterCallable<SnapshotResponse>(getConnection(),
2175         getRpcControllerFactory()) {
2176       @Override
2177       protected SnapshotResponse rpcCall() throws Exception {
2178         return master.snapshot(getRpcController(), request);
2179       }
2180     });
2181   }
2182
2183   @Override
2184   public boolean isSnapshotFinished(final SnapshotDescription snapshotDesc)
2185       throws IOException, HBaseSnapshotException, UnknownSnapshotException {
2186     final SnapshotProtos.SnapshotDescription snapshot =
2187         ProtobufUtil.createHBaseProtosSnapshotDesc(snapshotDesc);
2188     return executeCallable(new MasterCallable<IsSnapshotDoneResponse>(getConnection(),
2189         getRpcControllerFactory()) {
2190       @Override
2191       protected IsSnapshotDoneResponse rpcCall() throws Exception {
2192         return master.isSnapshotDone(getRpcController(),
2193           IsSnapshotDoneRequest.newBuilder().setSnapshot(snapshot).build());
2194       }
2195     }).getDone();
2196   }
2197
2198   @Override
2199   public void restoreSnapshot(final String snapshotName)
2200       throws IOException, RestoreSnapshotException {
2201     boolean takeFailSafeSnapshot =
2202         conf.getBoolean(HConstants.SNAPSHOT_RESTORE_TAKE_FAILSAFE_SNAPSHOT,
2203           HConstants.DEFAULT_SNAPSHOT_RESTORE_TAKE_FAILSAFE_SNAPSHOT);
2204     restoreSnapshot(snapshotName, takeFailSafeSnapshot);
2205   }
2206
2207   /**
2208    * Check whether the snapshot exists and contains disabled table
2209    *
2210    * @param snapshotName name of the snapshot to restore
2211    * @throws IOException if a remote or network exception occurs
2212    * @throws RestoreSnapshotException if no valid snapshot is found
2213    */
2214   private TableName getTableNameBeforeRestoreSnapshot(final String snapshotName)
2215       throws IOException, RestoreSnapshotException {
2216     TableName tableName = null;
2217     for (SnapshotDescription snapshotInfo: listSnapshots()) {
2218       if (snapshotInfo.getName().equals(snapshotName)) {
2219         tableName = snapshotInfo.getTableName();
2220         break;
2221       }
2222     }
2223
2224     if (tableName == null) {
2225       throw new RestoreSnapshotException(
2226         "Unable to find the table name for snapshot=" + snapshotName);
2227     }
2228     return tableName;
2229   }
2230
2231   @Override
2232   public void restoreSnapshot(String snapshotName, boolean takeFailSafeSnapshot)
2233       throws IOException, RestoreSnapshotException {
2234     restoreSnapshot(snapshotName, takeFailSafeSnapshot, false);
2235   }
2236
2237   @Override
2238   public void restoreSnapshot(final String snapshotName, final boolean takeFailSafeSnapshot,
2239       final boolean restoreAcl) throws IOException, RestoreSnapshotException {
2240     TableName tableName = getTableNameBeforeRestoreSnapshot(snapshotName);
2241
2242     // The table does not exists, switch to clone.
2243     if (!tableExists(tableName)) {
2244       cloneSnapshot(snapshotName, tableName, restoreAcl);
2245       return;
2246     }
2247
2248     // Check if the table is disabled
2249     if (!isTableDisabled(tableName)) {
2250       throw new TableNotDisabledException(tableName);
2251     }
2252
2253     // Take a snapshot of the current state
2254     String failSafeSnapshotSnapshotName = null;
2255     if (takeFailSafeSnapshot) {
2256       failSafeSnapshotSnapshotName = conf.get("hbase.snapshot.restore.failsafe.name",
2257         "hbase-failsafe-{snapshot.name}-{restore.timestamp}");
2258       failSafeSnapshotSnapshotName = failSafeSnapshotSnapshotName
2259         .replace("{snapshot.name}", snapshotName)
2260         .replace("{table.name}", tableName.toString().replace(TableName.NAMESPACE_DELIM, '.'))
2261         .replace("{restore.timestamp}", String.valueOf(EnvironmentEdgeManager.currentTime()));
2262       LOG.info("Taking restore-failsafe snapshot: " + failSafeSnapshotSnapshotName);
2263       snapshot(failSafeSnapshotSnapshotName, tableName);
2264     }
2265
2266     try {
2267       // Restore snapshot
2268       get(
2269         internalRestoreSnapshotAsync(snapshotName, tableName, restoreAcl),
2270         syncWaitTimeout,
2271         TimeUnit.MILLISECONDS);
2272     } catch (IOException e) {
2273       // Something went wrong during the restore...
2274       // if the pre-restore snapshot is available try to rollback
2275       if (takeFailSafeSnapshot) {
2276         try {
2277           get(
2278             internalRestoreSnapshotAsync(failSafeSnapshotSnapshotName, tableName, restoreAcl),
2279             syncWaitTimeout,
2280             TimeUnit.MILLISECONDS);
2281           String msg = "Restore snapshot=" + snapshotName +
2282             " failed. Rollback to snapshot=" + failSafeSnapshotSnapshotName + " succeeded.";
2283           LOG.error(msg, e);
2284           throw new RestoreSnapshotException(msg, e);
2285         } catch (IOException ex) {
2286           String msg = "Failed to restore and rollback to snapshot=" + failSafeSnapshotSnapshotName;
2287           LOG.error(msg, ex);
2288           throw new RestoreSnapshotException(msg, e);
2289         }
2290       } else {
2291         throw new RestoreSnapshotException("Failed to restore snapshot=" + snapshotName, e);
2292       }
2293     }
2294
2295     // If the restore is succeeded, delete the pre-restore snapshot
2296     if (takeFailSafeSnapshot) {
2297       try {
2298         LOG.info("Deleting restore-failsafe snapshot: " + failSafeSnapshotSnapshotName);
2299         deleteSnapshot(failSafeSnapshotSnapshotName);
2300       } catch (IOException e) {
2301         LOG.error("Unable to remove the failsafe snapshot: " + failSafeSnapshotSnapshotName, e);
2302       }
2303     }
2304   }
2305
2306   @Override
2307   public Future<Void> cloneSnapshotAsync(String snapshotName, TableName tableName,
2308       boolean restoreAcl) throws IOException, TableExistsException, RestoreSnapshotException {
2309     if (tableExists(tableName)) {
2310       throw new TableExistsException(tableName);
2311     }
2312     return internalRestoreSnapshotAsync(snapshotName, tableName, restoreAcl);
2313   }
2314
2315   @Override
2316   public byte[] execProcedureWithReturn(String signature, String instance, Map<String,
2317       String> props) throws IOException {
2318     ProcedureDescription desc = ProtobufUtil.buildProcedureDescription(signature, instance, props);
2319     final ExecProcedureRequest request =
2320         ExecProcedureRequest.newBuilder().setProcedure(desc).build();
2321     // run the procedure on the master
2322     ExecProcedureResponse response = executeCallable(
2323       new MasterCallable<ExecProcedureResponse>(getConnection(), getRpcControllerFactory()) {
2324         @Override
2325         protected ExecProcedureResponse rpcCall() throws Exception {
2326           return master.execProcedureWithRet(getRpcController(), request);
2327         }
2328       });
2329
2330     return response.hasReturnData() ? response.getReturnData().toByteArray() : null;
2331   }
2332
2333   @Override
2334   public void execProcedure(String signature, String instance, Map<String, String> props)
2335       throws IOException {
2336     ProcedureDescription desc = ProtobufUtil.buildProcedureDescription(signature, instance, props);
2337     final ExecProcedureRequest request =
2338         ExecProcedureRequest.newBuilder().setProcedure(desc).build();
2339     // run the procedure on the master
2340     ExecProcedureResponse response = executeCallable(new MasterCallable<ExecProcedureResponse>(
2341         getConnection(), getRpcControllerFactory()) {
2342       @Override
2343       protected ExecProcedureResponse rpcCall() throws Exception {
2344         return master.execProcedure(getRpcController(), request);
2345       }
2346     });
2347
2348     long start = EnvironmentEdgeManager.currentTime();
2349     long max = response.getExpectedTimeout();
2350     long maxPauseTime = max / this.numRetries;
2351     int tries = 0;
2352     LOG.debug("Waiting a max of " + max + " ms for procedure '" +
2353         signature + " : " + instance + "'' to complete. (max " + maxPauseTime + " ms per retry)");
2354     boolean done = false;
2355     while (tries == 0
2356         || ((EnvironmentEdgeManager.currentTime() - start) < max && !done)) {
2357       try {
2358         // sleep a backoff <= pauseTime amount
2359         long sleep = getPauseTime(tries++);
2360         sleep = sleep > maxPauseTime ? maxPauseTime : sleep;
2361         LOG.debug("(#" + tries + ") Sleeping: " + sleep +
2362           "ms while waiting for procedure completion.");
2363         Thread.sleep(sleep);
2364       } catch (InterruptedException e) {
2365         throw (InterruptedIOException)new InterruptedIOException("Interrupted").initCause(e);
2366       }
2367       LOG.debug("Getting current status of procedure from master...");
2368       done = isProcedureFinished(signature, instance, props);
2369     }
2370     if (!done) {
2371       throw new IOException("Procedure '" + signature + " : " + instance
2372           + "' wasn't completed in expectedTime:" + max + " ms");
2373     }
2374   }
2375
2376   @Override
2377   public boolean isProcedureFinished(String signature, String instance, Map<String, String> props)
2378       throws IOException {
2379     ProcedureDescription desc = ProtobufUtil.buildProcedureDescription(signature, instance, props);
2380     return executeCallable(
2381       new MasterCallable<IsProcedureDoneResponse>(getConnection(), getRpcControllerFactory()) {
2382         @Override
2383         protected IsProcedureDoneResponse rpcCall() throws Exception {
2384           return master.isProcedureDone(getRpcController(),
2385             IsProcedureDoneRequest.newBuilder().setProcedure(desc).build());
2386         }
2387       }).getDone();
2388   }
2389
2390   /**
2391    * Execute Restore/Clone snapshot and wait for the server to complete (blocking).
2392    * To check if the cloned table exists, use {@link #isTableAvailable} -- it is not safe to
2393    * create an HTable instance to this table before it is available.
2394    * @param snapshotName snapshot to restore
2395    * @param tableName table name to restore the snapshot on
2396    * @throws IOException if a remote or network exception occurs
2397    * @throws RestoreSnapshotException if snapshot failed to be restored
2398    * @throws IllegalArgumentException if the restore request is formatted incorrectly
2399    */
2400   private Future<Void> internalRestoreSnapshotAsync(final String snapshotName,
2401       final TableName tableName, final boolean restoreAcl)
2402       throws IOException, RestoreSnapshotException {
2403     final SnapshotProtos.SnapshotDescription snapshot =
2404         SnapshotProtos.SnapshotDescription.newBuilder()
2405         .setName(snapshotName).setTable(tableName.getNameAsString()).build();
2406
2407     // actually restore the snapshot
2408     ClientSnapshotDescriptionUtils.assertSnapshotRequestIsValid(snapshot);
2409
2410     RestoreSnapshotResponse response = executeCallable(
2411         new MasterCallable<RestoreSnapshotResponse>(getConnection(), getRpcControllerFactory()) {
2412           Long nonceGroup = ng.getNonceGroup();
2413           Long nonce = ng.newNonce();
2414       @Override
2415       protected RestoreSnapshotResponse rpcCall() throws Exception {
2416         final RestoreSnapshotRequest request = RestoreSnapshotRequest.newBuilder()
2417             .setSnapshot(snapshot)
2418             .setNonceGroup(nonceGroup)
2419             .setNonce(nonce)
2420             .setRestoreACL(restoreAcl)
2421             .build();
2422         return master.restoreSnapshot(getRpcController(), request);
2423       }
2424     });
2425
2426     return new RestoreSnapshotFuture(this, snapshot, tableName, response);
2427   }
2428
2429   private static class RestoreSnapshotFuture extends TableFuture<Void> {
2430     public RestoreSnapshotFuture(
2431         final HBaseAdmin admin,
2432         final SnapshotProtos.SnapshotDescription snapshot,
2433         final TableName tableName,
2434         final RestoreSnapshotResponse response) {
2435       super(admin, tableName,
2436           (response != null && response.hasProcId()) ? response.getProcId() : null);
2437
2438       if (response != null && !response.hasProcId()) {
2439         throw new UnsupportedOperationException("Client could not call old version of Server");
2440       }
2441     }
2442
2443     public RestoreSnapshotFuture(
2444         final HBaseAdmin admin,
2445         final TableName tableName,
2446         final Long procId) {
2447       super(admin, tableName, procId);
2448     }
2449
2450     @Override
2451     public String getOperationType() {
2452       return "MODIFY";
2453     }
2454   }
2455
2456   @Override
2457   public List<SnapshotDescription> listSnapshots() throws IOException {
2458     return executeCallable(new MasterCallable<List<SnapshotDescription>>(getConnection(),
2459         getRpcControllerFactory()) {
2460       @Override
2461       protected List<SnapshotDescription> rpcCall() throws Exception {
2462         List<SnapshotProtos.SnapshotDescription> snapshotsList = master
2463             .getCompletedSnapshots(getRpcController(),
2464                 GetCompletedSnapshotsRequest.newBuilder().build())
2465             .getSnapshotsList();
2466         List<SnapshotDescription> result = new ArrayList<>(snapshotsList.size());
2467         for (SnapshotProtos.SnapshotDescription snapshot : snapshotsList) {
2468           result.add(ProtobufUtil.createSnapshotDesc(snapshot));
2469         }
2470         return result;
2471       }
2472     });
2473   }
2474
2475   @Override
2476   public List<SnapshotDescription> listSnapshots(Pattern pattern) throws IOException {
2477     List<SnapshotDescription> matched = new LinkedList<>();
2478     List<SnapshotDescription> snapshots = listSnapshots();
2479     for (SnapshotDescription snapshot : snapshots) {
2480       if (pattern.matcher(snapshot.getName()).matches()) {
2481         matched.add(snapshot);
2482       }
2483     }
2484     return matched;
2485   }
2486
2487   @Override
2488   public List<SnapshotDescription> listTableSnapshots(Pattern tableNamePattern,
2489       Pattern snapshotNamePattern) throws IOException {
2490     TableName[] tableNames = listTableNames(tableNamePattern);
2491
2492     List<SnapshotDescription> tableSnapshots = new LinkedList<>();
2493     List<SnapshotDescription> snapshots = listSnapshots(snapshotNamePattern);
2494
2495     List<TableName> listOfTableNames = Arrays.asList(tableNames);
2496     for (SnapshotDescription snapshot : snapshots) {
2497       if (listOfTableNames.contains(snapshot.getTableName())) {
2498         tableSnapshots.add(snapshot);
2499       }
2500     }
2501     return tableSnapshots;
2502   }
2503
2504   @Override
2505   public void deleteSnapshot(final String snapshotName) throws IOException {
2506     // make sure the snapshot is possibly valid
2507     TableName.isLegalFullyQualifiedTableName(Bytes.toBytes(snapshotName));
2508     // do the delete
2509     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
2510       @Override
2511       protected Void rpcCall() throws Exception {
2512         master.deleteSnapshot(getRpcController(),
2513           DeleteSnapshotRequest.newBuilder().setSnapshot(
2514                 SnapshotProtos.SnapshotDescription.newBuilder().setName(snapshotName).build())
2515               .build()
2516         );
2517         return null;
2518       }
2519     });
2520   }
2521
2522   @Override
2523   public void deleteSnapshots(final Pattern pattern) throws IOException {
2524     List<SnapshotDescription> snapshots = listSnapshots(pattern);
2525     for (final SnapshotDescription snapshot : snapshots) {
2526       try {
2527         internalDeleteSnapshot(snapshot);
2528       } catch (IOException ex) {
2529         LOG.info("Failed to delete snapshot " + snapshot.getName() + " for table "
2530                 + snapshot.getTableNameAsString(), ex);
2531       }
2532     }
2533   }
2534
2535   private void internalDeleteSnapshot(final SnapshotDescription snapshot) throws IOException {
2536     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
2537       @Override
2538       protected Void rpcCall() throws Exception {
2539         this.master.deleteSnapshot(getRpcController(), DeleteSnapshotRequest.newBuilder()
2540           .setSnapshot(ProtobufUtil.createHBaseProtosSnapshotDesc(snapshot)).build());
2541         return null;
2542       }
2543     });
2544   }
2545
2546   @Override
2547   public void deleteTableSnapshots(Pattern tableNamePattern, Pattern snapshotNamePattern)
2548       throws IOException {
2549     List<SnapshotDescription> snapshots = listTableSnapshots(tableNamePattern, snapshotNamePattern);
2550     for (SnapshotDescription snapshot : snapshots) {
2551       try {
2552         internalDeleteSnapshot(snapshot);
2553         LOG.debug("Successfully deleted snapshot: " + snapshot.getName());
2554       } catch (IOException e) {
2555         LOG.error("Failed to delete snapshot: " + snapshot.getName(), e);
2556       }
2557     }
2558   }
2559
2560   @Override
2561   public void setQuota(final QuotaSettings quota) throws IOException {
2562     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
2563       @Override
2564       protected Void rpcCall() throws Exception {
2565         this.master.setQuota(getRpcController(), QuotaSettings.buildSetQuotaRequestProto(quota));
2566         return null;
2567       }
2568     });
2569   }
2570
2571   @Override
2572   public List<QuotaSettings> getQuota(QuotaFilter filter) throws IOException {
2573     List<QuotaSettings> quotas = new LinkedList<>();
2574     try (QuotaRetriever retriever = QuotaRetriever.open(conf, filter)) {
2575       Iterator<QuotaSettings> iterator = retriever.iterator();
2576       while (iterator.hasNext()) {
2577         quotas.add(iterator.next());
2578       }
2579     }
2580     return quotas;
2581   }
2582
2583   private <C extends RetryingCallable<V> & Closeable, V> V executeCallable(C callable)
2584       throws IOException {
2585     return executeCallable(callable, rpcCallerFactory, operationTimeout, rpcTimeout);
2586   }
2587
2588   static private <C extends RetryingCallable<V> & Closeable, V> V executeCallable(C callable,
2589              RpcRetryingCallerFactory rpcCallerFactory, int operationTimeout, int rpcTimeout)
2590   throws IOException {
2591     RpcRetryingCaller<V> caller = rpcCallerFactory.newCaller(rpcTimeout);
2592     try {
2593       return caller.callWithRetries(callable, operationTimeout);
2594     } finally {
2595       callable.close();
2596     }
2597   }
2598
2599   @Override
2600   // Coprocessor Endpoint against the Master.
2601   public CoprocessorRpcChannel coprocessorService() {
2602     return new SyncCoprocessorRpcChannel() {
2603       @Override
2604       protected Message callExecService(final RpcController controller,
2605           final Descriptors.MethodDescriptor method, final Message request,
2606           final Message responsePrototype)
2607       throws IOException {
2608         if (LOG.isTraceEnabled()) {
2609           LOG.trace("Call: " + method.getName() + ", " + request.toString());
2610         }
2611         // Try-with-resources so close gets called when we are done.
2612         try (MasterCallable<CoprocessorServiceResponse> callable =
2613             new MasterCallable<CoprocessorServiceResponse>(connection,
2614                 connection.getRpcControllerFactory()) {
2615           @Override
2616           protected CoprocessorServiceResponse rpcCall() throws Exception {
2617             CoprocessorServiceRequest csr =
2618                 CoprocessorRpcUtils.getCoprocessorServiceRequest(method, request);
2619             return this.master.execMasterService(getRpcController(), csr);
2620           }
2621         }) {
2622           // TODO: Are we retrying here? Does not seem so. We should use RetryingRpcCaller
2623           callable.prepare(false);
2624           int operationTimeout = connection.getConnectionConfiguration().getOperationTimeout();
2625           CoprocessorServiceResponse result = callable.call(operationTimeout);
2626           return CoprocessorRpcUtils.getResponse(result, responsePrototype);
2627         }
2628       }
2629     };
2630   }
2631
2632   @Override
2633   public CoprocessorRpcChannel coprocessorService(final ServerName serverName) {
2634     return new SyncCoprocessorRpcChannel() {
2635       @Override
2636       protected Message callExecService(RpcController controller,
2637           Descriptors.MethodDescriptor method, Message request, Message responsePrototype)
2638       throws IOException {
2639         if (LOG.isTraceEnabled()) {
2640           LOG.trace("Call: " + method.getName() + ", " + request.toString());
2641         }
2642         CoprocessorServiceRequest csr =
2643             CoprocessorRpcUtils.getCoprocessorServiceRequest(method, request);
2644         // TODO: Are we retrying here? Does not seem so. We should use RetryingRpcCaller
2645         // TODO: Make this same as RegionCoprocessorRpcChannel and MasterCoprocessorRpcChannel. They
2646         // are all different though should do same thing; e.g. RpcChannel setup.
2647         ClientProtos.ClientService.BlockingInterface stub = connection.getClient(serverName);
2648         CoprocessorServiceResponse result;
2649         try {
2650           result = stub.
2651               execRegionServerService(connection.getRpcControllerFactory().newController(), csr);
2652           return CoprocessorRpcUtils.getResponse(result, responsePrototype);
2653         } catch (ServiceException e) {
2654           throw ProtobufUtil.handleRemoteException(e);
2655         }
2656       }
2657     };
2658   }
2659
2660   @Override
2661   public void updateConfiguration(final ServerName server) throws IOException {
2662     final AdminService.BlockingInterface admin = this.connection.getAdmin(server);
2663     Callable<Void> callable = new Callable<Void>() {
2664       @Override
2665       public Void call() throws Exception {
2666         admin.updateConfiguration(null, UpdateConfigurationRequest.getDefaultInstance());
2667         return null;
2668       }
2669     };
2670     ProtobufUtil.call(callable);
2671   }
2672
2673   @Override
2674   public void updateConfiguration() throws IOException {
2675     ClusterMetrics status = getClusterMetrics(
2676       EnumSet.of(Option.LIVE_SERVERS, Option.MASTER, Option.BACKUP_MASTERS));
2677     for (ServerName server : status.getLiveServerMetrics().keySet()) {
2678       updateConfiguration(server);
2679     }
2680
2681     updateConfiguration(status.getMasterName());
2682
2683     for (ServerName server : status.getBackupMasterNames()) {
2684       updateConfiguration(server);
2685     }
2686   }
2687
2688   @Override
2689   public long getLastMajorCompactionTimestamp(final TableName tableName) throws IOException {
2690     return executeCallable(new MasterCallable<Long>(getConnection(), getRpcControllerFactory()) {
2691       @Override
2692       protected Long rpcCall() throws Exception {
2693         MajorCompactionTimestampRequest req =
2694             MajorCompactionTimestampRequest.newBuilder()
2695                 .setTableName(ProtobufUtil.toProtoTableName(tableName)).build();
2696         return master.getLastMajorCompactionTimestamp(getRpcController(), req).
2697             getCompactionTimestamp();
2698       }
2699     });
2700   }
2701
2702   @Override
2703   public long getLastMajorCompactionTimestampForRegion(final byte[] regionName) throws IOException {
2704     return executeCallable(new MasterCallable<Long>(getConnection(), getRpcControllerFactory()) {
2705       @Override
2706       protected Long rpcCall() throws Exception {
2707         MajorCompactionTimestampForRegionRequest req =
2708             MajorCompactionTimestampForRegionRequest.newBuilder().setRegion(RequestConverter
2709                       .buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName)).build();
2710         return master.getLastMajorCompactionTimestampForRegion(getRpcController(), req)
2711             .getCompactionTimestamp();
2712       }
2713     });
2714   }
2715
2716   /**
2717    * {@inheritDoc}
2718    */
2719   @Override
2720   public void compact(final TableName tableName, final byte[] columnFamily, CompactType compactType)
2721     throws IOException, InterruptedException {
2722     compact(tableName, columnFamily, false, compactType);
2723   }
2724
2725   /**
2726    * {@inheritDoc}
2727    */
2728   @Override
2729   public void compact(final TableName tableName, CompactType compactType)
2730     throws IOException, InterruptedException {
2731     compact(tableName, null, false, compactType);
2732   }
2733
2734   /**
2735    * {@inheritDoc}
2736    */
2737   @Override
2738   public void majorCompact(final TableName tableName, final byte[] columnFamily,
2739     CompactType compactType) throws IOException, InterruptedException {
2740     compact(tableName, columnFamily, true, compactType);
2741   }
2742
2743   /**
2744    * {@inheritDoc}
2745    */
2746   @Override
2747   public void majorCompact(final TableName tableName, CompactType compactType)
2748           throws IOException, InterruptedException {
2749     compact(tableName, null, true, compactType);
2750   }
2751
2752   /**
2753    * {@inheritDoc}
2754    */
2755   @Override
2756   public CompactionState getCompactionState(final TableName tableName, CompactType compactType)
2757       throws IOException {
2758     checkTableExists(tableName);
2759     if (!isTableEnabled(tableName)) {
2760       // If the table is disabled, the compaction state of the table should always be NONE
2761       return ProtobufUtil.createCompactionState(
2762         AdminProtos.GetRegionInfoResponse.CompactionState.NONE);
2763     }
2764
2765     AdminProtos.GetRegionInfoResponse.CompactionState state =
2766       AdminProtos.GetRegionInfoResponse.CompactionState.NONE;
2767
2768     // TODO: There is no timeout on this controller. Set one!
2769     HBaseRpcController rpcController = rpcControllerFactory.newController();
2770     switch (compactType) {
2771       case MOB:
2772         final AdminProtos.AdminService.BlockingInterface masterAdmin =
2773           this.connection.getAdminForMaster();
2774         Callable<AdminProtos.GetRegionInfoResponse.CompactionState> callable =
2775           new Callable<AdminProtos.GetRegionInfoResponse.CompactionState>() {
2776             @Override
2777             public AdminProtos.GetRegionInfoResponse.CompactionState call() throws Exception {
2778               RegionInfo info = RegionInfo.createMobRegionInfo(tableName);
2779               GetRegionInfoRequest request =
2780                 RequestConverter.buildGetRegionInfoRequest(info.getRegionName(), true);
2781               GetRegionInfoResponse response = masterAdmin.getRegionInfo(rpcController, request);
2782               return response.getCompactionState();
2783             }
2784           };
2785         state = ProtobufUtil.call(callable);
2786         break;
2787       case NORMAL:
2788         for (HRegionLocation loc : connection.locateRegions(tableName, false, false)) {
2789           ServerName sn = loc.getServerName();
2790           if (sn == null) {
2791             continue;
2792           }
2793           byte[] regionName = loc.getRegion().getRegionName();
2794           AdminService.BlockingInterface snAdmin = this.connection.getAdmin(sn);
2795           try {
2796             Callable<GetRegionInfoResponse> regionInfoCallable =
2797               new Callable<GetRegionInfoResponse>() {
2798                 @Override
2799                 public GetRegionInfoResponse call() throws Exception {
2800                   GetRegionInfoRequest request =
2801                     RequestConverter.buildGetRegionInfoRequest(regionName, true);
2802                   return snAdmin.getRegionInfo(rpcController, request);
2803                 }
2804               };
2805             GetRegionInfoResponse response = ProtobufUtil.call(regionInfoCallable);
2806             switch (response.getCompactionState()) {
2807               case MAJOR_AND_MINOR:
2808                 return CompactionState.MAJOR_AND_MINOR;
2809               case MAJOR:
2810                 if (state == AdminProtos.GetRegionInfoResponse.CompactionState.MINOR) {
2811                   return CompactionState.MAJOR_AND_MINOR;
2812                 }
2813                 state = AdminProtos.GetRegionInfoResponse.CompactionState.MAJOR;
2814                 break;
2815               case MINOR:
2816                 if (state == AdminProtos.GetRegionInfoResponse.CompactionState.MAJOR) {
2817                   return CompactionState.MAJOR_AND_MINOR;
2818                 }
2819                 state = AdminProtos.GetRegionInfoResponse.CompactionState.MINOR;
2820                 break;
2821               case NONE:
2822               default: // nothing, continue
2823             }
2824           } catch (NotServingRegionException e) {
2825             if (LOG.isDebugEnabled()) {
2826               LOG.debug("Trying to get compaction state of " + loc.getRegion() + ": " +
2827                 StringUtils.stringifyException(e));
2828             }
2829           } catch (RemoteException e) {
2830             if (e.getMessage().indexOf(NotServingRegionException.class.getName()) >= 0) {
2831               if (LOG.isDebugEnabled()) {
2832                 LOG.debug("Trying to get compaction state of " + loc.getRegion() + ": " +
2833                   StringUtils.stringifyException(e));
2834               }
2835             } else {
2836               throw e;
2837             }
2838           }
2839         }
2840         break;
2841       default:
2842         throw new IllegalArgumentException("Unknown compactType: " + compactType);
2843     }
2844     if (state != null) {
2845       return ProtobufUtil.createCompactionState(state);
2846     }
2847     return null;
2848   }
2849
2850   /**
2851    * Future that waits on a procedure result.
2852    * Returned by the async version of the Admin calls,
2853    * and used internally by the sync calls to wait on the result of the procedure.
2854    */
2855   @InterfaceAudience.Private
2856   @InterfaceStability.Evolving
2857   protected static class ProcedureFuture<V> implements Future<V> {
2858     private ExecutionException exception = null;
2859     private boolean procResultFound = false;
2860     private boolean done = false;
2861     private boolean cancelled = false;
2862     private V result = null;
2863
2864     private final HBaseAdmin admin;
2865     protected final Long procId;
2866
2867     public ProcedureFuture(final HBaseAdmin admin, final Long procId) {
2868       this.admin = admin;
2869       this.procId = procId;
2870     }
2871
2872     @Override
2873     public boolean cancel(boolean mayInterruptIfRunning) {
2874       AbortProcedureRequest abortProcRequest = AbortProcedureRequest.newBuilder()
2875           .setProcId(procId).setMayInterruptIfRunning(mayInterruptIfRunning).build();
2876       try {
2877         cancelled = abortProcedureResult(abortProcRequest).getIsProcedureAborted();
2878         if (cancelled) {
2879           done = true;
2880         }
2881       } catch (IOException e) {
2882         // Cancell thrown exception for some reason. At this time, we are not sure whether
2883         // the cancell succeeds or fails. We assume that it is failed, but print out a warning
2884         // for debugging purpose.
2885         LOG.warn(
2886           "Cancelling the procedure with procId=" + procId + " throws exception " + e.getMessage(),
2887           e);
2888         cancelled = false;
2889       }
2890       return cancelled;
2891     }
2892
2893     @Override
2894     public boolean isCancelled() {
2895       return cancelled;
2896     }
2897
2898     protected AbortProcedureResponse abortProcedureResult(
2899         final AbortProcedureRequest request) throws IOException {
2900       return admin.executeCallable(new MasterCallable<AbortProcedureResponse>(
2901           admin.getConnection(), admin.getRpcControllerFactory()) {
2902         @Override
2903         protected AbortProcedureResponse rpcCall() throws Exception {
2904           return master.abortProcedure(getRpcController(), request);
2905         }
2906       });
2907     }
2908
2909     @Override
2910     public V get() throws InterruptedException, ExecutionException {
2911       // TODO: should we ever spin forever?
2912       // fix HBASE-21715. TODO: If the function call get() without timeout limit is not allowed,
2913       // is it possible to compose instead of inheriting from the class Future for this class?
2914       try {
2915         return get(admin.getProcedureTimeout, TimeUnit.MILLISECONDS);
2916       } catch (TimeoutException e) {
2917         LOG.warn("Failed to get the procedure with procId=" + procId + " throws exception " + e
2918             .getMessage(), e);
2919         return null;
2920       }
2921     }
2922
2923     @Override
2924     public V get(long timeout, TimeUnit unit)
2925         throws InterruptedException, ExecutionException, TimeoutException {
2926       if (!done) {
2927         long deadlineTs = EnvironmentEdgeManager.currentTime() + unit.toMillis(timeout);
2928         try {
2929           try {
2930             // if the master support procedures, try to wait the result
2931             if (procId != null) {
2932               result = waitProcedureResult(procId, deadlineTs);
2933             }
2934             // if we don't have a proc result, try the compatibility wait
2935             if (!procResultFound) {
2936               result = waitOperationResult(deadlineTs);
2937             }
2938             result = postOperationResult(result, deadlineTs);
2939             done = true;
2940           } catch (IOException e) {
2941             result = postOperationFailure(e, deadlineTs);
2942             done = true;
2943           }
2944         } catch (IOException e) {
2945           exception = new ExecutionException(e);
2946           done = true;
2947         }
2948       }
2949       if (exception != null) {
2950         throw exception;
2951       }
2952       return result;
2953     }
2954
2955     @Override
2956     public boolean isDone() {
2957       return done;
2958     }
2959
2960     protected HBaseAdmin getAdmin() {
2961       return admin;
2962     }
2963
2964     private V waitProcedureResult(long procId, long deadlineTs)
2965         throws IOException, TimeoutException, InterruptedException {
2966       GetProcedureResultRequest request = GetProcedureResultRequest.newBuilder()
2967           .setProcId(procId)
2968           .build();
2969
2970       int tries = 0;
2971       IOException serviceEx = null;
2972       while (EnvironmentEdgeManager.currentTime() < deadlineTs) {
2973         GetProcedureResultResponse response = null;
2974         try {
2975           // Try to fetch the result
2976           response = getProcedureResult(request);
2977         } catch (IOException e) {
2978           serviceEx = unwrapException(e);
2979
2980           // the master may be down
2981           LOG.warn("failed to get the procedure result procId=" + procId, serviceEx);
2982
2983           // Not much to do, if we have a DoNotRetryIOException
2984           if (serviceEx instanceof DoNotRetryIOException) {
2985             // TODO: looks like there is no way to unwrap this exception and get the proper
2986             // UnsupportedOperationException aside from looking at the message.
2987             // anyway, if we fail here we just failover to the compatibility side
2988             // and that is always a valid solution.
2989             LOG.warn("Proc-v2 is unsupported on this master: " + serviceEx.getMessage(), serviceEx);
2990             procResultFound = false;
2991             return null;
2992           }
2993         }
2994
2995         // If the procedure is no longer running, we should have a result
2996         if (response != null && response.getState() != GetProcedureResultResponse.State.RUNNING) {
2997           procResultFound = response.getState() != GetProcedureResultResponse.State.NOT_FOUND;
2998           return convertResult(response);
2999         }
3000
3001         try {
3002           Thread.sleep(getAdmin().getPauseTime(tries++));
3003         } catch (InterruptedException e) {
3004           throw new InterruptedException(
3005             "Interrupted while waiting for the result of proc " + procId);
3006         }
3007       }
3008       if (serviceEx != null) {
3009         throw serviceEx;
3010       } else {
3011         throw new TimeoutException("The procedure " + procId + " is still running");
3012       }
3013     }
3014
3015     private static IOException unwrapException(IOException e) {
3016       if (e instanceof RemoteException) {
3017         return ((RemoteException)e).unwrapRemoteException();
3018       }
3019       return e;
3020     }
3021
3022     protected GetProcedureResultResponse getProcedureResult(final GetProcedureResultRequest request)
3023         throws IOException {
3024       return admin.executeCallable(new MasterCallable<GetProcedureResultResponse>(
3025           admin.getConnection(), admin.getRpcControllerFactory()) {
3026         @Override
3027         protected GetProcedureResultResponse rpcCall() throws Exception {
3028           return master.getProcedureResult(getRpcController(), request);
3029         }
3030       });
3031     }
3032
3033     /**
3034      * Convert the procedure result response to a specified type.
3035      * @param response the procedure result object to parse
3036      * @return the result data of the procedure.
3037      */
3038     protected V convertResult(final GetProcedureResultResponse response) throws IOException {
3039       if (response.hasException()) {
3040         throw ForeignExceptionUtil.toIOException(response.getException());
3041       }
3042       return null;
3043     }
3044
3045     /**
3046      * Fallback implementation in case the procedure is not supported by the server.
3047      * It should try to wait until the operation is completed.
3048      * @param deadlineTs the timestamp after which this method should throw a TimeoutException
3049      * @return the result data of the operation
3050      */
3051     protected V waitOperationResult(final long deadlineTs)
3052         throws IOException, TimeoutException {
3053       return null;
3054     }
3055
3056     /**
3057      * Called after the operation is completed and the result fetched. this allows to perform extra
3058      * steps after the procedure is completed. it allows to apply transformations to the result that
3059      * will be returned by get().
3060      * @param result the result of the procedure
3061      * @param deadlineTs the timestamp after which this method should throw a TimeoutException
3062      * @return the result of the procedure, which may be the same as the passed one
3063      */
3064     protected V postOperationResult(final V result, final long deadlineTs)
3065         throws IOException, TimeoutException {
3066       return result;
3067     }
3068
3069     /**
3070      * Called after the operation is terminated with a failure.
3071      * this allows to perform extra steps after the procedure is terminated.
3072      * it allows to apply transformations to the result that will be returned by get().
3073      * The default implementation will rethrow the exception
3074      * @param exception the exception got from fetching the result
3075      * @param deadlineTs the timestamp after which this method should throw a TimeoutException
3076      * @return the result of the procedure, which may be the same as the passed one
3077      */
3078     protected V postOperationFailure(final IOException exception, final long deadlineTs)
3079         throws IOException, TimeoutException {
3080       throw exception;
3081     }
3082
3083     protected interface WaitForStateCallable {
3084       boolean checkState(int tries) throws IOException;
3085       void throwInterruptedException() throws InterruptedIOException;
3086       void throwTimeoutException(long elapsed) throws TimeoutException;
3087     }
3088
3089     protected void waitForState(final long deadlineTs, final WaitForStateCallable callable)
3090         throws IOException, TimeoutException {
3091       int tries = 0;
3092       IOException serverEx = null;
3093       long startTime = EnvironmentEdgeManager.currentTime();
3094       while (EnvironmentEdgeManager.currentTime() < deadlineTs) {
3095         serverEx = null;
3096         try {
3097           if (callable.checkState(tries)) {
3098             return;
3099           }
3100         } catch (IOException e) {
3101           serverEx = e;
3102         }
3103         try {
3104           Thread.sleep(getAdmin().getPauseTime(tries++));
3105         } catch (InterruptedException e) {
3106           callable.throwInterruptedException();
3107         }
3108       }
3109       if (serverEx != null) {
3110         throw unwrapException(serverEx);
3111       } else {
3112         callable.throwTimeoutException(EnvironmentEdgeManager.currentTime() - startTime);
3113       }
3114     }
3115   }
3116
3117   @InterfaceAudience.Private
3118   @InterfaceStability.Evolving
3119   protected static abstract class TableFuture<V> extends ProcedureFuture<V> {
3120     private final TableName tableName;
3121
3122     public TableFuture(final HBaseAdmin admin, final TableName tableName, final Long procId) {
3123       super(admin, procId);
3124       this.tableName = tableName;
3125     }
3126
3127     @Override
3128     public String toString() {
3129       return getDescription();
3130     }
3131
3132     /**
3133      * @return the table name
3134      */
3135     protected TableName getTableName() {
3136       return tableName;
3137     }
3138
3139     /**
3140      * @return the table descriptor
3141      */
3142     protected TableDescriptor getDescriptor() throws IOException {
3143       return getAdmin().getDescriptor(getTableName());
3144     }
3145
3146     /**
3147      * @return the operation type like CREATE, DELETE, DISABLE etc.
3148      */
3149     public abstract String getOperationType();
3150
3151     /**
3152      * @return a description of the operation
3153      */
3154     protected String getDescription() {
3155       return "Operation: " + getOperationType() + ", " + "Table Name: " +
3156         tableName.getNameWithNamespaceInclAsString() + ", procId: " + procId;
3157     }
3158
3159     protected abstract class TableWaitForStateCallable implements WaitForStateCallable {
3160       @Override
3161       public void throwInterruptedException() throws InterruptedIOException {
3162         throw new InterruptedIOException("Interrupted while waiting for " + getDescription());
3163       }
3164
3165       @Override
3166       public void throwTimeoutException(long elapsedTime) throws TimeoutException {
3167         throw new TimeoutException(
3168           getDescription() + " has not completed after " + elapsedTime + "ms");
3169       }
3170     }
3171
3172     @Override
3173     protected V postOperationResult(final V result, final long deadlineTs)
3174         throws IOException, TimeoutException {
3175       LOG.info(getDescription() + " completed");
3176       return super.postOperationResult(result, deadlineTs);
3177     }
3178
3179     @Override
3180     protected V postOperationFailure(final IOException exception, final long deadlineTs)
3181         throws IOException, TimeoutException {
3182       LOG.info(getDescription() + " failed with " + exception.getMessage());
3183       return super.postOperationFailure(exception, deadlineTs);
3184     }
3185
3186     protected void waitForTableEnabled(final long deadlineTs)
3187         throws IOException, TimeoutException {
3188       waitForState(deadlineTs, new TableWaitForStateCallable() {
3189         @Override
3190         public boolean checkState(int tries) throws IOException {
3191           try {
3192             if (getAdmin().isTableAvailable(tableName)) {
3193               return true;
3194             }
3195           } catch (TableNotFoundException tnfe) {
3196             LOG.debug("Table " + tableName.getNameWithNamespaceInclAsString()
3197                 + " was not enabled, sleeping. tries=" + tries);
3198           }
3199           return false;
3200         }
3201       });
3202     }
3203
3204     protected void waitForTableDisabled(final long deadlineTs)
3205         throws IOException, TimeoutException {
3206       waitForState(deadlineTs, new TableWaitForStateCallable() {
3207         @Override
3208         public boolean checkState(int tries) throws IOException {
3209           return getAdmin().isTableDisabled(tableName);
3210         }
3211       });
3212     }
3213
3214     protected void waitTableNotFound(final long deadlineTs)
3215         throws IOException, TimeoutException {
3216       waitForState(deadlineTs, new TableWaitForStateCallable() {
3217         @Override
3218         public boolean checkState(int tries) throws IOException {
3219           return !getAdmin().tableExists(tableName);
3220         }
3221       });
3222     }
3223
3224     protected void waitForAllRegionsOnline(final long deadlineTs, final byte[][] splitKeys)
3225         throws IOException, TimeoutException {
3226       final TableDescriptor desc = getDescriptor();
3227       final AtomicInteger actualRegCount = new AtomicInteger(0);
3228       final MetaTableAccessor.Visitor visitor = new MetaTableAccessor.Visitor() {
3229         @Override
3230         public boolean visit(Result rowResult) throws IOException {
3231           RegionLocations list = MetaTableAccessor.getRegionLocations(rowResult);
3232           if (list == null) {
3233             LOG.warn("No serialized HRegionInfo in " + rowResult);
3234             return true;
3235           }
3236           HRegionLocation l = list.getRegionLocation();
3237           if (l == null) {
3238             return true;
3239           }
3240           if (!l.getRegion().getTable().equals(desc.getTableName())) {
3241             return false;
3242           }
3243           if (l.getRegion().isOffline() || l.getRegion().isSplit()) {
3244             return true;
3245           }
3246           HRegionLocation[] locations = list.getRegionLocations();
3247           for (HRegionLocation location : locations) {
3248             if (location == null) continue;
3249             ServerName serverName = location.getServerName();
3250             // Make sure that regions are assigned to server
3251             if (serverName != null && serverName.getAddress() != null) {
3252               actualRegCount.incrementAndGet();
3253             }
3254           }
3255           return true;
3256         }
3257       };
3258
3259       int tries = 0;
3260       int numRegs = (splitKeys == null ? 1 : splitKeys.length + 1) * desc.getRegionReplication();
3261       while (EnvironmentEdgeManager.currentTime() < deadlineTs) {
3262         actualRegCount.set(0);
3263         MetaTableAccessor.scanMetaForTableRegions(getAdmin().getConnection(), visitor,
3264           desc.getTableName());
3265         if (actualRegCount.get() == numRegs) {
3266           // all the regions are online
3267           return;
3268         }
3269
3270         try {
3271           Thread.sleep(getAdmin().getPauseTime(tries++));
3272         } catch (InterruptedException e) {
3273           throw new InterruptedIOException("Interrupted when opening" + " regions; "
3274               + actualRegCount.get() + " of " + numRegs + " regions processed so far");
3275         }
3276       }
3277       throw new TimeoutException("Only " + actualRegCount.get() + " of " + numRegs
3278           + " regions are online; retries exhausted.");
3279     }
3280   }
3281
3282   @InterfaceAudience.Private
3283   @InterfaceStability.Evolving
3284   protected static abstract class NamespaceFuture extends ProcedureFuture<Void> {
3285     private final String namespaceName;
3286
3287     public NamespaceFuture(final HBaseAdmin admin, final String namespaceName, final Long procId) {
3288       super(admin, procId);
3289       this.namespaceName = namespaceName;
3290     }
3291
3292     /**
3293      * @return the namespace name
3294      */
3295     protected String getNamespaceName() {
3296       return namespaceName;
3297     }
3298
3299     /**
3300      * @return the operation type like CREATE_NAMESPACE, DELETE_NAMESPACE, etc.
3301      */
3302     public abstract String getOperationType();
3303
3304     @Override
3305     public String toString() {
3306       return "Operation: " + getOperationType() + ", Namespace: " + getNamespaceName();
3307     }
3308   }
3309
3310   @InterfaceAudience.Private
3311   @InterfaceStability.Evolving
3312   private static class ReplicationFuture extends ProcedureFuture<Void> {
3313     private final String peerId;
3314     private final Supplier<String> getOperation;
3315
3316     public ReplicationFuture(HBaseAdmin admin, String peerId, Long procId,
3317         Supplier<String> getOperation) {
3318       super(admin, procId);
3319       this.peerId = peerId;
3320       this.getOperation = getOperation;
3321     }
3322
3323     @Override
3324     public String toString() {
3325       return "Operation: " + getOperation.get() + ", peerId: " + peerId;
3326     }
3327   }
3328
3329   @Override
3330   public List<SecurityCapability> getSecurityCapabilities() throws IOException {
3331     try {
3332       return executeCallable(new MasterCallable<List<SecurityCapability>>(getConnection(),
3333           getRpcControllerFactory()) {
3334         @Override
3335         protected List<SecurityCapability> rpcCall() throws Exception {
3336           SecurityCapabilitiesRequest req = SecurityCapabilitiesRequest.newBuilder().build();
3337           return ProtobufUtil.toSecurityCapabilityList(
3338             master.getSecurityCapabilities(getRpcController(), req).getCapabilitiesList());
3339         }
3340       });
3341     } catch (IOException e) {
3342       if (e instanceof RemoteException) {
3343         e = ((RemoteException)e).unwrapRemoteException();
3344       }
3345       throw e;
3346     }
3347   }
3348
3349   @Override
3350   public boolean splitSwitch(boolean enabled, boolean synchronous) throws IOException {
3351     return splitOrMergeSwitch(enabled, synchronous, MasterSwitchType.SPLIT);
3352   }
3353
3354   @Override
3355   public boolean mergeSwitch(boolean enabled, boolean synchronous) throws IOException {
3356     return splitOrMergeSwitch(enabled, synchronous, MasterSwitchType.MERGE);
3357   }
3358
3359   private boolean splitOrMergeSwitch(boolean enabled, boolean synchronous,
3360       MasterSwitchType switchType) throws IOException {
3361     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
3362       @Override
3363       protected Boolean rpcCall() throws Exception {
3364         MasterProtos.SetSplitOrMergeEnabledResponse response = master.setSplitOrMergeEnabled(
3365           getRpcController(),
3366           RequestConverter.buildSetSplitOrMergeEnabledRequest(enabled, synchronous, switchType));
3367         return response.getPrevValueList().get(0);
3368       }
3369     });
3370   }
3371
3372   @Override
3373   public boolean isSplitEnabled() throws IOException {
3374     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
3375       @Override
3376       protected Boolean rpcCall() throws Exception {
3377         return master.isSplitOrMergeEnabled(getRpcController(),
3378           RequestConverter.buildIsSplitOrMergeEnabledRequest(MasterSwitchType.SPLIT)).getEnabled();
3379       }
3380     });
3381   }
3382
3383   @Override
3384   public boolean isMergeEnabled() throws IOException {
3385     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
3386       @Override
3387       protected Boolean rpcCall() throws Exception {
3388         return master.isSplitOrMergeEnabled(getRpcController(),
3389           RequestConverter.buildIsSplitOrMergeEnabledRequest(MasterSwitchType.MERGE)).getEnabled();
3390       }
3391     });
3392   }
3393
3394   private RpcControllerFactory getRpcControllerFactory() {
3395     return this.rpcControllerFactory;
3396   }
3397
3398   @Override
3399   public Future<Void> addReplicationPeerAsync(String peerId, ReplicationPeerConfig peerConfig,
3400       boolean enabled) throws IOException {
3401     AddReplicationPeerResponse response = executeCallable(
3402       new MasterCallable<AddReplicationPeerResponse>(getConnection(), getRpcControllerFactory()) {
3403         @Override
3404         protected AddReplicationPeerResponse rpcCall() throws Exception {
3405           return master.addReplicationPeer(getRpcController(),
3406             RequestConverter.buildAddReplicationPeerRequest(peerId, peerConfig, enabled));
3407         }
3408       });
3409     return new ReplicationFuture(this, peerId, response.getProcId(), () -> "ADD_REPLICATION_PEER");
3410   }
3411
3412   @Override
3413   public Future<Void> removeReplicationPeerAsync(String peerId) throws IOException {
3414     RemoveReplicationPeerResponse response =
3415       executeCallable(new MasterCallable<RemoveReplicationPeerResponse>(getConnection(),
3416           getRpcControllerFactory()) {
3417         @Override
3418         protected RemoveReplicationPeerResponse rpcCall() throws Exception {
3419           return master.removeReplicationPeer(getRpcController(),
3420             RequestConverter.buildRemoveReplicationPeerRequest(peerId));
3421         }
3422       });
3423     return new ReplicationFuture(this, peerId, response.getProcId(),
3424       () -> "REMOVE_REPLICATION_PEER");
3425   }
3426
3427   @Override
3428   public Future<Void> enableReplicationPeerAsync(final String peerId) throws IOException {
3429     EnableReplicationPeerResponse response =
3430       executeCallable(new MasterCallable<EnableReplicationPeerResponse>(getConnection(),
3431           getRpcControllerFactory()) {
3432         @Override
3433         protected EnableReplicationPeerResponse rpcCall() throws Exception {
3434           return master.enableReplicationPeer(getRpcController(),
3435             RequestConverter.buildEnableReplicationPeerRequest(peerId));
3436         }
3437       });
3438     return new ReplicationFuture(this, peerId, response.getProcId(),
3439       () -> "ENABLE_REPLICATION_PEER");
3440   }
3441
3442   @Override
3443   public Future<Void> disableReplicationPeerAsync(final String peerId) throws IOException {
3444     DisableReplicationPeerResponse response =
3445       executeCallable(new MasterCallable<DisableReplicationPeerResponse>(getConnection(),
3446           getRpcControllerFactory()) {
3447         @Override
3448         protected DisableReplicationPeerResponse rpcCall() throws Exception {
3449           return master.disableReplicationPeer(getRpcController(),
3450             RequestConverter.buildDisableReplicationPeerRequest(peerId));
3451         }
3452       });
3453     return new ReplicationFuture(this, peerId, response.getProcId(),
3454       () -> "DISABLE_REPLICATION_PEER");
3455   }
3456
3457   @Override
3458   public ReplicationPeerConfig getReplicationPeerConfig(final String peerId) throws IOException {
3459     return executeCallable(new MasterCallable<ReplicationPeerConfig>(getConnection(),
3460         getRpcControllerFactory()) {
3461       @Override
3462       protected ReplicationPeerConfig rpcCall() throws Exception {
3463         GetReplicationPeerConfigResponse response = master.getReplicationPeerConfig(
3464           getRpcController(), RequestConverter.buildGetReplicationPeerConfigRequest(peerId));
3465         return ReplicationPeerConfigUtil.convert(response.getPeerConfig());
3466       }
3467     });
3468   }
3469
3470   @Override
3471   public Future<Void> updateReplicationPeerConfigAsync(final String peerId,
3472       final ReplicationPeerConfig peerConfig) throws IOException {
3473     UpdateReplicationPeerConfigResponse response =
3474       executeCallable(new MasterCallable<UpdateReplicationPeerConfigResponse>(getConnection(),
3475           getRpcControllerFactory()) {
3476         @Override
3477         protected UpdateReplicationPeerConfigResponse rpcCall() throws Exception {
3478           return master.updateReplicationPeerConfig(getRpcController(),
3479             RequestConverter.buildUpdateReplicationPeerConfigRequest(peerId, peerConfig));
3480         }
3481       });
3482     return new ReplicationFuture(this, peerId, response.getProcId(),
3483       () -> "UPDATE_REPLICATION_PEER_CONFIG");
3484   }
3485
3486   @Override
3487   public Future<Void> transitReplicationPeerSyncReplicationStateAsync(String peerId,
3488       SyncReplicationState state) throws IOException {
3489     TransitReplicationPeerSyncReplicationStateResponse response =
3490         executeCallable(new MasterCallable<TransitReplicationPeerSyncReplicationStateResponse>(
3491           getConnection(), getRpcControllerFactory()) {
3492           @Override
3493           protected TransitReplicationPeerSyncReplicationStateResponse rpcCall() throws Exception {
3494             return master.transitReplicationPeerSyncReplicationState(getRpcController(),
3495               RequestConverter.buildTransitReplicationPeerSyncReplicationStateRequest(peerId,
3496                 state));
3497           }
3498         });
3499     return new ReplicationFuture(this, peerId, response.getProcId(),
3500       () -> "TRANSIT_REPLICATION_PEER_SYNCHRONOUS_REPLICATION_STATE");
3501   }
3502
3503   @Override
3504   public List<ReplicationPeerDescription> listReplicationPeers() throws IOException {
3505     return listReplicationPeers((Pattern)null);
3506   }
3507
3508   @Override
3509   public List<ReplicationPeerDescription> listReplicationPeers(Pattern pattern)
3510       throws IOException {
3511     return executeCallable(new MasterCallable<List<ReplicationPeerDescription>>(getConnection(),
3512         getRpcControllerFactory()) {
3513       @Override
3514       protected List<ReplicationPeerDescription> rpcCall() throws Exception {
3515         List<ReplicationProtos.ReplicationPeerDescription> peersList = master.listReplicationPeers(
3516           getRpcController(), RequestConverter.buildListReplicationPeersRequest(pattern))
3517             .getPeerDescList();
3518         List<ReplicationPeerDescription> result = new ArrayList<>(peersList.size());
3519         for (ReplicationProtos.ReplicationPeerDescription peer : peersList) {
3520           result.add(ReplicationPeerConfigUtil.toReplicationPeerDescription(peer));
3521         }
3522         return result;
3523       }
3524     });
3525   }
3526
3527   @Override
3528   public void decommissionRegionServers(List<ServerName> servers, boolean offload)
3529       throws IOException {
3530     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
3531       @Override
3532       public Void rpcCall() throws ServiceException {
3533         master.decommissionRegionServers(getRpcController(),
3534           RequestConverter.buildDecommissionRegionServersRequest(servers, offload));
3535         return null;
3536       }
3537     });
3538   }
3539
3540   @Override
3541   public List<ServerName> listDecommissionedRegionServers() throws IOException {
3542     return executeCallable(new MasterCallable<List<ServerName>>(getConnection(),
3543               getRpcControllerFactory()) {
3544       @Override
3545       public List<ServerName> rpcCall() throws ServiceException {
3546         ListDecommissionedRegionServersRequest req =
3547             ListDecommissionedRegionServersRequest.newBuilder().build();
3548         List<ServerName> servers = new ArrayList<>();
3549         for (HBaseProtos.ServerName server : master
3550             .listDecommissionedRegionServers(getRpcController(), req).getServerNameList()) {
3551           servers.add(ProtobufUtil.toServerName(server));
3552         }
3553         return servers;
3554       }
3555     });
3556   }
3557
3558   @Override
3559   public void recommissionRegionServer(ServerName server, List<byte[]> encodedRegionNames)
3560       throws IOException {
3561     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
3562       @Override
3563       public Void rpcCall() throws ServiceException {
3564         master.recommissionRegionServer(getRpcController(),
3565           RequestConverter.buildRecommissionRegionServerRequest(server, encodedRegionNames));
3566         return null;
3567       }
3568     });
3569   }
3570
3571   @Override
3572   public List<TableCFs> listReplicatedTableCFs() throws IOException {
3573     List<TableCFs> replicatedTableCFs = new ArrayList<>();
3574     List<TableDescriptor> tables = listTableDescriptors();
3575     tables.forEach(table -> {
3576       Map<String, Integer> cfs = new HashMap<>();
3577       Stream.of(table.getColumnFamilies())
3578           .filter(column -> column.getScope() != HConstants.REPLICATION_SCOPE_LOCAL)
3579           .forEach(column -> {
3580             cfs.put(column.getNameAsString(), column.getScope());
3581           });
3582       if (!cfs.isEmpty()) {
3583         replicatedTableCFs.add(new TableCFs(table.getTableName(), cfs));
3584       }
3585     });
3586     return replicatedTableCFs;
3587   }
3588
3589   @Override
3590   public void enableTableReplication(final TableName tableName) throws IOException {
3591     if (tableName == null) {
3592       throw new IllegalArgumentException("Table name cannot be null");
3593     }
3594     if (!tableExists(tableName)) {
3595       throw new TableNotFoundException("Table '" + tableName.getNameAsString()
3596           + "' does not exists.");
3597     }
3598     byte[][] splits = getTableSplits(tableName);
3599     checkAndSyncTableDescToPeers(tableName, splits);
3600     setTableRep(tableName, true);
3601   }
3602
3603   @Override
3604   public void disableTableReplication(final TableName tableName) throws IOException {
3605     if (tableName == null) {
3606       throw new IllegalArgumentException("Table name is null");
3607     }
3608     if (!tableExists(tableName)) {
3609       throw new TableNotFoundException("Table '" + tableName.getNameAsString()
3610           + "' does not exists.");
3611     }
3612     setTableRep(tableName, false);
3613   }
3614
3615   /**
3616    * Connect to peer and check the table descriptor on peer:
3617    * <ol>
3618    * <li>Create the same table on peer when not exist.</li>
3619    * <li>Throw an exception if the table already has replication enabled on any of the column
3620    * families.</li>
3621    * <li>Throw an exception if the table exists on peer cluster but descriptors are not same.</li>
3622    * </ol>
3623    * @param tableName name of the table to sync to the peer
3624    * @param splits table split keys
3625    * @throws IOException if a remote or network exception occurs
3626    */
3627   private void checkAndSyncTableDescToPeers(final TableName tableName, final byte[][] splits)
3628       throws IOException {
3629     List<ReplicationPeerDescription> peers = listReplicationPeers();
3630     if (peers == null || peers.size() <= 0) {
3631       throw new IllegalArgumentException("Found no peer cluster for replication.");
3632     }
3633
3634     for (ReplicationPeerDescription peerDesc : peers) {
3635       if (peerDesc.getPeerConfig().needToReplicate(tableName)) {
3636         Configuration peerConf =
3637             ReplicationPeerConfigUtil.getPeerClusterConfiguration(this.conf, peerDesc);
3638         try (Connection conn = ConnectionFactory.createConnection(peerConf);
3639             Admin repHBaseAdmin = conn.getAdmin()) {
3640           TableDescriptor tableDesc = getDescriptor(tableName);
3641           TableDescriptor peerTableDesc = null;
3642           if (!repHBaseAdmin.tableExists(tableName)) {
3643             repHBaseAdmin.createTable(tableDesc, splits);
3644           } else {
3645             peerTableDesc = repHBaseAdmin.getDescriptor(tableName);
3646             if (peerTableDesc == null) {
3647               throw new IllegalArgumentException("Failed to get table descriptor for table "
3648                   + tableName.getNameAsString() + " from peer cluster " + peerDesc.getPeerId());
3649             }
3650             if (TableDescriptor.COMPARATOR_IGNORE_REPLICATION.compare(peerTableDesc,
3651               tableDesc) != 0) {
3652               throw new IllegalArgumentException("Table " + tableName.getNameAsString()
3653                   + " exists in peer cluster " + peerDesc.getPeerId()
3654                   + ", but the table descriptors are not same when compared with source cluster."
3655                   + " Thus can not enable the table's replication switch.");
3656             }
3657           }
3658         }
3659       }
3660     }
3661   }
3662
3663   /**
3664    * Set the table's replication switch if the table's replication switch is already not set.
3665    * @param tableName name of the table
3666    * @param enableRep is replication switch enable or disable
3667    * @throws IOException if a remote or network exception occurs
3668    */
3669   private void setTableRep(final TableName tableName, boolean enableRep) throws IOException {
3670     TableDescriptor tableDesc = getDescriptor(tableName);
3671     if (!tableDesc.matchReplicationScope(enableRep)) {
3672       int scope =
3673           enableRep ? HConstants.REPLICATION_SCOPE_GLOBAL : HConstants.REPLICATION_SCOPE_LOCAL;
3674       modifyTable(TableDescriptorBuilder.newBuilder(tableDesc).setReplicationScope(scope).build());
3675     }
3676   }
3677
3678   @Override
3679   public void clearCompactionQueues(final ServerName sn, final Set<String> queues)
3680     throws IOException, InterruptedException {
3681     if (queues == null || queues.size() == 0) {
3682       throw new IllegalArgumentException("queues cannot be null or empty");
3683     }
3684     final AdminService.BlockingInterface admin = this.connection.getAdmin(sn);
3685     Callable<Void> callable = new Callable<Void>() {
3686       @Override
3687       public Void call() throws Exception {
3688         // TODO: There is no timeout on this controller. Set one!
3689         HBaseRpcController controller = rpcControllerFactory.newController();
3690         ClearCompactionQueuesRequest request =
3691                 RequestConverter.buildClearCompactionQueuesRequest(queues);
3692         admin.clearCompactionQueues(controller, request);
3693         return null;
3694       }
3695     };
3696     ProtobufUtil.call(callable);
3697   }
3698
3699   @Override
3700   public List<ServerName> clearDeadServers(List<ServerName> servers) throws IOException {
3701     return executeCallable(new MasterCallable<List<ServerName>>(getConnection(),
3702             getRpcControllerFactory()) {
3703       @Override
3704       protected List<ServerName> rpcCall() throws Exception {
3705         ClearDeadServersRequest req = RequestConverter.
3706           buildClearDeadServersRequest(servers == null? Collections.emptyList(): servers);
3707         return ProtobufUtil.toServerNameList(
3708                 master.clearDeadServers(getRpcController(), req).getServerNameList());
3709       }
3710     });
3711   }
3712
3713   @Override
3714   public void cloneTableSchema(final TableName tableName, final TableName newTableName,
3715       final boolean preserveSplits) throws IOException {
3716     checkTableExists(tableName);
3717     if (tableExists(newTableName)) {
3718       throw new TableExistsException(newTableName);
3719     }
3720     TableDescriptor htd = TableDescriptorBuilder.copy(newTableName, getDescriptor(tableName));
3721     if (preserveSplits) {
3722       createTable(htd, getTableSplits(tableName));
3723     } else {
3724       createTable(htd);
3725     }
3726   }
3727
3728   @Override
3729   public boolean switchRpcThrottle(final boolean enable) throws IOException {
3730     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
3731       @Override
3732       protected Boolean rpcCall() throws Exception {
3733         return this.master
3734             .switchRpcThrottle(getRpcController(), MasterProtos.SwitchRpcThrottleRequest
3735                 .newBuilder().setRpcThrottleEnabled(enable).build())
3736             .getPreviousRpcThrottleEnabled();
3737       }
3738     });
3739   }
3740
3741   @Override
3742   public boolean isRpcThrottleEnabled() throws IOException {
3743     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
3744       @Override
3745       protected Boolean rpcCall() throws Exception {
3746         return this.master.isRpcThrottleEnabled(getRpcController(),
3747           IsRpcThrottleEnabledRequest.newBuilder().build()).getRpcThrottleEnabled();
3748       }
3749     });
3750   }
3751
3752   @Override
3753   public boolean exceedThrottleQuotaSwitch(final boolean enable) throws IOException {
3754     return executeCallable(new MasterCallable<Boolean>(getConnection(), getRpcControllerFactory()) {
3755       @Override
3756       protected Boolean rpcCall() throws Exception {
3757         return this.master
3758             .switchExceedThrottleQuota(getRpcController(),
3759               MasterProtos.SwitchExceedThrottleQuotaRequest.newBuilder()
3760                   .setExceedThrottleQuotaEnabled(enable).build())
3761             .getPreviousExceedThrottleQuotaEnabled();
3762       }
3763     });
3764   }
3765
3766   @Override
3767   public Map<TableName, Long> getSpaceQuotaTableSizes() throws IOException {
3768     return executeCallable(
3769       new MasterCallable<Map<TableName, Long>>(getConnection(), getRpcControllerFactory()) {
3770         @Override
3771         protected Map<TableName, Long> rpcCall() throws Exception {
3772           GetSpaceQuotaRegionSizesResponse resp = master.getSpaceQuotaRegionSizes(
3773             getRpcController(), RequestConverter.buildGetSpaceQuotaRegionSizesRequest());
3774           Map<TableName, Long> tableSizes = new HashMap<>();
3775           for (RegionSizes sizes : resp.getSizesList()) {
3776             TableName tn = ProtobufUtil.toTableName(sizes.getTableName());
3777             tableSizes.put(tn, sizes.getSize());
3778           }
3779           return tableSizes;
3780         }
3781       });
3782   }
3783
3784   @Override
3785   public Map<TableName, SpaceQuotaSnapshot> getRegionServerSpaceQuotaSnapshots(
3786       ServerName serverName) throws IOException {
3787     final AdminService.BlockingInterface admin = this.connection.getAdmin(serverName);
3788     Callable<GetSpaceQuotaSnapshotsResponse> callable =
3789       new Callable<GetSpaceQuotaSnapshotsResponse>() {
3790         @Override
3791         public GetSpaceQuotaSnapshotsResponse call() throws Exception {
3792           return admin.getSpaceQuotaSnapshots(rpcControllerFactory.newController(),
3793             RequestConverter.buildGetSpaceQuotaSnapshotsRequest());
3794         }
3795       };
3796     GetSpaceQuotaSnapshotsResponse resp = ProtobufUtil.call(callable);
3797     Map<TableName, SpaceQuotaSnapshot> snapshots = new HashMap<>();
3798     for (TableQuotaSnapshot snapshot : resp.getSnapshotsList()) {
3799       snapshots.put(ProtobufUtil.toTableName(snapshot.getTableName()),
3800         SpaceQuotaSnapshot.toSpaceQuotaSnapshot(snapshot.getSnapshot()));
3801     }
3802     return snapshots;
3803   }
3804
3805   @Override
3806   public SpaceQuotaSnapshot getCurrentSpaceQuotaSnapshot(String namespace) throws IOException {
3807     return executeCallable(
3808       new MasterCallable<SpaceQuotaSnapshot>(getConnection(), getRpcControllerFactory()) {
3809         @Override
3810         protected SpaceQuotaSnapshot rpcCall() throws Exception {
3811           GetQuotaStatesResponse resp = master.getQuotaStates(getRpcController(),
3812             RequestConverter.buildGetQuotaStatesRequest());
3813           for (GetQuotaStatesResponse.NamespaceQuotaSnapshot nsSnapshot : resp
3814             .getNsSnapshotsList()) {
3815             if (namespace.equals(nsSnapshot.getNamespace())) {
3816               return SpaceQuotaSnapshot.toSpaceQuotaSnapshot(nsSnapshot.getSnapshot());
3817             }
3818           }
3819           return null;
3820         }
3821       });
3822   }
3823
3824   @Override
3825   public SpaceQuotaSnapshot getCurrentSpaceQuotaSnapshot(TableName tableName) throws IOException {
3826     return executeCallable(
3827       new MasterCallable<SpaceQuotaSnapshot>(getConnection(), getRpcControllerFactory()) {
3828         @Override
3829         protected SpaceQuotaSnapshot rpcCall() throws Exception {
3830           GetQuotaStatesResponse resp = master.getQuotaStates(getRpcController(),
3831             RequestConverter.buildGetQuotaStatesRequest());
3832           HBaseProtos.TableName protoTableName = ProtobufUtil.toProtoTableName(tableName);
3833           for (GetQuotaStatesResponse.TableQuotaSnapshot tableSnapshot : resp
3834             .getTableSnapshotsList()) {
3835             if (protoTableName.equals(tableSnapshot.getTableName())) {
3836               return SpaceQuotaSnapshot.toSpaceQuotaSnapshot(tableSnapshot.getSnapshot());
3837             }
3838           }
3839           return null;
3840         }
3841       });
3842   }
3843
3844   @Override
3845   public void grant(UserPermission userPermission, boolean mergeExistingPermissions)
3846       throws IOException {
3847     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
3848       @Override
3849       protected Void rpcCall() throws Exception {
3850         GrantRequest req =
3851             ShadedAccessControlUtil.buildGrantRequest(userPermission, mergeExistingPermissions);
3852         this.master.grant(getRpcController(), req);
3853         return null;
3854       }
3855     });
3856   }
3857
3858   @Override
3859   public void revoke(UserPermission userPermission) throws IOException {
3860     executeCallable(new MasterCallable<Void>(getConnection(), getRpcControllerFactory()) {
3861       @Override
3862       protected Void rpcCall() throws Exception {
3863         RevokeRequest req = ShadedAccessControlUtil.buildRevokeRequest(userPermission);
3864         this.master.revoke(getRpcController(), req);
3865         return null;
3866       }
3867     });
3868   }
3869
3870   @Override
3871   public List<UserPermission>
3872       getUserPermissions(GetUserPermissionsRequest getUserPermissionsRequest) throws IOException {
3873     return executeCallable(
3874       new MasterCallable<List<UserPermission>>(getConnection(), getRpcControllerFactory()) {
3875         @Override
3876         protected List<UserPermission> rpcCall() throws Exception {
3877           AccessControlProtos.GetUserPermissionsRequest req =
3878               ShadedAccessControlUtil.buildGetUserPermissionsRequest(getUserPermissionsRequest);
3879           AccessControlProtos.GetUserPermissionsResponse response =
3880               this.master.getUserPermissions(getRpcController(), req);
3881           return response.getUserPermissionList().stream()
3882               .map(userPermission -> ShadedAccessControlUtil.toUserPermission(userPermission))
3883               .collect(Collectors.toList());
3884         }
3885       });
3886   }
3887
3888   @Override
3889   public List<Boolean> hasUserPermissions(String userName, List<Permission> permissions)
3890       throws IOException {
3891     return executeCallable(
3892       new MasterCallable<List<Boolean>>(getConnection(), getRpcControllerFactory()) {
3893         @Override
3894         protected List<Boolean> rpcCall() throws Exception {
3895           HasUserPermissionsRequest request =
3896               ShadedAccessControlUtil.buildHasUserPermissionsRequest(userName, permissions);
3897           return this.master.hasUserPermissions(getRpcController(), request)
3898               .getHasUserPermissionList();
3899         }
3900       });
3901   }
3902
3903   @Override
3904   public void close() {
3905   }
3906
3907   @Override
3908   public Future<Void> splitRegionAsync(byte[] regionName) throws IOException {
3909     return splitRegionAsync(regionName, null);
3910   }
3911
3912   @Override
3913   public List<RegionMetrics> getRegionMetrics(ServerName serverName) throws IOException {
3914     return getRegionMetrics(serverName, null);
3915   }
3916
3917   @Override
3918   public Future<Void> createTableAsync(TableDescriptor desc) throws IOException {
3919     return createTableAsync(desc, null);
3920   }
3921 }