Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id 205A9200C78 for ; Thu, 18 May 2017 17:00:02 +0200 (CEST) Received: by cust-asf.ponee.io (Postfix) id 1EC7C160BA7; Thu, 18 May 2017 15:00:02 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id 8E568160BD5 for ; Thu, 18 May 2017 16:59:59 +0200 (CEST) Received: (qmail 28894 invoked by uid 500); 18 May 2017 14:59:53 -0000 Mailing-List: contact commits-help@hbase.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@hbase.apache.org Delivered-To: mailing list commits@hbase.apache.org Received: (qmail 28643 invoked by uid 99); 18 May 2017 14:59:53 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 18 May 2017 14:59:53 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 4DA8CE0210; Thu, 18 May 2017 14:59:53 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: git-site-role@apache.org To: commits@hbase.apache.org Date: Thu, 18 May 2017 14:59:55 -0000 Message-Id: In-Reply-To: <68236e0982844296ad03cf1a364a5256@git.apache.org> References: <68236e0982844296ad03cf1a364a5256@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: [03/51] [partial] hbase-site git commit: Published site at 82d554e3783372cc6b05489452c815b57c06f6cd. archived-at: Thu, 18 May 2017 15:00:02 -0000 http://git-wip-us.apache.org/repos/asf/hbase-site/blob/8b4cf63f/devapidocs/src-html/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.MasterRpcCall.html ---------------------------------------------------------------------- diff --git a/devapidocs/src-html/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.MasterRpcCall.html b/devapidocs/src-html/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.MasterRpcCall.html index 4fd4af0..1b2d845 100644 --- a/devapidocs/src-html/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.MasterRpcCall.html +++ b/devapidocs/src-html/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.MasterRpcCall.html @@ -186,2271 +186,2272 @@ 178import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.RemoveReplicationPeerResponse; 179import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.UpdateReplicationPeerConfigRequest; 180import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.UpdateReplicationPeerConfigResponse; -181import org.apache.hadoop.hbase.snapshot.ClientSnapshotDescriptionUtils; -182import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException; -183import org.apache.hadoop.hbase.snapshot.SnapshotCreationException; -184import org.apache.hadoop.hbase.util.Bytes; -185import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; -186import org.apache.hadoop.hbase.util.ForeignExceptionUtil; -187import org.apache.hadoop.hbase.util.Pair; -188 -189/** -190 * The implementation of AsyncAdmin. -191 */ -192@InterfaceAudience.Private -193@InterfaceStability.Evolving -194public class AsyncHBaseAdmin implements AsyncAdmin { -195 public static final String FLUSH_TABLE_PROCEDURE_SIGNATURE = "flush-table-proc"; -196 -197 private static final Log LOG = LogFactory.getLog(AsyncHBaseAdmin.class); -198 -199 private final AsyncConnectionImpl connection; -200 -201 private final RawAsyncTable metaTable; -202 -203 private final long rpcTimeoutNs; -204 -205 private final long operationTimeoutNs; -206 -207 private final long pauseNs; -208 -209 private final int maxAttempts; -210 -211 private final int startLogErrorsCnt; -212 -213 private final NonceGenerator ng; -214 -215 AsyncHBaseAdmin(AsyncConnectionImpl connection) { -216 this.connection = connection; -217 this.metaTable = connection.getRawTable(META_TABLE_NAME); -218 this.rpcTimeoutNs = connection.connConf.getRpcTimeoutNs(); -219 this.operationTimeoutNs = connection.connConf.getOperationTimeoutNs(); -220 this.pauseNs = connection.connConf.getPauseNs(); -221 this.maxAttempts = connection.connConf.getMaxRetries(); -222 this.startLogErrorsCnt = connection.connConf.getStartLogErrorsCnt(); -223 this.ng = connection.getNonceGenerator(); -224 } -225 -226 private <T> MasterRequestCallerBuilder<T> newMasterCaller() { -227 return this.connection.callerFactory.<T> masterRequest() -228 .rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS) -229 .operationTimeout(operationTimeoutNs, TimeUnit.NANOSECONDS) -230 .pause(pauseNs, TimeUnit.NANOSECONDS).maxAttempts(maxAttempts) -231 .startLogErrorsCnt(startLogErrorsCnt); -232 } -233 -234 private <T> AdminRequestCallerBuilder<T> newAdminCaller() { -235 return this.connection.callerFactory.<T> adminRequest() -236 .rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS) -237 .operationTimeout(operationTimeoutNs, TimeUnit.NANOSECONDS) -238 .pause(pauseNs, TimeUnit.NANOSECONDS).maxAttempts(maxAttempts) -239 .startLogErrorsCnt(startLogErrorsCnt); -240 } -241 -242 @FunctionalInterface -243 private interface MasterRpcCall<RESP, REQ> { -244 void call(MasterService.Interface stub, HBaseRpcController controller, REQ req, -245 RpcCallback<RESP> done); -246 } -247 -248 @FunctionalInterface -249 private interface AdminRpcCall<RESP, REQ> { -250 void call(AdminService.Interface stub, HBaseRpcController controller, REQ req, -251 RpcCallback<RESP> done); -252 } -253 -254 @FunctionalInterface -255 private interface Converter<D, S> { -256 D convert(S src) throws IOException; -257 } -258 -259 private <PREQ, PRESP, RESP> CompletableFuture<RESP> call(HBaseRpcController controller, -260 MasterService.Interface stub, PREQ preq, MasterRpcCall<PRESP, PREQ> rpcCall, -261 Converter<RESP, PRESP> respConverter) { -262 CompletableFuture<RESP> future = new CompletableFuture<>(); -263 rpcCall.call(stub, controller, preq, new RpcCallback<PRESP>() { -264 -265 @Override -266 public void run(PRESP resp) { -267 if (controller.failed()) { -268 future.completeExceptionally(controller.getFailed()); -269 } else { -270 try { -271 future.complete(respConverter.convert(resp)); -272 } catch (IOException e) { -273 future.completeExceptionally(e); -274 } -275 } -276 } -277 }); -278 return future; -279 } -280 -281 //TODO abstract call and adminCall into a single method. -282 private <PREQ, PRESP, RESP> CompletableFuture<RESP> adminCall(HBaseRpcController controller, -283 AdminService.Interface stub, PREQ preq, AdminRpcCall<PRESP, PREQ> rpcCall, -284 Converter<RESP, PRESP> respConverter) { -285 -286 CompletableFuture<RESP> future = new CompletableFuture<>(); -287 rpcCall.call(stub, controller, preq, new RpcCallback<PRESP>() { -288 -289 @Override -290 public void run(PRESP resp) { -291 if (controller.failed()) { -292 future.completeExceptionally(new IOException(controller.errorText())); -293 } else { -294 try { -295 future.complete(respConverter.convert(resp)); -296 } catch (IOException e) { -297 future.completeExceptionally(e); -298 } -299 } -300 } -301 }); -302 return future; -303 } -304 -305 private <PREQ, PRESP> CompletableFuture<Void> procedureCall(PREQ preq, -306 MasterRpcCall<PRESP, PREQ> rpcCall, Converter<Long, PRESP> respConverter, -307 ProcedureBiConsumer consumer) { -308 CompletableFuture<Long> procFuture = this -309 .<Long> newMasterCaller() -310 .action( -311 (controller, stub) -> this.<PREQ, PRESP, Long> call(controller, stub, preq, rpcCall, -312 respConverter)).call(); -313 return waitProcedureResult(procFuture).whenComplete(consumer); -314 } -315 -316 @FunctionalInterface -317 private interface TableOperator { -318 CompletableFuture<Void> operate(TableName table); -319 } -320 -321 private CompletableFuture<TableDescriptor[]> batchTableOperations(Pattern pattern, -322 TableOperator operator, String operationType) { -323 CompletableFuture<TableDescriptor[]> future = new CompletableFuture<>(); -324 List<TableDescriptor> failed = new LinkedList<>(); -325 listTables(pattern, false).whenComplete( -326 (tables, error) -> { -327 if (error != null) { -328 future.completeExceptionally(error); -329 return; -330 } -331 CompletableFuture[] futures = Arrays.stream(tables) -332 .map((table) -> operator.operate(table.getTableName()).whenComplete((v, ex) -> { -333 if (ex != null) { -334 LOG.info("Failed to " + operationType + " table " + table.getTableName(), ex); -335 failed.add(table); -336 } -337 })).<CompletableFuture> toArray(size -> new CompletableFuture[size]); -338 CompletableFuture.allOf(futures).thenAccept((v) -> { -339 future.complete(failed.toArray(new TableDescriptor[failed.size()])); -340 }); -341 }); -342 return future; -343 } -344 -345 @Override -346 public AsyncConnectionImpl getConnection() { -347 return this.connection; -348 } -349 -350 @Override -351 public CompletableFuture<Boolean> tableExists(TableName tableName) { -352 return AsyncMetaTableAccessor.tableExists(metaTable, tableName); -353 } -354 -355 @Override -356 public CompletableFuture<TableDescriptor[]> listTables() { -357 return listTables((Pattern) null, false); -358 } -359 -360 @Override -361 public CompletableFuture<TableDescriptor[]> listTables(String regex, boolean includeSysTables) { -362 return listTables(Pattern.compile(regex), false); -363 } -364 -365 @Override -366 public CompletableFuture<TableDescriptor[]> listTables(Pattern pattern, boolean includeSysTables) { -367 return this -368 .<TableDescriptor[]>newMasterCaller() -369 .action( -370 (controller, stub) -> this -371 .<GetTableDescriptorsRequest, GetTableDescriptorsResponse, TableDescriptor[]> call( -372 controller, stub, RequestConverter.buildGetTableDescriptorsRequest(pattern, -373 includeSysTables), (s, c, req, done) -> s.getTableDescriptors(c, req, done), ( -374 resp) -> ProtobufUtil.getTableDescriptorArray(resp))).call(); -375 } -376 -377 @Override -378 public CompletableFuture<TableName[]> listTableNames() { -379 return listTableNames((Pattern) null, false); -380 } -381 -382 @Override -383 public CompletableFuture<TableName[]> listTableNames(String regex, boolean includeSysTables) { -384 return listTableNames(Pattern.compile(regex), false); -385 } -386 -387 @Override -388 public CompletableFuture<TableName[]> listTableNames(Pattern pattern, boolean includeSysTables) { -389 return this -390 .<TableName[]>newMasterCaller() -391 .action( -392 (controller, stub) -> this -393 .<GetTableNamesRequest, GetTableNamesResponse, TableName[]> call(controller, stub, -394 RequestConverter.buildGetTableNamesRequest(pattern, includeSysTables), (s, c, req, -395 done) -> s.getTableNames(c, req, done), (resp) -> ProtobufUtil -396 .getTableNameArray(resp.getTableNamesList()))).call(); -397 } -398 -399 @Override -400 public CompletableFuture<TableDescriptor> getTableDescriptor(TableName tableName) { -401 CompletableFuture<TableDescriptor> future = new CompletableFuture<>(); -402 this.<List<TableSchema>> newMasterCaller() -403 .action( -404 (controller, stub) -> this -405 .<GetTableDescriptorsRequest, GetTableDescriptorsResponse, List<TableSchema>> call( -406 controller, stub, RequestConverter.buildGetTableDescriptorsRequest(tableName), (s, -407 c, req, done) -> s.getTableDescriptors(c, req, done), (resp) -> resp -408 .getTableSchemaList())).call().whenComplete((tableSchemas, error) -> { -409 if (error != null) { -410 future.completeExceptionally(error); -411 return; -412 } -413 if (!tableSchemas.isEmpty()) { -414 future.complete(ProtobufUtil.convertToTableDesc(tableSchemas.get(0))); -415 } else { -416 future.completeExceptionally(new TableNotFoundException(tableName.getNameAsString())); -417 } -418 }); -419 return future; -420 } -421 -422 @Override -423 public CompletableFuture<Void> createTable(TableDescriptor desc) { -424 return createTable(desc, null); -425 } -426 -427 @Override -428 public CompletableFuture<Void> createTable(TableDescriptor desc, byte[] startKey, byte[] endKey, -429 int numRegions) { -430 try { -431 return createTable(desc, getSplitKeys(startKey, endKey, numRegions)); -432 } catch (IllegalArgumentException e) { -433 return failedFuture(e); -434 } -435 } -436 -437 @Override -438 public CompletableFuture<Void> createTable(TableDescriptor desc, byte[][] splitKeys) { -439 if (desc.getTableName() == null) { -440 return failedFuture(new IllegalArgumentException("TableName cannot be null")); -441 } -442 if (splitKeys != null && splitKeys.length > 0) { -443 Arrays.sort(splitKeys, Bytes.BYTES_COMPARATOR); -444 // Verify there are no duplicate split keys -445 byte[] lastKey = null; -446 for (byte[] splitKey : splitKeys) { -447 if (Bytes.compareTo(splitKey, HConstants.EMPTY_BYTE_ARRAY) == 0) { -448 return failedFuture(new IllegalArgumentException( -449 "Empty split key must not be passed in the split keys.")); -450 } -451 if (lastKey != null && Bytes.equals(splitKey, lastKey)) { -452 return failedFuture(new IllegalArgumentException("All split keys must be unique, " -453 + "found duplicate: " + Bytes.toStringBinary(splitKey) + ", " -454 + Bytes.toStringBinary(lastKey))); -455 } -456 lastKey = splitKey; -457 } -458 } -459 -460 return this.<CreateTableRequest, CreateTableResponse> procedureCall( -461 RequestConverter.buildCreateTableRequest(desc, splitKeys, ng.getNonceGroup(), ng.newNonce()), -462 (s, c, req, done) -> s.createTable(c, req, done), (resp) -> resp.getProcId(), -463 new CreateTableProcedureBiConsumer(this, desc.getTableName())); -464 } -465 -466 @Override -467 public CompletableFuture<Void> deleteTable(TableName tableName) { -468 return this.<DeleteTableRequest, DeleteTableResponse> procedureCall(RequestConverter -469 .buildDeleteTableRequest(tableName, ng.getNonceGroup(), ng.newNonce()), -470 (s, c, req, done) -> s.deleteTable(c, req, done), (resp) -> resp.getProcId(), -471 new DeleteTableProcedureBiConsumer(this, tableName)); -472 } -473 -474 @Override -475 public CompletableFuture<TableDescriptor[]> deleteTables(String regex) { -476 return deleteTables(Pattern.compile(regex)); -477 } -478 -479 @Override -480 public CompletableFuture<TableDescriptor[]> deleteTables(Pattern pattern) { -481 return batchTableOperations(pattern, (table) -> deleteTable(table), "DELETE"); -482 } -483 -484 @Override -485 public CompletableFuture<Void> truncateTable(TableName tableName, boolean preserveSplits) { -486 return this.<TruncateTableRequest, TruncateTableResponse> procedureCall( -487 RequestConverter.buildTruncateTableRequest(tableName, preserveSplits, ng.getNonceGroup(), -488 ng.newNonce()), (s, c, req, done) -> s.truncateTable(c, req, done), -489 (resp) -> resp.getProcId(), new TruncateTableProcedureBiConsumer(this, tableName)); -490 } -491 -492 @Override -493 public CompletableFuture<Void> enableTable(TableName tableName) { -494 return this.<EnableTableRequest, EnableTableResponse> procedureCall(RequestConverter -495 .buildEnableTableRequest(tableName, ng.getNonceGroup(), ng.newNonce()), -496 (s, c, req, done) -> s.enableTable(c, req, done), (resp) -> resp.getProcId(), -497 new EnableTableProcedureBiConsumer(this, tableName)); -498 } -499 -500 @Override -501 public CompletableFuture<TableDescriptor[]> enableTables(String regex) { -502 return enableTables(Pattern.compile(regex)); -503 } -504 -505 @Override -506 public CompletableFuture<TableDescriptor[]> enableTables(Pattern pattern) { -507 return batchTableOperations(pattern, (table) -> enableTable(table), "ENABLE"); -508 } -509 -510 @Override -511 public CompletableFuture<Void> disableTable(TableName tableName) { -512 return this.<DisableTableRequest, DisableTableResponse> procedureCall(RequestConverter -513 .buildDisableTableRequest(tableName, ng.getNonceGroup(), ng.newNonce()), -514 (s, c, req, done) -> s.disableTable(c, req, done), (resp) -> resp.getProcId(), -515 new DisableTableProcedureBiConsumer(this, tableName)); -516 } -517 -518 @Override -519 public CompletableFuture<TableDescriptor[]> disableTables(String regex) { -520 return disableTables(Pattern.compile(regex)); -521 } -522 -523 @Override -524 public CompletableFuture<TableDescriptor[]> disableTables(Pattern pattern) { -525 return batchTableOperations(pattern, (table) -> disableTable(table), "DISABLE"); -526 } -527 +181import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos; +182import org.apache.hadoop.hbase.snapshot.ClientSnapshotDescriptionUtils; +183import org.apache.hadoop.hbase.snapshot.RestoreSnapshotException; +184import org.apache.hadoop.hbase.snapshot.SnapshotCreationException; +185import org.apache.hadoop.hbase.util.Bytes; +186import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +187import org.apache.hadoop.hbase.util.ForeignExceptionUtil; +188import org.apache.hadoop.hbase.util.Pair; +189 +190/** +191 * The implementation of AsyncAdmin. +192 */ +193@InterfaceAudience.Private +194@InterfaceStability.Evolving +195public class AsyncHBaseAdmin implements AsyncAdmin { +196 public static final String FLUSH_TABLE_PROCEDURE_SIGNATURE = "flush-table-proc"; +197 +198 private static final Log LOG = LogFactory.getLog(AsyncHBaseAdmin.class); +199 +200 private final AsyncConnectionImpl connection; +201 +202 private final RawAsyncTable metaTable; +203 +204 private final long rpcTimeoutNs; +205 +206 private final long operationTimeoutNs; +207 +208 private final long pauseNs; +209 +210 private final int maxAttempts; +211 +212 private final int startLogErrorsCnt; +213 +214 private final NonceGenerator ng; +215 +216 AsyncHBaseAdmin(AsyncConnectionImpl connection) { +217 this.connection = connection; +218 this.metaTable = connection.getRawTable(META_TABLE_NAME); +219 this.rpcTimeoutNs = connection.connConf.getRpcTimeoutNs(); +220 this.operationTimeoutNs = connection.connConf.getOperationTimeoutNs(); +221 this.pauseNs = connection.connConf.getPauseNs(); +222 this.maxAttempts = connection.connConf.getMaxRetries(); +223 this.startLogErrorsCnt = connection.connConf.getStartLogErrorsCnt(); +224 this.ng = connection.getNonceGenerator(); +225 } +226 +227 private <T> MasterRequestCallerBuilder<T> newMasterCaller() { +228 return this.connection.callerFactory.<T> masterRequest() +229 .rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS) +230 .operationTimeout(operationTimeoutNs, TimeUnit.NANOSECONDS) +231 .pause(pauseNs, TimeUnit.NANOSECONDS).maxAttempts(maxAttempts) +232 .startLogErrorsCnt(startLogErrorsCnt); +233 } +234 +235 private <T> AdminRequestCallerBuilder<T> newAdminCaller() { +236 return this.connection.callerFactory.<T> adminRequest() +237 .rpcTimeout(rpcTimeoutNs, TimeUnit.NANOSECONDS) +238 .operationTimeout(operationTimeoutNs, TimeUnit.NANOSECONDS) +239 .pause(pauseNs, TimeUnit.NANOSECONDS).maxAttempts(maxAttempts) +240 .startLogErrorsCnt(startLogErrorsCnt); +241 } +242 +243 @FunctionalInterface +244 private interface MasterRpcCall<RESP, REQ> { +245 void call(MasterService.Interface stub, HBaseRpcController controller, REQ req, +246 RpcCallback<RESP> done); +247 } +248 +249 @FunctionalInterface +250 private interface AdminRpcCall<RESP, REQ> { +251 void call(AdminService.Interface stub, HBaseRpcController controller, REQ req, +252 RpcCallback<RESP> done); +253 } +254 +255 @FunctionalInterface +256 private interface Converter<D, S> { +257 D convert(S src) throws IOException; +258 } +259 +260 private <PREQ, PRESP, RESP> CompletableFuture<RESP> call(HBaseRpcController controller, +261 MasterService.Interface stub, PREQ preq, MasterRpcCall<PRESP, PREQ> rpcCall, +262 Converter<RESP, PRESP> respConverter) { +263 CompletableFuture<RESP> future = new CompletableFuture<>(); +264 rpcCall.call(stub, controller, preq, new RpcCallback<PRESP>() { +265 +266 @Override +267 public void run(PRESP resp) { +268 if (controller.failed()) { +269 future.completeExceptionally(controller.getFailed()); +270 } else { +271 try { +272 future.complete(respConverter.convert(resp)); +273 } catch (IOException e) { +274 future.completeExceptionally(e); +275 } +276 } +277 } +278 }); +279 return future; +280 } +281 +282 //TODO abstract call and adminCall into a single method. +283 private <PREQ, PRESP, RESP> CompletableFuture<RESP> adminCall(HBaseRpcController controller, +284 AdminService.Interface stub, PREQ preq, AdminRpcCall<PRESP, PREQ> rpcCall, +285 Converter<RESP, PRESP> respConverter) { +286 +287 CompletableFuture<RESP> future = new CompletableFuture<>(); +288 rpcCall.call(stub, controller, preq, new RpcCallback<PRESP>() { +289 +290 @Override +291 public void run(PRESP resp) { +292 if (controller.failed()) { +293 future.completeExceptionally(new IOException(controller.errorText())); +294 } else { +295 try { +296 future.complete(respConverter.convert(resp)); +297 } catch (IOException e) { +298 future.completeExceptionally(e); +299 } +300 } +301 } +302 }); +303 return future; +304 } +305 +306 private <PREQ, PRESP> CompletableFuture<Void> procedureCall(PREQ preq, +307 MasterRpcCall<PRESP, PREQ> rpcCall, Converter<Long, PRESP> respConverter, +308 ProcedureBiConsumer consumer) { +309 CompletableFuture<Long> procFuture = this +310 .<Long> newMasterCaller() +311 .action( +312 (controller, stub) -> this.<PREQ, PRESP, Long> call(controller, stub, preq, rpcCall, +313 respConverter)).call(); +314 return waitProcedureResult(procFuture).whenComplete(consumer); +315 } +316 +317 @FunctionalInterface +318 private interface TableOperator { +319 CompletableFuture<Void> operate(TableName table); +320 } +321 +322 private CompletableFuture<TableDescriptor[]> batchTableOperations(Pattern pattern, +323 TableOperator operator, String operationType) { +324 CompletableFuture<TableDescriptor[]> future = new CompletableFuture<>(); +325 List<TableDescriptor> failed = new LinkedList<>(); +326 listTables(pattern, false).whenComplete( +327 (tables, error) -> { +328 if (error != null) { +329 future.completeExceptionally(error); +330 return; +331 } +332 CompletableFuture[] futures = Arrays.stream(tables) +333 .map((table) -> operator.operate(table.getTableName()).whenComplete((v, ex) -> { +334 if (ex != null) { +335 LOG.info("Failed to " + operationType + " table " + table.getTableName(), ex); +336 failed.add(table); +337 } +338 })).<CompletableFuture> toArray(size -> new CompletableFuture[size]); +339 CompletableFuture.allOf(futures).thenAccept((v) -> { +340 future.complete(failed.toArray(new TableDescriptor[failed.size()])); +341 }); +342 }); +343 return future; +344 } +345 +346 @Override +347 public AsyncConnectionImpl getConnection() { +348 return this.connection; +349 } +350 +351 @Override +352 public CompletableFuture<Boolean> tableExists(TableName tableName) { +353 return AsyncMetaTableAccessor.tableExists(metaTable, tableName); +354 } +355 +356 @Override +357 public CompletableFuture<TableDescriptor[]> listTables() { +358 return listTables((Pattern) null, false); +359 } +360 +361 @Override +362 public CompletableFuture<TableDescriptor[]> listTables(String regex, boolean includeSysTables) { +363 return listTables(Pattern.compile(regex), false); +364 } +365 +366 @Override +367 public CompletableFuture<TableDescriptor[]> listTables(Pattern pattern, boolean includeSysTables) { +368 return this +369 .<TableDescriptor[]>newMasterCaller() +370 .action( +371 (controller, stub) -> this +372 .<GetTableDescriptorsRequest, GetTableDescriptorsResponse, TableDescriptor[]> call( +373 controller, stub, RequestConverter.buildGetTableDescriptorsRequest(pattern, +374 includeSysTables), (s, c, req, done) -> s.getTableDescriptors(c, req, done), ( +375 resp) -> ProtobufUtil.getTableDescriptorArray(resp))).call(); +376 } +377 +378 @Override +379 public CompletableFuture<TableName[]> listTableNames() { +380 return listTableNames((Pattern) null, false); +381 } +382 +383 @Override +384 public CompletableFuture<TableName[]> listTableNames(String regex, boolean includeSysTables) { +385 return listTableNames(Pattern.compile(regex), false); +386 } +387 +388 @Override +389 public CompletableFuture<TableName[]> listTableNames(Pattern pattern, boolean includeSysTables) { +390 return this +391 .<TableName[]>newMasterCaller() +392 .action( +393 (controller, stub) -> this +394 .<GetTableNamesRequest, GetTableNamesResponse, TableName[]> call(controller, stub, +395 RequestConverter.buildGetTableNamesRequest(pattern, includeSysTables), (s, c, req, +396 done) -> s.getTableNames(c, req, done), (resp) -> ProtobufUtil +397 .getTableNameArray(resp.getTableNamesList()))).call(); +398 } +399 +400 @Override +401 public CompletableFuture<TableDescriptor> getTableDescriptor(TableName tableName) { +402 CompletableFuture<TableDescriptor> future = new CompletableFuture<>(); +403 this.<List<TableSchema>> newMasterCaller() +404 .action( +405 (controller, stub) -> this +406 .<GetTableDescriptorsRequest, GetTableDescriptorsResponse, List<TableSchema>> call( +407 controller, stub, RequestConverter.buildGetTableDescriptorsRequest(tableName), (s, +408 c, req, done) -> s.getTableDescriptors(c, req, done), (resp) -> resp +409 .getTableSchemaList())).call().whenComplete((tableSchemas, error) -> { +410 if (error != null) { +411 future.completeExceptionally(error); +412 return; +413 } +414 if (!tableSchemas.isEmpty()) { +415 future.complete(ProtobufUtil.convertToTableDesc(tableSchemas.get(0))); +416 } else { +417 future.completeExceptionally(new TableNotFoundException(tableName.getNameAsString())); +418 } +419 }); +420 return future; +421 } +422 +423 @Override +424 public CompletableFuture<Void> createTable(TableDescriptor desc) { +425 return createTable(desc, null); +426 } +427 +428 @Override +429 public CompletableFuture<Void> createTable(TableDescriptor desc, byte[] startKey, byte[] endKey, +430 int numRegions) { +431 try { +432 return createTable(desc, getSplitKeys(startKey, endKey, numRegions)); +433 } catch (IllegalArgumentException e) { +434 return failedFuture(e); +435 } +436 } +437 +438 @Override +439 public CompletableFuture<Void> createTable(TableDescriptor desc, byte[][] splitKeys) { +440 if (desc.getTableName() == null) { +441 return failedFuture(new IllegalArgumentException("TableName cannot be null")); +442 } +443 if (splitKeys != null && splitKeys.length > 0) { +444 Arrays.sort(splitKeys, Bytes.BYTES_COMPARATOR); +445 // Verify there are no duplicate split keys +446 byte[] lastKey = null; +447 for (byte[] splitKey : splitKeys) { +448 if (Bytes.compareTo(splitKey, HConstants.EMPTY_BYTE_ARRAY) == 0) { +449 return failedFuture(new IllegalArgumentException( +450 "Empty split key must not be passed in the split keys.")); +451 } +452 if (lastKey != null && Bytes.equals(splitKey, lastKey)) { +453 return failedFuture(new IllegalArgumentException("All split keys must be unique, " +454 + "found duplicate: " + Bytes.toStringBinary(splitKey) + ", " +455 + Bytes.toStringBinary(lastKey))); +456 } +457 lastKey = splitKey; +458 } +459 } +460 +461 return this.<CreateTableRequest, CreateTableResponse> procedureCall( +462 RequestConverter.buildCreateTableRequest(desc, splitKeys, ng.getNonceGroup(), ng.newNonce()), +463 (s, c, req, done) -> s.createTable(c, req, done), (resp) -> resp.getProcId(), +464 new CreateTableProcedureBiConsumer(this, desc.getTableName())); +465 } +466 +467 @Override +468 public CompletableFuture<Void> deleteTable(TableName tableName) { +469 return this.<DeleteTableRequest, DeleteTableResponse> procedureCall(RequestConverter +470 .buildDeleteTableRequest(tableName, ng.getNonceGroup(), ng.newNonce()), +471 (s, c, req, done) -> s.deleteTable(c, req, done), (resp) -> resp.getProcId(), +472 new DeleteTableProcedureBiConsumer(this, tableName)); +473 } +474 +475 @Override +476 public CompletableFuture<TableDescriptor[]> deleteTables(String regex) { +477 return deleteTables(Pattern.compile(regex)); +478 } +479 +480 @Override +481 public CompletableFuture<TableDescriptor[]> deleteTables(Pattern pattern) { +482 return batchTableOperations(pattern, (table) -> deleteTable(table), "DELETE"); +483 } +484 +485 @Override +486 public CompletableFuture<Void> truncateTable(TableName tableName, boolean preserveSplits) { +487 return this.<TruncateTableRequest, TruncateTableResponse> procedureCall( +488 RequestConverter.buildTruncateTableRequest(tableName, preserveSplits, ng.getNonceGroup(), +489 ng.newNonce()), (s, c, req, done) -> s.truncateTable(c, req, done), +490 (resp) -> resp.getProcId(), new TruncateTableProcedureBiConsumer(this, tableName)); +491 } +492 +493 @Override +494 public CompletableFuture<Void> enableTable(TableName tableName) { +495 return this.<EnableTableRequest, EnableTableResponse> procedureCall(RequestConverter +496 .buildEnableTableRequest(tableName, ng.getNonceGroup(), ng.newNonce()), +497 (s, c, req, done) -> s.enableTable(c, req, done), (resp) -> resp.getProcId(), +498 new EnableTableProcedureBiConsumer(this, tableName)); +499 } +500 +501 @Override +502 public CompletableFuture<TableDescriptor[]> enableTables(String regex) { +503 return enableTables(Pattern.compile(regex)); +504 } +505 +506 @Override +507 public CompletableFuture<TableDescriptor[]> enableTables(Pattern pattern) { +508 return batchTableOperations(pattern, (table) -> enableTable(table), "ENABLE"); +509 } +510 +511 @Override +512 public CompletableFuture<Void> disableTable(TableName tableName) { +513 return this.<DisableTableRequest, DisableTableResponse> procedureCall(RequestConverter +514 .buildDisableTableRequest(tableName, ng.getNonceGroup(), ng.newNonce()), +515 (s, c, req, done) -> s.disableTable(c, req, done), (resp) -> resp.getProcId(), +516 new DisableTableProcedureBiConsumer(this, tableName)); +517 } +518 +519 @Override +520 public CompletableFuture<TableDescriptor[]> disableTables(String regex) { +521 return disableTables(Pattern.compile(regex)); +522 } +523 +524 @Override +525 public CompletableFuture<TableDescriptor[]> disableTables(Pattern pattern) { +526 return batchTableOperations(pattern, (table) -> disableTable(table), "DISABLE"); +527 } 528 -529 @Override -530 public CompletableFuture<Boolean> isTableEnabled(TableName tableName) { -531 CompletableFuture<Boolean> future = new CompletableFuture<>(); -532 AsyncMetaTableAccessor.getTableState(metaTable, tableName).whenComplete((state, error) -> { -533 if (error != null) { -534 future.completeExceptionally(error); -535 return; -536 } -537 if (state.isPresent()) { -538 future.complete(state.get().inStates(TableState.State.ENABLED)); -539 } else { -540 future.completeExceptionally(new TableNotFoundException(tableName)); -541 } -542 }); -543 return future; -544 } -545 -546 @Override -547 public CompletableFuture<Boolean> isTableDisabled(TableName tableName) { -548 CompletableFuture<Boolean> future = new CompletableFuture<>(); -549 AsyncMetaTableAccessor.getTableState(metaTable, tableName).whenComplete((state, error) -> { -550 if (error != null) { -551 future.completeExceptionally(error); -552 return; -553 } -554 if (state.isPresent()) { -555 future.complete(state.get().inStates(TableState.State.DISABLED)); -556 } else { -557 future.completeExceptionally(new TableNotFoundException(tableName)); -558 } -559 }); -560 return future; -561 } -562 -563 @Override -564 public CompletableFuture<Boolean> isTableAvailable(TableName tableName) { -565 return isTableAvailable(tableName, null); -566 } -567 -568 @Override -569 public CompletableFuture<Boolean> isTableAvailable(TableName tableName, byte[][] splitKeys) { -570 CompletableFuture<Boolean> future = new CompletableFuture<>(); -571 isTableEnabled(tableName).whenComplete( -572 (enabled, error) -> { -573 if (error != null) { -574 future.completeExceptionally(error); -575 return; -576 } -577 if (!enabled) { -578 future.complete(false); -579 } else { -580 AsyncMetaTableAccessor.getTableRegionsAndLocations(metaTable, Optional.of(tableName)) -581 .whenComplete( -582 (locations, error1) -> { -583 if (error1 != null) { -584 future.completeExceptionally(error1); -585 return; -586 } -587 int notDeployed = 0; -588 int regionCount = 0; -589 for (Pair<HRegionInfo, ServerName> pair : locations) { -590 HRegionInfo info = pair.getFirst(); -591 if (pair.getSecond() == null) { -592 if (LOG.isDebugEnabled()) { -593 LOG.debug("Table " + tableName + " has not deployed region " -594 + pair.getFirst().getEncodedName()); -595 } -596 notDeployed++; -597 } else if (splitKeys != null -598 && !Bytes.equals(info.getStartKey(), HConstants.EMPTY_BYTE_ARRAY)) { -599 for (byte[] splitKey : splitKeys) { -600 // Just check if the splitkey is available -601 if (Bytes.equals(info.getStartKey(), splitKey)) { -602 regionCount++; -603 break; -604 } -605 } -606 } else { -607 // Always empty start row should be counted -608 regionCount++; -609 } -610 } -611 if (notDeployed > 0) { -612 if (LOG.isDebugEnabled()) { -613 LOG.debug("Table " + tableName + " has " + notDeployed + " regions"); -614 } -615 future.complete(false); -616 } else if (splitKeys != null && regionCount != splitKeys.length + 1) { -617 if (LOG.isDebugEnabled()) { -618 LOG.debug("Table " + tableName + " expected to have " -619 + (splitKeys.length + 1) + " regions, but only " + regionCount -620 + " available"); -621 } -622 future.complete(false); -623 } else { -624 if (LOG.isDebugEnabled()) { -625 LOG.debug("Table " + tableName + " should be available"); -626 } -627 future.complete(true); -628 } -629 }); -630 } -631 }); -632 return future; -633 } -634 -635 @Override -636 public CompletableFuture<Pair<Integer, Integer>> getAlterStatus(TableName tableName) { -637 return this -638 .<Pair<Integer, Integer>>newMasterCaller() -639 .action( -640 (controller, stub) -> this -641 .<GetSchemaAlterStatusRequest, GetSchemaAlterStatusResponse, Pair<Integer, Integer>> call( -642 controller, stub, RequestConverter.buildGetSchemaAlterStatusRequest(tableName), (s, -643 c, req, done) -> s.getSchemaAlterStatus(c, req, done), (resp) -> new Pair<>( -644 resp.getYetToUpdateRegions(), resp.getTotalRegions()))).call(); -645 } -646 -647 @Override -648 public CompletableFuture<Void> addColumnFamily(TableName tableName, HColumnDescriptor columnFamily) { -649 return this.<AddColumnRequest, AddColumnResponse> procedureCall( -650 RequestConverter.buildAddColumnRequest(tableName, columnFamily, ng.getNonceGroup(), -651 ng.newNonce()), (s, c, req, done) -> s.addColumn(c, req, done), (resp) -> resp.getProcId(), -652 new AddColumnFamilyProcedureBiConsumer(this, tableName)); -653 } -654 -655 @Override -656 public CompletableFuture<Void> deleteColumnFamily(TableName tableName, byte[] columnFamily) { -657 return this.<DeleteColumnRequest, DeleteColumnResponse> procedureCall( -658 RequestConverter.buildDeleteColumnRequest(tableName, columnFamily, ng.getNonceGroup(), -659 ng.newNonce()), (s, c, req, done) -> s.deleteColumn(c, req, done), -660 (resp) -> resp.getProcId(), new DeleteColumnFamilyProcedureBiConsumer(this, tableName)); -661 } -662 -663 @Override -664 public CompletableFuture<Void> modifyColumnFamily(TableName tableName, -665 HColumnDescriptor columnFamily) { -666 return this.<ModifyColumnRequest, ModifyColumnResponse> procedureCall( -667 RequestConverter.buildModifyColumnRequest(tableName, columnFamily, ng.getNonceGroup(), -668 ng.newNonce()), (s, c, req, done) -> s.modifyColumn(c, req, done), -669 (resp) -> resp.getProcId(), new ModifyColumnFamilyProcedureBiConsumer(this, tableName)); -670 } -671 -672 @Override -673 public CompletableFuture<Void> createNamespace(NamespaceDescriptor descriptor) { -674 return this.<CreateNamespaceRequest, CreateNamespaceResponse> procedureCall( -675 RequestConverter.buildCreateNamespaceRequest(descriptor), -676 (s, c, req, done) -> s.createNamespace(c, req, done), (resp) -> resp.getProcId(), -677 new CreateNamespaceProcedureBiConsumer(this, descriptor.getName())); -678 } -679 -680 @Override -681 public CompletableFuture<Void> modifyNamespace(NamespaceDescriptor descriptor) { -682 return this.<ModifyNamespaceRequest, ModifyNamespaceResponse> procedureCall( -683 RequestConverter.buildModifyNamespaceRequest(descriptor), -684 (s, c, req, done) -> s.modifyNamespace(c, req, done), (resp) -> resp.getProcId(), -685 new ModifyNamespaceProcedureBiConsumer(this, descriptor.getName())); -686 } -687 -688 @Override -689 public CompletableFuture<Void> deleteNamespace(String name) { -690 return this.<DeleteNamespaceRequest, DeleteNamespaceResponse> procedureCall( -691 RequestConverter.buildDeleteNamespaceRequest(name), -692 (s, c, req, done) -> s.deleteNamespace(c, req, done), (resp) -> resp.getProcId(), -693 new DeleteNamespaceProcedureBiConsumer(this, name)); -694 } -695 -696 @Override