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