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