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