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