From 0206dc67d643e4a248a319c724cd6e58f0e77603 Mon Sep 17 00:00:00 2001 From: stack Date: Fri, 5 Aug 2016 15:18:48 -0700 Subject: [PATCH] Revert "HBASE-16308 Contain protobuf references Gather up the pb references into a few locations only rather than have pb references distributed all about the code base." This reverts commit ed87a81b4b61c4842c12572a47c97ae23773012f. --- .../hbase/client/AbstractRegionServerCallable.java | 23 +- .../apache/hadoop/hbase/client/ClientScanner.java | 2 +- .../hadoop/hbase/client/ClientSimpleScanner.java | 3 +- .../hadoop/hbase/client/ClientSmallScanner.java | 42 +- ...tryingCallable.java => ConnectionCallable.java} | 25 +- .../hbase/client/ConnectionImplementation.java | 40 +- .../hadoop/hbase/client/FlushRegionCallable.java | 26 +- .../org/apache/hadoop/hbase/client/HBaseAdmin.java | 1110 +++++++++++--------- .../org/apache/hadoop/hbase/client/HTable.java | 455 ++++---- .../apache/hadoop/hbase/client/MasterCallable.java | 37 +- .../hbase/client/MasterKeepAliveConnection.java | 3 +- .../hadoop/hbase/client/MultiServerCallable.java | 35 +- .../client/PayloadCarryingServerCallable.java | 44 +- .../hbase/client/RegionAdminServiceCallable.java | 54 +- .../hadoop/hbase/client/RegionServerCallable.java | 72 +- .../hadoop/hbase/client/RetryingTimeTracker.java | 12 +- .../hbase/client/ReversedScannerCallable.java | 4 +- .../hadoop/hbase/client/RpcRetryingCaller.java | 5 +- .../hbase/client/RpcRetryingCallerFactory.java | 1 - .../client/RpcRetryingCallerWithReadReplicas.java | 26 +- .../hadoop/hbase/client/ScannerCallable.java | 140 ++- .../hadoop/hbase/client/SecureBulkLoadClient.java | 80 +- .../hbase/ipc/MasterCoprocessorRpcChannel.java | 3 +- .../hbase/ipc/PayloadCarryingRpcController.java | 139 +-- .../hbase/ipc/RegionCoprocessorRpcChannel.java | 23 +- ...ntroller.java => TimeLimitedRpcController.java} | 130 +-- .../apache/hadoop/hbase/protobuf/ProtobufUtil.java | 73 +- .../hadoop/hbase/client/TestClientScanner.java | 1 + .../org/apache/hadoop/hbase/HBaseIOException.java | 3 +- .../apache/hadoop/hbase/util/ExceptionUtil.java | 2 +- .../org/apache/hadoop/hbase/ipc/RpcServer.java | 6 +- .../hbase/mapreduce/LoadIncrementalHFiles.java | 36 +- .../hbase/master/ExpiredMobFileCleanerChore.java | 6 + .../apache/hadoop/hbase/master/ServerManager.java | 5 +- .../hadoop/hbase/mob/ExpiredMobFileCleaner.java | 12 +- .../apache/hadoop/hbase/mob/mapreduce/Sweeper.java | 6 +- .../hadoop/hbase/regionserver/RSRpcServices.java | 15 +- .../hbase/regionserver/wal/WALEditsReplaySink.java | 43 +- .../RegionReplicaReplicationEndpoint.java | 54 +- .../java/org/apache/hadoop/hbase/util/Merge.java | 13 +- .../org/apache/hadoop/hbase/TestNamespace.java | 7 +- .../org/apache/hadoop/hbase/client/TestAdmin2.java | 8 +- .../hadoop/hbase/client/TestClientTimeouts.java | 7 +- .../org/apache/hadoop/hbase/client/TestHCM.java | 37 +- .../hbase/client/TestReplicaWithCluster.java | 35 +- .../regionserver/TestHRegionServerBulkLoad.java | 23 +- .../TestHRegionServerBulkLoadWithOldClient.java | 13 +- ...HRegionServerBulkLoadWithOldSecureEndpoint.java | 27 +- .../hadoop/hbase/spark/SparkSQLPushDownFilter.java | 4 +- 49 files changed, 1461 insertions(+), 1509 deletions(-) rename hbase-client/src/main/java/org/apache/hadoop/hbase/client/{RpcRetryingCallable.java => ConnectionCallable.java} (69%) copy hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/{PayloadCarryingRpcController.java => TimeLimitedRpcController.java} (50%) diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractRegionServerCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractRegionServerCallable.java index 5a1f5ccfce..7279d816a9 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractRegionServerCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AbstractRegionServerCallable.java @@ -18,7 +18,8 @@ package org.apache.hadoop.hbase.client; import java.io.IOException; - +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.ServerName; @@ -28,15 +29,26 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.util.Bytes; /** - * Added by HBASE-15745 Refactor of RPC classes to better accept async changes. - * Temporary. + * Implementations call a RegionServer. + * Passed to a {@link RpcRetryingCaller} so we retry on fail. + * TODO: this class is actually tied to one region, because most of the paths make use of + * the regioninfo part of location when building requests. The only reason it works for + * multi-region requests (e.g. batch) is that they happen to not use the region parts. + * This could be done cleaner (e.g. having a generic parameter and 2 derived classes, + * RegionCallable and actual RegionServerCallable with ServerName. + * @param the class that the ServerCallable handles */ @InterfaceAudience.Private abstract class AbstractRegionServerCallable implements RetryingCallable { + // Public because used outside of this package over in ipc. + private static final Log LOG = LogFactory.getLog(AbstractRegionServerCallable.class); + protected final Connection connection; protected final TableName tableName; protected final byte[] row; + protected HRegionLocation location; + protected final static int MIN_WAIT_DEAD_SERVER = 10000; /** @@ -115,7 +127,8 @@ abstract class AbstractRegionServerCallable implements RetryingCallable { @Override public void prepare(final boolean reload) throws IOException { // check table state if this is a retry - if (reload && !tableName.equals(TableName.META_TABLE_NAME) && + if (reload && + !tableName.equals(TableName.META_TABLE_NAME) && getConnection().isTableDisabled(tableName)) { throw new TableNotEnabledException(tableName.getNameAsString() + " is disabled."); } @@ -135,4 +148,4 @@ abstract class AbstractRegionServerCallable implements RetryingCallable { * @throws IOException When client could not be created */ abstract void setClientByServiceName(ServerName serviceName) throws IOException; -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java index 3e676c719f..cb4c71454b 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientScanner.java @@ -847,4 +847,4 @@ public abstract class ClientScanner extends AbstractClientScanner { Cell[] list = Arrays.copyOfRange(result.rawCells(), index, result.rawCells().length); return Result.create(list, result.getExists(), result.isStale(), result.isPartial()); } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSimpleScanner.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSimpleScanner.java index ecf083b8f0..f8869716fa 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSimpleScanner.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSimpleScanner.java @@ -32,6 +32,7 @@ import java.util.concurrent.ExecutorService; */ @InterfaceAudience.Private public class ClientSimpleScanner extends ClientScanner { + public ClientSimpleScanner(Configuration configuration, Scan scan, TableName name, ClusterConnection connection, RpcRetryingCallerFactory rpcCallerFactory, RpcControllerFactory rpcControllerFactory, ExecutorService pool, @@ -49,4 +50,4 @@ public class ClientSimpleScanner extends ClientScanner { public Result next() throws IOException { return nextWithSyncCache(); } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSmallScanner.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSmallScanner.java index 429c4cfc9c..f9bdd55d12 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSmallScanner.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ClientSmallScanner.java @@ -18,10 +18,8 @@ */ package org.apache.hadoop.hbase.client; -import java.io.IOException; -import java.io.InterruptedIOException; -import java.util.concurrent.ExecutorService; - +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.ServiceException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; @@ -31,15 +29,17 @@ import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.client.metrics.ScanMetrics; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; import org.apache.hadoop.hbase.ipc.RpcControllerFactory; +import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.ResponseConverter; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanResponse; import org.apache.hadoop.hbase.util.Bytes; -import com.google.common.annotations.VisibleForTesting; +import java.io.IOException; +import java.io.InterruptedIOException; +import java.util.concurrent.ExecutorService; /** * Client scanner for small scan. Generally, only one RPC is called to fetch the @@ -185,7 +185,7 @@ public class ClientSmallScanner extends ClientSimpleScanner { } @Override - protected Result[] call(PayloadCarryingRpcController controller) throws Exception { + public Result[] call(int timeout) throws IOException { if (this.closed) return null; if (Thread.interrupted()) { throw new InterruptedIOException(); @@ -193,17 +193,25 @@ public class ClientSmallScanner extends ClientSimpleScanner { ScanRequest request = RequestConverter.buildScanRequest(getLocation() .getRegionInfo().getRegionName(), getScan(), getCaching(), true); ScanResponse response = null; - response = getStub().scan(controller, request); - Result[] results = ResponseConverter.getResults(controller.cellScanner(), response); - if (response.hasMoreResultsInRegion()) { - setHasMoreResultsContext(true); - setServerHasMoreResults(response.getMoreResultsInRegion()); - } else { - setHasMoreResultsContext(false); + controller = controllerFactory.newController(); + try { + controller.setPriority(getTableName()); + controller.setCallTimeout(timeout); + response = getStub().scan(controller, request); + Result[] results = ResponseConverter.getResults(controller.cellScanner(), + response); + if (response.hasMoreResultsInRegion()) { + setHasMoreResultsContext(true); + setServerHasMoreResults(response.getMoreResultsInRegion()); + } else { + setHasMoreResultsContext(false); + } + // We need to update result metrics since we are overriding call() + updateResultsMetrics(results); + return results; + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } - // We need to update result metrics since we are overriding call() - updateResultsMetrics(results); - return results; } @Override diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionCallable.java similarity index 69% rename from hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallable.java rename to hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionCallable.java index 68a4aa2f45..3f44927463 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionCallable.java @@ -21,13 +21,17 @@ package org.apache.hadoop.hbase.client; import java.io.Closeable; import java.io.IOException; -import org.apache.hadoop.hbase.protobuf.ProtobufUtil; - /** - * A RetryingCallable for RPC connection operations. + * A RetryingCallable for generic connection operations. * @param return type */ -abstract class RpcRetryingCallable implements RetryingCallable, Closeable { +abstract class ConnectionCallable implements RetryingCallable, Closeable { + protected Connection connection; + + public ConnectionCallable(final Connection connection) { + this.connection = connection; + } + @Override public void prepare(boolean reload) throws IOException { } @@ -49,17 +53,4 @@ abstract class RpcRetryingCallable implements RetryingCallable, Closeable public long sleep(long pause, int tries) { return ConnectionUtils.getPauseTime(pause, tries); } - - @Override - // Same trick as in RegionServerCallable so users don't have to copy/paste so much boilerplate - // and so we contain references to protobuf. - public V call(int callTimeout) throws IOException { - try { - return rpcCall(callTimeout); - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); - } - } - - protected abstract V rpcCall(int callTimeout) throws Exception; } \ No newline at end of file diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java index 638050f5a7..8dcda13f84 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java @@ -20,6 +20,11 @@ package org.apache.hadoop.hbase.client; import static org.apache.hadoop.hbase.client.MetricsConnection.CLIENT_SIDE_METRICS_ENABLED_KEY; +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.BlockingRpcChannel; +import com.google.protobuf.RpcController; +import com.google.protobuf.ServiceException; + import java.io.Closeable; import java.io.IOException; import java.io.InterruptedIOException; @@ -63,7 +68,6 @@ import org.apache.hadoop.hbase.exceptions.RegionMovedException; import org.apache.hadoop.hbase.ipc.RpcClient; import org.apache.hadoop.hbase.ipc.RpcClientFactory; import org.apache.hadoop.hbase.ipc.RpcControllerFactory; -import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos; @@ -91,11 +95,6 @@ import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher; import org.apache.hadoop.ipc.RemoteException; import org.apache.zookeeper.KeeperException; -import com.google.common.annotations.VisibleForTesting; -import com.google.protobuf.BlockingRpcChannel; -import com.google.protobuf.RpcController; -import com.google.protobuf.ServiceException; - /** * Main implementation of {@link Connection} and {@link ClusterConnection} interfaces. * Encapsulates connection to zookeeper and regionservers. @@ -934,13 +933,9 @@ class ConnectionImplementation implements ClusterConnection, Closeable { this.stub = null; } - boolean isMasterRunning() throws IOException { - MasterProtos.IsMasterRunningResponse response = null; - try { - response = this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest()); - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); - } + boolean isMasterRunning() throws ServiceException { + MasterProtos.IsMasterRunningResponse response = + this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest()); return response != null? response.getIsMasterRunning(): false; } } @@ -1063,14 +1058,14 @@ class ConnectionImplementation implements ClusterConnection, Closeable { /** * Once setup, check it works by doing isMasterRunning check. */ - protected abstract void isMasterRunning() throws IOException; + protected abstract void isMasterRunning() throws ServiceException; /** * Create a stub. Try once only. It is not typed because there is no common type to * protobuf services nor their interfaces. Let the caller do appropriate casting. * @return A stub for master services. */ - private Object makeStubNoRetries() throws IOException, KeeperException { + private Object makeStubNoRetries() throws IOException, KeeperException, ServiceException { ZooKeeperKeepAliveConnection zkw; try { zkw = getKeepAliveZooKeeperWatcher(); @@ -1110,7 +1105,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable { } /** - * Create a stub against the master. Retry if necessary. + * Create a stub against the master. Retry if necessary. * @return A stub to do intf against the master * @throws org.apache.hadoop.hbase.MasterNotRunningException if master is not running */ @@ -1126,7 +1121,10 @@ class ConnectionImplementation implements ClusterConnection, Closeable { exceptionCaught = e; } catch (KeeperException e) { exceptionCaught = e; + } catch (ServiceException e) { + exceptionCaught = e; } + throw new MasterNotRunningException(exceptionCaught); } else { throw new DoNotRetryIOException("Connection was closed while trying to get master"); @@ -1157,12 +1155,8 @@ class ConnectionImplementation implements ClusterConnection, Closeable { } @Override - protected void isMasterRunning() throws IOException { - try { - this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest()); - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); - } + protected void isMasterRunning() throws ServiceException { + this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest()); } } @@ -1707,7 +1701,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable { // java.net.ConnectException but they're not declared. So we catch it... LOG.info("Master connection is not running anymore", e.getUndeclaredThrowable()); return false; - } catch (IOException se) { + } catch (ServiceException se) { LOG.warn("Checking master connection", se); return false; } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/FlushRegionCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/FlushRegionCallable.java index c7bf804214..73bdb740ca 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/FlushRegionCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/FlushRegionCallable.java @@ -27,18 +27,23 @@ import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; import org.apache.hadoop.hbase.ipc.RpcControllerFactory; +import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.FlushRegionRequest; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.FlushRegionResponse; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +import com.google.protobuf.ServiceException; + /** * A Callable for flushRegion() RPC. */ @InterfaceAudience.Private public class FlushRegionCallable extends RegionAdminServiceCallable { + private static final Log LOG = LogFactory.getLog(FlushRegionCallable.class); + private final byte[] regionName; private final boolean writeFlushWalMarker; private boolean reload; @@ -59,14 +64,18 @@ public class FlushRegionCallable extends RegionAdminServiceCallable abortProcedureAsync(final long procId, final boolean mayInterruptIfRunning) - throws IOException { + public Future abortProcedureAsync( + final long procId, + final boolean mayInterruptIfRunning) throws IOException { Boolean abortProcResponse = executeCallable( - new MasterCallable(getConnection(), getRpcControllerFactory()) { + new MasterCallable(getConnection()) { @Override - protected AbortProcedureResponse call(PayloadCarryingRpcController rpcController) - throws Exception { + public AbortProcedureResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); AbortProcedureRequest abortProcRequest = AbortProcedureRequest.newBuilder().setProcId(procId).build(); - return master.abortProcedure(rpcController, abortProcRequest); + return master.abortProcedure(controller, abortProcRequest); } }).getIsProcedureAborted(); @@ -314,9 +324,9 @@ public class HBaseAdmin implements Admin { @Override public boolean tableExists(final TableName tableName) throws IOException { - return executeCallable(new RpcRetryingCallable() { + return executeCallable(new ConnectionCallable(getConnection()) { @Override - protected Boolean rpcCall(int callTimeout) throws Exception { + public Boolean call(int callTimeout) throws ServiceException, IOException { return MetaTableAccessor.tableExists(connection, tableName); } }); @@ -340,15 +350,14 @@ public class HBaseAdmin implements Admin { @Override public HTableDescriptor[] listTables(final Pattern pattern, final boolean includeSysTables) throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected HTableDescriptor[] call(PayloadCarryingRpcController rpcController) - throws Exception { + public HTableDescriptor[] call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); GetTableDescriptorsRequest req = RequestConverter.buildGetTableDescriptorsRequest(pattern, includeSysTables); - return ProtobufUtil.getHTableDescriptorArray(master.getTableDescriptors(rpcController, - req)); + return ProtobufUtil.getHTableDescriptorArray(master.getTableDescriptors(controller, req)); } }); } @@ -377,13 +386,14 @@ public class HBaseAdmin implements Admin { @Override public TableName[] listTableNames(final Pattern pattern, final boolean includeSysTables) throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected TableName[] call(PayloadCarryingRpcController rpcController) throws Exception { + public TableName[] call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); GetTableNamesRequest req = RequestConverter.buildGetTableNamesRequest(pattern, includeSysTables); - return ProtobufUtil.getTableNameArray(master.getTableNames(rpcController, req) + return ProtobufUtil.getTableNameArray(master.getTableNames(controller, req) .getTableNamesList()); } }); @@ -404,25 +414,27 @@ public class HBaseAdmin implements Admin { static HTableDescriptor getTableDescriptor(final TableName tableName, Connection connection, RpcRetryingCallerFactory rpcCallerFactory, final RpcControllerFactory rpcControllerFactory, int operationTimeout, int rpcTimeout) throws IOException { - if (tableName == null) return null; - HTableDescriptor htd = - executeCallable(new MasterCallable(connection, rpcControllerFactory) { - @Override - protected HTableDescriptor call(PayloadCarryingRpcController rpcController) - throws Exception { - GetTableDescriptorsRequest req = - RequestConverter.buildGetTableDescriptorsRequest(tableName); - GetTableDescriptorsResponse htds = master.getTableDescriptors(rpcController, req); - if (!htds.getTableSchemaList().isEmpty()) { - return ProtobufUtil.convertToHTableDesc(htds.getTableSchemaList().get(0)); + if (tableName == null) return null; + HTableDescriptor htd = executeCallable(new MasterCallable(connection) { + @Override + public HTableDescriptor call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + GetTableDescriptorsResponse htds; + GetTableDescriptorsRequest req = + RequestConverter.buildGetTableDescriptorsRequest(tableName); + htds = master.getTableDescriptors(controller, req); + + if (!htds.getTableSchemaList().isEmpty()) { + return ProtobufUtil.convertToHTableDesc(htds.getTableSchemaList().get(0)); + } + return null; } - return null; + }, rpcCallerFactory, operationTimeout, rpcTimeout); + if (htd != null) { + return htd; } - }, rpcCallerFactory, operationTimeout, rpcTimeout); - if (htd != null) { - return htd; - } - throw new TableNotFoundException(tableName.getNameAsString()); + throw new TableNotFoundException(tableName.getNameAsString()); } private long getPauseTime(int tries) { @@ -490,14 +502,15 @@ public class HBaseAdmin implements Admin { } CreateTableResponse response = executeCallable( - new MasterCallable(getConnection(), getRpcControllerFactory()) { + new MasterCallable(getConnection()) { @Override - protected CreateTableResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(desc.getTableName()); + public CreateTableResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(desc.getTableName()); CreateTableRequest request = RequestConverter.buildCreateTableRequest( desc, splitKeys, ng.getNonceGroup(), ng.newNonce()); - return master.createTable(rpcController, request); + return master.createTable(controller, request); } }); return new CreateTableFuture(this, desc, splitKeys, response); @@ -541,14 +554,15 @@ public class HBaseAdmin implements Admin { @Override public Future deleteTableAsync(final TableName tableName) throws IOException { DeleteTableResponse response = executeCallable( - new MasterCallable(getConnection(), getRpcControllerFactory()) { + new MasterCallable(getConnection()) { @Override - protected DeleteTableResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public DeleteTableResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); DeleteTableRequest req = RequestConverter.buildDeleteTableRequest(tableName, ng.getNonceGroup(),ng.newNonce()); - return master.deleteTable(rpcController,req); + return master.deleteTable(controller,req); } }); return new DeleteTableFuture(this, tableName, response); @@ -622,16 +636,16 @@ public class HBaseAdmin implements Admin { public Future truncateTableAsync(final TableName tableName, final boolean preserveSplits) throws IOException { TruncateTableResponse response = - executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected TruncateTableResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public TruncateTableResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); LOG.info("Started truncating " + tableName); TruncateTableRequest req = RequestConverter.buildTruncateTableRequest( tableName, preserveSplits, ng.getNonceGroup(), ng.newNonce()); - return master.truncateTable(rpcController, req); + return master.truncateTable(controller, req); } }); return new TruncateTableFuture(this, tableName, preserveSplits, response); @@ -687,15 +701,17 @@ public class HBaseAdmin implements Admin { public Future enableTableAsync(final TableName tableName) throws IOException { TableName.isLegalFullyQualifiedTableName(tableName.getName()); EnableTableResponse response = executeCallable( - new MasterCallable(getConnection(), getRpcControllerFactory()) { + new MasterCallable(getConnection()) { @Override - protected EnableTableResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public EnableTableResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); + LOG.info("Started enable of " + tableName); EnableTableRequest req = RequestConverter.buildEnableTableRequest(tableName, ng.getNonceGroup(),ng.newNonce()); - return master.enableTable(rpcController,req); + return master.enableTable(controller,req); } }); return new EnableTableFuture(this, tableName, response); @@ -751,16 +767,18 @@ public class HBaseAdmin implements Admin { public Future disableTableAsync(final TableName tableName) throws IOException { TableName.isLegalFullyQualifiedTableName(tableName.getName()); DisableTableResponse response = executeCallable( - new MasterCallable(getConnection(), getRpcControllerFactory()) { + new MasterCallable(getConnection()) { @Override - protected DisableTableResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public DisableTableResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); + LOG.info("Started disable of " + tableName); DisableTableRequest req = RequestConverter.buildDisableTableRequest( tableName, ng.getNonceGroup(), ng.newNonce()); - return master.disableTable(rpcController, req); + return master.disableTable(controller, req); } }); return new DisableTableFuture(this, tableName, response); @@ -809,9 +827,9 @@ public class HBaseAdmin implements Admin { @Override public boolean isTableEnabled(final TableName tableName) throws IOException { checkTableExists(tableName); - return executeCallable(new RpcRetryingCallable() { + return executeCallable(new ConnectionCallable(getConnection()) { @Override - protected Boolean rpcCall(int callTimeout) throws Exception { + public Boolean call(int callTimeout) throws ServiceException, IOException { TableState tableState = MetaTableAccessor.getTableState(connection, tableName); if (tableState == null) throw new TableNotFoundException(tableName); @@ -838,15 +856,16 @@ public class HBaseAdmin implements Admin { @Override public Pair getAlterStatus(final TableName tableName) throws IOException { - return executeCallable(new MasterCallable>(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable>(getConnection()) { @Override - protected Pair call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public Pair call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); + GetSchemaAlterStatusRequest req = RequestConverter .buildGetSchemaAlterStatusRequest(tableName); - GetSchemaAlterStatusResponse ret = master.getSchemaAlterStatus(rpcController, req); + GetSchemaAlterStatusResponse ret = master.getSchemaAlterStatus(controller, req); Pair pair = new Pair<>(ret.getYetToUpdateRegions(), ret.getTotalRegions()); return pair; @@ -875,16 +894,17 @@ public class HBaseAdmin implements Admin { public Future addColumnFamily(final TableName tableName, final HColumnDescriptor columnFamily) throws IOException { AddColumnResponse response = - executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected AddColumnResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public AddColumnResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); + AddColumnRequest req = RequestConverter.buildAddColumnRequest(tableName, columnFamily, ng.getNonceGroup(), ng.newNonce()); - return master.addColumn(rpcController, req); + return master.addColumn(controller, req); } }); return new AddColumnFamilyFuture(this, tableName, response); @@ -919,16 +939,17 @@ public class HBaseAdmin implements Admin { public Future deleteColumnFamily(final TableName tableName, final byte[] columnFamily) throws IOException { DeleteColumnResponse response = - executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected DeleteColumnResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public DeleteColumnResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); + DeleteColumnRequest req = RequestConverter.buildDeleteColumnRequest(tableName, columnFamily, ng.getNonceGroup(), ng.newNonce()); - master.deleteColumn(rpcController, req); + master.deleteColumn(controller, req); return null; } }); @@ -964,16 +985,17 @@ public class HBaseAdmin implements Admin { public Future modifyColumnFamily(final TableName tableName, final HColumnDescriptor columnFamily) throws IOException { ModifyColumnResponse response = - executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected ModifyColumnResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public ModifyColumnResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); + ModifyColumnRequest req = RequestConverter.buildModifyColumnRequest(tableName, columnFamily, ng.getNonceGroup(), ng.newNonce()); - master.modifyColumn(rpcController, req); + master.modifyColumn(controller, req); return null; } }); @@ -1022,26 +1044,28 @@ public class HBaseAdmin implements Admin { @Override public boolean closeRegionWithEncodedRegionName(final String encodedRegionName, final String serverName) throws IOException { - if (null == serverName || ("").equals(serverName.trim())) { - throw new IllegalArgumentException("The servername cannot be null or empty."); + if (null == serverName || ("").equals(serverName.trim())) { + throw new IllegalArgumentException( + "The servername cannot be null or empty."); } - final ServerName sn = ServerName.valueOf(serverName); - final AdminService.BlockingInterface admin = connection.getAdmin(sn); - final PayloadCarryingRpcController controller = rpcControllerFactory.newController(); - return executeCallable(new RpcRetryingCallable() { - @Override - protected Boolean rpcCall(int callTimeout) throws Exception { - controller.setCallTimeout(callTimeout); - CloseRegionRequest request = - RequestConverter.buildCloseRegionRequest(sn, encodedRegionName); - CloseRegionResponse response = admin.closeRegion(controller, request); - boolean closed = response.getClosed(); - if (false == closed) { - LOG.error("Not able to close the region " + encodedRegionName + "."); - } - return closed; + ServerName sn = ServerName.valueOf(serverName); + AdminService.BlockingInterface admin = this.connection.getAdmin(sn); + // Close the region without updating zk state. + CloseRegionRequest request = + RequestConverter.buildCloseRegionRequest(sn, encodedRegionName); + try { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + + // TODO: this does not do retries, it should. Set priority and timeout in controller + CloseRegionResponse response = admin.closeRegion(controller, request); + boolean isRegionClosed = response.getClosed(); + if (false == isRegionClosed) { + LOG.error("Not able to close the region " + encodedRegionName + "."); } - }); + return isRegionClosed; + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } @Override @@ -1080,20 +1104,20 @@ public class HBaseAdmin implements Admin { if (regionServerPair.getSecond() == null) { throw new NoServerForRegionException(Bytes.toStringBinary(regionName)); } - final HRegionInfo hRegionInfo = regionServerPair.getFirst(); + HRegionInfo hRegionInfo = regionServerPair.getFirst(); ServerName serverName = regionServerPair.getSecond(); - final PayloadCarryingRpcController controller = rpcControllerFactory.newController(); - final AdminService.BlockingInterface admin = this.connection.getAdmin(serverName); - executeCallable(new RpcRetryingCallable() { - @Override - protected Void rpcCall(int callTimeout) throws Exception { - controller.setCallTimeout(callTimeout); - FlushRegionRequest request = - RequestConverter.buildFlushRegionRequest(hRegionInfo.getRegionName()); - admin.flushRegion(controller, request); - return null; - } - }); + + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + + AdminService.BlockingInterface admin = this.connection.getAdmin(serverName); + FlushRegionRequest request = + RequestConverter.buildFlushRegionRequest(hRegionInfo.getRegionName()); + try { + // TODO: this does not do retries, it should. Set priority and timeout in controller + admin.flushRegion(controller, request); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } /** @@ -1244,45 +1268,67 @@ public class HBaseAdmin implements Admin { private void compact(final ServerName sn, final HRegionInfo hri, final boolean major, final byte [] family) throws IOException { - final PayloadCarryingRpcController controller = rpcControllerFactory.newController(); - final AdminService.BlockingInterface admin = this.connection.getAdmin(sn); - executeCallable(new RpcRetryingCallable() { - @Override - protected Void rpcCall(int callTimeout) throws Exception { - controller.setCallTimeout(callTimeout); - CompactRegionRequest request = - RequestConverter.buildCompactRegionRequest(hri.getRegionName(), major, family); - admin.compactRegion(controller, request); - return null; - } - }); + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + AdminService.BlockingInterface admin = this.connection.getAdmin(sn); + CompactRegionRequest request = + RequestConverter.buildCompactRegionRequest(hri.getRegionName(), major, family); + try { + // TODO: this does not do retries, it should. Set priority and timeout in controller + admin.compactRegion(controller, request); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } @Override public void move(final byte [] encodedRegionName, final byte [] destServerName) - throws IOException { - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + throws IOException { + + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - rpcController.setPriority(encodedRegionName); - MoveRegionRequest request = - RequestConverter.buildMoveRegionRequest(encodedRegionName, destServerName); - master.moveRegion(rpcController, request); + public Void call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + // Hard to know the table name, at least check if meta + if (isMetaRegion(encodedRegionName)) { + controller.setPriority(TableName.META_TABLE_NAME); + } + + try { + MoveRegionRequest request = + RequestConverter.buildMoveRegionRequest(encodedRegionName, destServerName); + master.moveRegion(controller, request); + } catch (DeserializationException de) { + LOG.error("Could not parse destination server name: " + de); + throw new ServiceException(new DoNotRetryIOException(de)); + } return null; } }); } + private boolean isMetaRegion(final byte[] regionName) { + return Bytes.equals(regionName, HRegionInfo.FIRST_META_REGIONINFO.getRegionName()) + || Bytes.equals(regionName, HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes()); + } + @Override - public void assign(final byte [] regionName) throws MasterNotRunningException, + public void assign(final byte[] regionName) throws MasterNotRunningException, ZooKeeperConnectionException, IOException { - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + final byte[] toBeAssigned = getRegionName(regionName); + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - rpcController.setPriority(regionName); + public Void call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + // Hard to know the table name, at least check if meta + if (isMetaRegion(regionName)) { + controller.setPriority(TableName.META_TABLE_NAME); + } + AssignRegionRequest request = - RequestConverter.buildAssignRegionRequest(getRegionName(regionName)); - master.assignRegion(rpcController, request); + RequestConverter.buildAssignRegionRequest(toBeAssigned); + master.assignRegion(controller,request); return null; } }); @@ -1292,13 +1338,18 @@ public class HBaseAdmin implements Admin { public void unassign(final byte [] regionName, final boolean force) throws MasterNotRunningException, ZooKeeperConnectionException, IOException { final byte[] toBeUnassigned = getRegionName(regionName); - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - rpcController.setPriority(regionName); + public Void call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + // Hard to know the table name, at least check if meta + if (isMetaRegion(regionName)) { + controller.setPriority(TableName.META_TABLE_NAME); + } UnassignRegionRequest request = - RequestConverter.buildUnassignRegionRequest(toBeUnassigned, force); - master.unassignRegion(rpcController, request); + RequestConverter.buildUnassignRegionRequest(toBeUnassigned, force); + master.unassignRegion(controller, request); return null; } }); @@ -1307,11 +1358,16 @@ public class HBaseAdmin implements Admin { @Override public void offline(final byte [] regionName) throws IOException { - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - rpcController.setPriority(regionName); - master.offlineRegion(rpcController, RequestConverter.buildOfflineRegionRequest(regionName)); + public Void call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + // Hard to know the table name, at least check if meta + if (isMetaRegion(regionName)) { + controller.setPriority(TableName.META_TABLE_NAME); + } + master.offlineRegion(controller, RequestConverter.buildOfflineRegionRequest(regionName)); return null; } }); @@ -1320,44 +1376,56 @@ public class HBaseAdmin implements Admin { @Override public boolean setBalancerRunning(final boolean on, final boolean synchronous) throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + SetBalancerRunningRequest req = RequestConverter.buildSetBalancerRunningRequest(on, synchronous); - return master.setBalancerRunning(rpcController, req).getPrevBalanceValue(); + return master.setBalancerRunning(controller, req).getPrevBalanceValue(); } }); } @Override public boolean balancer() throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { - return master.balance(rpcController, - RequestConverter.buildBalanceRequest(false)).getBalancerRan(); + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + return master.balance(controller, + RequestConverter.buildBalanceRequest(false)).getBalancerRan(); } }); } @Override public boolean balancer(final boolean force) throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { - return master.balance(rpcController, - RequestConverter.buildBalanceRequest(force)).getBalancerRan(); + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + return master.balance(controller, + RequestConverter.buildBalanceRequest(force)).getBalancerRan(); } }); } @Override public boolean isBalancerEnabled() throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { - return master.isBalancerEnabled(rpcController, + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + return master.isBalancerEnabled(controller, RequestConverter.buildIsBalancerEnabledRequest()).getEnabled(); } }); @@ -1365,10 +1433,13 @@ public class HBaseAdmin implements Admin { @Override public boolean normalize() throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { - return master.normalize(rpcController, + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + return master.normalize(controller, RequestConverter.buildNormalizeRequest()).getNormalizerRan(); } }); @@ -1376,10 +1447,13 @@ public class HBaseAdmin implements Admin { @Override public boolean isNormalizerEnabled() throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { - return master.isNormalizerEnabled(rpcController, + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + return master.isNormalizerEnabled(controller, RequestConverter.buildIsNormalizerEnabledRequest()).getEnabled(); } }); @@ -1387,22 +1461,28 @@ public class HBaseAdmin implements Admin { @Override public boolean setNormalizerRunning(final boolean on) throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + SetNormalizerRunningRequest req = RequestConverter.buildSetNormalizerRunningRequest(on); - return master.setNormalizerRunning(rpcController, req).getPrevNormalizerValue(); + return master.setNormalizerRunning(controller, req).getPrevNormalizerValue(); } }); } @Override public boolean enableCatalogJanitor(final boolean enable) throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { - return master.enableCatalogJanitor(rpcController, + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + return master.enableCatalogJanitor(controller, RequestConverter.buildEnableCatalogJanitorRequest(enable)).getPrevValue(); } }); @@ -1410,10 +1490,13 @@ public class HBaseAdmin implements Admin { @Override public int runCatalogScan() throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Integer call(PayloadCarryingRpcController rpcController) throws Exception { - return master.runCatalogScan(rpcController, + public Integer call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + return master.runCatalogScan(controller, RequestConverter.buildCatalogScanRequest()).getScanResult(); } }); @@ -1421,10 +1504,13 @@ public class HBaseAdmin implements Admin { @Override public boolean isCatalogJanitorEnabled() throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { - return master.isCatalogJanitorEnabled(rpcController, + public Boolean call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + return master.isCatalogJanitorEnabled(controller, RequestConverter.buildIsCatalogJanitorEnabledRequest()).getValue(); } }); @@ -1530,19 +1616,25 @@ public class HBaseAdmin implements Admin { } DispatchMergingRegionsResponse response = - executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected DispatchMergingRegionsResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - DispatchMergingRegionsRequest request = RequestConverter - .buildDispatchMergingRegionsRequest( + public DispatchMergingRegionsResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + + try { + DispatchMergingRegionsRequest request = RequestConverter + .buildDispatchMergingRegionsRequest( encodedNameOfRegionA, encodedNameOfRegionB, forcible, ng.getNonceGroup(), ng.newNonce()); - return master.dispatchMergingRegions(rpcController, request); + return master.dispatchMergingRegions(controller, request); + } catch (DeserializationException de) { + LOG.error("Could not parse destination server name: " + de); + throw new ServiceException(new DoNotRetryIOException(de)); + } } }); return new DispatchMergingRegionsFuture(this, tableName, response); @@ -1654,17 +1746,21 @@ public class HBaseAdmin implements Admin { throw new IllegalArgumentException("the specified table name '" + tableName + "' doesn't match with the HTD one: " + htd.getTableName()); } + ModifyTableResponse response = executeCallable( - new MasterCallable(getConnection(), getRpcControllerFactory()) { + new MasterCallable(getConnection()) { @Override - protected ModifyTableResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - rpcController.setPriority(tableName); + public ModifyTableResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(tableName); + ModifyTableRequest request = RequestConverter.buildModifyTableRequest( tableName, htd, ng.getNonceGroup(), ng.newNonce()); - return master.modifyTable(rpcController, request); + return master.modifyTable(controller, request); } }); + return new ModifyTableFuture(this, tableName, response); } @@ -1779,9 +1875,9 @@ public class HBaseAdmin implements Admin { */ private TableName checkTableExists(final TableName tableName) throws IOException { - return executeCallable(new RpcRetryingCallable() { + return executeCallable(new ConnectionCallable(getConnection()) { @Override - protected TableName rpcCall(int callTimeout) throws Exception { + public TableName call(int callTimeout) throws ServiceException, IOException { if (!MetaTableAccessor.tableExists(connection, tableName)) { throw new TableNotFoundException(tableName); } @@ -1792,11 +1888,13 @@ public class HBaseAdmin implements Admin { @Override public synchronized void shutdown() throws IOException { - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - rpcController.setPriority(HConstants.HIGH_QOS); - master.shutdown(rpcController, ShutdownRequest.newBuilder().build()); + public Void call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(HConstants.HIGH_QOS); + master.shutdown(controller, ShutdownRequest.newBuilder().build()); return null; } }); @@ -1804,11 +1902,13 @@ public class HBaseAdmin implements Admin { @Override public synchronized void stopMaster() throws IOException { - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - rpcController.setPriority(HConstants.HIGH_QOS); - master.stopMaster(rpcController, StopMasterRequest.newBuilder().build()); + public Void call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + controller.setPriority(HConstants.HIGH_QOS); + master.stopMaster(controller, StopMasterRequest.newBuilder().build()); return null; } }); @@ -1819,41 +1919,43 @@ public class HBaseAdmin implements Admin { throws IOException { String hostname = Addressing.parseHostname(hostnamePort); int port = Addressing.parsePort(hostnamePort); - final AdminService.BlockingInterface admin = + AdminService.BlockingInterface admin = this.connection.getAdmin(ServerName.valueOf(hostname, port, 0)); - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { - @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - rpcController.setPriority(HConstants.HIGH_QOS); - StopServerRequest request = RequestConverter.buildStopServerRequest( - "Called by admin client " + this.connection.toString()); - admin.stopServer(rpcController, request); - return null; - } - }); + StopServerRequest request = RequestConverter.buildStopServerRequest( + "Called by admin client " + this.connection.toString()); + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + + controller.setPriority(HConstants.HIGH_QOS); + try { + // TODO: this does not do retries, it should. Set priority and timeout in controller + admin.stopServer(controller, request); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } @Override public boolean isMasterInMaintenanceMode() throws IOException { - return executeCallable(new MasterCallable(getConnection(), - this.rpcControllerFactory) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected IsInMaintenanceModeResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - return master.isMasterInMaintenanceMode(rpcController, - IsInMaintenanceModeRequest.newBuilder().build()); + public IsInMaintenanceModeResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return master.isMasterInMaintenanceMode( + controller, IsInMaintenanceModeRequest.newBuilder().build()); } }).getInMaintenanceMode(); } @Override public ClusterStatus getClusterStatus() throws IOException { - return executeCallable(new MasterCallable(getConnection(), - this.rpcControllerFactory) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected ClusterStatus call(PayloadCarryingRpcController rpcController) throws Exception { + public ClusterStatus call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); GetClusterStatusRequest req = RequestConverter.buildGetClusterStatusRequest(); - return ProtobufUtil.convert(master.getClusterStatus(rpcController, req).getClusterStatus()); + return ProtobufUtil.convert(master.getClusterStatus(controller, req).getClusterStatus()); } }); } @@ -1894,16 +1996,19 @@ public class HBaseAdmin implements Admin { public Future createNamespaceAsync(final NamespaceDescriptor descriptor) throws IOException { CreateNamespaceResponse response = - executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { - @Override - protected CreateNamespaceResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - return master.createNamespace(rpcController, - CreateNamespaceRequest.newBuilder().setNamespaceDescriptor(ProtobufUtil. - toProtoNamespaceDescriptor(descriptor)).build()); - } - }); + executeCallable(new MasterCallable(getConnection()) { + @Override + public CreateNamespaceResponse call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + // TODO: set priority based on NS? + return master.createNamespace(controller, + CreateNamespaceRequest.newBuilder() + .setNamespaceDescriptor(ProtobufUtil + .toProtoNamespaceDescriptor(descriptor)).build() + ); + } + }); return new NamespaceFuture(this, descriptor.getName(), response.getProcId()) { @Override public String getOperationType() { @@ -1922,16 +2027,16 @@ public class HBaseAdmin implements Admin { public Future modifyNamespaceAsync(final NamespaceDescriptor descriptor) throws IOException { ModifyNamespaceResponse response = - executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { - @Override - protected ModifyNamespaceResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - // TODO: set priority based on NS? - return master.modifyNamespace(rpcController, ModifyNamespaceRequest.newBuilder(). - setNamespaceDescriptor(ProtobufUtil.toProtoNamespaceDescriptor(descriptor)).build()); - } - }); + executeCallable(new MasterCallable(getConnection()) { + @Override + public ModifyNamespaceResponse call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + // TODO: set priority based on NS? + return master.modifyNamespace(controller, ModifyNamespaceRequest.newBuilder(). + setNamespaceDescriptor(ProtobufUtil.toProtoNamespaceDescriptor(descriptor)).build()); + } + }); return new NamespaceFuture(this, descriptor.getName(), response.getProcId()) { @Override public String getOperationType() { @@ -1950,16 +2055,16 @@ public class HBaseAdmin implements Admin { public Future deleteNamespaceAsync(final String name) throws IOException { DeleteNamespaceResponse response = - executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { - @Override - protected DeleteNamespaceResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - // TODO: set priority based on NS? - return master.deleteNamespace(rpcController, DeleteNamespaceRequest.newBuilder(). - setNamespaceName(name).build()); - } - }); + executeCallable(new MasterCallable(getConnection()) { + @Override + public DeleteNamespaceResponse call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + // TODO: set priority based on NS? + return master.deleteNamespace(controller, DeleteNamespaceRequest.newBuilder(). + setNamespaceName(name).build()); + } + }); return new NamespaceFuture(this, name, response.getProcId()) { @Override public String getOperationType() { @@ -1970,94 +2075,100 @@ public class HBaseAdmin implements Admin { @Override public NamespaceDescriptor getNamespaceDescriptor(final String name) throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { - @Override - protected NamespaceDescriptor call(PayloadCarryingRpcController rpcController) - throws Exception { - return ProtobufUtil.toNamespaceDescriptor( - master.getNamespaceDescriptor(rpcController, GetNamespaceDescriptorRequest.newBuilder(). + return + executeCallable(new MasterCallable(getConnection()) { + @Override + public NamespaceDescriptor call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return ProtobufUtil.toNamespaceDescriptor( + master.getNamespaceDescriptor(controller, GetNamespaceDescriptorRequest.newBuilder(). setNamespaceName(name).build()).getNamespaceDescriptor()); - } - }); + } + }); } @Override public NamespaceDescriptor[] listNamespaceDescriptors() throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { - @Override - protected NamespaceDescriptor[] call(PayloadCarryingRpcController rpcController) - throws Exception { - List list = - master.listNamespaceDescriptors(rpcController, - ListNamespaceDescriptorsRequest.newBuilder().build()).getNamespaceDescriptorList(); - NamespaceDescriptor[] res = new NamespaceDescriptor[list.size()]; - for(int i = 0; i < list.size(); i++) { - res[i] = ProtobufUtil.toNamespaceDescriptor(list.get(i)); - } - return res; - } - }); + return + executeCallable(new MasterCallable(getConnection()) { + @Override + public NamespaceDescriptor[] call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + List list = + master.listNamespaceDescriptors(controller, + ListNamespaceDescriptorsRequest.newBuilder().build()) + .getNamespaceDescriptorList(); + NamespaceDescriptor[] res = new NamespaceDescriptor[list.size()]; + for(int i = 0; i < list.size(); i++) { + res[i] = ProtobufUtil.toNamespaceDescriptor(list.get(i)); + } + return res; + } + }); } @Override public ProcedureInfo[] listProcedures() throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { - @Override - protected ProcedureInfo[] call(PayloadCarryingRpcController rpcController) - throws Exception { - List procList = master.listProcedures( - rpcController, ListProceduresRequest.newBuilder().build()).getProcedureList(); - ProcedureInfo[] procInfoList = new ProcedureInfo[procList.size()]; - for (int i = 0; i < procList.size(); i++) { - procInfoList[i] = ProcedureUtil.convert(procList.get(i)); - } - return procInfoList; - } - }); + return + executeCallable(new MasterCallable(getConnection()) { + @Override + public ProcedureInfo[] call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + List procList = master.listProcedures( + controller, ListProceduresRequest.newBuilder().build()).getProcedureList(); + ProcedureInfo[] procInfoList = new ProcedureInfo[procList.size()]; + for (int i = 0; i < procList.size(); i++) { + procInfoList[i] = ProcedureUtil.convert(procList.get(i)); + } + return procInfoList; + } + }); } @Override public HTableDescriptor[] listTableDescriptorsByNamespace(final String name) throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { - @Override - protected HTableDescriptor[] call(PayloadCarryingRpcController rpcController) - throws Exception { - List list = - master.listTableDescriptorsByNamespace(rpcController, - ListTableDescriptorsByNamespaceRequest.newBuilder().setNamespaceName(name) - .build()).getTableSchemaList(); - HTableDescriptor[] res = new HTableDescriptor[list.size()]; - for(int i=0; i < list.size(); i++) { - - res[i] = ProtobufUtil.convertToHTableDesc(list.get(i)); - } - return res; - } - }); + return + executeCallable(new MasterCallable(getConnection()) { + @Override + public HTableDescriptor[] call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + List list = + master.listTableDescriptorsByNamespace(controller, + ListTableDescriptorsByNamespaceRequest.newBuilder().setNamespaceName(name) + .build()).getTableSchemaList(); + HTableDescriptor[] res = new HTableDescriptor[list.size()]; + for(int i=0; i < list.size(); i++) { + + res[i] = ProtobufUtil.convertToHTableDesc(list.get(i)); + } + return res; + } + }); } @Override public TableName[] listTableNamesByNamespace(final String name) throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { - @Override - protected TableName[] call(PayloadCarryingRpcController rpcController) - throws Exception { - List tableNames = - master.listTableNamesByNamespace(rpcController, ListTableNamesByNamespaceRequest. + return + executeCallable(new MasterCallable(getConnection()) { + @Override + public TableName[] call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + List tableNames = + master.listTableNamesByNamespace(controller, ListTableNamesByNamespaceRequest. newBuilder().setNamespaceName(name).build()) - .getTableNameList(); - TableName[] result = new TableName[tableNames.size()]; - for (int i = 0; i < tableNames.size(); i++) { - result[i] = ProtobufUtil.toTableName(tableNames.get(i)); - } - return result; - } - }); + .getTableNameList(); + TableName[] result = new TableName[tableNames.size()]; + for (int i = 0; i < tableNames.size(); i++) { + result[i] = ProtobufUtil.toTableName(tableNames.get(i)); + } + return result; + } + }); } /** @@ -2065,26 +2176,10 @@ public class HBaseAdmin implements Admin { * @param conf system configuration * @throws MasterNotRunningException if the master is not running * @throws ZooKeeperConnectionException if unable to connect to zookeeper - * @deprecated since hbase-2.0.0 because throws a ServiceException. We don't want to have - * protobuf as part of our public API. Use {@link #available(Configuration)} */ // Used by tests and by the Merge tool. Merge tool uses it to figure if HBase is up or not. - // MOB uses it too. - // NOTE: hbase-2.0.0 removes ServiceException from the throw. - @Deprecated public static void checkHBaseAvailable(Configuration conf) - throws MasterNotRunningException, ZooKeeperConnectionException, IOException, - com.google.protobuf.ServiceException { - available(conf); - } - - /** - * Is HBase available? Throw an exception if not. - * @param conf system configuration - * @throws ZooKeeperConnectionException if unable to connect to zookeeper] - */ - public static void available(final Configuration conf) - throws ZooKeeperConnectionException, InterruptedIOException { + throws MasterNotRunningException, ZooKeeperConnectionException, ServiceException, IOException { Configuration copyOfConf = HBaseConfiguration.create(conf); // We set it to make it fail as soon as possible if HBase is not available copyOfConf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1); @@ -2096,6 +2191,7 @@ public class HBaseAdmin implements Admin { (ClusterConnection) ConnectionFactory.createConnection(copyOfConf); ZooKeeperKeepAliveConnection zkw = ((ConnectionImplementation) connection). getKeepAliveZooKeeperWatcher();) { + // This is NASTY. FIX!!!! Dependent on internal implementation! TODO zkw.getRecoverableZooKeeper().getZooKeeper().exists(zkw.baseZNode, false); connection.isMasterRunning(); @@ -2135,15 +2231,14 @@ public class HBaseAdmin implements Admin { @Override public HTableDescriptor[] getTableDescriptorsByTableName(final List tableNames) throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected HTableDescriptor[] call(PayloadCarryingRpcController rpcController) - throws Exception { + public HTableDescriptor[] call(int callTimeout) throws Exception { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); GetTableDescriptorsRequest req = RequestConverter.buildGetTableDescriptorsRequest(tableNames); - return ProtobufUtil. - getHTableDescriptorArray(master.getTableDescriptors(rpcController, req)); + return ProtobufUtil.getHTableDescriptorArray(master.getTableDescriptors(controller, req)); } }); } @@ -2181,16 +2276,16 @@ public class HBaseAdmin implements Admin { private RollWALWriterResponse rollWALWriterImpl(final ServerName sn) throws IOException, FailedLogCloseException { - final AdminService.BlockingInterface admin = this.connection.getAdmin(sn); - Callable callable = new Callable() { - @Override - public RollWALWriterResponse call() throws Exception { - RollWALWriterRequest request = RequestConverter.buildRollWALWriterRequest(); - PayloadCarryingRpcController controller = rpcControllerFactory.newController(); - return admin.rollWALWriter(controller, request); - } - }; - return ProtobufUtil.call(callable); + AdminService.BlockingInterface admin = this.connection.getAdmin(sn); + RollWALWriterRequest request = RequestConverter.buildRollWALWriterRequest(); + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + + try { + // TODO: this does not do retries, it should. Set priority and timeout in controller + return admin.rollWALWriter(controller, request); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } /** @@ -2226,7 +2321,8 @@ public class HBaseAdmin implements Admin { } byte[][] regionsToFlush = new byte[regionCount][]; for (int i = 0; i < regionCount; i++) { - regionsToFlush[i] = ProtobufUtil.toBytes(response.getRegionToFlush(i)); + ByteString region = response.getRegionToFlush(i); + regionsToFlush[i] = region.toByteArray(); } return regionsToFlush; } @@ -2256,31 +2352,28 @@ public class HBaseAdmin implements Admin { @Override public CompactionState getCompactionStateForRegion(final byte[] regionName) throws IOException { - final Pair regionServerPair = getRegion(regionName); - if (regionServerPair == null) { - throw new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName)); - } - if (regionServerPair.getSecond() == null) { - throw new NoServerForRegionException(Bytes.toStringBinary(regionName)); - } - ServerName sn = regionServerPair.getSecond(); - final AdminService.BlockingInterface admin = this.connection.getAdmin(sn); - Callable callable = new Callable() { - @Override - public CompactionState call() throws Exception { - PayloadCarryingRpcController controller = rpcControllerFactory.newController(); - GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest( - regionServerPair.getFirst().getRegionName(), true); - - // TODO: this does not do retries, it should. Set priority and timeout in controller - GetRegionInfoResponse response = admin.getRegionInfo(controller, request); - if (response.getCompactionState() != null) { - return ProtobufUtil.createCompactionState(response.getCompactionState()); - } - return null; + try { + Pair regionServerPair = getRegion(regionName); + if (regionServerPair == null) { + throw new IllegalArgumentException("Invalid region: " + Bytes.toStringBinary(regionName)); + } + if (regionServerPair.getSecond() == null) { + throw new NoServerForRegionException(Bytes.toStringBinary(regionName)); + } + ServerName sn = regionServerPair.getSecond(); + AdminService.BlockingInterface admin = this.connection.getAdmin(sn); + GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest( + regionServerPair.getFirst().getRegionName(), true); + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + // TODO: this does not do retries, it should. Set priority and timeout in controller + GetRegionInfoResponse response = admin.getRegionInfo(controller, request); + if (response.getCompactionState() != null) { + return ProtobufUtil.createCompactionState(response.getCompactionState()); } - }; - return ProtobufUtil.call(callable); + return null; + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } @Override @@ -2332,12 +2425,12 @@ public class HBaseAdmin implements Admin { throw (InterruptedIOException)new InterruptedIOException("Interrupted").initCause(e); } LOG.debug("Getting current status of snapshot from master..."); - done = executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + done = executeCallable(new MasterCallable(getConnection()) { @Override - protected IsSnapshotDoneResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - return master.isSnapshotDone(rpcController, request); + public IsSnapshotDoneResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return master.isSnapshotDone(controller, request); } }); } @@ -2383,12 +2476,12 @@ public class HBaseAdmin implements Admin { final SnapshotRequest request = SnapshotRequest.newBuilder().setSnapshot(snapshot) .build(); // run the snapshot on the master - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected SnapshotResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - return master.snapshot(rpcController, request); + public SnapshotResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return master.snapshot(controller, request); } }); } @@ -2397,12 +2490,12 @@ public class HBaseAdmin implements Admin { public boolean isSnapshotFinished(final SnapshotDescription snapshotDesc) throws IOException, HBaseSnapshotException, UnknownSnapshotException { final HBaseProtos.SnapshotDescription snapshot = createHBaseProtosSnapshotDesc(snapshotDesc); - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected IsSnapshotDoneResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - return master.isSnapshotDone(rpcController, + public IsSnapshotDoneResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return master.isSnapshotDone(controller, IsSnapshotDoneRequest.newBuilder().setSnapshot(snapshot).build()); } }).getDone(); @@ -2581,11 +2674,12 @@ public class HBaseAdmin implements Admin { .setProcedure(builder.build()).build(); // run the procedure on the master ExecProcedureResponse response = executeCallable(new MasterCallable( - getConnection(), getRpcControllerFactory()) { + getConnection()) { @Override - protected ExecProcedureResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - return master.execProcedureWithRet(rpcController, request); + public ExecProcedureResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return master.execProcedureWithRet(controller, request); } }); @@ -2607,11 +2701,12 @@ public class HBaseAdmin implements Admin { .setProcedure(builder.build()).build(); // run the procedure on the master ExecProcedureResponse response = executeCallable(new MasterCallable( - getConnection(), getRpcControllerFactory()) { + getConnection()) { @Override - protected ExecProcedureResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - return master.execProcedure(rpcController, request); + public ExecProcedureResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return master.execProcedure(controller, request); } }); @@ -2655,11 +2750,12 @@ public class HBaseAdmin implements Admin { } final ProcedureDescription desc = builder.build(); return executeCallable( - new MasterCallable(getConnection(), getRpcControllerFactory()) { + new MasterCallable(getConnection()) { @Override - protected IsProcedureDoneResponse call(PayloadCarryingRpcController rpcController) - throws Exception { - return master.isProcedureDone(rpcController, IsProcedureDoneRequest + public IsProcedureDoneResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return master.isProcedureDone(controller, IsProcedureDoneRequest .newBuilder().setProcedure(desc).build()); } }).getDone(); @@ -2685,16 +2781,17 @@ public class HBaseAdmin implements Admin { ClientSnapshotDescriptionUtils.assertSnapshotRequestIsValid(snapshot); RestoreSnapshotResponse response = executeCallable( - new MasterCallable(getConnection(), getRpcControllerFactory()) { + new MasterCallable(getConnection()) { @Override - protected RestoreSnapshotResponse call(PayloadCarryingRpcController rpcController) - throws Exception { + public RestoreSnapshotResponse call(int callTimeout) throws ServiceException { final RestoreSnapshotRequest request = RestoreSnapshotRequest.newBuilder() .setSnapshot(snapshot) .setNonceGroup(ng.getNonceGroup()) .setNonce(ng.newNonce()) .build(); - return master.restoreSnapshot(rpcController, request); + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + return master.restoreSnapshot(controller, request); } }); @@ -2731,13 +2828,13 @@ public class HBaseAdmin implements Admin { @Override public List listSnapshots() throws IOException { - return executeCallable(new MasterCallable>(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable>(getConnection()) { @Override - protected List call(PayloadCarryingRpcController rpcController) - throws Exception { + public List call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); List snapshotsList = master - .getCompletedSnapshots(rpcController, GetCompletedSnapshotsRequest.newBuilder().build()) + .getCompletedSnapshots(controller, GetCompletedSnapshotsRequest.newBuilder().build()) .getSnapshotsList(); List result = new ArrayList(snapshotsList.size()); for (HBaseProtos.SnapshotDescription snapshot : snapshotsList) { @@ -2800,11 +2897,14 @@ public class HBaseAdmin implements Admin { // make sure the snapshot is possibly valid TableName.isLegalFullyQualifiedTableName(Bytes.toBytes(snapshotName)); // do the delete - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - master.deleteSnapshot(rpcController, - DeleteSnapshotRequest.newBuilder().setSnapshot( + public Void call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + master.deleteSnapshot(controller, + DeleteSnapshotRequest.newBuilder(). + setSnapshot( HBaseProtos.SnapshotDescription.newBuilder().setName(snapshotName).build()) .build() ); @@ -2833,10 +2933,12 @@ public class HBaseAdmin implements Admin { } private void internalDeleteSnapshot(final SnapshotDescription snapshot) throws IOException { - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { - this.master.deleteSnapshot(rpcController, DeleteSnapshotRequest.newBuilder() + public Void call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); + this.master.deleteSnapshot(controller, DeleteSnapshotRequest.newBuilder() .setSnapshot(createHBaseProtosSnapshotDesc(snapshot)).build()); return null; } @@ -2865,10 +2967,11 @@ public class HBaseAdmin implements Admin { @Override public void setQuota(final QuotaSettings quota) throws IOException { - executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + executeCallable(new MasterCallable(getConnection()) { @Override - protected Void call(PayloadCarryingRpcController rpcController) throws Exception { + public Void call(int callTimeout) throws ServiceException { PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); this.master.setQuota(controller, QuotaSettings.buildSetQuotaRequestProto(quota)); return null; } @@ -2886,8 +2989,8 @@ public class HBaseAdmin implements Admin { } static private & Closeable, V> V executeCallable(C callable, - RpcRetryingCallerFactory rpcCallerFactory, int operationTimeout, int rpcTimeout) - throws IOException { + RpcRetryingCallerFactory rpcCallerFactory, int operationTimeout, + int rpcTimeout) throws IOException { RpcRetryingCaller caller = rpcCallerFactory.newCaller(rpcTimeout); try { return caller.callWithRetries(callable, operationTimeout); @@ -2905,6 +3008,7 @@ public class HBaseAdmin implements Admin { * Simple {@link Abortable}, throwing RuntimeException on abort. */ private static class ThrowableAbortable implements Abortable { + @Override public void abort(String why, Throwable e) { throw new RuntimeException(why, e); @@ -2922,16 +3026,13 @@ public class HBaseAdmin implements Admin { } @Override - public void updateConfiguration(final ServerName server) throws IOException { - final AdminService.BlockingInterface admin = this.connection.getAdmin(server); - Callable callable = new Callable() { - @Override - public Void call() throws Exception { - admin.updateConfiguration(null, UpdateConfigurationRequest.getDefaultInstance()); - return null; - } - }; - ProtobufUtil.call(callable); + public void updateConfiguration(ServerName server) throws IOException { + try { + this.connection.getAdmin(server).updateConfiguration(null, + UpdateConfigurationRequest.getDefaultInstance()); + } catch (ServiceException e) { + throw ProtobufUtil.getRemoteException(e); + } } @Override @@ -2944,7 +3045,8 @@ public class HBaseAdmin implements Admin { @Override public int getMasterInfoPort() throws IOException { // TODO: Fix! Reaching into internal implementation!!!! - ConnectionImplementation connection = (ConnectionImplementation)this.connection; + ConnectionImplementation connection = + (ConnectionImplementation)this.connection; ZooKeeperKeepAliveConnection zkw = connection.getKeepAliveZooKeeperWatcher(); try { return MasterAddressTracker.getMasterInfoPort(zkw); @@ -2955,7 +3057,8 @@ public class HBaseAdmin implements Admin { private ServerName getMasterAddress() throws IOException { // TODO: Fix! Reaching into internal implementation!!!! - ConnectionImplementation connection = (ConnectionImplementation)this.connection; + ConnectionImplementation connection = + (ConnectionImplementation)this.connection; ZooKeeperKeepAliveConnection zkw = connection.getKeepAliveZooKeeperWatcher(); try { return MasterAddressTracker.getMasterAddress(zkw); @@ -2966,26 +3069,33 @@ public class HBaseAdmin implements Admin { @Override public long getLastMajorCompactionTimestamp(final TableName tableName) throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Long call(PayloadCarryingRpcController rpcController) throws Exception { + public Long call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); MajorCompactionTimestampRequest req = MajorCompactionTimestampRequest.newBuilder() .setTableName(ProtobufUtil.toProtoTableName(tableName)).build(); - return master.getLastMajorCompactionTimestamp(rpcController, req).getCompactionTimestamp(); + return master.getLastMajorCompactionTimestamp(controller, req).getCompactionTimestamp(); } }); } @Override public long getLastMajorCompactionTimestampForRegion(final byte[] regionName) throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Long call(PayloadCarryingRpcController rpcController) throws Exception { + public Long call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); MajorCompactionTimestampForRegionRequest req = - MajorCompactionTimestampForRegionRequest.newBuilder().setRegion(RequestConverter + MajorCompactionTimestampForRegionRequest + .newBuilder() + .setRegion( + RequestConverter .buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName)).build(); - return master.getLastMajorCompactionTimestampForRegion(rpcController, req) + return master.getLastMajorCompactionTimestampForRegion(controller, req) .getCompactionTimestamp(); } }); @@ -3024,35 +3134,32 @@ public class HBaseAdmin implements Admin { @Override public void majorCompact(final TableName tableName, CompactType compactType) throws IOException, InterruptedException { - compact(tableName, null, true, compactType); + compact(tableName, null, true, compactType); } /** * {@inheritDoc} */ @Override - public CompactionState getCompactionState(final TableName tableName, + public CompactionState getCompactionState(TableName tableName, CompactType compactType) throws IOException { AdminProtos.GetRegionInfoResponse.CompactionState state = AdminProtos.GetRegionInfoResponse.CompactionState.NONE; checkTableExists(tableName); - final PayloadCarryingRpcController rpcController = rpcControllerFactory.newController(); + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); switch (compactType) { case MOB: - final AdminProtos.AdminService.BlockingInterface masterAdmin = - this.connection.getAdmin(getMasterAddress()); - Callable callable = - new Callable() { - @Override - public AdminProtos.GetRegionInfoResponse.CompactionState call() throws Exception { - HRegionInfo info = getMobRegionInfo(tableName); - GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest( - info.getRegionName(), true); - GetRegionInfoResponse response = masterAdmin.getRegionInfo(rpcController, request); - return response.getCompactionState(); - } - }; - state = ProtobufUtil.call(callable); + try { + ServerName master = getMasterAddress(); + HRegionInfo info = getMobRegionInfo(tableName); + GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest( + info.getRegionName(), true); + GetRegionInfoResponse response = this.connection.getAdmin(master) + .getRegionInfo(controller, request); + state = response.getCompactionState(); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } break; case NORMAL: default: @@ -3066,23 +3173,15 @@ public class HBaseAdmin implements Admin { } else { pairs = MetaTableAccessor.getTableRegionsAndLocations(connection, tableName); } - for (Pair pair: pairs) { + for (Pair pair : pairs) { if (pair.getFirst().isOffline()) continue; if (pair.getSecond() == null) continue; - final ServerName sn = pair.getSecond(); - final byte [] regionName = pair.getFirst().getRegionName(); - final AdminService.BlockingInterface snAdmin = this.connection.getAdmin(sn); try { - Callable regionInfoCallable = - new Callable() { - @Override - public GetRegionInfoResponse call() throws Exception { - GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest( - regionName, true); - return snAdmin.getRegionInfo(rpcController, request); - } - }; - GetRegionInfoResponse response = ProtobufUtil.call(regionInfoCallable); + ServerName sn = pair.getSecond(); + AdminService.BlockingInterface admin = this.connection.getAdmin(sn); + GetRegionInfoRequest request = RequestConverter.buildGetRegionInfoRequest( + pair.getFirst().getRegionName(), true); + GetRegionInfoResponse response = admin.getRegionInfo(controller, request); switch (response.getCompactionState()) { case MAJOR_AND_MINOR: return CompactionState.MAJOR_AND_MINOR; @@ -3118,6 +3217,8 @@ public class HBaseAdmin implements Admin { } } } + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } finally { if (zookeeper != null) { zookeeper.close(); @@ -3182,11 +3283,12 @@ public class HBaseAdmin implements Admin { protected AbortProcedureResponse abortProcedureResult( final AbortProcedureRequest request) throws IOException { return admin.executeCallable(new MasterCallable( - admin.getConnection(), admin.getRpcControllerFactory()) { + admin.getConnection()) { @Override - protected AbortProcedureResponse call(PayloadCarryingRpcController rcpController) - throws Exception { - return master.abortProcedure(rcpController, request); + public AbortProcedureResponse call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = admin.getRpcControllerFactory().newController(); + controller.setCallTimeout(callTimeout); + return master.abortProcedure(controller, request); } }); } @@ -3299,10 +3401,9 @@ public class HBaseAdmin implements Admin { protected GetProcedureResultResponse getProcedureResult(final GetProcedureResultRequest request) throws IOException { return admin.executeCallable(new MasterCallable( - admin.getConnection(), admin.getRpcControllerFactory()) { + admin.getConnection()) { @Override - protected GetProcedureResultResponse call(PayloadCarryingRpcController rpcController) - throws Exception { + public GetProcedureResultResponse call(int callTimeout) throws ServiceException { return master.getProcedureResult(null, request); } }); @@ -3598,14 +3699,14 @@ public class HBaseAdmin implements Admin { @Override public List getSecurityCapabilities() throws IOException { try { - return executeCallable(new MasterCallable>(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable>(getConnection()) { @Override - protected List call(PayloadCarryingRpcController rpcContoller) - throws Exception { + public List call(int callTimeout) throws ServiceException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setCallTimeout(callTimeout); SecurityCapabilitiesRequest req = SecurityCapabilitiesRequest.newBuilder().build(); return ProtobufUtil.toSecurityCapabilityList( - master.getSecurityCapabilities(rpcContoller, req).getCapabilitiesList()); + master.getSecurityCapabilities(controller, req).getCapabilitiesList()); } }); } catch (IOException e) { @@ -3620,10 +3721,9 @@ public class HBaseAdmin implements Admin { public boolean[] setSplitOrMergeEnabled(final boolean enabled, final boolean synchronous, final MasterSwitchType... switchTypes) throws IOException { - return executeCallable(new MasterCallable(getConnection(), - getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected boolean[] call(PayloadCarryingRpcController rpcController) throws Exception { + public boolean[] call(int callTimeout) throws ServiceException { MasterProtos.SetSplitOrMergeEnabledResponse response = master.setSplitOrMergeEnabled(null, RequestConverter.buildSetSplitOrMergeEnabledRequest(enabled, synchronous, switchTypes)); boolean[] result = new boolean[switchTypes.length]; @@ -3638,9 +3738,9 @@ public class HBaseAdmin implements Admin { @Override public boolean isSplitOrMergeEnabled(final MasterSwitchType switchType) throws IOException { - return executeCallable(new MasterCallable(getConnection(), getRpcControllerFactory()) { + return executeCallable(new MasterCallable(getConnection()) { @Override - protected Boolean call(PayloadCarryingRpcController rpcController) throws Exception { + public Boolean call(int callTimeout) throws ServiceException { return master.isSplitOrMergeEnabled(null, RequestConverter.buildIsSplitOrMergeEnabledRequest(switchType)).getEnabled(); } @@ -3653,6 +3753,6 @@ public class HBaseAdmin implements Admin { } private RpcControllerFactory getRpcControllerFactory() { - return this.rpcControllerFactory; + return rpcControllerFactory; } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java index 1b3e11102a..fbd9f51cf7 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java @@ -18,6 +18,12 @@ */ package org.apache.hadoop.hbase.client; +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.Descriptors; +import com.google.protobuf.Message; +import com.google.protobuf.Service; +import com.google.protobuf.ServiceException; + import java.io.IOException; import java.io.InterruptedIOException; import java.util.ArrayList; @@ -37,6 +43,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.DoNotRetryIOException; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.HTableDescriptor; @@ -67,16 +74,6 @@ import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.hbase.util.ReflectionUtils; import org.apache.hadoop.hbase.util.Threads; -import com.google.common.annotations.VisibleForTesting; - -// DO NOT MAKE USE OF THESE IMPORTS! THEY ARE HERE FOR COPROCESSOR ENDPOINTS ONLY. -// Internally, we use shaded protobuf. This below are part of our public API. -import com.google.protobuf.Descriptors; -import com.google.protobuf.Message; -import com.google.protobuf.ServiceException; -import com.google.protobuf.Service; -// SEE ABOVE NOTE! - /** * An implementation of {@link Table}. Used to communicate with a single HBase table. * Lightweight. Get as needed and just close when done. @@ -414,16 +411,23 @@ public class HTable implements Table { if (get.getConsistency() == Consistency.STRONG) { // Good old call. - final Get configuredGet = get; + final Get getReq = get; RegionServerCallable callable = new RegionServerCallable(this.connection, - this.rpcControllerFactory, getName(), get.getRow()) { + getName(), get.getRow()) { @Override - protected Result call(PayloadCarryingRpcController controller) throws Exception { - ClientProtos.GetRequest request = RequestConverter.buildGetRequest( - getLocation().getRegionInfo().getRegionName(), configuredGet); - ClientProtos.GetResponse response = getStub().get(controller, request); - if (response == null) return null; - return ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); + public Result call(int callTimeout) throws IOException { + ClientProtos.GetRequest request = + RequestConverter.buildGetRequest(getLocation().getRegionInfo().getRegionName(), getReq); + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setPriority(tableName); + controller.setCallTimeout(callTimeout); + try { + ClientProtos.GetResponse response = getStub().get(controller, request); + if (response == null) return null; + return ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } }; return rpcCallerFactory.newCaller(rpcTimeout).callWithRetries(callable, @@ -439,6 +443,7 @@ public class HTable implements Table { return callable.call(operationTimeout); } + /** * {@inheritDoc} */ @@ -449,14 +454,16 @@ public class HTable implements Table { } try { Object[] r1 = new Object[gets.size()]; - batch((List)gets, r1); - // Translate. + batch((List) gets, r1); + + // translate. Result [] results = new Result[r1.length]; - int i = 0; - for (Object obj: r1) { - // Batch ensures if there is a failure we get an exception instead - results[i++] = (Result)obj; + int i=0; + for (Object o : r1) { + // batch ensures if there is a failure we get an exception instead + results[i++] = (Result) o; } + return results; } catch (InterruptedException e) { throw (InterruptedIOException)new InterruptedIOException().initCause(e); @@ -504,13 +511,21 @@ public class HTable implements Table { public void delete(final Delete delete) throws IOException { RegionServerCallable callable = new RegionServerCallable(connection, - this.rpcControllerFactory, getName(), delete.getRow()) { + tableName, delete.getRow()) { @Override - protected Boolean call(PayloadCarryingRpcController controller) throws Exception { - MutateRequest request = RequestConverter.buildMutateRequest( - getLocation().getRegionInfo().getRegionName(), delete); - MutateResponse response = getStub().mutate(controller, request); - return Boolean.valueOf(response.getProcessed()); + public Boolean call(int callTimeout) throws IOException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setPriority(tableName); + controller.setCallTimeout(callTimeout); + + try { + MutateRequest request = RequestConverter.buildMutateRequest( + getLocation().getRegionInfo().getRegionName(), delete); + MutateResponse response = getStub().mutate(controller, request); + return Boolean.valueOf(response.getProcessed()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } }; rpcCallerFactory. newCaller(rpcTimeout).callWithRetries(callable, @@ -566,28 +581,41 @@ public class HTable implements Table { */ @Override public void mutateRow(final RowMutations rm) throws IOException { + final RetryingTimeTracker tracker = new RetryingTimeTracker(); PayloadCarryingServerCallable callable = - new PayloadCarryingServerCallable(this.connection, getName(), rm.getRow(), + new PayloadCarryingServerCallable(connection, getName(), rm.getRow(), rpcControllerFactory) { - @Override - protected MultiResponse call(PayloadCarryingRpcController controller) throws Exception { - RegionAction.Builder regionMutationBuilder = RequestConverter.buildRegionAction( - getLocation().getRegionInfo().getRegionName(), rm); - regionMutationBuilder.setAtomic(true); - MultiRequest request = - MultiRequest.newBuilder().addRegionAction(regionMutationBuilder.build()).build(); - ClientProtos.MultiResponse response = getStub().multi(controller, request); - ClientProtos.RegionActionResult res = response.getRegionActionResultList().get(0); - if (res.hasException()) { - Throwable ex = ProtobufUtil.toException(res.getException()); - if (ex instanceof IOException) { - throw (IOException) ex; + @Override + public MultiResponse call(int callTimeout) throws IOException { + tracker.start(); + controller.setPriority(tableName); + int remainingTime = tracker.getRemainingTime(callTimeout); + if (remainingTime == 0) { + throw new DoNotRetryIOException("Timeout for mutate row"); + } + controller.setCallTimeout(remainingTime); + try { + RegionAction.Builder regionMutationBuilder = RequestConverter.buildRegionAction( + getLocation().getRegionInfo().getRegionName(), rm); + regionMutationBuilder.setAtomic(true); + MultiRequest request = + MultiRequest.newBuilder().addRegionAction(regionMutationBuilder.build()).build(); + ClientProtos.MultiResponse response = getStub().multi(controller, request); + ClientProtos.RegionActionResult res = response.getRegionActionResultList().get(0); + if (res.hasException()) { + Throwable ex = ProtobufUtil.toException(res.getException()); + if (ex instanceof IOException) { + throw (IOException) ex; + } + throw new IOException("Failed to mutate row: " + + Bytes.toStringBinary(rm.getRow()), ex); + } + return ResponseConverter.getResults(request, response, controller.cellScanner()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } - throw new IOException("Failed to mutate row: " + Bytes.toStringBinary(rm.getRow()), ex); } - return ResponseConverter.getResults(request, response, controller.cellScanner()); - } - }; + }; AsyncRequestFuture ars = multiAp.submitAll(pool, tableName, rm.getMutations(), null, null, callable, operationTimeout); ars.waitUntilDone(); @@ -596,31 +624,38 @@ public class HTable implements Table { } } - private static void checkHasFamilies(final Mutation mutation) throws IOException { - if (mutation.numFamilies() == 0) { - throw new IOException("Invalid arguments to " + mutation + ", zero columns specified"); - } - } - /** * {@inheritDoc} */ @Override public Result append(final Append append) throws IOException { - checkHasFamilies(append); - RegionServerCallable callable = new RegionServerCallable(this.connection, - this.rpcControllerFactory, getName(), append.getRow()) { - @Override - protected Result call(PayloadCarryingRpcController controller) throws Exception { - MutateRequest request = RequestConverter.buildMutateRequest( - getLocation().getRegionInfo().getRegionName(), append, getNonceGroup(), getNewNonce()); - MutateResponse response = getStub().mutate(controller, request); - if (!response.hasResult()) return null; - return ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); - } - }; - return rpcCallerFactory. newCaller(this.rpcTimeout). - callWithRetries(callable, this.operationTimeout); + if (append.numFamilies() == 0) { + throw new IOException( + "Invalid arguments to append, no columns specified"); + } + + NonceGenerator ng = this.connection.getNonceGenerator(); + final long nonceGroup = ng.getNonceGroup(), nonce = ng.newNonce(); + RegionServerCallable callable = + new RegionServerCallable(this.connection, getName(), append.getRow()) { + @Override + public Result call(int callTimeout) throws IOException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setPriority(getTableName()); + controller.setCallTimeout(callTimeout); + try { + MutateRequest request = RequestConverter.buildMutateRequest( + getLocation().getRegionInfo().getRegionName(), append, nonceGroup, nonce); + MutateResponse response = getStub().mutate(controller, request); + if (!response.hasResult()) return null; + return ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + } + }; + return rpcCallerFactory. newCaller(rpcTimeout).callWithRetries(callable, + this.operationTimeout); } /** @@ -628,16 +663,27 @@ public class HTable implements Table { */ @Override public Result increment(final Increment increment) throws IOException { - checkHasFamilies(increment); + if (!increment.hasFamilies()) { + throw new IOException( + "Invalid arguments to increment, no columns specified"); + } + NonceGenerator ng = this.connection.getNonceGenerator(); + final long nonceGroup = ng.getNonceGroup(), nonce = ng.newNonce(); RegionServerCallable callable = new RegionServerCallable(this.connection, - this.rpcControllerFactory, getName(), increment.getRow()) { + getName(), increment.getRow()) { @Override - protected Result call(PayloadCarryingRpcController controller) throws Exception { - MutateRequest request = RequestConverter.buildMutateRequest( - getLocation().getRegionInfo().getRegionName(), increment, getNonceGroup(), getNewNonce()); - MutateResponse response = getStub().mutate(controller, request); - // Should this check for null like append does? - return ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); + public Result call(int callTimeout) throws IOException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setPriority(getTableName()); + controller.setCallTimeout(callTimeout); + try { + MutateRequest request = RequestConverter.buildMutateRequest( + getLocation().getRegionInfo().getRegionName(), increment, nonceGroup, nonce); + MutateResponse response = getStub().mutate(controller, request); + return ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } }; return rpcCallerFactory. newCaller(rpcTimeout).callWithRetries(callable, @@ -676,20 +722,28 @@ public class HTable implements Table { NonceGenerator ng = this.connection.getNonceGenerator(); final long nonceGroup = ng.getNonceGroup(), nonce = ng.newNonce(); - RegionServerCallable callable = new RegionServerCallable(this.connection, - this.rpcControllerFactory, getName(), row) { - @Override - protected Long call(PayloadCarryingRpcController controller) throws Exception { - MutateRequest request = RequestConverter.buildIncrementRequest( - getLocation().getRegionInfo().getRegionName(), row, family, - qualifier, amount, durability, nonceGroup, nonce); - MutateResponse response = getStub().mutate(controller, request); - Result result = ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); - return Long.valueOf(Bytes.toLong(result.getValue(family, qualifier))); - } - }; - return rpcCallerFactory. newCaller(rpcTimeout). - callWithRetries(callable, this.operationTimeout); + RegionServerCallable callable = + new RegionServerCallable(connection, getName(), row) { + @Override + public Long call(int callTimeout) throws IOException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setPriority(getTableName()); + controller.setCallTimeout(callTimeout); + try { + MutateRequest request = RequestConverter.buildIncrementRequest( + getLocation().getRegionInfo().getRegionName(), row, family, + qualifier, amount, durability, nonceGroup, nonce); + MutateResponse response = getStub().mutate(controller, request); + Result result = + ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); + return Long.valueOf(Bytes.toLong(result.getValue(family, qualifier))); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + } + }; + return rpcCallerFactory. newCaller(rpcTimeout).callWithRetries(callable, + this.operationTimeout); } /** @@ -700,19 +754,26 @@ public class HTable implements Table { final byte [] family, final byte [] qualifier, final byte [] value, final Put put) throws IOException { - RegionServerCallable callable = new RegionServerCallable(this.connection, - this.rpcControllerFactory, getName(), row) { - @Override - protected Boolean call(PayloadCarryingRpcController controller) throws Exception { - MutateRequest request = RequestConverter.buildMutateRequest( - getLocation().getRegionInfo().getRegionName(), row, family, qualifier, - new BinaryComparator(value), CompareType.EQUAL, put); - MutateResponse response = getStub().mutate(controller, request); - return Boolean.valueOf(response.getProcessed()); - } - }; - return rpcCallerFactory. newCaller(rpcTimeout). - callWithRetries(callable, this.operationTimeout); + RegionServerCallable callable = + new RegionServerCallable(connection, getName(), row) { + @Override + public Boolean call(int callTimeout) throws IOException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setPriority(tableName); + controller.setCallTimeout(callTimeout); + try { + MutateRequest request = RequestConverter.buildMutateRequest( + getLocation().getRegionInfo().getRegionName(), row, family, qualifier, + new BinaryComparator(value), CompareType.EQUAL, put); + MutateResponse response = getStub().mutate(controller, request); + return Boolean.valueOf(response.getProcessed()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + } + }; + return rpcCallerFactory. newCaller(rpcTimeout).callWithRetries(callable, + this.operationTimeout); } /** @@ -723,42 +784,57 @@ public class HTable implements Table { final byte [] qualifier, final CompareOp compareOp, final byte [] value, final Put put) throws IOException { - RegionServerCallable callable = new RegionServerCallable(this.connection, - this.rpcControllerFactory, getName(), row) { - @Override - protected Boolean call(PayloadCarryingRpcController controller) throws Exception { - CompareType compareType = CompareType.valueOf(compareOp.name()); - MutateRequest request = RequestConverter.buildMutateRequest( - getLocation().getRegionInfo().getRegionName(), row, family, qualifier, - new BinaryComparator(value), compareType, put); - MutateResponse response = getStub().mutate(controller, request); - return Boolean.valueOf(response.getProcessed()); - } - }; - return rpcCallerFactory. newCaller(rpcTimeout). - callWithRetries(callable, this.operationTimeout); + RegionServerCallable callable = + new RegionServerCallable(connection, getName(), row) { + @Override + public Boolean call(int callTimeout) throws IOException { + PayloadCarryingRpcController controller = new PayloadCarryingRpcController(); + controller.setPriority(tableName); + controller.setCallTimeout(callTimeout); + try { + CompareType compareType = CompareType.valueOf(compareOp.name()); + MutateRequest request = RequestConverter.buildMutateRequest( + getLocation().getRegionInfo().getRegionName(), row, family, qualifier, + new BinaryComparator(value), compareType, put); + MutateResponse response = getStub().mutate(controller, request); + return Boolean.valueOf(response.getProcessed()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + } + }; + return rpcCallerFactory. newCaller(rpcTimeout).callWithRetries(callable, + this.operationTimeout); } /** * {@inheritDoc} */ @Override - public boolean checkAndDelete(final byte [] row, final byte [] family, final byte [] qualifier, - final byte [] value, final Delete delete) + public boolean checkAndDelete(final byte [] row, + final byte [] family, final byte [] qualifier, final byte [] value, + final Delete delete) throws IOException { - RegionServerCallable callable = new RegionServerCallable(this.connection, - this.rpcControllerFactory, getName(), row) { - @Override - protected Boolean call(PayloadCarryingRpcController controller) throws Exception { - MutateRequest request = RequestConverter.buildMutateRequest( - getLocation().getRegionInfo().getRegionName(), row, family, qualifier, - new BinaryComparator(value), CompareType.EQUAL, delete); - MutateResponse response = getStub().mutate(controller, request); - return Boolean.valueOf(response.getProcessed()); - } - }; - return rpcCallerFactory. newCaller(rpcTimeout). - callWithRetries(callable, this.operationTimeout); + RegionServerCallable callable = + new RegionServerCallable(connection, getName(), row) { + @Override + public Boolean call(int callTimeout) throws IOException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setPriority(tableName); + controller.setCallTimeout(callTimeout); + try { + MutateRequest request = RequestConverter.buildMutateRequest( + getLocation().getRegionInfo().getRegionName(), row, family, qualifier, + new BinaryComparator(value), CompareType.EQUAL, delete); + MutateResponse response = getStub().mutate(controller, request); + return Boolean.valueOf(response.getProcessed()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + } + }; + return rpcCallerFactory. newCaller(rpcTimeout).callWithRetries(callable, + this.operationTimeout); } /** @@ -769,18 +845,25 @@ public class HTable implements Table { final byte [] qualifier, final CompareOp compareOp, final byte [] value, final Delete delete) throws IOException { - RegionServerCallable callable = new RegionServerCallable(this.connection, - this.rpcControllerFactory, getName(), row) { - @Override - protected Boolean call(PayloadCarryingRpcController controller) throws Exception { - CompareType compareType = CompareType.valueOf(compareOp.name()); - MutateRequest request = RequestConverter.buildMutateRequest( - getLocation().getRegionInfo().getRegionName(), row, family, qualifier, - new BinaryComparator(value), compareType, delete); - MutateResponse response = getStub().mutate(controller, request); - return Boolean.valueOf(response.getProcessed()); - } - }; + RegionServerCallable callable = + new RegionServerCallable(connection, getName(), row) { + @Override + public Boolean call(int callTimeout) throws IOException { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(); + controller.setPriority(tableName); + controller.setCallTimeout(callTimeout); + try { + CompareType compareType = CompareType.valueOf(compareOp.name()); + MutateRequest request = RequestConverter.buildMutateRequest( + getLocation().getRegionInfo().getRegionName(), row, family, qualifier, + new BinaryComparator(value), compareType, delete); + MutateResponse response = getStub().mutate(controller, request); + return Boolean.valueOf(response.getProcessed()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + } + }; return rpcCallerFactory. newCaller(rpcTimeout).callWithRetries(callable, this.operationTimeout); } @@ -792,28 +875,40 @@ public class HTable implements Table { public boolean checkAndMutate(final byte [] row, final byte [] family, final byte [] qualifier, final CompareOp compareOp, final byte [] value, final RowMutations rm) throws IOException { + final RetryingTimeTracker tracker = new RetryingTimeTracker(); PayloadCarryingServerCallable callable = new PayloadCarryingServerCallable(connection, getName(), rm.getRow(), rpcControllerFactory) { @Override - protected MultiResponse call(PayloadCarryingRpcController controller) throws Exception { - CompareType compareType = CompareType.valueOf(compareOp.name()); - MultiRequest request = RequestConverter.buildMutateRequest( - getLocation().getRegionInfo().getRegionName(), row, family, qualifier, - new BinaryComparator(value), compareType, rm); - ClientProtos.MultiResponse response = getStub().multi(controller, request); - ClientProtos.RegionActionResult res = response.getRegionActionResultList().get(0); - if (res.hasException()) { - Throwable ex = ProtobufUtil.toException(res.getException()); - if (ex instanceof IOException) { - throw (IOException)ex; + public MultiResponse call(int callTimeout) throws IOException { + tracker.start(); + controller.setPriority(tableName); + int remainingTime = tracker.getRemainingTime(callTimeout); + if (remainingTime == 0) { + throw new DoNotRetryIOException("Timeout for mutate row"); + } + controller.setCallTimeout(remainingTime); + try { + CompareType compareType = CompareType.valueOf(compareOp.name()); + MultiRequest request = RequestConverter.buildMutateRequest( + getLocation().getRegionInfo().getRegionName(), row, family, qualifier, + new BinaryComparator(value), compareType, rm); + ClientProtos.MultiResponse response = getStub().multi(controller, request); + ClientProtos.RegionActionResult res = response.getRegionActionResultList().get(0); + if (res.hasException()) { + Throwable ex = ProtobufUtil.toException(res.getException()); + if(ex instanceof IOException) { + throw (IOException)ex; + } + throw new IOException("Failed to checkAndMutate row: "+ + Bytes.toStringBinary(rm.getRow()), ex); } - throw new IOException("Failed to checkAndMutate row: "+ Bytes.toStringBinary(rm.getRow()), ex); + return ResponseConverter.getResults(request, response, controller.cellScanner()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } - return ResponseConverter.getResults(request, response, controller.cellScanner()); } }; - /** * Currently, we use one array to store 'processed' flag which is returned by server. * It is excessive to send such a large array, but that is required by the framework right now @@ -873,6 +968,7 @@ public class HTable implements Table { } /** + * {@inheritDoc} * @throws IOException */ void flushCommits() throws IOException { @@ -1049,18 +1145,19 @@ public class HTable implements Table { for (final byte[] r : keys) { final RegionCoprocessorRpcChannel channel = new RegionCoprocessorRpcChannel(connection, tableName, r); - Future future = pool.submit(new Callable() { - @Override - public R call() throws Exception { - T instance = ProtobufUtil.newServiceStub(service, channel); - R result = callable.call(instance); - byte[] region = channel.getLastRegion(); - if (callback != null) { - callback.update(region, r, result); - } - return result; - } - }); + Future future = pool.submit( + new Callable() { + @Override + public R call() throws Exception { + T instance = ProtobufUtil.newServiceStub(service, channel); + R result = callable.call(instance); + byte[] region = channel.getLastRegion(); + if (callback != null) { + callback.update(region, r, result); + } + return result; + } + }); futures.put(r, future); } for (Map.Entry> e : futures.entrySet()) { @@ -1113,6 +1210,9 @@ public class HTable implements Table { return tableName + ";" + connection; } + /** + * {@inheritDoc} + */ @Override public Map batchCoprocessorService( Descriptors.MethodDescriptor methodDescriptor, Message request, @@ -1121,13 +1221,14 @@ public class HTable implements Table { Bytes.BYTES_COMPARATOR)); batchCoprocessorService(methodDescriptor, request, startKey, endKey, responsePrototype, new Callback() { - @Override - public void update(byte[] region, byte[] row, R result) { - if (region != null) { - results.put(region, result); - } - } - }); + + @Override + public void update(byte[] region, byte[] row, R result) { + if (region != null) { + results.put(region, result); + } + } + }); return results; } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MasterCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MasterCallable.java index ae622556e8..66d3c21d68 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MasterCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MasterCallable.java @@ -21,24 +21,16 @@ package org.apache.hadoop.hbase.client; import java.io.Closeable; import java.io.IOException; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; -import org.apache.hadoop.hbase.ipc.RpcControllerFactory; -import org.apache.hadoop.hbase.protobuf.ProtobufUtil; - /** * A RetryingCallable for master operations. * @param return type */ -// Like RegionServerCallable abstract class MasterCallable implements RetryingCallable, Closeable { protected ClusterConnection connection; protected MasterKeepAliveConnection master; - private final PayloadCarryingRpcController rpcController; - MasterCallable(final Connection connection, - final RpcControllerFactory rpcConnectionFactory) { + public MasterCallable(final Connection connection) { this.connection = (ClusterConnection) connection; - this.rpcController = rpcConnectionFactory.newController(); } @Override @@ -67,31 +59,4 @@ abstract class MasterCallable implements RetryingCallable, Closeable { public long sleep(long pause, int tries) { return ConnectionUtils.getPauseTime(pause, tries); } - - /** - * Override that changes Exception from {@link Exception} to {@link IOException}. It also does - * setup of an rpcController and calls through to the unimplemented - * call(PayloadCarryingRpcController) method; implement this method to add your rpc invocation. - */ - @Override - // Same trick as in RegionServerCallable so users don't have to copy/paste so much boilerplate - // and so we contain references to protobuf. We can't set priority on the rpcController as - // we do in RegionServerCallable because we don't always have a Table when we call. - public V call(int callTimeout) throws IOException { - try { - this.rpcController.setCallTimeout(callTimeout); - return call(this.rpcController); - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); - } - } - - /** - * Run RPC call. - * @param rpcController PayloadCarryingRpcController is a mouthful but it at a minimum is a - * facade on protobuf so we don't have to put protobuf everywhere; we can keep it behind this - * class. - * @throws Exception - */ - protected abstract V call(PayloadCarryingRpcController rpcController) throws Exception; } \ No newline at end of file diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MasterKeepAliveConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MasterKeepAliveConnection.java index 47693f40c3..e445b785f0 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MasterKeepAliveConnection.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MasterKeepAliveConnection.java @@ -33,7 +33,8 @@ import org.apache.hadoop.hbase.protobuf.generated.MasterProtos; * against the master on the MasterProtos.MasterService.BlockingInterface; but not by * final user code. Hence it's package protected. */ -interface MasterKeepAliveConnection extends MasterProtos.MasterService.BlockingInterface { +interface MasterKeepAliveConnection +extends MasterProtos.MasterService.BlockingInterface { // Do this instead of implement Closeable because closeable returning IOE is PITA. void close(); } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiServerCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiServerCallable.java index a3162f4995..e764ceb45c 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiServerCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MultiServerCallable.java @@ -30,9 +30,8 @@ import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.TableName; -import org.apache.hadoop.hbase.classification.InterfaceAudience; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; import org.apache.hadoop.hbase.ipc.RpcControllerFactory; +import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.ResponseConverter; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos; @@ -42,14 +41,14 @@ import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.RegionAction; import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos; import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.ServiceException; /** * Callable that handles the multi method call going against a single - * regionserver; i.e. A RegionServerCallable for the multi call (It is NOT a - * RegionServerCallable that goes against multiple regions). + * regionserver; i.e. A {@link RegionServerCallable} for the multi call (It is not a + * {@link RegionServerCallable} that goes against multiple regions. * @param */ -@InterfaceAudience.Private class MultiServerCallable extends PayloadCarryingServerCallable { private final MultiAction multiAction; private final boolean cellBlock; @@ -80,7 +79,7 @@ class MultiServerCallable extends PayloadCarryingServerCallable extends PayloadCarryingServerCallable(countOfActions); // Send data in cellblocks. The call to buildNoDataMultiRequest will skip RowMutations. // They have already been handled above. Guess at count of cells @@ -115,18 +116,18 @@ class MultiServerCallable extends PayloadCarryingServerCallable extends PayloadCarryingServerCallable extends RegionServerCallable - implements Cancellable { - private final RetryingTimeTracker tracker = new RetryingTimeTracker(); - - PayloadCarryingServerCallable(Connection connection, TableName tableName, byte[] row, - RpcControllerFactory rpcControllerFactory) { - super(connection, rpcControllerFactory, tableName, row); - } +public abstract class PayloadCarryingServerCallable + extends RegionServerCallable implements Cancellable { + protected PayloadCarryingRpcController controller; - /* Override so can mess with the callTimeout. - * (non-Javadoc) - * @see org.apache.hadoop.hbase.client.RegionServerCallable#rpcCall(int) - */ - @Override - public T call(int callTimeout) throws IOException { - // It is expected (it seems) that tracker.start can be called multiple times (on each trip - // through the call when retrying). Also, we can call start and no need of a stop. - this.tracker.start(); - int remainingTime = tracker.getRemainingTime(callTimeout); - if (remainingTime == 0) { - throw new DoNotRetryIOException("Timeout for mutate row"); - } - return super.call(remainingTime); + public PayloadCarryingServerCallable(Connection connection, TableName tableName, byte[] row, + RpcControllerFactory rpcControllerFactory) { + super(connection, tableName, row); + this.controller = rpcControllerFactory.newController(); } @Override public void cancel() { - getRpcController().startCancel(); + controller.startCancel(); } @Override public boolean isCancelled() { - return getRpcController().isCanceled(); + return controller.isCanceled(); } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionAdminServiceCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionAdminServiceCallable.java index 4e347ddeff..54c93a0b2b 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionAdminServiceCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionAdminServiceCallable.java @@ -27,30 +27,31 @@ import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.RegionLocations; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.classification.InterfaceAudience; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; import org.apache.hadoop.hbase.ipc.RpcControllerFactory; -import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService; import org.apache.hadoop.hbase.util.Bytes; /** - * Similar to RegionServerCallable but for the AdminService interface. This service callable + * Similar to {@link RegionServerCallable} but for the AdminService interface. This service callable * assumes a Table and row and thus does region locating similar to RegionServerCallable. - * Works against Admin stub rather than Client stub. */ @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD", justification="stub used by ipc") @InterfaceAudience.Private public abstract class RegionAdminServiceCallable implements RetryingCallable { - protected AdminService.BlockingInterface stub; - protected final RpcControllerFactory rpcControllerFactory; - private PayloadCarryingRpcController controller = null; protected final ClusterConnection connection; + + protected final RpcControllerFactory rpcControllerFactory; + + protected AdminService.BlockingInterface stub; + protected HRegionLocation location; + protected final TableName tableName; protected final byte[] row; protected final int replicaId; + protected final static int MIN_WAIT_DEAD_SERVER = 10000; public RegionAdminServiceCallable(ClusterConnection connection, @@ -81,13 +82,16 @@ public abstract class RegionAdminServiceCallable implements RetryingCallable< if (Thread.interrupted()) { throw new InterruptedIOException(); } + if (reload || location == null) { location = getLocation(!reload); } + if (location == null) { // With this exception, there will be a retry. throw new HBaseIOException(getExceptionMessage()); } + this.setStub(connection.getAdmin(location.getServerName())); } @@ -163,39 +167,7 @@ public abstract class RegionAdminServiceCallable implements RetryingCallable< if (rl == null) { throw new RetriesExhaustedException("Can't get the locations"); } - return rl; - } - - /** - * Override that changes Exception from {@link Exception} to {@link IOException}. It also does - * setup of an rpcController and calls through to the unimplemented - * call(PayloadCarryingRpcController) method; implement this method to add your rpc invocation. - */ - @Override - // Same trick as in RegionServerCallable so users don't have to copy/paste so much boilerplate - // and so we contain references to protobuf. We can't set priority on the rpcController as - // we do in RegionServerCallable because we don't always have a Table when we call. - public T call(int callTimeout) throws IOException { - this.controller = rpcControllerFactory.newController(); - this.controller.setPriority(this.tableName); - this.controller.setCallTimeout(callTimeout); - try { - return call(this.controller); - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); - } - } - PayloadCarryingRpcController getCurrentPayloadCarryingRpcController() { - return this.controller; + return rl; } - - /** - * Run RPC call. - * @param rpcController PayloadCarryingRpcController is a mouthful but it at a minimum is a - * facade on protobuf so we don't have to put protobuf everywhere; we can keep it behind this - * class. - * @throws Exception - */ - protected abstract T call(PayloadCarryingRpcController rpcController) throws Exception; -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionServerCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionServerCallable.java index 861b3759af..d878bae94e 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionServerCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionServerCallable.java @@ -1,4 +1,5 @@ /** + * * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information @@ -23,20 +24,12 @@ import java.io.IOException; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.classification.InterfaceAudience; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; -import org.apache.hadoop.hbase.ipc.RpcControllerFactory; -import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; /** - * Implementations make an rpc call against a RegionService via a protobuf Service. - * Implement #rpcCall(RpcController) and then call {@link #call(int)} to - * trigger the rpc. The {@link #call(int)} eventually invokes your - * #rpcCall(RpcController) meanwhile saving you having to write a bunch of - * boilerplate. The {@link #call(int)} implementation is from {@link RpcRetryingCaller} so rpcs are - * retried on fail. - * - *

TODO: this class is actually tied to one region, because most of the paths make use of + * Implementations call a RegionServer and implement {@link #call(int)}. + * Passed to a {@link RpcRetryingCaller} so we retry on fail. + * TODO: this class is actually tied to one region, because most of the paths make use of * the regioninfo part of location when building requests. The only reason it works for * multi-region requests (e.g. batch) is that they happen to not use the region parts. * This could be done cleaner (e.g. having a generic parameter and 2 derived classes, @@ -44,27 +37,18 @@ import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService; * @param the class that the ServerCallable handles */ @InterfaceAudience.Private -public abstract class RegionServerCallable extends AbstractRegionServerCallable { +public abstract class RegionServerCallable extends AbstractRegionServerCallable implements + RetryingCallable { + private ClientService.BlockingInterface stub; - private final PayloadCarryingRpcController rpcController; /** * @param connection Connection to use. * @param tableName Table name to which row belongs. * @param row The row we want in tableName. */ - public RegionServerCallable(Connection connection, RpcControllerFactory rpcControllerFactory, - TableName tableName, byte [] row) { - this(connection, rpcControllerFactory.newController(), tableName, row); - } - - public RegionServerCallable(Connection connection, PayloadCarryingRpcController rpcController, - TableName tableName, byte [] row) { + public RegionServerCallable(Connection connection, TableName tableName, byte [] row) { super(connection, tableName, row); - this.rpcController = rpcController; - if (this.rpcController != null) { - this.rpcController.setPriority(tableName); - } } void setClientByServiceName(ServerName service) throws IOException { @@ -85,42 +69,4 @@ public abstract class RegionServerCallable extends AbstractRegionServerCallab void setStub(final ClientService.BlockingInterface stub) { this.stub = stub; } - - /** - * Override that changes Exception from {@link Exception} to {@link IOException}. It also does - * setup of an rpcController and calls through to the unimplemented - * call(PayloadCarryingRpcController) method; implement this method to add your rpc invocation. - */ - @Override - public T call(int callTimeout) throws IOException { - if (this.rpcController != null) { - this.rpcController.setCallTimeout(callTimeout); - } - try { - return call(this.rpcController); - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); - } - } - - /** - * Run RPC call. - * @param rpcController PayloadCarryingRpcController is a mouthful but it at a minimum is a - * facade on protobuf so we don't have to put protobuf everywhere; we can keep it behind this - * class. - * @throws Exception - */ - protected abstract T call(PayloadCarryingRpcController rpcController) throws Exception; - - public PayloadCarryingRpcController getRpcController() { - return this.rpcController; - } - - long getNonceGroup() { - return getConnection().getNonceGenerator().getNonceGroup(); - } - - long getNewNonce() { - return getConnection().getNonceGenerator().newNonce(); - } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RetryingTimeTracker.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RetryingTimeTracker.java index b9438e6589..24288e6f32 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RetryingTimeTracker.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RetryingTimeTracker.java @@ -22,6 +22,7 @@ import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; * Tracks the amount of time remaining for an operation. */ class RetryingTimeTracker { + private long globalStartTime = -1; public void start() { @@ -37,19 +38,16 @@ class RetryingTimeTracker { if (callTimeout == Integer.MAX_VALUE) { return Integer.MAX_VALUE; } - long remaining = EnvironmentEdgeManager.currentTime() - this.globalStartTime; - long remainingTime = callTimeout - remaining; + int remainingTime = (int) ( + callTimeout - + (EnvironmentEdgeManager.currentTime() - this.globalStartTime)); if (remainingTime < 1) { // If there is no time left, we're trying anyway. It's too late. // 0 means no timeout, and it's not the intent here. So we secure both cases by // resetting to the minimum. remainingTime = 1; } - if (remainingTime > Integer.MAX_VALUE) { - throw new RuntimeException("remainingTime=" + remainingTime + - " which is > Integer.MAX_VALUE"); - } - return (int)remainingTime; + return remainingTime; } } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ReversedScannerCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ReversedScannerCallable.java index 644337d006..0c2d3459d5 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ReversedScannerCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ReversedScannerCallable.java @@ -176,9 +176,9 @@ public class ReversedScannerCallable extends ScannerCallable { @Override public ScannerCallable getScannerCallableForReplica(int id) { - ReversedScannerCallable r = new ReversedScannerCallable(getConnection(), this.tableName, + ReversedScannerCallable r = new ReversedScannerCallable(this.cConnection, this.tableName, this.getScan(), this.scanMetrics, this.locateStartRow, controllerFactory, id); r.setCaching(this.getCaching()); return r; } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCaller.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCaller.java index 2b2e4c8ffd..b4cd2ef774 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCaller.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCaller.java @@ -22,6 +22,9 @@ import org.apache.hadoop.hbase.classification.InterfaceStability; import java.io.IOException; +/** + * + */ @InterfaceAudience.Public @InterfaceStability.Evolving public interface RpcRetryingCaller { @@ -49,4 +52,4 @@ public interface RpcRetryingCaller { */ T callWithoutRetries(RetryingCallable callable, int callTimeout) throws IOException, RuntimeException; -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerFactory.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerFactory.java index f92aeae0df..1c723c5d7a 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerFactory.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerFactory.java @@ -36,7 +36,6 @@ public class RpcRetryingCallerFactory { private final int rpcTimeout; private final RetryingCallerInterceptor interceptor; private final int startLogErrorsCnt; - /* These below data members are UNUSED!!!*/ private final boolean enableBackPressure; private ServerStatisticTracker stats; diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerWithReadReplicas.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerWithReadReplicas.java index 27856482b8..65dbb10744 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerWithReadReplicas.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RpcRetryingCallerWithReadReplicas.java @@ -29,6 +29,8 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.DoNotRetryIOException; import org.apache.hadoop.hbase.HBaseIOException; @@ -44,6 +46,8 @@ import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.generated.ClientProtos; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +import com.google.protobuf.ServiceException; + /** * Caller that goes to replica if the primary region does no answer within a configurable @@ -53,6 +57,8 @@ import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; */ @InterfaceAudience.Private public class RpcRetryingCallerWithReadReplicas { + private static final Log LOG = LogFactory.getLog(RpcRetryingCallerWithReadReplicas.class); + protected final ExecutorService pool; protected final ClusterConnection cConnection; protected final Configuration conf; @@ -92,7 +98,7 @@ public class RpcRetryingCallerWithReadReplicas { private final PayloadCarryingRpcController controller; public ReplicaRegionServerCallable(int id, HRegionLocation location) { - super(RpcRetryingCallerWithReadReplicas.this.cConnection, rpcControllerFactory, + super(RpcRetryingCallerWithReadReplicas.this.cConnection, RpcRetryingCallerWithReadReplicas.this.tableName, get.getRow()); this.id = id; this.location = location; @@ -135,20 +141,28 @@ public class RpcRetryingCallerWithReadReplicas { } @Override - protected Result call(PayloadCarryingRpcController controller) throws Exception { + public Result call(int callTimeout) throws Exception { if (controller.isCanceled()) return null; + if (Thread.interrupted()) { throw new InterruptedIOException(); } + byte[] reg = location.getRegionInfo().getRegionName(); + ClientProtos.GetRequest request = RequestConverter.buildGetRequest(reg, get); controller.setCallTimeout(callTimeout); - ClientProtos.GetResponse response = getStub().get(controller, request); - if (response == null) { - return null; + + try { + ClientProtos.GetResponse response = getStub().get(controller, request); + if (response == null) { + return null; + } + return ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } - return ProtobufUtil.toResult(response.getResult(), controller.cellScanner()); } @Override diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java index 1689d11404..72d69ec8ee 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ScannerCallable.java @@ -52,6 +52,9 @@ import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException; import org.apache.hadoop.ipc.RemoteException; import org.apache.hadoop.net.DNS; +import com.google.protobuf.ServiceException; +import com.google.protobuf.TextFormat; + /** * Scanner operations such as create, next, etc. * Used by {@link ResultScanner}s made by {@link Table}. Passed to a retrying caller such as @@ -71,6 +74,7 @@ public class ScannerCallable extends RegionServerCallable { protected boolean renew = false; private Scan scan; private int caching = 1; + protected final ClusterConnection cConnection; protected ScanMetrics scanMetrics; private boolean logScannerActivity = false; private int logCutOffLatency = 1000; @@ -121,8 +125,9 @@ public class ScannerCallable extends RegionServerCallable { */ public ScannerCallable(ClusterConnection connection, TableName tableName, Scan scan, ScanMetrics scanMetrics, RpcControllerFactory rpcControllerFactory, int id) { - super(connection, rpcControllerFactory, tableName, scan.getStartRow()); + super(connection, tableName, scan.getStartRow()); this.id = id; + this.cConnection = connection; this.scan = scan; this.scanMetrics = scanMetrics; Configuration conf = connection.getConfiguration(); @@ -180,16 +185,25 @@ public class ScannerCallable extends RegionServerCallable { } } - protected Result [] call(PayloadCarryingRpcController controller) throws Exception { + + @Override + public Result [] call(int callTimeout) throws IOException { if (Thread.interrupted()) { throw new InterruptedIOException(); } - if (this.closed) { - if (this.scannerId != -1) { + + if (controller == null) { + controller = controllerFactory.newController(); + controller.setPriority(getTableName()); + controller.setCallTimeout(callTimeout); + } + + if (closed) { + if (scannerId != -1) { close(); } } else { - if (this.scannerId == -1L) { + if (scannerId == -1L) { this.scannerId = openScanner(); } else { Result [] rrs = null; @@ -198,56 +212,61 @@ public class ScannerCallable extends RegionServerCallable { setHeartbeatMessage(false); try { incRPCcallsMetrics(); - request = RequestConverter.buildScanRequest(scannerId, caching, false, nextCallSeq, + request = + RequestConverter.buildScanRequest(scannerId, caching, false, nextCallSeq, this.scanMetrics != null, renew); ScanResponse response = null; - response = getStub().scan(controller, request); - // Client and RS maintain a nextCallSeq number during the scan. Every next() call - // from client to server will increment this number in both sides. Client passes this - // number along with the request and at RS side both the incoming nextCallSeq and its - // nextCallSeq will be matched. In case of a timeout this increment at the client side - // should not happen. If at the server side fetching of next batch of data was over, - // there will be mismatch in the nextCallSeq number. Server will throw - // OutOfOrderScannerNextException and then client will reopen the scanner with startrow - // as the last successfully retrieved row. - // See HBASE-5974 - nextCallSeq++; - long timestamp = System.currentTimeMillis(); - setHeartbeatMessage(response.hasHeartbeatMessage() && response.getHeartbeatMessage()); - // Results are returned via controller - CellScanner cellScanner = controller.cellScanner(); - rrs = ResponseConverter.getResults(cellScanner, response); - if (logScannerActivity) { - long now = System.currentTimeMillis(); - if (now - timestamp > logCutOffLatency) { - int rows = rrs == null ? 0 : rrs.length; - LOG.info("Took " + (now-timestamp) + "ms to fetch " + try { + response = getStub().scan(controller, request); + // Client and RS maintain a nextCallSeq number during the scan. Every next() call + // from client to server will increment this number in both sides. Client passes this + // number along with the request and at RS side both the incoming nextCallSeq and its + // nextCallSeq will be matched. In case of a timeout this increment at the client side + // should not happen. If at the server side fetching of next batch of data was over, + // there will be mismatch in the nextCallSeq number. Server will throw + // OutOfOrderScannerNextException and then client will reopen the scanner with startrow + // as the last successfully retrieved row. + // See HBASE-5974 + nextCallSeq++; + long timestamp = System.currentTimeMillis(); + setHeartbeatMessage(response.hasHeartbeatMessage() && response.getHeartbeatMessage()); + // Results are returned via controller + CellScanner cellScanner = controller.cellScanner(); + rrs = ResponseConverter.getResults(cellScanner, response); + if (logScannerActivity) { + long now = System.currentTimeMillis(); + if (now - timestamp > logCutOffLatency) { + int rows = rrs == null ? 0 : rrs.length; + LOG.info("Took " + (now-timestamp) + "ms to fetch " + rows + " rows from scanner=" + scannerId); + } } - } - updateServerSideMetrics(response); - // moreResults is only used for the case where a filter exhausts all elements - if (response.hasMoreResults() && !response.getMoreResults()) { - this.scannerId = -1L; - this.closed = true; - // Implied that no results were returned back, either. - return null; - } - // moreResultsInRegion explicitly defines when a RS may choose to terminate a batch due - // to size or quantity of results in the response. - if (response.hasMoreResultsInRegion()) { - // Set what the RS said - setHasMoreResultsContext(true); - setServerHasMoreResults(response.getMoreResultsInRegion()); - } else { - // Server didn't respond whether it has more results or not. - setHasMoreResultsContext(false); + updateServerSideMetrics(response); + // moreResults is only used for the case where a filter exhausts all elements + if (response.hasMoreResults() && !response.getMoreResults()) { + scannerId = -1L; + closed = true; + // Implied that no results were returned back, either. + return null; + } + // moreResultsInRegion explicitly defines when a RS may choose to terminate a batch due + // to size or quantity of results in the response. + if (response.hasMoreResultsInRegion()) { + // Set what the RS said + setHasMoreResultsContext(true); + setServerHasMoreResults(response.getMoreResultsInRegion()); + } else { + // Server didn't respond whether it has more results or not. + setHasMoreResultsContext(false); + } + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } updateResultsMetrics(rrs); } catch (IOException e) { if (logScannerActivity) { - LOG.info("Got exception making request " + ProtobufUtil.toText(request) + " to " + - getLocation(), e); + LOG.info("Got exception making request " + TextFormat.shortDebugString(request) + + " to " + getLocation(), e); } IOException ioe = e; if (e instanceof RemoteException) { @@ -256,9 +275,9 @@ public class ScannerCallable extends RegionServerCallable { if (logScannerActivity && (ioe instanceof UnknownScannerException)) { try { HRegionLocation location = - getConnection().relocateRegion(getTableName(), scan.getStartRow()); - LOG.info("Scanner=" + scannerId + " expired, current region location is " + - location.toString()); + getConnection().relocateRegion(getTableName(), scan.getStartRow()); + LOG.info("Scanner=" + scannerId + + " expired, current region location is " + location.toString()); } catch (Throwable t) { LOG.info("Failed to relocate region", t); } @@ -357,8 +376,8 @@ public class ScannerCallable extends RegionServerCallable { RequestConverter.buildScanRequest(this.scannerId, 0, true, this.scanMetrics != null); try { getStub().scan(controller, request); - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } } catch (IOException e) { LOG.warn("Ignore, probably already closed", e); @@ -368,8 +387,10 @@ public class ScannerCallable extends RegionServerCallable { protected long openScanner() throws IOException { incRPCcallsMetrics(); - ScanRequest request = RequestConverter.buildScanRequest( - getLocation().getRegionInfo().getRegionName(), this.scan, 0, false); + ScanRequest request = + RequestConverter.buildScanRequest( + getLocation().getRegionInfo().getRegionName(), + this.scan, 0, false); try { ScanResponse response = getStub().scan(controller, request); long id = response.getScannerId(); @@ -378,8 +399,8 @@ public class ScannerCallable extends RegionServerCallable { + " on region " + getLocation().toString()); } return id; - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } } @@ -422,6 +443,11 @@ public class ScannerCallable extends RegionServerCallable { return caching; } + @Override + public ClusterConnection getConnection() { + return cConnection; + } + /** * Set the number of rows that will be fetched on next * @param caching the number of rows for caching @@ -462,4 +488,4 @@ public class ScannerCallable extends RegionServerCallable { protected void setHasMoreResultsContext(boolean serverHasMoreResultsContext) { this.serverHasMoreResultsContext = serverHasMoreResultsContext; } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SecureBulkLoadClient.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SecureBulkLoadClient.java index d6896e13f2..7b1547d36a 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SecureBulkLoadClient.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/SecureBulkLoadClient.java @@ -22,9 +22,6 @@ import java.io.IOException; import java.util.List; import org.apache.hadoop.hbase.classification.InterfaceAudience; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; -import org.apache.hadoop.hbase.ipc.RpcControllerFactory; -import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; @@ -41,35 +38,41 @@ import org.apache.hadoop.hbase.security.SecureBulkLoadUtil; import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.security.token.Token; +import com.google.protobuf.ServiceException; + /** * Client proxy for SecureBulkLoadProtocol */ @InterfaceAudience.Private public class SecureBulkLoadClient { private Table table; - private final RpcControllerFactory rpcControllerFactory; - public SecureBulkLoadClient(final Configuration conf, Table table) { + public SecureBulkLoadClient(Table table) { this.table = table; - this.rpcControllerFactory = new RpcControllerFactory(conf); } public String prepareBulkLoad(final Connection conn) throws IOException { try { - RegionServerCallable callable = new RegionServerCallable(conn, - this.rpcControllerFactory, table.getName(), HConstants.EMPTY_START_ROW) { - @Override - protected String call(PayloadCarryingRpcController controller) throws Exception { - byte[] regionName = getLocation().getRegionInfo().getRegionName(); - RegionSpecifier region = - RequestConverter.buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName); - PrepareBulkLoadRequest request = PrepareBulkLoadRequest.newBuilder() - .setTableName(ProtobufUtil.toProtoTableName(table.getName())) - .setRegion(region).build(); - PrepareBulkLoadResponse response = getStub().prepareBulkLoad(null, request); - return response.getBulkToken(); - } - }; + RegionServerCallable callable = + new RegionServerCallable(conn, table.getName(), HConstants.EMPTY_START_ROW) { + @Override + public String call(int callTimeout) throws IOException { + byte[] regionName = getLocation().getRegionInfo().getRegionName(); + RegionSpecifier region = + RequestConverter + .buildRegionSpecifier(RegionSpecifierType.REGION_NAME, regionName); + try { + PrepareBulkLoadRequest request = + PrepareBulkLoadRequest.newBuilder() + .setTableName(ProtobufUtil.toProtoTableName(table.getName())) + .setRegion(region).build(); + PrepareBulkLoadResponse response = getStub().prepareBulkLoad(null, request); + return response.getBulkToken(); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + } + }; return RpcRetryingCallerFactory.instantiate(conn.getConfiguration(), null) . newCaller().callWithRetries(callable, Integer.MAX_VALUE); } catch (Throwable throwable) { @@ -79,19 +82,24 @@ public class SecureBulkLoadClient { public void cleanupBulkLoad(final Connection conn, final String bulkToken) throws IOException { try { - RegionServerCallable callable = new RegionServerCallable(conn, - this.rpcControllerFactory, table.getName(), HConstants.EMPTY_START_ROW) { - @Override - protected Void call(PayloadCarryingRpcController controller) throws Exception { - byte[] regionName = getLocation().getRegionInfo().getRegionName(); - RegionSpecifier region = RequestConverter.buildRegionSpecifier( - RegionSpecifierType.REGION_NAME, regionName); - CleanupBulkLoadRequest request = - CleanupBulkLoadRequest.newBuilder().setRegion(region).setBulkToken(bulkToken).build(); - getStub().cleanupBulkLoad(null, request); - return null; - } - }; + RegionServerCallable callable = + new RegionServerCallable(conn, table.getName(), HConstants.EMPTY_START_ROW) { + @Override + public Void call(int callTimeout) throws IOException { + byte[] regionName = getLocation().getRegionInfo().getRegionName(); + RegionSpecifier region = RequestConverter.buildRegionSpecifier( + RegionSpecifierType.REGION_NAME, regionName); + try { + CleanupBulkLoadRequest request = + CleanupBulkLoadRequest.newBuilder().setRegion(region) + .setBulkToken(bulkToken).build(); + getStub().cleanupBulkLoad(null, request); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + return null; + } + }; RpcRetryingCallerFactory.instantiate(conn.getConfiguration(), null) . newCaller().callWithRetries(callable, Integer.MAX_VALUE); } catch (Throwable throwable) { @@ -122,12 +130,12 @@ public class SecureBulkLoadClient { try { BulkLoadHFileResponse response = client.bulkLoadHFile(null, request); return response.getLoaded(); - } catch (Exception se) { - throw ProtobufUtil.handleRemoteException(se); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); } } public Path getStagingPath(String bulkToken, byte[] family) throws IOException { return SecureBulkLoadUtil.getStagingPath(table.getConfiguration(), bulkToken, family); } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/MasterCoprocessorRpcChannel.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/MasterCoprocessorRpcChannel.java index a6384e36ab..6fae5cb7ba 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/MasterCoprocessorRpcChannel.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/MasterCoprocessorRpcChannel.java @@ -77,4 +77,5 @@ public class MasterCoprocessorRpcChannel extends SyncCoprocessorRpcChannel { } return response; } -} \ No newline at end of file + +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/PayloadCarryingRpcController.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/PayloadCarryingRpcController.java index 6c290a618f..f4f18b3ea5 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/PayloadCarryingRpcController.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/PayloadCarryingRpcController.java @@ -17,39 +17,24 @@ */ package org.apache.hadoop.hbase.ipc; -import java.io.IOException; import java.util.List; -import java.util.concurrent.atomic.AtomicReference; import org.apache.hadoop.hbase.CellScannable; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.HConstants; -import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.classification.InterfaceAudience; -import org.apache.hadoop.hbase.util.Bytes; - -import com.google.protobuf.RpcCallback; -import com.google.protobuf.RpcController; /** * Optionally carries Cells across the proxy/service interface down into ipc. On its - * way out it optionally carries a set of result Cell data. We stick the Cells here when we want - * to avoid having to protobuf them (for performance reasons). This class is used ferrying data - * across the proxy/protobuf service chasm. Also does call timeout. Used by client and server - * ipc'ing. + * way out it optionally carries a set of result Cell data. We stick the Cells here when we want + * to avoid having to protobuf them. This class is used ferrying data across the proxy/protobuf + * service chasm. Used by client and server ipc'ing. */ @InterfaceAudience.Private -public class PayloadCarryingRpcController implements RpcController, CellScannable { - /** - * The time, in ms before the call should expire. - */ - protected volatile Integer callTimeout; - protected volatile boolean cancelled = false; - protected final AtomicReference> cancellationCb = new AtomicReference<>(null); - protected final AtomicReference> failureCb = new AtomicReference<>(null); - private IOException exception; +public class PayloadCarryingRpcController + extends TimeLimitedRpcController implements CellScannable { public static final int PRIORITY_UNSET = -1; /** @@ -108,123 +93,15 @@ public class PayloadCarryingRpcController implements RpcController, CellScannabl } /** - * @param regionName RegionName. If hbase:meta, we'll set high priority. - */ - public void setPriority(final byte [] regionName) { - if (isMetaRegion(regionName)) { - setPriority(TableName.META_TABLE_NAME); - } - } - - private static boolean isMetaRegion(final byte[] regionName) { - return Bytes.equals(regionName, HRegionInfo.FIRST_META_REGIONINFO.getRegionName()) - || Bytes.equals(regionName, HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes()); - } - - /** * @return The priority of this request */ public int getPriority() { return priority; } - @Override - public void reset() { + @Override public void reset() { + super.reset(); priority = 0; cellScanner = null; - exception = null; - cancelled = false; - failureCb.set(null); - cancellationCb.set(null); - callTimeout = null; - } - - public int getCallTimeout() { - if (callTimeout != null) { - return callTimeout; - } else { - return 0; - } - } - - public void setCallTimeout(int callTimeout) { - this.callTimeout = callTimeout; - } - - public boolean hasCallTimeout(){ - return callTimeout != null; - } - - @Override - public String errorText() { - if (exception != null) { - return exception.getMessage(); - } else { - return null; - } - } - - /** - * For use in async rpc clients - * @return true if failed - */ - @Override - public boolean failed() { - return this.exception != null; - } - - @Override - public boolean isCanceled() { - return cancelled; - } - - @Override - public void notifyOnCancel(RpcCallback cancellationCb) { - this.cancellationCb.set(cancellationCb); - if (this.cancelled) { - cancellationCb.run(null); - } - } - - /** - * Notify a callback on error. - * For use in async rpc clients - * - * @param failureCb the callback to call on error - */ - public void notifyOnFail(RpcCallback failureCb) { - this.failureCb.set(failureCb); - if (this.exception != null) { - failureCb.run(this.exception); - } - } - - @Override - public void setFailed(String reason) { - this.exception = new IOException(reason); - if (this.failureCb.get() != null) { - this.failureCb.get().run(this.exception); - } - } - - /** - * Set failed with an exception to pass on. - * For use in async rpc clients - * - * @param e exception to set with - */ - public void setFailed(IOException e) { - this.exception = e; - if (this.failureCb.get() != null) { - this.failureCb.get().run(this.exception); - } - } - - @Override - public void startCancel() { - cancelled = true; - if (cancellationCb.get() != null) { - cancellationCb.get().run(null); - } } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RegionCoprocessorRpcChannel.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RegionCoprocessorRpcChannel.java index dbc9041f77..55d63756cd 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RegionCoprocessorRpcChannel.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/RegionCoprocessorRpcChannel.java @@ -76,23 +76,30 @@ public class RegionCoprocessorRpcChannel extends SyncCoprocessorRpcChannel { Descriptors.MethodDescriptor method, Message request, Message responsePrototype) throws IOException { if (LOG.isTraceEnabled()) { - LOG.trace("Call: " + method.getName() + ", " + request.toString()); + LOG.trace("Call: "+method.getName()+", "+request.toString()); } + if (row == null) { throw new IllegalArgumentException("Missing row property for remote region location"); } + + final RpcController rpcController = controller == null + ? rpcControllerFactory.newController() : controller; + final ClientProtos.CoprocessorServiceCall call = CoprocessorRpcUtils.buildServiceCall(row, method, request); RegionServerCallable callable = - new RegionServerCallable(connection, - controller == null? this.rpcControllerFactory.newController(): - (PayloadCarryingRpcController)controller, - table, row) { + new RegionServerCallable(connection, table, row) { @Override - protected CoprocessorServiceResponse call(PayloadCarryingRpcController controller) - throws Exception { + public CoprocessorServiceResponse call(int callTimeout) throws Exception { + if (rpcController instanceof PayloadCarryingRpcController) { + ((PayloadCarryingRpcController) rpcController).setPriority(tableName); + } + if (rpcController instanceof TimeLimitedRpcController) { + ((TimeLimitedRpcController) rpcController).setCallTimeout(callTimeout); + } byte[] regionName = getLocation().getRegionInfo().getRegionName(); - return ProtobufUtil.execService(getRpcController(), getStub(), call, regionName); + return ProtobufUtil.execService(rpcController, getStub(), call, regionName); } }; CoprocessorServiceResponse result = rpcCallerFactory. newCaller() diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/PayloadCarryingRpcController.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/TimeLimitedRpcController.java similarity index 50% copy from hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/PayloadCarryingRpcController.java copy to hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/TimeLimitedRpcController.java index 6c290a618f..cf08ea9cea 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/PayloadCarryingRpcController.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/TimeLimitedRpcController.java @@ -15,129 +15,32 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.hadoop.hbase.ipc; +import com.google.protobuf.RpcCallback; +import com.google.protobuf.RpcController; + import java.io.IOException; -import java.util.List; import java.util.concurrent.atomic.AtomicReference; -import org.apache.hadoop.hbase.CellScannable; -import org.apache.hadoop.hbase.CellScanner; -import org.apache.hadoop.hbase.CellUtil; -import org.apache.hadoop.hbase.HConstants; -import org.apache.hadoop.hbase.HRegionInfo; -import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.classification.InterfaceAudience; -import org.apache.hadoop.hbase.util.Bytes; -import com.google.protobuf.RpcCallback; -import com.google.protobuf.RpcController; - -/** - * Optionally carries Cells across the proxy/service interface down into ipc. On its - * way out it optionally carries a set of result Cell data. We stick the Cells here when we want - * to avoid having to protobuf them (for performance reasons). This class is used ferrying data - * across the proxy/protobuf service chasm. Also does call timeout. Used by client and server - * ipc'ing. - */ @InterfaceAudience.Private -public class PayloadCarryingRpcController implements RpcController, CellScannable { +public class TimeLimitedRpcController implements RpcController { + /** * The time, in ms before the call should expire. */ protected volatile Integer callTimeout; protected volatile boolean cancelled = false; - protected final AtomicReference> cancellationCb = new AtomicReference<>(null); - protected final AtomicReference> failureCb = new AtomicReference<>(null); - private IOException exception; - - public static final int PRIORITY_UNSET = -1; - /** - * Priority to set on this request. Set it here in controller so available composing the - * request. This is the ordained way of setting priorities going forward. We will be - * undoing the old annotation-based mechanism. - */ - private int priority = PRIORITY_UNSET; - - /** - * They are optionally set on construction, cleared after we make the call, and then optionally - * set on response with the result. We use this lowest common denominator access to Cells because - * sometimes the scanner is backed by a List of Cells and other times, it is backed by an - * encoded block that implements CellScanner. - */ - private CellScanner cellScanner; - - public PayloadCarryingRpcController() { - this((CellScanner)null); - } - - public PayloadCarryingRpcController(final CellScanner cellScanner) { - this.cellScanner = cellScanner; - } - - public PayloadCarryingRpcController(final List cellIterables) { - this.cellScanner = cellIterables == null? null: CellUtil.createCellScanner(cellIterables); - } - - /** - * @return One-shot cell scanner (you cannot back it up and restart) - */ - @Override - public CellScanner cellScanner() { - return cellScanner; - } - - public void setCellScanner(final CellScanner cellScanner) { - this.cellScanner = cellScanner; - } - - /** - * @param priority Priority for this request; should fall roughly in the range - * {@link HConstants#NORMAL_QOS} to {@link HConstants#HIGH_QOS} - */ - public void setPriority(int priority) { - this.priority = priority; - } - - /** - * @param tn Set priority based off the table we are going against. - */ - public void setPriority(final TableName tn) { - this.priority = - (tn != null && tn.isSystemTable())? HConstants.SYSTEMTABLE_QOS: HConstants.NORMAL_QOS; - } + protected final AtomicReference> cancellationCb = + new AtomicReference<>(null); - /** - * @param regionName RegionName. If hbase:meta, we'll set high priority. - */ - public void setPriority(final byte [] regionName) { - if (isMetaRegion(regionName)) { - setPriority(TableName.META_TABLE_NAME); - } - } + protected final AtomicReference> failureCb = + new AtomicReference<>(null); - private static boolean isMetaRegion(final byte[] regionName) { - return Bytes.equals(regionName, HRegionInfo.FIRST_META_REGIONINFO.getRegionName()) - || Bytes.equals(regionName, HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes()); - } - - /** - * @return The priority of this request - */ - public int getPriority() { - return priority; - } - - @Override - public void reset() { - priority = 0; - cellScanner = null; - exception = null; - cancelled = false; - failureCb.set(null); - cancellationCb.set(null); - callTimeout = null; - } + private IOException exception; public int getCallTimeout() { if (callTimeout != null) { @@ -200,6 +103,15 @@ public class PayloadCarryingRpcController implements RpcController, CellScannabl } @Override + public void reset() { + exception = null; + cancelled = false; + failureCb.set(null); + cancellationCb.set(null); + callTimeout = null; + } + + @Override public void setFailed(String reason) { this.exception = new IOException(reason); if (this.failureCb.get() != null) { @@ -227,4 +139,4 @@ public class PayloadCarryingRpcController implements RpcController, CellScannabl cancellationCb.get().run(null); } } -} \ No newline at end of file +} diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java index 623acd5fdf..5ba0572eb5 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java @@ -17,9 +17,6 @@ */ package org.apache.hadoop.hbase.protobuf; -import static org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier -.RegionSpecifierType.REGION_NAME; - import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; @@ -41,11 +38,14 @@ import java.util.Map; import java.util.Map.Entry; import java.util.NavigableSet; import java.util.Set; -import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; + +import static org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier +.RegionSpecifierType.REGION_NAME; + import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.CellUtil; @@ -53,7 +53,6 @@ import org.apache.hadoop.hbase.ClusterId; import org.apache.hadoop.hbase.ClusterStatus; import org.apache.hadoop.hbase.DoNotRetryIOException; import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.HBaseIOException; import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionInfo; @@ -125,8 +124,8 @@ import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos; import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.LiveServerInfo; import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionInTransition; import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionLoad; -import org.apache.hadoop.hbase.protobuf.generated.ComparatorProtos; import org.apache.hadoop.hbase.protobuf.generated.FSProtos.HBaseVersionFileContent; +import org.apache.hadoop.hbase.protobuf.generated.ComparatorProtos; import org.apache.hadoop.hbase.protobuf.generated.FilterProtos; import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos; import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.BytesBytesPair; @@ -172,9 +171,11 @@ import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.DynamicClassLoader; import org.apache.hadoop.hbase.util.ExceptionUtil; import org.apache.hadoop.hbase.util.Methods; +import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.hbase.util.VersionInfo; import org.apache.hadoop.io.Text; import org.apache.hadoop.ipc.RemoteException; +import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.security.token.Token; import com.google.common.collect.ArrayListMultimap; @@ -333,32 +334,17 @@ public final class ProtobufUtil { * a new IOException that wraps the unexpected ServiceException. */ public static IOException getRemoteException(ServiceException se) { - return makeIOExceptionOfException(se); - } - - /** - * Like {@link #getRemoteException(ServiceException)} but more generic, able to handle more than - * just {@link ServiceException}. Prefer this method to - * {@link #getRemoteException(ServiceException)} because trying to - * contain direct protobuf references. - * @param e - */ - public static IOException handleRemoteException(Exception e) { - return makeIOExceptionOfException(e); - } - - private static IOException makeIOExceptionOfException(Exception e) { - Throwable t = e; - if (e instanceof ServiceException) { - t = e.getCause(); + Throwable e = se.getCause(); + if (e == null) { + return new IOException(se); } - if (ExceptionUtil.isInterrupt(t)) { - return ExceptionUtil.asInterrupt(t); + if (ExceptionUtil.isInterrupt(e)) { + return ExceptionUtil.asInterrupt(e); } - if (t instanceof RemoteException) { - t = ((RemoteException)t).unwrapRemoteException(); + if (e instanceof RemoteException) { + e = ((RemoteException) e).unwrapRemoteException(); } - return t instanceof IOException? (IOException)t: new HBaseIOException(t); + return e instanceof IOException ? (IOException) e : new IOException(se); } /** @@ -1266,6 +1252,7 @@ public final class ProtobufUtil { return toMutation(type, mutation, builder, HConstants.NO_NONCE); } + @SuppressWarnings("deprecation") public static MutationProto toMutation(final MutationType type, final Mutation mutation, MutationProto.Builder builder, long nonce) throws IOException { @@ -2671,11 +2658,13 @@ public final class ProtobufUtil { } } + @SuppressWarnings("deprecation") public static CompactionDescriptor toCompactionDescriptor(HRegionInfo info, byte[] family, List inputPaths, List outputPaths, Path storeDir) { return toCompactionDescriptor(info, null, family, inputPaths, outputPaths, storeDir); } + @SuppressWarnings("deprecation") public static CompactionDescriptor toCompactionDescriptor(HRegionInfo info, byte[] regionName, byte[] family, List inputPaths, List outputPaths, Path storeDir) { // compaction descriptor contains relative paths. @@ -3674,28 +3663,4 @@ public final class ProtobufUtil { return new RegionLoadStats(stats.getMemstoreLoad(), stats.getHeapOccupancy(), stats.getCompactionPressure()); } - - /** - * @param msg - * @return A String version of the passed in msg - */ - public static String toText(Message msg) { - return TextFormat.shortDebugString(msg); - } - - public static byte [] toBytes(ByteString bs) { - return bs.toByteArray(); - } - - /** - * Contain ServiceException inside here. Take a callable that is doing our pb rpc and run it. - * @throws IOException - */ - public static T call(Callable callable) throws IOException { - try { - return callable.call(); - } catch (Exception e) { - throw ProtobufUtil.handleRemoteException(e); - } - } -} \ No newline at end of file +} diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestClientScanner.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestClientScanner.java index fd2a39393b..f083001f11 100644 --- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestClientScanner.java +++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestClientScanner.java @@ -65,6 +65,7 @@ public class TestClientScanner { RpcControllerFactory controllerFactory; @Before + @SuppressWarnings("deprecation") public void setup() throws IOException { clusterConn = Mockito.mock(ClusterConnection.class); rpcFactory = Mockito.mock(RpcRetryingCallerFactory.class); diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/HBaseIOException.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/HBaseIOException.java index edcbdc5041..9c3367ed83 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/HBaseIOException.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/HBaseIOException.java @@ -45,5 +45,4 @@ public class HBaseIOException extends IOException { public HBaseIOException(Throwable cause) { super(cause); - } -} \ No newline at end of file + }} diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ExceptionUtil.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ExceptionUtil.java index 7e6c5d6398..688b51a02a 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ExceptionUtil.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/ExceptionUtil.java @@ -73,4 +73,4 @@ public class ExceptionUtil { private ExceptionUtil() { } -} \ No newline at end of file +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java index ec28315b6c..73226aa4c7 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java @@ -75,17 +75,20 @@ import org.apache.hadoop.hbase.DoNotRetryIOException; import org.apache.hadoop.hbase.HBaseIOException; import org.apache.hadoop.hbase.HBaseInterfaceAudience; import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.Server; +import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.classification.InterfaceStability; +import org.apache.hadoop.hbase.client.Operation; import org.apache.hadoop.hbase.client.VersionInfoUtil; import org.apache.hadoop.hbase.codec.Codec; import org.apache.hadoop.hbase.conf.ConfigurationObserver; import org.apache.hadoop.hbase.exceptions.RegionMovedException; import org.apache.hadoop.hbase.io.ByteBufferInputStream; -import org.apache.hadoop.hbase.io.ByteBufferListOutputStream; import org.apache.hadoop.hbase.io.ByteBufferOutputStream; import org.apache.hadoop.hbase.io.ByteBufferPool; +import org.apache.hadoop.hbase.io.ByteBufferListOutputStream; import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler; import org.apache.hadoop.hbase.monitoring.TaskMonitor; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; @@ -96,6 +99,7 @@ import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ExceptionResponse; import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.RequestHeader; import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.ResponseHeader; import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.UserInformation; +import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.security.AccessDeniedException; import org.apache.hadoop.hbase.security.AuthMethod; import org.apache.hadoop.hbase.security.HBasePolicyProvider; diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java index 8ddbe18025..09dedec8bd 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java @@ -87,8 +87,6 @@ import org.apache.hadoop.hbase.io.hfile.HFileContext; import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder; import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoder; import org.apache.hadoop.hbase.io.hfile.HFileScanner; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; -import org.apache.hadoop.hbase.ipc.RpcControllerFactory; import org.apache.hadoop.hbase.regionserver.BloomType; import org.apache.hadoop.hbase.regionserver.HStore; import org.apache.hadoop.hbase.regionserver.StoreFileInfo; @@ -104,6 +102,7 @@ import org.apache.hadoop.util.ToolRunner; /** * Tool to load the output of HFileOutputFormat into an existing table. + * @see #usage() */ @InterfaceAudience.Public @InterfaceStability.Stable @@ -131,13 +130,11 @@ public class LoadIncrementalHFiles extends Configured implements Tool { private String bulkToken; private UserProvider userProvider; private int nrThreads; - private RpcControllerFactory rpcControllerFactory; private LoadIncrementalHFiles() {} public LoadIncrementalHFiles(Configuration conf) throws Exception { super(conf); - this.rpcControllerFactory = new RpcControllerFactory(conf); initialize(); } @@ -325,7 +322,7 @@ public class LoadIncrementalHFiles extends Configured implements Tool { // LQI queue does not need to be threadsafe -- all operations on this queue // happen in this thread Deque queue = new LinkedList<>(); - SecureBulkLoadClient secureClient = new SecureBulkLoadClient(table.getConfiguration(), table); + SecureBulkLoadClient secureClient = new SecureBulkLoadClient(table); try { /* @@ -476,11 +473,9 @@ public class LoadIncrementalHFiles extends Configured implements Tool { /** * Used by the replication sink to load the hfiles from the source cluster. It does the following, - *
    - *
  1. LoadIncrementalHFiles#groupOrSplitPhase(Table, ExecutorService, Deque, Pair)}
    2. - *
  2. LoadIncrementalHFiles#bulkLoadPhase(Table, Connection, ExecutorService, Deque, Multimap) - *
    3. - *
+ * 1. {@link LoadIncrementalHFiles#groupOrSplitPhase(Table, ExecutorService, Deque, Pair)} 2. + * {@link + * LoadIncrementalHFiles#bulkLoadPhase(Table, Connection, ExecutorService, Deque, Multimap)} * @param table Table to which these hfiles should be loaded to * @param conn Connection to use * @param queue {@link LoadQueueItem} has hfiles yet to be loaded @@ -781,23 +776,27 @@ public class LoadIncrementalHFiles extends Configured implements Tool { protected List tryAtomicRegionLoad(final Connection conn, final TableName tableName, final byte[] first, final Collection lqis) throws IOException { - final List> famPaths = new ArrayList<>(lqis.size()); + final List> famPaths = + new ArrayList<>(lqis.size()); for (LoadQueueItem lqi : lqis) { famPaths.add(Pair.newPair(lqi.family, lqi.hfilePath.toString())); } - final RegionServerCallable svrCallable = new RegionServerCallable(conn, - rpcControllerFactory, tableName, first) { + + final RegionServerCallable svrCallable = + new RegionServerCallable(conn, tableName, first) { @Override - protected Boolean call(PayloadCarryingRpcController controller) throws Exception { + public Boolean call(int callTimeout) throws Exception { SecureBulkLoadClient secureClient = null; boolean success = false; + try { LOG.debug("Going to connect to server " + getLocation() + " for row " + Bytes.toStringBinary(getRow()) + " with hfile group " + famPaths); byte[] regionName = getLocation().getRegionInfo().getRegionName(); try (Table table = conn.getTable(getTableName())) { - secureClient = new SecureBulkLoadClient(getConf(), table); - success = secureClient.secureBulkLoadHFiles(getStub(), famPaths, regionName, + secureClient = new SecureBulkLoadClient(table); + success = + secureClient.secureBulkLoadHFiles(getStub(), famPaths, regionName, assignSeqIds, fsDelegationToken.getUserToken(), bulkToken); } return success; @@ -1079,7 +1078,7 @@ public class LoadIncrementalHFiles extends Configured implements Tool { /** * Called from replication sink, where it manages bulkToken(staging directory) by itself. This is - * used only when SecureBulkLoadEndpoint is configured in hbase.coprocessor.region.classes + * used only when {@link SecureBulkLoadEndpoint} is configured in hbase.coprocessor.region.classes * property. This directory is used as a temporary directory where all files are initially * copied/moved from user given directory, set all the required file permissions and then from * their it is finally loaded into a table. This should be set only when, one would like to manage @@ -1089,4 +1088,5 @@ public class LoadIncrementalHFiles extends Configured implements Tool { public void setBulkToken(String stagingDir) { this.bulkToken = stagingDir; } -} \ No newline at end of file + +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ExpiredMobFileCleanerChore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ExpiredMobFileCleanerChore.java index 3261bd6dff..a21edcc2cb 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ExpiredMobFileCleanerChore.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ExpiredMobFileCleanerChore.java @@ -35,6 +35,8 @@ import org.apache.hadoop.hbase.mob.ExpiredMobFileCleaner; import org.apache.hadoop.hbase.mob.MobConstants; import org.apache.hadoop.hbase.mob.MobUtils; +import com.google.protobuf.ServiceException; + /** * The Class ExpiredMobFileCleanerChore for running cleaner regularly to remove the expired * mob files. @@ -84,6 +86,10 @@ public class ExpiredMobFileCleanerChore extends ScheduledChore { } catch (LockTimeoutException e) { LOG.info("Fail to acquire the lock because of timeout, maybe a" + " MobCompactor is running", e); + } catch (ServiceException e) { + LOG.error( + "Fail to clean the expired mob files for the column " + hcd.getNameAsString() + + " in the table " + htd.getNameAsString(), e); } catch (IOException e) { LOG.error( "Fail to clean the expired mob files for the column " + hcd.getNameAsString() diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java index d7ba4f35cf..531883ad10 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java @@ -61,9 +61,9 @@ import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ServerInfo; +import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupRequest; import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionStoreSequenceIds; import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.StoreSequenceId; -import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupRequest; import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode; import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.regionserver.RegionOpeningState; @@ -454,7 +454,8 @@ public class ServerManager { /** * Adds the onlineServers list. onlineServers should be locked. * @param serverName The remote servers name. - * @param s + * @param sl + * @return Server load from the removed server, if any. */ @VisibleForTesting void recordNewServerWithLock(final ServerName serverName, final ServerLoad sl) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/ExpiredMobFileCleaner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/ExpiredMobFileCleaner.java index d4a54bb958..3c965cb6de 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/ExpiredMobFileCleaner.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/ExpiredMobFileCleaner.java @@ -41,6 +41,8 @@ import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; +import com.google.protobuf.ServiceException; + /** * The cleaner to delete the expired MOB files. */ @@ -58,8 +60,11 @@ public class ExpiredMobFileCleaner extends Configured implements Tool { * directory. * @param tableName The current table name. * @param family The current family. + * @throws ServiceException + * @throws IOException */ - public void cleanExpiredMobFiles(String tableName, HColumnDescriptor family) throws IOException { + public void cleanExpiredMobFiles(String tableName, HColumnDescriptor family) + throws ServiceException, IOException { Configuration conf = getConf(); TableName tn = TableName.valueOf(tableName); FileSystem fs = FileSystem.get(conf); @@ -94,7 +99,7 @@ public class ExpiredMobFileCleaner extends Configured implements Tool { String tableName = args[0]; String familyName = args[1]; TableName tn = TableName.valueOf(tableName); - HBaseAdmin.available(getConf()); + HBaseAdmin.checkHBaseAvailable(getConf()); Connection connection = ConnectionFactory.createConnection(getConf()); Admin admin = connection.getAdmin(); try { @@ -122,4 +127,5 @@ public class ExpiredMobFileCleaner extends Configured implements Tool { } } } -} \ No newline at end of file + +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/mapreduce/Sweeper.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/mapreduce/Sweeper.java index c27e8aecf1..8547c8cc6e 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/mapreduce/Sweeper.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/mapreduce/Sweeper.java @@ -38,6 +38,8 @@ import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.apache.zookeeper.KeeperException; +import com.google.protobuf.ServiceException; + /** * The sweep tool. It deletes the mob files that are not used and merges the small mob files to * bigger ones. Each run of this sweep tool only handles one column family. The runs on @@ -62,10 +64,10 @@ public class Sweeper extends Configured implements Tool { * @throws ServiceException */ int sweepFamily(String tableName, String familyName) throws IOException, InterruptedException, - ClassNotFoundException, KeeperException { + ClassNotFoundException, KeeperException, ServiceException { Configuration conf = getConf(); // make sure the target HBase exists. - HBaseAdmin.available(conf); + HBaseAdmin.checkHBaseAvailable(conf); Connection connection = ConnectionFactory.createConnection(getConf()); Admin admin = connection.getAdmin(); try { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java index fb9a605613..d87ada4f05 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java @@ -87,6 +87,7 @@ import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; import org.apache.hadoop.hbase.ipc.RpcServerInterface; import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException; import org.apache.hadoop.hbase.ipc.ServerRpcController; +import org.apache.hadoop.hbase.ipc.TimeLimitedRpcController; import org.apache.hadoop.hbase.master.MasterRpcServices; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.RequestConverter; @@ -2764,15 +2765,13 @@ public class RSRpcServices implements HBaseRPCErrorHandler, timeLimitDelta = scannerLeaseTimeoutPeriod > 0 ? scannerLeaseTimeoutPeriod : rpcTimeout; } - if (controller instanceof PayloadCarryingRpcController) { - PayloadCarryingRpcController pRpcController = - (PayloadCarryingRpcController)controller; - if (pRpcController.getCallTimeout() > 0) { - timeLimitDelta = Math.min(timeLimitDelta, pRpcController.getCallTimeout()); + if (controller instanceof TimeLimitedRpcController) { + TimeLimitedRpcController timeLimitedRpcController = + (TimeLimitedRpcController)controller; + if (timeLimitedRpcController.getCallTimeout() > 0) { + timeLimitDelta = Math.min(timeLimitDelta, + timeLimitedRpcController.getCallTimeout()); } - } else { - throw new UnsupportedOperationException("We only do " + - "PayloadCarryingRpcControllers! FIX IF A PROBLEM"); } // Use half of whichever timeout value was more restrictive... But don't allow // the time limit to be less than the allowable minimum (could cause an diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALEditsReplaySink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALEditsReplaySink.java index c71153d7be..3eb85bd337 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALEditsReplaySink.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALEditsReplaySink.java @@ -18,6 +18,8 @@ */ package org.apache.hadoop.hbase.regionserver.wal; +import com.google.protobuf.ServiceException; + import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; @@ -42,6 +44,7 @@ import org.apache.hadoop.hbase.client.RegionServerCallable; import org.apache.hadoop.hbase.client.RpcRetryingCallerFactory; import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; import org.apache.hadoop.hbase.ipc.RpcControllerFactory; +import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.ReplicationProtbufUtil; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService; @@ -58,8 +61,10 @@ import org.apache.hadoop.hbase.wal.WAL.Entry; */ @InterfaceAudience.Private public class WALEditsReplaySink { + private static final Log LOG = LogFactory.getLog(WALEditsReplaySink.class); private static final int MAX_BATCH_SIZE = 1024; + private final Configuration conf; private final ClusterConnection conn; private final TableName tableName; @@ -161,8 +166,8 @@ public class WALEditsReplaySink { try { RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate(conf, null); ReplayServerCallable callable = - new ReplayServerCallable(this.conn, this.rpcControllerFactory, - this.tableName, regionLoc, entries); + new ReplayServerCallable(this.conn, this.tableName, regionLoc, + regionInfo, entries); factory. newCaller().callWithRetries(callable, this.replayTimeout); } catch (IOException ie) { if (skipErrors) { @@ -179,19 +184,31 @@ public class WALEditsReplaySink { * @param */ class ReplayServerCallable extends RegionServerCallable { + private HRegionInfo regionInfo; private List entries; - ReplayServerCallable(final Connection connection, RpcControllerFactory rpcControllerFactory, - final TableName tableName, final HRegionLocation regionLoc, final List entries) { - super(connection, rpcControllerFactory, tableName, null); + ReplayServerCallable(final Connection connection, final TableName tableName, + final HRegionLocation regionLoc, final HRegionInfo regionInfo, + final List entries) { + super(connection, tableName, null); this.entries = entries; + this.regionInfo = regionInfo; setLocation(regionLoc); } @Override - protected ReplicateWALEntryResponse call(PayloadCarryingRpcController controller) - throws Exception { - if (entries.isEmpty()) return null; + public ReplicateWALEntryResponse call(int callTimeout) throws IOException { + try { + replayToServer(this.regionInfo, this.entries); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } + return null; + } + + private void replayToServer(HRegionInfo regionInfo, List entries) + throws IOException, ServiceException { + if (entries.isEmpty()) return; Entry[] entriesArray = new Entry[entries.size()]; entriesArray = entries.toArray(entriesArray); @@ -199,8 +216,12 @@ public class WALEditsReplaySink { Pair p = ReplicationProtbufUtil.buildReplicateWALEntryRequest(entriesArray); - controller.setCellScanner(p.getSecond()); - return remoteSvr.replay(controller, p.getFirst()); + PayloadCarryingRpcController controller = rpcControllerFactory.newController(p.getSecond()); + try { + remoteSvr.replay(controller, p.getFirst()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } @Override @@ -224,4 +245,4 @@ public class WALEditsReplaySink { } } } -} \ No newline at end of file +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java index c756294a2c..b0fd176f93 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/RegionReplicaReplicationEndpoint.java @@ -34,6 +34,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.CellScanner; @@ -45,21 +46,27 @@ import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.RegionLocations; -import org.apache.hadoop.hbase.TableDescriptors; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.TableNotFoundException; -import org.apache.hadoop.hbase.classification.InterfaceAudience; -import org.apache.hadoop.hbase.client.ClusterConnection; +import org.apache.hadoop.hbase.TableDescriptors; import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.RegionAdminServiceCallable; +import org.apache.hadoop.hbase.client.ClusterConnection; import org.apache.hadoop.hbase.client.RegionReplicaUtil; import org.apache.hadoop.hbase.client.RetryingCallable; import org.apache.hadoop.hbase.client.RpcRetryingCallerFactory; import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; import org.apache.hadoop.hbase.ipc.RpcControllerFactory; +import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.ReplicationProtbufUtil; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ReplicateWALEntryResponse; +import org.apache.hadoop.hbase.wal.WAL.Entry; +import org.apache.hadoop.hbase.wal.WALSplitter.EntryBuffers; +import org.apache.hadoop.hbase.wal.WALSplitter.OutputSink; +import org.apache.hadoop.hbase.wal.WALSplitter.PipelineController; +import org.apache.hadoop.hbase.wal.WALSplitter.RegionEntryBuffer; +import org.apache.hadoop.hbase.wal.WALSplitter.SinkWriter; import org.apache.hadoop.hbase.replication.BaseWALEntryFilter; import org.apache.hadoop.hbase.replication.ChainWALEntryFilter; import org.apache.hadoop.hbase.replication.HBaseReplicationEndpoint; @@ -67,17 +74,12 @@ import org.apache.hadoop.hbase.replication.WALEntryFilter; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.hbase.util.Threads; -import org.apache.hadoop.hbase.wal.WAL.Entry; -import org.apache.hadoop.hbase.wal.WALSplitter.EntryBuffers; -import org.apache.hadoop.hbase.wal.WALSplitter.OutputSink; -import org.apache.hadoop.hbase.wal.WALSplitter.PipelineController; -import org.apache.hadoop.hbase.wal.WALSplitter.RegionEntryBuffer; -import org.apache.hadoop.hbase.wal.WALSplitter.SinkWriter; import org.apache.hadoop.util.StringUtils; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; import com.google.common.collect.Lists; +import com.google.protobuf.ServiceException; /** * A {@link org.apache.hadoop.hbase.replication.ReplicationEndpoint} endpoint @@ -609,8 +611,9 @@ public class RegionReplicaReplicationEndpoint extends HBaseReplicationEndpoint { * Calls replay on the passed edits for the given set of entries belonging to the region. It skips * the entry if the region boundaries have changed or the region is gone. */ - static class RegionReplicaReplayCallable extends - RegionAdminServiceCallable { + static class RegionReplicaReplayCallable + extends RegionAdminServiceCallable { + private final List entries; private final byte[] initialEncodedRegionName; private final AtomicLong skippedEntries; @@ -625,25 +628,38 @@ public class RegionReplicaReplicationEndpoint extends HBaseReplicationEndpoint { this.initialEncodedRegionName = regionInfo.getEncodedNameAsBytes(); } - public ReplicateWALEntryResponse call(PayloadCarryingRpcController controller) throws Exception { - // Check whether we should still replay this entry. If the regions are changed, or the + @Override + public ReplicateWALEntryResponse call(int timeout) throws IOException { + return replayToServer(this.entries, timeout); + } + + private ReplicateWALEntryResponse replayToServer(List entries, int timeout) + throws IOException { + // check whether we should still replay this entry. If the regions are changed, or the // entry is not coming form the primary region, filter it out because we do not need it. // Regions can change because of (1) region split (2) region merge (3) table recreated boolean skip = false; + if (!Bytes.equals(location.getRegionInfo().getEncodedNameAsBytes(), - initialEncodedRegionName)) { + initialEncodedRegionName)) { skip = true; } - if (!this.entries.isEmpty() && !skip) { - Entry[] entriesArray = new Entry[this.entries.size()]; - entriesArray = this.entries.toArray(entriesArray); + if (!entries.isEmpty() && !skip) { + Entry[] entriesArray = new Entry[entries.size()]; + entriesArray = entries.toArray(entriesArray); // set the region name for the target region replica Pair p = ReplicationProtbufUtil.buildReplicateWALEntryRequest(entriesArray, location .getRegionInfo().getEncodedNameAsBytes(), null, null, null); - controller.setCellScanner(p.getSecond()); - return stub.replay(controller, p.getFirst()); + try { + PayloadCarryingRpcController controller = rpcControllerFactory.newController(p.getSecond()); + controller.setCallTimeout(timeout); + controller.setPriority(tableName); + return stub.replay(controller, p.getFirst()); + } catch (ServiceException se) { + throw ProtobufUtil.getRemoteException(se); + } } if (skip) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/Merge.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/Merge.java index 3c81cfe51c..d708edc8cf 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/Merge.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/Merge.java @@ -23,18 +23,19 @@ import java.io.IOException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.hbase.MetaTableAccessor; +import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HBaseInterfaceAudience; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.hbase.MetaTableAccessor; -import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.MasterNotRunningException; import org.apache.hadoop.hbase.ZooKeeperConnectionException; -import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.client.HBaseAdmin; @@ -79,11 +80,13 @@ public class Merge extends Configured implements Tool { // Verify HBase is down LOG.info("Verifying that HBase is not running..."); try { - HBaseAdmin.available(getConf()); + HBaseAdmin.checkHBaseAvailable(getConf()); LOG.fatal("HBase cluster must be off-line, and is not. Aborting."); return -1; } catch (ZooKeeperConnectionException zkce) { // If no zk, presume no master. + } catch (MasterNotRunningException e) { + // Expected. Ignore. } // Initialize MetaUtils and and get the root of the HBase installation diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/TestNamespace.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/TestNamespace.java index 2dca6b16f8..d778fa9ce2 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/TestNamespace.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/TestNamespace.java @@ -60,6 +60,7 @@ public class TestNamespace { private static ZKNamespaceManager zkNamespaceManager; private String prefix = "TestNamespace"; + @BeforeClass public static void setUp() throws Exception { TEST_UTIL = new HBaseTestingUtility(); @@ -300,8 +301,7 @@ public class TestNamespace { runWithExpectedException(new Callable() { @Override public Void call() throws Exception { - HTableDescriptor htd = - new HTableDescriptor(TableName.valueOf("non_existing_namespace", "table1")); + HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("non_existing_namespace", "table1")); htd.addFamily(new HColumnDescriptor("family1")); admin.createTable(htd); return null; @@ -387,4 +387,5 @@ public class TestNamespace { } fail("Should have thrown exception " + exceptionClass); } -} \ No newline at end of file + +} diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java index 171662228b..d088fc4695 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAdmin2.java @@ -37,6 +37,7 @@ import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.HRegionLocation; import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.MasterNotRunningException; import org.apache.hadoop.hbase.MiniHBaseCluster; import org.apache.hadoop.hbase.NotServingRegionException; import org.apache.hadoop.hbase.ProcedureInfo; @@ -58,6 +59,7 @@ import org.apache.hadoop.hbase.testclassification.LargeTests; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.hbase.wal.AbstractFSWALProvider; +import org.apache.hadoop.hbase.wal.FSHLogProvider; import org.junit.After; import org.junit.AfterClass; import org.junit.Before; @@ -65,6 +67,8 @@ import org.junit.BeforeClass; import org.junit.Test; import org.junit.experimental.categories.Category; +import com.google.protobuf.ServiceException; + /** * Class to test HBaseAdmin. @@ -639,9 +643,11 @@ public class TestAdmin2 { long start = System.currentTimeMillis(); try { - HBaseAdmin.available(conf); + HBaseAdmin.checkHBaseAvailable(conf); assertTrue(false); + } catch (MasterNotRunningException ignored) { } catch (ZooKeeperConnectionException ignored) { + } catch (ServiceException ignored) { } catch (IOException ignored) { } long end = System.currentTimeMillis(); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestClientTimeouts.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestClientTimeouts.java index f49c55843a..679d9c91da 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestClientTimeouts.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestClientTimeouts.java @@ -28,10 +28,13 @@ import java.net.UnknownHostException; import java.util.Random; import java.util.concurrent.atomic.AtomicInteger; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.MasterNotRunningException; import org.apache.hadoop.hbase.ServerName; import org.apache.hadoop.hbase.ZooKeeperConnectionException; import org.apache.hadoop.hbase.ipc.AbstractRpcClient; @@ -53,6 +56,7 @@ import com.google.protobuf.ServiceException; @Category({MediumTests.class, ClientTests.class}) public class TestClientTimeouts { + private static final Log LOG = LogFactory.getLog(TestClientTimeouts.class); private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility(); protected static int SLAVES = 1; @@ -83,6 +87,7 @@ public class TestClientTimeouts { */ @Test public void testAdminTimeout() throws Exception { + Connection lastConnection = null; boolean lastFailed = false; int initialInvocations = RandomTimeoutBlockingRpcChannel.invokations.get(); RandomTimeoutRpcClient rpcClient = (RandomTimeoutRpcClient) RpcClientFactory @@ -100,7 +105,7 @@ public class TestClientTimeouts { connection = ConnectionFactory.createConnection(conf); admin = connection.getAdmin(); // run some admin commands - HBaseAdmin.available(conf); + HBaseAdmin.checkHBaseAvailable(conf); admin.setBalancerRunning(false, false); } catch (ZooKeeperConnectionException ex) { // Since we are randomly throwing SocketTimeoutExceptions, it is possible to get diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestHCM.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestHCM.java index 33af5dea24..1b20b76622 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestHCM.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestHCM.java @@ -61,7 +61,6 @@ import org.apache.hadoop.hbase.exceptions.DeserializationException; import org.apache.hadoop.hbase.exceptions.RegionMovedException; import org.apache.hadoop.hbase.filter.Filter; import org.apache.hadoop.hbase.filter.FilterBase; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; import org.apache.hadoop.hbase.ipc.RpcClient; import org.apache.hadoop.hbase.ipc.RpcControllerFactory; import org.apache.hadoop.hbase.master.HMaster; @@ -104,6 +103,8 @@ public class TestHCM { TableName.valueOf("test2"); private static final TableName TABLE_NAME3 = TableName.valueOf("test3"); + private static final TableName TABLE_NAME4 = + TableName.valueOf("test4"); private static final byte[] FAM_NAM = Bytes.toBytes("f"); private static final byte[] ROW = Bytes.toBytes("bbb"); private static final byte[] ROW_X = Bytes.toBytes("xxx"); @@ -406,11 +407,10 @@ public class TestHCM { long pauseTime; long baseTime = 100; TableName tableName = TableName.valueOf("HCM-testCallableSleep"); + Table table = TEST_UTIL.createTable(tableName, FAM_NAM); RegionServerCallable regionServerCallable = new RegionServerCallable( - TEST_UTIL.getConnection(), new RpcControllerFactory(TEST_UTIL.getConfiguration()), - tableName, ROW) { - @Override - protected Object call(PayloadCarryingRpcController controller) throws Exception { + TEST_UTIL.getConnection(), tableName, ROW) { + public Object call(int timeout) throws IOException { return null; } }; @@ -424,10 +424,9 @@ public class TestHCM { RegionAdminServiceCallable regionAdminServiceCallable = new RegionAdminServiceCallable( - (ClusterConnection) TEST_UTIL.getConnection(), - new RpcControllerFactory(TEST_UTIL.getConfiguration()), tableName, ROW) { - @Override - public Object call(PayloadCarryingRpcController controller) throws Exception { + (ClusterConnection) TEST_UTIL.getConnection(), new RpcControllerFactory( + TEST_UTIL.getConfiguration()), tableName, ROW) { + public Object call(int timeout) throws IOException { return null; } }; @@ -439,21 +438,16 @@ public class TestHCM { assertTrue(pauseTime <= (baseTime * HConstants.RETRY_BACKOFF[i] * 1.01f)); } - MasterCallable masterCallable = new MasterCallable(TEST_UTIL.getConnection(), - new RpcControllerFactory(TEST_UTIL.getConfiguration())) { - @Override - protected Object call(PayloadCarryingRpcController rpcController) throws Exception { + MasterCallable masterCallable = new MasterCallable(TEST_UTIL.getConnection()) { + public Object call(int timeout) throws IOException { return null; } }; - try { - for (int i = 0; i < HConstants.RETRY_BACKOFF.length; i++) { - pauseTime = masterCallable.sleep(baseTime, i); - assertTrue(pauseTime >= (baseTime * HConstants.RETRY_BACKOFF[i])); - assertTrue(pauseTime <= (baseTime * HConstants.RETRY_BACKOFF[i] * 1.01f)); - } - } finally { - masterCallable.close(); + + for (int i = 0; i < HConstants.RETRY_BACKOFF.length; i++) { + pauseTime = masterCallable.sleep(baseTime, i); + assertTrue(pauseTime >= (baseTime * HConstants.RETRY_BACKOFF[i])); + assertTrue(pauseTime <= (baseTime * HConstants.RETRY_BACKOFF[i] * 1.01f)); } } @@ -1155,6 +1149,7 @@ public class TestHCM { ManualEnvironmentEdge timeMachine = new ManualEnvironmentEdge(); EnvironmentEdgeManager.injectEdge(timeMachine); try { + long timeBase = timeMachine.currentTime(); long largeAmountOfTime = ANY_PAUSE * 1000; ConnectionImplementation.ServerErrorTracker tracker = new ConnectionImplementation.ServerErrorTracker(largeAmountOfTime, 100); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java index d99d2ee462..354f0a8aa5 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestReplicaWithCluster.java @@ -35,8 +35,6 @@ import org.apache.hadoop.hbase.client.replication.ReplicationAdmin; import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver; import org.apache.hadoop.hbase.coprocessor.ObserverContext; import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; -import org.apache.hadoop.hbase.ipc.RpcControllerFactory; import org.apache.hadoop.hbase.regionserver.StorefileRefresherChore; import org.apache.hadoop.hbase.regionserver.TestHRegionServerBulkLoad; import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; @@ -334,27 +332,26 @@ public class TestReplicaWithCluster { // bulk load HFiles LOG.debug("Loading test data"); + @SuppressWarnings("deprecation") final ClusterConnection conn = (ClusterConnection) HTU.getAdmin().getConnection(); table = conn.getTable(hdt.getTableName()); - final String bulkToken = - new SecureBulkLoadClient(HTU.getConfiguration(), table).prepareBulkLoad(conn); - RegionServerCallable callable = new RegionServerCallable(conn, - new RpcControllerFactory(HTU.getConfiguration()), hdt.getTableName(), - TestHRegionServerBulkLoad.rowkey(0)) { - @Override - protected Void call(PayloadCarryingRpcController controller) throws Exception { - LOG.debug("Going to connect to server " + getLocation() + " for row " + final String bulkToken = new SecureBulkLoadClient(table).prepareBulkLoad(conn); + RegionServerCallable callable = new RegionServerCallable( + conn, hdt.getTableName(), TestHRegionServerBulkLoad.rowkey(0)) { + @Override + public Void call(int timeout) throws Exception { + LOG.debug("Going to connect to server " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); - SecureBulkLoadClient secureClient = null; - byte[] regionName = getLocation().getRegionInfo().getRegionName(); - try (Table table = conn.getTable(getTableName())) { - secureClient = new SecureBulkLoadClient(HTU.getConfiguration(), table); - secureClient.secureBulkLoadHFiles(getStub(), famPaths, regionName, - true, null, bulkToken); + SecureBulkLoadClient secureClient = null; + byte[] regionName = getLocation().getRegionInfo().getRegionName(); + try (Table table = conn.getTable(getTableName())) { + secureClient = new SecureBulkLoadClient(table); + secureClient.secureBulkLoadHFiles(getStub(), famPaths, regionName, + true, null, bulkToken); + } + return null; } - return null; - } - }; + }; RpcRetryingCallerFactory factory = new RpcRetryingCallerFactory(HTU.getConfiguration()); RpcRetryingCaller caller = factory.newCaller(); caller.callWithRetries(callable, 10000); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java index 30805c0fc7..6e6820120e 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoad.java @@ -62,8 +62,6 @@ import org.apache.hadoop.hbase.io.hfile.CacheConfig; import org.apache.hadoop.hbase.io.hfile.HFile; import org.apache.hadoop.hbase.io.hfile.HFileContext; import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; -import org.apache.hadoop.hbase.ipc.RpcControllerFactory; import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CompactRegionRequest; @@ -200,20 +198,19 @@ public class TestHRegionServerBulkLoad { } // bulk load HFiles - final ClusterConnection conn = (ClusterConnection)UTIL.getConnection(); + final ClusterConnection conn = (ClusterConnection) UTIL.getAdmin().getConnection(); Table table = conn.getTable(tableName); - final String bulkToken = new SecureBulkLoadClient(UTIL.getConfiguration(), table). - prepareBulkLoad(conn); - RegionServerCallable callable = new RegionServerCallable(conn, - new RpcControllerFactory(UTIL.getConfiguration()), tableName, Bytes.toBytes("aaa")) { + final String bulkToken = new SecureBulkLoadClient(table).prepareBulkLoad(conn); + RegionServerCallable callable = + new RegionServerCallable(conn, tableName, Bytes.toBytes("aaa")) { @Override - public Void call(PayloadCarryingRpcController controller) throws Exception { + public Void call(int callTimeout) throws Exception { LOG.debug("Going to connect to server " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); SecureBulkLoadClient secureClient = null; byte[] regionName = getLocation().getRegionInfo().getRegionName(); try (Table table = conn.getTable(getTableName())) { - secureClient = new SecureBulkLoadClient(UTIL.getConfiguration(), table); + secureClient = new SecureBulkLoadClient(table); secureClient.secureBulkLoadHFiles(getStub(), famPaths, regionName, true, null, bulkToken); } @@ -227,15 +224,15 @@ public class TestHRegionServerBulkLoad { // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 5 == 0) { // 5 * 50 = 250 open file handles! - callable = new RegionServerCallable(conn, - new RpcControllerFactory(UTIL.getConfiguration()), tableName, Bytes.toBytes("aaa")) { + callable = new RegionServerCallable(conn, tableName, Bytes.toBytes("aaa")) { @Override - protected Void call(PayloadCarryingRpcController controller) throws Exception { + public Void call(int callTimeout) throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = conn.getAdmin(getLocation().getServerName()); - CompactRegionRequest request = RequestConverter.buildCompactRegionRequest( + CompactRegionRequest request = + RequestConverter.buildCompactRegionRequest( getLocation().getRegionInfo().getRegionName(), true, null); server.compactRegion(null, request); numCompactions.incrementAndGet(); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java index 7560a419bc..d55adefe62 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldClient.java @@ -33,8 +33,6 @@ import org.apache.hadoop.hbase.client.ClusterConnection; import org.apache.hadoop.hbase.client.RegionServerCallable; import org.apache.hadoop.hbase.client.RpcRetryingCaller; import org.apache.hadoop.hbase.client.RpcRetryingCallerFactory; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; -import org.apache.hadoop.hbase.ipc.RpcControllerFactory; import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CompactRegionRequest; @@ -91,12 +89,10 @@ public class TestHRegionServerBulkLoadWithOldClient extends TestHRegionServerBul // bulk load HFiles final ClusterConnection conn = (ClusterConnection) UTIL.getAdmin().getConnection(); - RpcControllerFactory rpcControllerFactory = new RpcControllerFactory(UTIL.getConfiguration()); RegionServerCallable callable = - new RegionServerCallable(conn, rpcControllerFactory, tableName, - Bytes.toBytes("aaa")) { + new RegionServerCallable(conn, tableName, Bytes.toBytes("aaa")) { @Override - protected Void call(PayloadCarryingRpcController controller) throws Exception { + public Void call(int callTimeout) throws Exception { LOG.info("Non-secure old client"); byte[] regionName = getLocation().getRegionInfo().getRegionName(); BulkLoadHFileRequest request = @@ -113,10 +109,9 @@ public class TestHRegionServerBulkLoadWithOldClient extends TestHRegionServerBul // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 5 == 0) { // 5 * 50 = 250 open file handles! - callable = new RegionServerCallable(conn, rpcControllerFactory, tableName, - Bytes.toBytes("aaa")) { + callable = new RegionServerCallable(conn, tableName, Bytes.toBytes("aaa")) { @Override - protected Void call(PayloadCarryingRpcController controller) throws Exception { + public Void call(int callTimeout) throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java index 0bc9498bc6..6de6261b94 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionServerBulkLoadWithOldSecureEndpoint.java @@ -33,13 +33,13 @@ import org.apache.hadoop.hbase.client.ClusterConnection; import org.apache.hadoop.hbase.client.RegionServerCallable; import org.apache.hadoop.hbase.client.RpcRetryingCaller; import org.apache.hadoop.hbase.client.RpcRetryingCallerFactory; +import org.apache.hadoop.hbase.client.SecureBulkLoadClient; import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.coprocessor.CoprocessorHost; -import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController; -import org.apache.hadoop.hbase.ipc.RpcControllerFactory; import org.apache.hadoop.hbase.protobuf.RequestConverter; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.CompactRegionRequest; +import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileRequest; import org.apache.hadoop.hbase.testclassification.LargeTests; import org.apache.hadoop.hbase.testclassification.RegionServerTests; import org.apache.hadoop.hbase.util.Bytes; @@ -62,8 +62,7 @@ public class TestHRegionServerBulkLoadWithOldSecureEndpoint extends TestHRegionS super(duration); } - private static final Log LOG = - LogFactory.getLog(TestHRegionServerBulkLoadWithOldSecureEndpoint.class); + private static final Log LOG = LogFactory.getLog(TestHRegionServerBulkLoadWithOldSecureEndpoint.class); @BeforeClass public static void setUpBeforeClass() throws IOException { @@ -104,17 +103,16 @@ public class TestHRegionServerBulkLoadWithOldSecureEndpoint extends TestHRegionS final ClusterConnection conn = (ClusterConnection) UTIL.getAdmin().getConnection(); Table table = conn.getTable(tableName); final String bulkToken = new SecureBulkLoadEndpointClient(table).prepareBulkLoad(tableName); - RpcControllerFactory rpcControllerFactory = new RpcControllerFactory(UTIL.getConfiguration()); RegionServerCallable callable = - new RegionServerCallable(conn, rpcControllerFactory, tableName, - Bytes.toBytes("aaa")) { + new RegionServerCallable(conn, tableName, Bytes.toBytes("aaa")) { @Override - protected Void call(PayloadCarryingRpcController controller) throws Exception { - LOG.debug("Going to connect to server " + getLocation() + " for row " + - Bytes.toStringBinary(getRow())); + public Void call(int callTimeout) throws Exception { + LOG.debug("Going to connect to server " + getLocation() + " for row " + + Bytes.toStringBinary(getRow())); try (Table table = conn.getTable(getTableName())) { - boolean loaded = new SecureBulkLoadEndpointClient(table).bulkLoadHFiles(famPaths, - null, bulkToken, getLocation().getRegionInfo().getStartKey()); + boolean loaded = + new SecureBulkLoadEndpointClient(table).bulkLoadHFiles(famPaths, null, + bulkToken, getLocation().getRegionInfo().getStartKey()); } return null; } @@ -126,10 +124,9 @@ public class TestHRegionServerBulkLoadWithOldSecureEndpoint extends TestHRegionS // Periodically do compaction to reduce the number of open file handles. if (numBulkLoads.get() % 5 == 0) { // 5 * 50 = 250 open file handles! - callable = new RegionServerCallable(conn, rpcControllerFactory, tableName, - Bytes.toBytes("aaa")) { + callable = new RegionServerCallable(conn, tableName, Bytes.toBytes("aaa")) { @Override - protected Void call(PayloadCarryingRpcController controller) throws Exception { + public Void call(int callTimeout) throws Exception { LOG.debug("compacting " + getLocation() + " for row " + Bytes.toStringBinary(getRow())); AdminProtos.AdminService.BlockingInterface server = diff --git a/hbase-spark/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java b/hbase-spark/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java index 3e90fe14ee..fa66d69c6f 100644 --- a/hbase-spark/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java +++ b/hbase-spark/src/main/java/org/apache/hadoop/hbase/spark/SparkSQLPushDownFilter.java @@ -17,6 +17,8 @@ package org.apache.hadoop.hbase.spark; +import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.hbase.Cell; @@ -35,8 +37,6 @@ import java.io.IOException; import java.util.HashMap; import java.util.List; import java.util.Map; -import com.google.protobuf.InvalidProtocolBufferException; -import com.google.protobuf.ByteString; /** * This filter will push down all qualifier logic given to us -- 2.11.4.GIT