Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-23898 Add trace support for simple apis in async client (#2813) 

Signed-off-by: Guanghao Zhang <zghao@apache.org>
This commit is contained in:
Duo Zhang 2020-12-30 14:58:57 +08:00 committed by Duo Zhang
parent 57960fa8fa
commit 805b2ae2ad
16 changed files with 1169 additions and 346 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
hbase-client/pom.xml
View File

@ -144,6 +144,16 @@
<groupId>org.jruby.joni</groupId>
<artifactId>joni</artifactId>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk-testing</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>jcl-over-slf4j</artifactId>

4
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnection.java
View File

@ -64,8 +64,8 @@ public interface AsyncConnection extends Closeable {
/**
* Retrieve an {@link AsyncTable} implementation for accessing a table.
* <p>
* The returned instance will use default configs. Use {@link #getTableBuilder(TableName)} if
* you want to customize some configs.
* The returned instance will use default configs. Use {@link #getTableBuilder(TableName)} if you
* want to customize some configs.
* <p>
* This method no longer checks table existence. An exception will be thrown if the table does not
* exist only when the first operation is attempted.

114
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java
View File

@ -27,6 +27,8 @@ import static org.apache.hadoop.hbase.client.MetricsConnection.CLIENT_SIDE_METRI
import static org.apache.hadoop.hbase.client.NonceGenerator.CLIENT_NONCES_ENABLED_KEY;
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.context.Scope;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.Optional;
@ -51,6 +53,7 @@ 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.security.User;
import org.apache.hadoop.hbase.trace.TraceUtil;
import org.apache.hadoop.hbase.util.ConcurrentMapUtils;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.security.UserGroupInformation;
@ -121,7 +124,7 @@ class AsyncConnectionImpl implements AsyncConnection {
private volatile ConnectionOverAsyncConnection conn;
public AsyncConnectionImpl(Configuration conf, ConnectionRegistry registry, String clusterId,
SocketAddress localAddress, User user) {
SocketAddress localAddress, User user) {
this.conf = conf;
this.user = user;
@ -135,8 +138,8 @@ class AsyncConnectionImpl implements AsyncConnection {
} else {
this.metrics = Optional.empty();
}
this.rpcClient = RpcClientFactory.createClient(
conf, clusterId, localAddress, metrics.orElse(null));
this.rpcClient =
RpcClientFactory.createClient(conf, clusterId, localAddress, metrics.orElse(null));
this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
this.rpcTimeout =
(int) Math.min(Integer.MAX_VALUE, TimeUnit.NANOSECONDS.toMillis(connConf.getRpcTimeoutNs()));
@ -159,14 +162,13 @@ class AsyncConnectionImpl implements AsyncConnection {
LOG.warn("{} is true, but {} is not set", STATUS_PUBLISHED, STATUS_LISTENER_CLASS);
} else {
try {
listener = new ClusterStatusListener(
new ClusterStatusListener.DeadServerHandler() {
@Override
public void newDead(ServerName sn) {
locator.clearCache(sn);
rpcClient.cancelConnections(sn);
}
}, conf, listenerClass);
listener = new ClusterStatusListener(new ClusterStatusListener.DeadServerHandler() {
@Override
public void newDead(ServerName sn) {
locator.clearCache(sn);
rpcClient.cancelConnections(sn);
}
}, conf, listenerClass);
} catch (IOException e) {
LOG.warn("Failed create of ClusterStatusListener, not a critical, ignoring...", e);
}
@ -206,28 +208,30 @@ class AsyncConnectionImpl implements AsyncConnection {
@Override
public void close() {
if (!closed.compareAndSet(false, true)) {
return;
}
LOG.info("Connection has been closed by {}.", Thread.currentThread().getName());
if(LOG.isDebugEnabled()){
logCallStack(Thread.currentThread().getStackTrace());
}
IOUtils.closeQuietly(clusterStatusListener,
e -> LOG.warn("failed to close clusterStatusListener", e));
IOUtils.closeQuietly(rpcClient, e -> LOG.warn("failed to close rpcClient", e));
IOUtils.closeQuietly(registry, e -> LOG.warn("failed to close registry", e));
synchronized (this) {
if (choreService != null) {
choreService.shutdown();
choreService = null;
TraceUtil.trace(() -> {
if (!closed.compareAndSet(false, true)) {
return;
}
}
metrics.ifPresent(MetricsConnection::shutdown);
ConnectionOverAsyncConnection c = this.conn;
if (c != null) {
c.closePool();
}
LOG.info("Connection has been closed by {}.", Thread.currentThread().getName());
if (LOG.isDebugEnabled()) {
logCallStack(Thread.currentThread().getStackTrace());
}
IOUtils.closeQuietly(clusterStatusListener,
e -> LOG.warn("failed to close clusterStatusListener", e));
IOUtils.closeQuietly(rpcClient, e -> LOG.warn("failed to close rpcClient", e));
IOUtils.closeQuietly(registry, e -> LOG.warn("failed to close registry", e));
synchronized (this) {
if (choreService != null) {
choreService.shutdown();
choreService = null;
}
}
metrics.ifPresent(MetricsConnection::shutdown);
ConnectionOverAsyncConnection c = this.conn;
if (c != null) {
c.closePool();
}
}, "AsyncConnection.close");
}
private void logCallStack(StackTraceElement[] stackTraceElements) {
@ -341,7 +345,7 @@ class AsyncConnectionImpl implements AsyncConnection {
@Override
public AsyncTableBuilder<ScanResultConsumer> getTableBuilder(TableName tableName,
ExecutorService pool) {
ExecutorService pool) {
return new AsyncTableBuilderBase<ScanResultConsumer>(tableName, connConf) {
@Override
@ -382,7 +386,7 @@ class AsyncConnectionImpl implements AsyncConnection {
@Override
public AsyncBufferedMutatorBuilder getBufferedMutatorBuilder(TableName tableName,
ExecutorService pool) {
ExecutorService pool) {
return new AsyncBufferedMutatorBuilderImpl(connConf, getTableBuilder(tableName, pool),
RETRY_TIMER);
}
@ -406,28 +410,36 @@ class AsyncConnectionImpl implements AsyncConnection {
@Override
public CompletableFuture<Hbck> getHbck() {
CompletableFuture<Hbck> future = new CompletableFuture<>();
addListener(registry.getActiveMaster(), (sn, error) -> {
if (error != null) {
future.completeExceptionally(error);
} else {
try {
future.complete(getHbck(sn));
} catch (IOException e) {
future.completeExceptionally(e);
return TraceUtil.tracedFuture(() -> {
CompletableFuture<Hbck> future = new CompletableFuture<>();
addListener(registry.getActiveMaster(), (sn, error) -> {
if (error != null) {
future.completeExceptionally(error);
} else {
try {
future.complete(getHbck(sn));
} catch (IOException e) {
future.completeExceptionally(e);
}
}
}
});
return future;
});
return future;
}, getClass().getName() + ".getHbck");
}
@Override
public Hbck getHbck(ServerName masterServer) throws IOException {
// we will not create a new connection when creating a new protobuf stub, and for hbck there
// will be no performance consideration, so for simplification we will create a new stub every
// time instead of caching the stub here.
return new HBaseHbck(MasterProtos.HbckService.newBlockingStub(
rpcClient.createBlockingRpcChannel(masterServer, user, rpcTimeout)), rpcControllerFactory);
Span span = TraceUtil.createSpan(getClass().getName() + ".getHbck")
.setAttribute(TraceUtil.SERVER_NAME_KEY, masterServer.getServerName());
try (Scope scope = span.makeCurrent()) {
// we will not create a new connection when creating a new protobuf stub, and for hbck there
// will be no performance consideration, so for simplification we will create a new stub every
// time instead of caching the stub here.
return new HBaseHbck(
MasterProtos.HbckService
.newBlockingStub(rpcClient.createBlockingRpcChannel(masterServer, user, rpcTimeout)),
rpcControllerFactory);
}
}
Optional<MetricsConnection> getConnectionMetrics() {

166
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRegionLocator.java
View File

@ -18,16 +18,28 @@
package org.apache.hadoop.hbase.client;
import static org.apache.hadoop.hbase.TableName.META_TABLE_NAME;
import static org.apache.hadoop.hbase.trace.TraceUtil.REGION_NAMES_KEY;
import static org.apache.hadoop.hbase.trace.TraceUtil.SERVER_NAME_KEY;
import static org.apache.hadoop.hbase.trace.TraceUtil.createSpan;
import static org.apache.hadoop.hbase.trace.TraceUtil.createTableSpan;
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Scope;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.RegionLocations;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
import org.apache.hadoop.hbase.trace.TraceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.FutureUtils;
import org.apache.yetus.audience.InterfaceAudience;
@ -61,7 +73,7 @@ class AsyncRegionLocator {
}
private <T> CompletableFuture<T> withTimeout(CompletableFuture<T> future, long timeoutNs,
Supplier<String> timeoutMsg) {
Supplier<String> timeoutMsg) {
if (future.isDone() || timeoutNs <= 0) {
return future;
}
@ -84,64 +96,101 @@ class AsyncRegionLocator {
return TableName.isMetaTableName(tableName);
}
private <T> CompletableFuture<T> tracedLocationFuture(Supplier<CompletableFuture<T>> action,
Function<T, List<String>> getRegionNames, TableName tableName, String methodName) {
Span span = createTableSpan(getClass().getSimpleName() + "." + methodName, tableName);
try (Scope scope = span.makeCurrent()) {
CompletableFuture<T> future = action.get();
FutureUtils.addListener(future, (resp, error) -> {
if (error != null) {
span.recordException(error);
span.setStatus(StatusCode.ERROR);
} else {
List<String> regionNames = getRegionNames.apply(resp);
if (!regionNames.isEmpty()) {
span.setAttribute(REGION_NAMES_KEY, regionNames);
}
span.setStatus(StatusCode.OK);
}
span.end();
});
return future;
}
}
private List<String> getRegionName(RegionLocations locs) {
List<String> names = new ArrayList<>();
for (HRegionLocation loc : locs.getRegionLocations()) {
if (loc != null) {
names.add(loc.getRegion().getRegionNameAsString());
}
}
return names;
}
CompletableFuture<RegionLocations> getRegionLocations(TableName tableName, byte[] row,
RegionLocateType type, boolean reload, long timeoutNs) {
CompletableFuture<RegionLocations> future = isMeta(tableName)
? metaRegionLocator.getRegionLocations(RegionReplicaUtil.DEFAULT_REPLICA_ID, reload)
: nonMetaRegionLocator.getRegionLocations(tableName, row,
RegionReplicaUtil.DEFAULT_REPLICA_ID, type, reload);
return withTimeout(future, timeoutNs,
() -> "Timeout(" + TimeUnit.NANOSECONDS.toMillis(timeoutNs) +
"ms) waiting for region locations for " + tableName + ", row='" +
Bytes.toStringBinary(row) + "'");
RegionLocateType type, boolean reload, long timeoutNs) {
return tracedLocationFuture(() -> {
CompletableFuture<RegionLocations> future = isMeta(tableName) ?
metaRegionLocator.getRegionLocations(RegionReplicaUtil.DEFAULT_REPLICA_ID, reload) :
nonMetaRegionLocator.getRegionLocations(tableName, row,
RegionReplicaUtil.DEFAULT_REPLICA_ID, type, reload);
return withTimeout(future, timeoutNs,
() -> "Timeout(" + TimeUnit.NANOSECONDS.toMillis(timeoutNs) +
"ms) waiting for region locations for " + tableName + ", row='" +
Bytes.toStringBinary(row) + "'");
}, this::getRegionName, tableName, "getRegionLocations");
}
CompletableFuture<HRegionLocation> getRegionLocation(TableName tableName, byte[] row,
int replicaId, RegionLocateType type, boolean reload, long timeoutNs) {
// meta region can not be split right now so we always call the same method.
// Change it later if the meta table can have more than one regions.
CompletableFuture<HRegionLocation> future = new CompletableFuture<>();
CompletableFuture<RegionLocations> locsFuture =
isMeta(tableName) ? metaRegionLocator.getRegionLocations(replicaId, reload)
: nonMetaRegionLocator.getRegionLocations(tableName, row, replicaId, type, reload);
addListener(locsFuture, (locs, error) -> {
if (error != null) {
future.completeExceptionally(error);
return;
}
HRegionLocation loc = locs.getRegionLocation(replicaId);
if (loc == null) {
future.completeExceptionally(
new RegionOfflineException("No location for " + tableName + ", row='" +
Bytes.toStringBinary(row) + "', locateType=" + type + ", replicaId=" + replicaId));
} else if (loc.getServerName() == null) {
future.completeExceptionally(
new RegionOfflineException("No server address listed for region '" +
loc.getRegion().getRegionNameAsString() + ", row='" + Bytes.toStringBinary(row) +
"', locateType=" + type + ", replicaId=" + replicaId));
} else {
future.complete(loc);
}
});
return withTimeout(future, timeoutNs,
() -> "Timeout(" + TimeUnit.NANOSECONDS.toMillis(timeoutNs) +
"ms) waiting for region location for " + tableName + ", row='" + Bytes.toStringBinary(row) +
"', replicaId=" + replicaId);
int replicaId, RegionLocateType type, boolean reload, long timeoutNs) {
return tracedLocationFuture(() -> {
// meta region can not be split right now so we always call the same method.
// Change it later if the meta table can have more than one regions.
CompletableFuture<HRegionLocation> future = new CompletableFuture<>();
CompletableFuture<RegionLocations> locsFuture =
isMeta(tableName) ? metaRegionLocator.getRegionLocations(replicaId, reload) :
nonMetaRegionLocator.getRegionLocations(tableName, row, replicaId, type, reload);
addListener(locsFuture, (locs, error) -> {
if (error != null) {
future.completeExceptionally(error);
return;
}
HRegionLocation loc = locs.getRegionLocation(replicaId);
if (loc == null) {
future.completeExceptionally(
new RegionOfflineException("No location for " + tableName + ", row='" +
Bytes.toStringBinary(row) + "', locateType=" + type + ", replicaId=" + replicaId));
} else if (loc.getServerName() == null) {
future.completeExceptionally(
new RegionOfflineException("No server address listed for region '" +
loc.getRegion().getRegionNameAsString() + ", row='" + Bytes.toStringBinary(row) +
"', locateType=" + type + ", replicaId=" + replicaId));
} else {
future.complete(loc);
}
});
return withTimeout(future, timeoutNs,
() -> "Timeout(" + TimeUnit.NANOSECONDS.toMillis(timeoutNs) +
"ms) waiting for region location for " + tableName + ", row='" +
Bytes.toStringBinary(row) + "', replicaId=" + replicaId);
}, loc -> Arrays.asList(loc.getRegion().getRegionNameAsString()), tableName,
"getRegionLocation");
}
CompletableFuture<HRegionLocation> getRegionLocation(TableName tableName, byte[] row,
int replicaId, RegionLocateType type, long timeoutNs) {
int replicaId, RegionLocateType type, long timeoutNs) {
return getRegionLocation(tableName, row, replicaId, type, false, timeoutNs);
}
CompletableFuture<HRegionLocation> getRegionLocation(TableName tableName, byte[] row,
RegionLocateType type, boolean reload, long timeoutNs) {
RegionLocateType type, boolean reload, long timeoutNs) {
return getRegionLocation(tableName, row, RegionReplicaUtil.DEFAULT_REPLICA_ID, type, reload,
timeoutNs);
}
CompletableFuture<HRegionLocation> getRegionLocation(TableName tableName, byte[] row,
RegionLocateType type, long timeoutNs) {
RegionLocateType type, long timeoutNs) {
return getRegionLocation(tableName, row, type, false, timeoutNs);
}
@ -154,24 +203,31 @@ class AsyncRegionLocator {
}
void clearCache(TableName tableName) {
LOG.debug("Clear meta cache for {}", tableName);
if (tableName.equals(META_TABLE_NAME)) {
metaRegionLocator.clearCache();
} else {
nonMetaRegionLocator.clearCache(tableName);
}
TraceUtil.trace(() -> {
LOG.debug("Clear meta cache for {}", tableName);
if (tableName.equals(META_TABLE_NAME)) {
metaRegionLocator.clearCache();
} else {
nonMetaRegionLocator.clearCache(tableName);
}
}, () -> createTableSpan("AsyncRegionLocator.clearCache", tableName));
}
void clearCache(ServerName serverName) {
LOG.debug("Clear meta cache for {}", serverName);
metaRegionLocator.clearCache(serverName);
nonMetaRegionLocator.clearCache(serverName);
conn.getConnectionMetrics().ifPresent(MetricsConnection::incrMetaCacheNumClearServer);
TraceUtil.trace(() -> {
LOG.debug("Clear meta cache for {}", serverName);
metaRegionLocator.clearCache(serverName);
nonMetaRegionLocator.clearCache(serverName);
conn.getConnectionMetrics().ifPresent(MetricsConnection::incrMetaCacheNumClearServer);
}, () -> createSpan("AsyncRegionLocator.clearCache").setAttribute(SERVER_NAME_KEY,
serverName.getServerName()));
}
void clearCache() {
metaRegionLocator.clearCache();
nonMetaRegionLocator.clearCache();
TraceUtil.trace(() -> {
metaRegionLocator.clearCache();
nonMetaRegionLocator.clearCache();
}, "AsyncRegionLocator.clearCache");
}
AsyncNonMetaRegionLocator getNonMetaRegionLocator() {

41
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTable.java
View File

@ -70,6 +70,7 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
* Gets the {@link AsyncTableRegionLocator} for this table.
*/
AsyncTableRegionLocator getRegionLocator();
/**
* Get timeout of each rpc request in this Table instance. It will be overridden by a more
* specific rpc timeout config such as readRpcTimeout or writeRpcTimeout.
@ -184,7 +185,7 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
* {@link CompletableFuture}.
*/
default CompletableFuture<Long> incrementColumnValue(byte[] row, byte[] family, byte[] qualifier,
long amount) {
long amount) {
return incrementColumnValue(row, family, qualifier, amount, Durability.SYNC_WAL);
}
@ -204,12 +205,12 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
* {@link CompletableFuture}.
*/
default CompletableFuture<Long> incrementColumnValue(byte[] row, byte[] family, byte[] qualifier,
long amount, Durability durability) {
long amount, Durability durability) {
Preconditions.checkNotNull(row, "row is null");
Preconditions.checkNotNull(family, "family is null");
return increment(
new Increment(row).addColumn(family, qualifier, amount).setDurability(durability))
.thenApply(r -> Bytes.toLong(r.getValue(family, qualifier)));
.thenApply(r -> Bytes.toLong(r.getValue(family, qualifier)));
}
/**
@ -233,16 +234,15 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
* </pre>
*
* @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
* any more.
* any more.
*/
@Deprecated
CheckAndMutateBuilder checkAndMutate(byte[] row, byte[] family);
/**
* A helper class for sending checkAndMutate request.
*
* @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
* any more.
* any more.
*/
@Deprecated
interface CheckAndMutateBuilder {
@ -319,16 +319,15 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
* </pre>
*
* @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
* any more.
* any more.
*/
@Deprecated
CheckAndMutateWithFilterBuilder checkAndMutate(byte[] row, Filter filter);
/**
* A helper class for sending checkAndMutate request with a filter.
*
* @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
* any more.
* any more.
*/
@Deprecated
interface CheckAndMutateWithFilterBuilder {
@ -361,9 +360,8 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
}
/**
* checkAndMutate that atomically checks if a row matches the specified condition. If it does,
* it performs the specified action.
*
* checkAndMutate that atomically checks if a row matches the specified condition. If it does, it
* performs the specified action.
* @param checkAndMutate The CheckAndMutate object.
* @return A {@link CompletableFuture}s that represent the result for the CheckAndMutate.
*/
@ -373,22 +371,19 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
* Batch version of checkAndMutate. The specified CheckAndMutates are batched only in the sense
* that they are sent to a RS in one RPC, but each CheckAndMutate operation is still executed
* atomically (and thus, each may fail independently of others).
*
* @param checkAndMutates The list of CheckAndMutate.
* @return A list of {@link CompletableFuture}s that represent the result for each
* CheckAndMutate.
* @return A list of {@link CompletableFuture}s that represent the result for each CheckAndMutate.
*/
List<CompletableFuture<CheckAndMutateResult>> checkAndMutate(
List<CheckAndMutate> checkAndMutates);
List<CompletableFuture<CheckAndMutateResult>>
checkAndMutate(List<CheckAndMutate> checkAndMutates);
/**
* A simple version of batch checkAndMutate. It will fail if there are any failures.
*
* @param checkAndMutates The list of rows to apply.
* @return A {@link CompletableFuture} that wrapper the result list.
*/
default CompletableFuture<List<CheckAndMutateResult>> checkAndMutateAll(
List<CheckAndMutate> checkAndMutates) {
default CompletableFuture<List<CheckAndMutateResult>>
checkAndMutateAll(List<CheckAndMutate> checkAndMutates) {
return allOf(checkAndMutate(checkAndMutates));
}
@ -484,7 +479,7 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
*/
default List<CompletableFuture<Boolean>> exists(List<Get> gets) {
return get(toCheckExistenceOnly(gets)).stream()
.<CompletableFuture<Boolean>> map(f -> f.thenApply(r -> r.getExists())).collect(toList());
.<CompletableFuture<Boolean>> map(f -> f.thenApply(r -> r.getExists())).collect(toList());
}
/**
@ -592,7 +587,7 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
* @see ServiceCaller
*/
<S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
ServiceCaller<S, R> callable, byte[] row);
ServiceCaller<S, R> callable, byte[] row);
/**
* The callback when we want to execute a coprocessor call on a range of regions.
@ -731,5 +726,5 @@ public interface AsyncTable<C extends ScanResultConsumerBase> {
* for more details.
*/
<S, R> CoprocessorServiceBuilder<S, R> coprocessorService(Function<RpcChannel, S> stubMaker,
ServiceCaller<S, R> callable, CoprocessorCallback<R> callback);
ServiceCaller<S, R> callable, CoprocessorCallback<R> callback);
}

18
hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncTableRegionLocatorImpl.java
View File

@ -17,6 +17,8 @@
*/
package org.apache.hadoop.hbase.client;
import static org.apache.hadoop.hbase.trace.TraceUtil.tracedFuture;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@ -47,19 +49,21 @@ class AsyncTableRegionLocatorImpl implements AsyncTableRegionLocator {
@Override
public CompletableFuture<HRegionLocation> getRegionLocation(byte[] row, int replicaId,
boolean reload) {
boolean reload) {
return conn.getLocator().getRegionLocation(tableName, row, replicaId, RegionLocateType.CURRENT,
reload, -1L);
}
@Override
public CompletableFuture<List<HRegionLocation>> getAllRegionLocations() {
if (TableName.isMetaTableName(tableName)) {
return conn.registry.getMetaRegionLocations()
.thenApply(locs -> Arrays.asList(locs.getRegionLocations()));
}
return ClientMetaTableAccessor
.getTableHRegionLocations(conn.getTable(TableName.META_TABLE_NAME), tableName);
return tracedFuture(() -> {
if (TableName.isMetaTableName(tableName)) {
return conn.registry.getMetaRegionLocations()
.thenApply(locs -> Arrays.asList(locs.getRegionLocations()));
}
return ClientMetaTableAccessor
.getTableHRegionLocations(conn.getTable(TableName.META_TABLE_NAME), tableName);
}, getClass().getSimpleName() + ".getAllRegionLocations");
}
@Override

55
hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionFactory.java
View File

@ -30,6 +30,7 @@ import org.apache.hadoop.hbase.AuthUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.security.UserProvider;
import org.apache.hadoop.hbase.trace.TraceUtil;
import org.apache.hadoop.hbase.util.FutureUtils;
import org.apache.hadoop.hbase.util.ReflectionUtils;
import org.apache.yetus.audience.InterfaceAudience;
@ -277,31 +278,33 @@ public class ConnectionFactory {
* @return AsyncConnection object wrapped by CompletableFuture
*/
public static CompletableFuture<AsyncConnection> createAsyncConnection(Configuration conf,
final User user) {
CompletableFuture<AsyncConnection> future = new CompletableFuture<>();
ConnectionRegistry registry = ConnectionRegistryFactory.getRegistry(conf);
addListener(registry.getClusterId(), (clusterId, error) -> {
if (error != null) {
registry.close();
future.completeExceptionally(error);
return;
}
if (clusterId == null) {
registry.close();
future.completeExceptionally(new IOException("clusterid came back null"));
return;
}
Class<? extends AsyncConnection> clazz = conf.getClass(HBASE_CLIENT_ASYNC_CONNECTION_IMPL,
AsyncConnectionImpl.class, AsyncConnection.class);
try {
future.complete(
user.runAs((PrivilegedExceptionAction<? extends AsyncConnection>) () -> ReflectionUtils
.newInstance(clazz, conf, registry, clusterId, null, user)));
} catch (Exception e) {
registry.close();
future.completeExceptionally(e);
}
});
return future;
final User user) {
return TraceUtil.tracedFuture(() -> {
CompletableFuture<AsyncConnection> future = new CompletableFuture<>();
ConnectionRegistry registry = ConnectionRegistryFactory.getRegistry(conf);
addListener(registry.getClusterId(), (clusterId, error) -> {
if (error != null) {
registry.close();
future.completeExceptionally(error);
return;
}
if (clusterId == null) {
registry.close();
future.completeExceptionally(new IOException("clusterid came back null"));
return;
}
Class<? extends AsyncConnection> clazz = conf.getClass(HBASE_CLIENT_ASYNC_CONNECTION_IMPL,
AsyncConnectionImpl.class, AsyncConnection.class);
try {
future.complete(
user.runAs((PrivilegedExceptionAction<? extends AsyncConnection>) () -> ReflectionUtils
.newInstance(clazz, conf, registry, clusterId, null, user)));
} catch (Exception e) {
registry.close();
future.completeExceptionally(e);
}
});
return future;
}, ConnectionFactory.class.getSimpleName() + ".createAsyncConnection");
}
}

351
hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java
View File

@ -23,6 +23,8 @@ import static org.apache.hadoop.hbase.client.ConnectionUtils.isEmptyStopRow;
import static org.apache.hadoop.hbase.client.ConnectionUtils.timelineConsistentRead;
import static org.apache.hadoop.hbase.client.ConnectionUtils.validatePut;
import static org.apache.hadoop.hbase.client.ConnectionUtils.validatePutsInRowMutations;
import static org.apache.hadoop.hbase.trace.TraceUtil.tracedFuture;
import static org.apache.hadoop.hbase.trace.TraceUtil.tracedFutures;
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
import java.io.IOException;
@ -132,8 +134,8 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
}
this.maxAttempts = builder.maxAttempts;
this.startLogErrorsCnt = builder.startLogErrorsCnt;
this.defaultScannerCaching = tableName.isSystemTable() ? conn.connConf.getMetaScannerCaching()
: conn.connConf.getScannerCaching();
this.defaultScannerCaching = tableName.isSystemTable() ? conn.connConf.getMetaScannerCaching() :
conn.connConf.getScannerCaching();
this.defaultScannerMaxResultSize = conn.connConf.getScannerMaxResultSize();
}
@ -158,23 +160,23 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
}
private static <REQ, RESP> CompletableFuture<RESP> mutate(HBaseRpcController controller,
HRegionLocation loc, ClientService.Interface stub, REQ req,
Converter<MutateRequest, byte[], REQ> reqConvert,
Converter<RESP, HBaseRpcController, MutateResponse> respConverter) {
HRegionLocation loc, ClientService.Interface stub, REQ req,
Converter<MutateRequest, byte[], REQ> reqConvert,
Converter<RESP, HBaseRpcController, MutateResponse> respConverter) {
return ConnectionUtils.call(controller, loc, stub, req, reqConvert,
(s, c, r, done) -> s.mutate(c, r, done), respConverter);
}
private static <REQ> CompletableFuture<Void> voidMutate(HBaseRpcController controller,
HRegionLocation loc, ClientService.Interface stub, REQ req,
Converter<MutateRequest, byte[], REQ> reqConvert) {
HRegionLocation loc, ClientService.Interface stub, REQ req,
Converter<MutateRequest, byte[], REQ> reqConvert) {
return mutate(controller, loc, stub, req, reqConvert, (c, resp) -> {
return null;
});
}
private static Result toResult(HBaseRpcController controller, MutateResponse resp)
throws IOException {
throws IOException {
if (!resp.hasResult()) {
return null;
}
@ -187,9 +189,9 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
}
private <REQ, RESP> CompletableFuture<RESP> noncedMutate(long nonceGroup, long nonce,
HBaseRpcController controller, HRegionLocation loc, ClientService.Interface stub, REQ req,
NoncedConverter<MutateRequest, byte[], REQ> reqConvert,
Converter<RESP, HBaseRpcController, MutateResponse> respConverter) {
HBaseRpcController controller, HRegionLocation loc, ClientService.Interface stub, REQ req,
NoncedConverter<MutateRequest, byte[], REQ> reqConvert,
Converter<RESP, HBaseRpcController, MutateResponse> respConverter) {
return mutate(controller, loc, stub, req,
(info, src) -> reqConvert.convert(info, src, nonceGroup, nonce), respConverter);
}
@ -202,8 +204,8 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
.maxAttempts(maxAttempts).startLogErrorsCnt(startLogErrorsCnt);
}
private <T, R extends OperationWithAttributes & Row> SingleRequestCallerBuilder<T> newCaller(
R row, long rpcTimeoutNs) {
private <T, R extends OperationWithAttributes & Row> SingleRequestCallerBuilder<T>
newCaller(R row, long rpcTimeoutNs) {
return newCaller(row.getRow(), row.getPriority(), rpcTimeoutNs);
}
@ -218,50 +220,58 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
@Override
public CompletableFuture<Result> get(Get get) {
return timelineConsistentRead(conn.getLocator(), tableName, get, get.getRow(),
RegionLocateType.CURRENT, replicaId -> get(get, replicaId), readRpcTimeoutNs,
conn.connConf.getPrimaryCallTimeoutNs(), retryTimer, conn.getConnectionMetrics());
return tracedFuture(
() -> timelineConsistentRead(conn.getLocator(), tableName, get, get.getRow(),
RegionLocateType.CURRENT, replicaId -> get(get, replicaId), readRpcTimeoutNs,
conn.connConf.getPrimaryCallTimeoutNs(), retryTimer, conn.getConnectionMetrics()),
"AsyncTable.get", tableName);
}
@Override
public CompletableFuture<Void> put(Put put) {
validatePut(put, conn.connConf.getMaxKeyValueSize());
return this.<Void, Put> newCaller(put, writeRpcTimeoutNs)
return tracedFuture(() -> this.<Void, Put> newCaller(put, writeRpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.<Put> voidMutate(controller, loc, stub,
put, RequestConverter::buildMutateRequest))
.call();
.call(), "AsyncTable.put", tableName);
}
@Override
public CompletableFuture<Void> delete(Delete delete) {
return this.<Void, Delete> newCaller(delete, writeRpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.<Delete> voidMutate(controller, loc,
stub, delete, RequestConverter::buildMutateRequest))
.call();
return tracedFuture(
() -> this.<Void, Delete> newCaller(delete, writeRpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.<Delete> voidMutate(controller, loc,
stub, delete, RequestConverter::buildMutateRequest))
.call(),
"AsyncTable.delete", tableName);
}
@Override
public CompletableFuture<Result> append(Append append) {
checkHasFamilies(append);
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return this.<Result, Append> newCaller(append, rpcTimeoutNs)
.action(
(controller, loc, stub) -> this.<Append, Result> noncedMutate(nonceGroup, nonce, controller,
loc, stub, append, RequestConverter::buildMutateRequest, RawAsyncTableImpl::toResult))
.call();
return tracedFuture(() -> {
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return this.<Result, Append> newCaller(append, rpcTimeoutNs)
.action((controller, loc, stub) -> this.<Append, Result> noncedMutate(nonceGroup, nonce,
controller, loc, stub, append, RequestConverter::buildMutateRequest,
RawAsyncTableImpl::toResult))
.call();
}, "AsyncTable.append", tableName);
}
@Override
public CompletableFuture<Result> increment(Increment increment) {
checkHasFamilies(increment);
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return this.<Result, Increment> newCaller(increment, rpcTimeoutNs)
.action((controller, loc, stub) -> this.<Increment, Result> noncedMutate(nonceGroup, nonce,
controller, loc, stub, increment, RequestConverter::buildMutateRequest,
RawAsyncTableImpl::toResult))
.call();
return tracedFuture(() -> {
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return this.<Result, Increment> newCaller(increment, rpcTimeoutNs)
.action((controller, loc, stub) -> this.<Increment, Result> noncedMutate(nonceGroup, nonce,
controller, loc, stub, increment, RequestConverter::buildMutateRequest,
RawAsyncTableImpl::toResult))
.call();
}, "AsyncTable.increment", tableName);
}
private final class CheckAndMutateBuilderImpl implements CheckAndMutateBuilder {
@ -319,39 +329,43 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
public CompletableFuture<Boolean> thenPut(Put put) {
validatePut(put, conn.connConf.getMaxKeyValueSize());
preCheck();
return RawAsyncTableImpl.this.<Boolean> newCaller(row, put.getPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc,
stub, put,
(rn, p) -> RequestConverter.buildMutateRequest(rn, row, family, qualifier, op, value,
null, timeRange, p, HConstants.NO_NONCE, HConstants.NO_NONCE),
(c, r) -> r.getProcessed()))
.call();
return tracedFuture(
() -> RawAsyncTableImpl.this.<Boolean> newCaller(row, put.getPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, put,
(rn, p) -> RequestConverter.buildMutateRequest(rn, row, family, qualifier, op, value,
null, timeRange, p, HConstants.NO_NONCE, HConstants.NO_NONCE),
(c, r) -> r.getProcessed()))
.call(),
"AsyncTable.CheckAndMutateBuilder.thenPut", tableName);
}
@Override
public CompletableFuture<Boolean> thenDelete(Delete delete) {
preCheck();
return RawAsyncTableImpl.this.<Boolean> newCaller(row, delete.getPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller,
loc, stub, delete,
(rn, d) -> RequestConverter.buildMutateRequest(rn, row, family, qualifier, op, value,
null, timeRange, d, HConstants.NO_NONCE, HConstants.NO_NONCE),
(c, r) -> r.getProcessed()))
.call();
return tracedFuture(
() -> RawAsyncTableImpl.this.<Boolean> newCaller(row, delete.getPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, delete,
(rn, d) -> RequestConverter.buildMutateRequest(rn, row, family, qualifier, op, value,
null, timeRange, d, HConstants.NO_NONCE, HConstants.NO_NONCE),
(c, r) -> r.getProcessed()))
.call(),
"AsyncTable.CheckAndMutateBuilder.thenDelete", tableName);
}
@Override
public CompletableFuture<Boolean> thenMutate(RowMutations mutation) {
preCheck();
validatePutsInRowMutations(mutation, conn.connConf.getMaxKeyValueSize());
return RawAsyncTableImpl.this.<Boolean> newCaller(row, mutation.getMaxPriority(),
rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.this.mutateRow(controller,
loc, stub, mutation,
(rn, rm) -> RequestConverter.buildMultiRequest(rn, row, family, qualifier, op, value,
null, timeRange, rm, HConstants.NO_NONCE, HConstants.NO_NONCE),
CheckAndMutateResult::isSuccess))
.call();
return tracedFuture(
() -> RawAsyncTableImpl.this
.<Boolean> newCaller(row, mutation.getMaxPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.this.mutateRow(controller, loc, stub,
mutation,
(rn, rm) -> RequestConverter.buildMultiRequest(rn, row, family, qualifier, op, value,
null, timeRange, rm, HConstants.NO_NONCE, HConstants.NO_NONCE),
CheckAndMutateResult::isSuccess))
.call(),
"AsyncTable.CheckAndMutateBuilder.thenMutate", tableName);
}
}
@ -383,37 +397,42 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
@Override
public CompletableFuture<Boolean> thenPut(Put put) {
validatePut(put, conn.connConf.getMaxKeyValueSize());
return RawAsyncTableImpl.this.<Boolean> newCaller(row, put.getPriority(), rpcTimeoutNs)
return tracedFuture(
() -> RawAsyncTableImpl.this.<Boolean> newCaller(row, put.getPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc,
stub, put,
(rn, p) -> RequestConverter.buildMutateRequest(rn, row, null, null, null, null,
filter, timeRange, p, HConstants.NO_NONCE, HConstants.NO_NONCE),
(c, r) -> r.getProcessed()))
.call();
.call(),
"AsyncTable.CheckAndMutateWithFilterBuilder.thenPut", tableName);
}
@Override
public CompletableFuture<Boolean> thenDelete(Delete delete) {
return RawAsyncTableImpl.this.<Boolean> newCaller(row, delete.getPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller,
loc, stub, delete,
(rn, d) -> RequestConverter.buildMutateRequest(rn, row, null, null, null, null,
filter, timeRange, d, HConstants.NO_NONCE, HConstants.NO_NONCE),
(c, r) -> r.getProcessed()))
.call();
return tracedFuture(
() -> RawAsyncTableImpl.this.<Boolean> newCaller(row, delete.getPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, delete,
(rn, d) -> RequestConverter.buildMutateRequest(rn, row, null, null, null, null, filter,
timeRange, d, HConstants.NO_NONCE, HConstants.NO_NONCE),
(c, r) -> r.getProcessed()))
.call(),
"AsyncTable.CheckAndMutateWithFilterBuilder.thenDelete", tableName);
}
@Override
public CompletableFuture<Boolean> thenMutate(RowMutations mutation) {
validatePutsInRowMutations(mutation, conn.connConf.getMaxKeyValueSize());
return RawAsyncTableImpl.this.<Boolean> newCaller(row, mutation.getMaxPriority(),
rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.this.mutateRow(controller,
loc, stub, mutation,
(rn, rm) -> RequestConverter.buildMultiRequest(rn, row, null, null, null, null,
filter, timeRange, rm, HConstants.NO_NONCE, HConstants.NO_NONCE),
CheckAndMutateResult::isSuccess))
.call();
return tracedFuture(
() -> RawAsyncTableImpl.this
.<Boolean> newCaller(row, mutation.getMaxPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.this.mutateRow(controller, loc, stub,
mutation,
(rn, rm) -> RequestConverter.buildMultiRequest(rn, row, null, null, null, null, filter,
timeRange, rm, HConstants.NO_NONCE, HConstants.NO_NONCE),
CheckAndMutateResult::isSuccess))
.call(),
"AsyncTable.CheckAndMutateWithFilterBuilder.thenMutate", tableName);
}
}
@ -424,63 +443,69 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
@Override
public CompletableFuture<CheckAndMutateResult> checkAndMutate(CheckAndMutate checkAndMutate) {
if (checkAndMutate.getAction() instanceof Put || checkAndMutate.getAction() instanceof Delete
|| checkAndMutate.getAction() instanceof Increment
|| checkAndMutate.getAction() instanceof Append) {
Mutation mutation = (Mutation) checkAndMutate.getAction();
if (mutation instanceof Put) {
validatePut((Put) mutation, conn.connConf.getMaxKeyValueSize());
return tracedFuture(() -> {
if (checkAndMutate.getAction() instanceof Put ||
checkAndMutate.getAction() instanceof Delete ||
checkAndMutate.getAction() instanceof Increment ||
checkAndMutate.getAction() instanceof Append) {
Mutation mutation = (Mutation) checkAndMutate.getAction();
if (mutation instanceof Put) {
validatePut((Put) mutation, conn.connConf.getMaxKeyValueSize());
}
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return RawAsyncTableImpl.this
.<CheckAndMutateResult> newCaller(checkAndMutate.getRow(), mutation.getPriority(),
rpcTimeoutNs)
.action(
(controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, mutation,
(rn, m) -> RequestConverter.buildMutateRequest(rn, checkAndMutate.getRow(),
checkAndMutate.getFamily(), checkAndMutate.getQualifier(),
checkAndMutate.getCompareOp(), checkAndMutate.getValue(),
checkAndMutate.getFilter(), checkAndMutate.getTimeRange(), m, nonceGroup, nonce),
(c, r) -> ResponseConverter.getCheckAndMutateResult(r, c.cellScanner())))
.call();
} else if (checkAndMutate.getAction() instanceof RowMutations) {
RowMutations rowMutations = (RowMutations) checkAndMutate.getAction();
validatePutsInRowMutations(rowMutations, conn.connConf.getMaxKeyValueSize());
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return RawAsyncTableImpl.this
.<CheckAndMutateResult> newCaller(checkAndMutate.getRow(), rowMutations.getMaxPriority(),
rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.this
.<CheckAndMutateResult, CheckAndMutateResult> mutateRow(controller, loc, stub,
rowMutations,
(rn, rm) -> RequestConverter.buildMultiRequest(rn, checkAndMutate.getRow(),
checkAndMutate.getFamily(), checkAndMutate.getQualifier(),
checkAndMutate.getCompareOp(), checkAndMutate.getValue(),
checkAndMutate.getFilter(), checkAndMutate.getTimeRange(), rm, nonceGroup, nonce),
resp -> resp))
.call();
} else {
CompletableFuture<CheckAndMutateResult> future = new CompletableFuture<>();
future.completeExceptionally(new DoNotRetryIOException(
"CheckAndMutate doesn't support " + checkAndMutate.getAction().getClass().getName()));
return future;
}
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return RawAsyncTableImpl.this.<CheckAndMutateResult> newCaller(checkAndMutate.getRow(),
mutation.getPriority(), rpcTimeoutNs)
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller,
loc, stub, mutation,
(rn, m) -> RequestConverter.buildMutateRequest(rn, checkAndMutate.getRow(),
checkAndMutate.getFamily(), checkAndMutate.getQualifier(),
checkAndMutate.getCompareOp(), checkAndMutate.getValue(), checkAndMutate.getFilter(),
checkAndMutate.getTimeRange(), m, nonceGroup, nonce),
(c, r) -> ResponseConverter.getCheckAndMutateResult(r, c.cellScanner())))
.call();
} else if (checkAndMutate.getAction() instanceof RowMutations) {
RowMutations rowMutations = (RowMutations) checkAndMutate.getAction();
validatePutsInRowMutations(rowMutations, conn.connConf.getMaxKeyValueSize());
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return RawAsyncTableImpl.this.<CheckAndMutateResult> newCaller(checkAndMutate.getRow(),
rowMutations.getMaxPriority(), rpcTimeoutNs)
.action((controller, loc, stub) ->
RawAsyncTableImpl.this.<CheckAndMutateResult, CheckAndMutateResult> mutateRow(
controller, loc, stub, rowMutations,
(rn, rm) -> RequestConverter.buildMultiRequest(rn, checkAndMutate.getRow(),
checkAndMutate.getFamily(), checkAndMutate.getQualifier(),
checkAndMutate.getCompareOp(), checkAndMutate.getValue(), checkAndMutate.getFilter(),
checkAndMutate.getTimeRange(), rm, nonceGroup, nonce),
resp -> resp))
.call();
} else {
CompletableFuture<CheckAndMutateResult> future = new CompletableFuture<>();
future.completeExceptionally(new DoNotRetryIOException(
"CheckAndMutate doesn't support " + checkAndMutate.getAction().getClass().getName()));
return future;
}
}, "AsyncTable.checkAndMutate", tableName);
}
@Override
public List<CompletableFuture<CheckAndMutateResult>> checkAndMutate(
List<CheckAndMutate> checkAndMutates) {
return batch(checkAndMutates, rpcTimeoutNs).stream()
.map(f -> f.thenApply(r -> (CheckAndMutateResult) r)).collect(toList());
public List<CompletableFuture<CheckAndMutateResult>>
checkAndMutate(List<CheckAndMutate> checkAndMutates) {
return tracedFutures(
() -> batch(checkAndMutates, rpcTimeoutNs).stream()
.map(f -> f.thenApply(r -> (CheckAndMutateResult) r)).collect(toList()),
"AsyncTable.checkAndMutateList", tableName);
}
// We need the MultiRequest when constructing the org.apache.hadoop.hbase.client.MultiResponse,
// so here I write a new method as I do not want to change the abstraction of call method.
@SuppressWarnings("unchecked")
private <RES, RESP> CompletableFuture<RESP> mutateRow(HBaseRpcController controller,
HRegionLocation loc, ClientService.Interface stub, RowMutations mutation,
Converter<MultiRequest, byte[], RowMutations> reqConvert,
Function<RES, RESP> respConverter) {
HRegionLocation loc, ClientService.Interface stub, RowMutations mutation,
Converter<MultiRequest, byte[], RowMutations> reqConvert, Function<RES, RESP> respConverter) {
CompletableFuture<RESP> future = new CompletableFuture<>();
try {
byte[] regionName = loc.getRegion().getRegionName();
@ -499,12 +524,12 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
loc.getServerName(), multiResp);
Throwable ex = multiResp.getException(regionName);
if (ex != null) {
future.completeExceptionally(ex instanceof IOException ? ex
: new IOException(
future.completeExceptionally(ex instanceof IOException ? ex :
new IOException(
"Failed to mutate row: " + Bytes.toStringBinary(mutation.getRow()), ex));
} else {
future.complete(respConverter
.apply((RES) multiResp.getResults().get(regionName).result.get(0)));
future.complete(
respConverter.apply((RES) multiResp.getResults().get(regionName).result.get(0)));
}
} catch (IOException e) {
future.completeExceptionally(e);
@ -523,12 +548,14 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
validatePutsInRowMutations(mutations, conn.connConf.getMaxKeyValueSize());
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
long nonce = conn.getNonceGenerator().newNonce();
return this.<Result> newCaller(mutations.getRow(), mutations.getMaxPriority(),
writeRpcTimeoutNs).action((controller, loc, stub) ->
this.<Result, Result> mutateRow(controller, loc, stub, mutations,
(rn, rm) -> RequestConverter.buildMultiRequest(rn, rm, nonceGroup, nonce),
return tracedFuture(
() -> this
.<Result> newCaller(mutations.getRow(), mutations.getMaxPriority(), writeRpcTimeoutNs)
.action((controller, loc, stub) -> this.<Result, Result> mutateRow(controller, loc, stub,
mutations, (rn, rm) -> RequestConverter.buildMultiRequest(rn, rm, nonceGroup, nonce),
resp -> resp))
.call();
.call(),
"AsyncTable.mutateRow", tableName);
}
private Scan setDefaultScanConfig(Scan scan) {
@ -564,46 +591,48 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
@Override
public CompletableFuture<List<Result>> scanAll(Scan scan) {
CompletableFuture<List<Result>> future = new CompletableFuture<>();
List<Result> scanResults = new ArrayList<>();
scan(scan, new AdvancedScanResultConsumer() {
return tracedFuture(() -> {
CompletableFuture<List<Result>> future = new CompletableFuture<>();
List<Result> scanResults = new ArrayList<>();
scan(scan, new AdvancedScanResultConsumer() {
@Override
public void onNext(Result[] results, ScanController controller) {
scanResults.addAll(Arrays.asList(results));
}
@Override
public void onNext(Result[] results, ScanController controller) {
scanResults.addAll(Arrays.asList(results));
}
@Override
public void onError(Throwable error) {
future.completeExceptionally(error);
}
@Override
public void onError(Throwable error) {
future.completeExceptionally(error);
}
@Override
public void onComplete() {
future.complete(scanResults);
}
});
return future;
@Override
public void onComplete() {
future.complete(scanResults);
}
});
return future;
}, "AsyncTable.scanAll", tableName);
}
@Override
public List<CompletableFuture<Result>> get(List<Get> gets) {
return batch(gets, readRpcTimeoutNs);
return tracedFutures(() -> batch(gets, readRpcTimeoutNs), "AsyncTable.getList", tableName);
}
@Override
public List<CompletableFuture<Void>> put(List<Put> puts) {
return voidMutate(puts);
return tracedFutures(() -> voidMutate(puts), "AsyncTable.putList", tableName);
}
@Override
public List<CompletableFuture<Void>> delete(List<Delete> deletes) {
return voidMutate(deletes);
return tracedFutures(() -> voidMutate(deletes), "AsyncTable.deleteList", tableName);
}
@Override
public <T> List<CompletableFuture<T>> batch(List<? extends Row> actions) {
return batch(actions, rpcTimeoutNs);
return tracedFutures(() -> batch(actions, rpcTimeoutNs), "AsyncTable.batch", tableName);
}
private List<CompletableFuture<Void>> voidMutate(List<? extends Row> actions) {
@ -660,7 +689,7 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
}
private <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
ServiceCaller<S, R> callable, RegionInfo region, byte[] row) {
ServiceCaller<S, R> callable, RegionInfo region, byte[] row) {
RegionCoprocessorRpcChannelImpl channel = new RegionCoprocessorRpcChannelImpl(conn, tableName,
region, row, rpcTimeoutNs, operationTimeoutNs);
S stub = stubMaker.apply(channel);
@ -678,7 +707,7 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
@Override
public <S, R> CompletableFuture<R> coprocessorService(Function<RpcChannel, S> stubMaker,
ServiceCaller<S, R> callable, byte[] row) {
ServiceCaller<S, R> callable, byte[] row) {
return coprocessorService(stubMaker, callable, null, row);
}
@ -700,9 +729,9 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
}
private <S, R> void onLocateComplete(Function<RpcChannel, S> stubMaker,
ServiceCaller<S, R> callable, CoprocessorCallback<R> callback, List<HRegionLocation> locs,
byte[] endKey, boolean endKeyInclusive, AtomicBoolean locateFinished,
AtomicInteger unfinishedRequest, HRegionLocation loc, Throwable error) {
ServiceCaller<S, R> callable, CoprocessorCallback<R> callback, List<HRegionLocation> locs,
byte[] endKey, boolean endKeyInclusive, AtomicBoolean locateFinished,
AtomicInteger unfinishedRequest, HRegionLocation loc, Throwable error) {
if (error != null) {
callback.onError(error);
return;
@ -731,7 +760,7 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
}
private final class CoprocessorServiceBuilderImpl<S, R>
implements CoprocessorServiceBuilder<S, R> {
implements CoprocessorServiceBuilder<S, R> {
private final Function<RpcChannel, S> stubMaker;
@ -748,7 +777,7 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
private boolean endKeyInclusive;
public CoprocessorServiceBuilderImpl(Function<RpcChannel, S> stubMaker,
ServiceCaller<S, R> callable, CoprocessorCallback<R> callback) {
ServiceCaller<S, R> callable, CoprocessorCallback<R> callback) {
this.stubMaker = Preconditions.checkNotNull(stubMaker, "stubMaker is null");
this.callable = Preconditions.checkNotNull(callable, "callable is null");
this.callback = Preconditions.checkNotNull(callback, "callback is null");
@ -785,8 +814,8 @@ class RawAsyncTableImpl implements AsyncTable<AdvancedScanResultConsumer> {
@Override
public <S, R> CoprocessorServiceBuilder<S, R> coprocessorService(
Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
CoprocessorCallback<R> callback) {
Function<RpcChannel, S> stubMaker, ServiceCaller<S, R> callable,
CoprocessorCallback<R> callback) {
return new CoprocessorServiceBuilderImpl<>(stubMaker, callable, callback);
}
}

9
hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java
View File

@ -393,10 +393,11 @@ public abstract class AbstractRpcClient<T extends RpcConnection> implements RpcC
}
private Call callMethod(final Descriptors.MethodDescriptor md, final HBaseRpcController hrc,
final Message param, Message returnType, final User ticket,
final Address addr, final RpcCallback<Message> callback) {
Span span = TraceUtil.getGlobalTracer().spanBuilder("RpcClient.callMethod." + md.getFullName())
.startSpan();
final Message param, Message returnType, final User ticket, final Address addr,
final RpcCallback<Message> callback) {
Span span = TraceUtil.createSpan("RpcClient.callMethod")
.setAttribute(TraceUtil.RPC_SERVICE_KEY, md.getService().getName())
.setAttribute(TraceUtil.RPC_METHOD_KEY, md.getName());
try (Scope scope = span.makeCurrent()) {
final MetricsConnection.CallStats cs = MetricsConnection.newCallStats();
cs.setStartTime(EnvironmentEdgeManager.currentTime());

157
hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncRegionLocatorTracing.java Normal file
View File

@ -0,0 +1,157 @@
/**
* 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
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.hbase.client;
import static org.junit.Assert.assertEquals;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.sdk.testing.junit4.OpenTelemetryRule;
import io.opentelemetry.sdk.trace.data.SpanData;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.RegionLocations;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.Waiter;
import org.apache.hadoop.hbase.security.UserProvider;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.trace.TraceUtil;
import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
@Category({ ClientTests.class, MediumTests.class })
public class TestAsyncRegionLocatorTracing {
@ClassRule
public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestAsyncRegionLocatorTracing.class);
private static Configuration CONF = HBaseConfiguration.create();
private AsyncConnectionImpl conn;
private RegionLocations locs;
@Rule
public OpenTelemetryRule traceRule = OpenTelemetryRule.create();
@Before
public void setUp() throws IOException {
RegionInfo metaRegionInfo = RegionInfoBuilder.newBuilder(TableName.META_TABLE_NAME).build();
locs = new RegionLocations(
new HRegionLocation(metaRegionInfo,
ServerName.valueOf("127.0.0.1", 12345, System.currentTimeMillis())),
new HRegionLocation(RegionReplicaUtil.getRegionInfoForReplica(metaRegionInfo, 1),
ServerName.valueOf("127.0.0.2", 12345, System.currentTimeMillis())),
new HRegionLocation(RegionReplicaUtil.getRegionInfoForReplica(metaRegionInfo, 2),
ServerName.valueOf("127.0.0.3", 12345, System.currentTimeMillis())));
conn = new AsyncConnectionImpl(CONF, new DoNothingConnectionRegistry(CONF) {
@Override
public CompletableFuture<RegionLocations> getMetaRegionLocations() {
return CompletableFuture.completedFuture(locs);
}
}, "test", null, UserProvider.instantiate(CONF).getCurrent());
}
@After
public void tearDown() throws IOException {
Closeables.close(conn, true);
}
private SpanData waitSpan(String name) {
Waiter.waitFor(CONF, 1000,
() -> traceRule.getSpans().stream().map(SpanData::getName).anyMatch(s -> s.equals(name)));
return traceRule.getSpans().stream().filter(s -> s.getName().equals(name)).findFirst().get();
}
@Test
public void testClearCache() {
conn.getLocator().clearCache();
SpanData span = waitSpan("AsyncRegionLocator.clearCache");
assertEquals(StatusCode.OK, span.getStatus().getStatusCode());
}
@Test
public void testClearCacheServerName() {
ServerName sn = ServerName.valueOf("127.0.0.1", 12345, System.currentTimeMillis());
conn.getLocator().clearCache(sn);
SpanData span = waitSpan("AsyncRegionLocator.clearCache");
assertEquals(StatusCode.OK, span.getStatus().getStatusCode());
assertEquals(sn.toString(), span.getAttributes().get(TraceUtil.SERVER_NAME_KEY));
}
@Test
public void testClearCacheTableName() {
conn.getLocator().clearCache(TableName.META_TABLE_NAME);
SpanData span = waitSpan("AsyncRegionLocator.clearCache");
assertEquals(StatusCode.OK, span.getStatus().getStatusCode());
assertEquals(TableName.META_TABLE_NAME.getNamespaceAsString(),
span.getAttributes().get(TraceUtil.NAMESPACE_KEY));
assertEquals(TableName.META_TABLE_NAME.getNameAsString(),
span.getAttributes().get(TraceUtil.TABLE_KEY));
}
@Test
public void testGetRegionLocation() {
conn.getLocator().getRegionLocation(TableName.META_TABLE_NAME, HConstants.EMPTY_START_ROW,
RegionLocateType.CURRENT, TimeUnit.SECONDS.toNanos(1)).join();
SpanData span = waitSpan("AsyncRegionLocator.getRegionLocation");
assertEquals(StatusCode.OK, span.getStatus().getStatusCode());
assertEquals(TableName.META_TABLE_NAME.getNamespaceAsString(),
span.getAttributes().get(TraceUtil.NAMESPACE_KEY));
assertEquals(TableName.META_TABLE_NAME.getNameAsString(),
span.getAttributes().get(TraceUtil.TABLE_KEY));
List<String> regionNames = span.getAttributes().get(TraceUtil.REGION_NAMES_KEY);
assertEquals(1, regionNames.size());
assertEquals(locs.getDefaultRegionLocation().getRegion().getRegionNameAsString(),
regionNames.get(0));
}
@Test
public void testGetRegionLocations() {
conn.getLocator().getRegionLocations(TableName.META_TABLE_NAME, HConstants.EMPTY_START_ROW,
RegionLocateType.CURRENT, false, TimeUnit.SECONDS.toNanos(1)).join();
SpanData span = waitSpan("AsyncRegionLocator.getRegionLocations");
assertEquals(StatusCode.OK, span.getStatus().getStatusCode());
assertEquals(TableName.META_TABLE_NAME.getNamespaceAsString(),
span.getAttributes().get(TraceUtil.NAMESPACE_KEY));
assertEquals(TableName.META_TABLE_NAME.getNameAsString(),
span.getAttributes().get(TraceUtil.TABLE_KEY));
List<String> regionNames = span.getAttributes().get(TraceUtil.REGION_NAMES_KEY);
assertEquals(3, regionNames.size());
for (int i = 0; i < 3; i++) {
assertEquals(locs.getRegionLocation(i).getRegion().getRegionNameAsString(),
regionNames.get(i));
}
}
}

414
hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableTracing.java Normal file
View File

@ -0,0 +1,414 @@
/**
* 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
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.hbase.client;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import io.opentelemetry.api.trace.Span.Kind;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.sdk.testing.junit4.OpenTelemetryRule;
import io.opentelemetry.sdk.trace.data.SpanData;
import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.Cell.Type;
import org.apache.hadoop.hbase.CellBuilderFactory;
import org.apache.hadoop.hbase.CellBuilderType;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.Waiter;
import org.apache.hadoop.hbase.ipc.HBaseRpcController;
import org.apache.hadoop.hbase.security.UserProvider;
import org.apache.hadoop.hbase.testclassification.ClientTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.trace.TraceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ClientService;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.GetRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.GetResponse;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutateRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutateResponse;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutationProto;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutationProto.ColumnValue;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutationProto.ColumnValue.QualifierValue;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.RegionActionResult;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ResultOrException;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanResponse;
@Category({ ClientTests.class, MediumTests.class })
public class TestAsyncTableTracing {
@ClassRule
public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestAsyncTableTracing.class);
private static Configuration CONF = HBaseConfiguration.create();
private ClientService.Interface stub;
private AsyncConnection conn;
private AsyncTable<?> table;
@Rule
public OpenTelemetryRule traceRule = OpenTelemetryRule.create();
@Before
public void setUp() throws IOException {
stub = mock(ClientService.Interface.class);
AtomicInteger scanNextCalled = new AtomicInteger(0);
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
ScanRequest req = invocation.getArgument(1);
RpcCallback<ScanResponse> done = invocation.getArgument(2);
if (!req.hasScannerId()) {
done.run(ScanResponse.newBuilder().setScannerId(1).setTtl(800)
.setMoreResultsInRegion(true).setMoreResults(true).build());
} else {
if (req.hasCloseScanner() && req.getCloseScanner()) {
done.run(ScanResponse.getDefaultInstance());
} else {
Cell cell = CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setType(Type.Put)
.setRow(Bytes.toBytes(scanNextCalled.incrementAndGet()))
.setFamily(Bytes.toBytes("cf")).setQualifier(Bytes.toBytes("cq"))
.setValue(Bytes.toBytes("v")).build();
Result result = Result.create(Arrays.asList(cell));
ScanResponse.Builder builder = ScanResponse.newBuilder().setScannerId(1).setTtl(800)
.addResults(ProtobufUtil.toResult(result));
if (req.getLimitOfRows() == 1) {
builder.setMoreResultsInRegion(false).setMoreResults(false);
} else {
builder.setMoreResultsInRegion(true).setMoreResults(true);
}
ForkJoinPool.commonPool().execute(() -> done.run(builder.build()));
}
}
return null;
}
}).when(stub).scan(any(HBaseRpcController.class), any(ScanRequest.class), any());
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
ClientProtos.MultiResponse resp =
ClientProtos.MultiResponse.newBuilder()
.addRegionActionResult(RegionActionResult.newBuilder().addResultOrException(
ResultOrException.newBuilder().setResult(ProtobufUtil.toResult(new Result()))))
.build();
RpcCallback<ClientProtos.MultiResponse> done = invocation.getArgument(2);
ForkJoinPool.commonPool().execute(() -> done.run(resp));
return null;
}
}).when(stub).multi(any(HBaseRpcController.class), any(ClientProtos.MultiRequest.class), any());
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
MutationProto req = ((MutateRequest) invocation.getArgument(1)).getMutation();
MutateResponse resp;
switch (req.getMutateType()) {
case INCREMENT:
ColumnValue value = req.getColumnValue(0);
QualifierValue qvalue = value.getQualifierValue(0);
Cell cell = CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setType(Type.Put)
.setRow(req.getRow().toByteArray()).setFamily(value.getFamily().toByteArray())
.setQualifier(qvalue.getQualifier().toByteArray())
.setValue(qvalue.getValue().toByteArray()).build();
resp = MutateResponse.newBuilder()
.setResult(ProtobufUtil.toResult(Result.create(Arrays.asList(cell)))).build();
break;
default:
resp = MutateResponse.getDefaultInstance();
break;
}
RpcCallback<MutateResponse> done = invocation.getArgument(2);
ForkJoinPool.commonPool().execute(() -> done.run(resp));
return null;
}
}).when(stub).mutate(any(HBaseRpcController.class), any(MutateRequest.class), any());
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
RpcCallback<GetResponse> done = invocation.getArgument(2);
ForkJoinPool.commonPool().execute(() -> done.run(GetResponse.getDefaultInstance()));
return null;
}
}).when(stub).get(any(HBaseRpcController.class), any(GetRequest.class), any());
conn = new AsyncConnectionImpl(CONF, new DoNothingConnectionRegistry(CONF), "test", null,
UserProvider.instantiate(CONF).getCurrent()) {
@Override
AsyncRegionLocator getLocator() {
AsyncRegionLocator locator = mock(AsyncRegionLocator.class);
Answer<CompletableFuture<HRegionLocation>> answer =
new Answer<CompletableFuture<HRegionLocation>>() {
@Override
public CompletableFuture<HRegionLocation> answer(InvocationOnMock invocation)
throws Throwable {
TableName tableName = invocation.getArgument(0);
RegionInfo info = RegionInfoBuilder.newBuilder(tableName).build();
ServerName serverName = ServerName.valueOf("rs", 16010, 12345);
HRegionLocation loc = new HRegionLocation(info, serverName);
return CompletableFuture.completedFuture(loc);
}
};
doAnswer(answer).when(locator).getRegionLocation(any(TableName.class), any(byte[].class),
any(RegionLocateType.class), anyLong());
doAnswer(answer).when(locator).getRegionLocation(any(TableName.class), any(byte[].class),
anyInt(), any(RegionLocateType.class), anyLong());
return locator;
}
@Override
ClientService.Interface getRegionServerStub(ServerName serverName) throws IOException {
return stub;
}
};
table = conn.getTable(TableName.valueOf("table"), ForkJoinPool.commonPool());
}
@After
public void tearDown() throws IOException {
Closeables.close(conn, true);
}
private void assertTrace(String methodName) {
Waiter.waitFor(CONF, 1000,
() -> traceRule.getSpans().stream()
.anyMatch(span -> span.getName().equals("AsyncTable." + methodName) &&
span.getKind() == Kind.INTERNAL && span.hasEnded()));
SpanData data = traceRule.getSpans().stream()
.filter(s -> s.getName().equals("AsyncTable." + methodName)).findFirst().get();
assertEquals(StatusCode.OK, data.getStatus().getStatusCode());
TableName tableName = table.getName();
assertEquals(tableName.getNamespaceAsString(),
data.getAttributes().get(TraceUtil.NAMESPACE_KEY));
assertEquals(tableName.getNameAsString(), data.getAttributes().get(TraceUtil.TABLE_KEY));
}
@Test
public void testExists() {
table.exists(new Get(Bytes.toBytes(0))).join();
assertTrace("get");
}
@Test
public void testGet() {
table.get(new Get(Bytes.toBytes(0))).join();
assertTrace("get");
}
@Test
public void testPut() {
table.put(new Put(Bytes.toBytes(0)).addColumn(Bytes.toBytes("cf"), Bytes.toBytes("cq"),
Bytes.toBytes("v"))).join();
assertTrace("put");
}
@Test
public void testDelete() {
table.delete(new Delete(Bytes.toBytes(0))).join();
assertTrace("delete");
}
@Test
public void testAppend() {
table.append(new Append(Bytes.toBytes(0)).addColumn(Bytes.toBytes("cf"), Bytes.toBytes("cq"),
Bytes.toBytes("v"))).join();
assertTrace("append");
}
@Test
public void testIncrement() {
table
.increment(
new Increment(Bytes.toBytes(0)).addColumn(Bytes.toBytes("cf"), Bytes.toBytes("cq"), 1))
.join();
assertTrace("increment");
}
@Test
public void testIncrementColumnValue1() {
table.incrementColumnValue(Bytes.toBytes(0), Bytes.toBytes("cf"), Bytes.toBytes("cq"), 1)
.join();
assertTrace("increment");
}
@Test
public void testIncrementColumnValue2() {
table.incrementColumnValue(Bytes.toBytes(0), Bytes.toBytes("cf"), Bytes.toBytes("cq"), 1,
Durability.ASYNC_WAL).join();
assertTrace("increment");
}
@Test
public void testCheckAndMutate() {
table.checkAndMutate(CheckAndMutate.newBuilder(Bytes.toBytes(0))
.ifEquals(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v"))
.build(new Delete(Bytes.toBytes(0)))).join();
assertTrace("checkAndMutate");
}
@Test
public void testCheckAndMutateList() {
CompletableFuture
.allOf(table.checkAndMutate(Arrays.asList(CheckAndMutate.newBuilder(Bytes.toBytes(0))
.ifEquals(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v"))
.build(new Delete(Bytes.toBytes(0))))).toArray(new CompletableFuture[0]))
.join();
assertTrace("checkAndMutateList");
}
@Test
public void testCheckAndMutateAll() {
table.checkAndMutateAll(Arrays.asList(CheckAndMutate.newBuilder(Bytes.toBytes(0))
.ifEquals(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v"))
.build(new Delete(Bytes.toBytes(0))))).join();
assertTrace("checkAndMutateList");
}
@Test
public void testMutateRow() throws IOException {
table.mutateRow(new RowMutations(Bytes.toBytes(0)).add(new Delete(Bytes.toBytes(0))));
assertTrace("mutateRow");
}
@Test
public void testScanAll() throws IOException {
table.scanAll(new Scan().setCaching(1).setMaxResultSize(1).setLimit(1)).join();
assertTrace("scanAll");
}
@Test
public void testExistsList() {
CompletableFuture
.allOf(
table.exists(Arrays.asList(new Get(Bytes.toBytes(0)))).toArray(new CompletableFuture[0]))
.join();
assertTrace("getList");
}
@Test
public void testExistsAll() {
table.existsAll(Arrays.asList(new Get(Bytes.toBytes(0)))).join();
assertTrace("getList");
}
@Test
public void testGetList() {
CompletableFuture
.allOf(table.get(Arrays.asList(new Get(Bytes.toBytes(0)))).toArray(new CompletableFuture[0]))
.join();
assertTrace("getList");
}
@Test
public void testGetAll() {
table.getAll(Arrays.asList(new Get(Bytes.toBytes(0)))).join();
assertTrace("getList");
}
@Test
public void testPutList() {
CompletableFuture
.allOf(table.put(Arrays.asList(new Put(Bytes.toBytes(0)).addColumn(Bytes.toBytes("cf"),
Bytes.toBytes("cq"), Bytes.toBytes("v")))).toArray(new CompletableFuture[0]))
.join();
assertTrace("putList");
}
@Test
public void testPutAll() {
table.putAll(Arrays.asList(new Put(Bytes.toBytes(0)).addColumn(Bytes.toBytes("cf"),
Bytes.toBytes("cq"), Bytes.toBytes("v")))).join();
assertTrace("putList");
}
@Test
public void testDeleteList() {
CompletableFuture
.allOf(
table.delete(Arrays.asList(new Delete(Bytes.toBytes(0)))).toArray(new CompletableFuture[0]))
.join();
assertTrace("deleteList");
}
@Test
public void testDeleteAll() {
table.deleteAll(Arrays.asList(new Delete(Bytes.toBytes(0)))).join();
assertTrace("deleteList");
}
@Test
public void testBatch() {
CompletableFuture
.allOf(
table.batch(Arrays.asList(new Delete(Bytes.toBytes(0)))).toArray(new CompletableFuture[0]))
.join();
assertTrace("batch");
}
@Test
public void testBatchAll() {
table.batchAll(Arrays.asList(new Delete(Bytes.toBytes(0)))).join();
assertTrace("batch");
}
@Test
public void testConnClose() throws IOException {
conn.close();
Waiter.waitFor(CONF, 1000,
() -> traceRule.getSpans().stream()
.anyMatch(span -> span.getName().equals("AsyncConnection.close") &&
span.getKind() == Kind.INTERNAL && span.hasEnded()));
SpanData data = traceRule.getSpans().stream()
.filter(s -> s.getName().equals("AsyncConnection.close")).findFirst().get();
assertEquals(StatusCode.OK, data.getStatus().getStatusCode());
}
}

134
hbase-common/src/main/java/org/apache/hadoop/hbase/trace/TraceUtil.java
View File

@ -18,7 +18,19 @@
package org.apache.hadoop.hbase.trace;
import io.opentelemetry.api.GlobalOpenTelemetry;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.Span.Kind;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.api.trace.attributes.SemanticAttributes;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.util.FutureUtils;
import org.apache.yetus.audience.InterfaceAudience;
@InterfaceAudience.Private
@ -26,10 +38,132 @@ public final class TraceUtil {
private static final String INSTRUMENTATION_NAME = "io.opentelemetry.contrib.hbase";
public static final AttributeKey<String> NAMESPACE_KEY = SemanticAttributes.DB_HBASE_NAMESPACE;
public static final AttributeKey<String> TABLE_KEY = AttributeKey.stringKey("db.hbase.table");
public static final AttributeKey<List<String>> REGION_NAMES_KEY =
AttributeKey.stringArrayKey("db.hbase.regions");
public static final AttributeKey<String> RPC_SERVICE_KEY =
AttributeKey.stringKey("db.hbase.rpc.service");
public static final AttributeKey<String> RPC_METHOD_KEY =
AttributeKey.stringKey("db.hbase.rpc.method");
public static final AttributeKey<String> SERVER_NAME_KEY =
AttributeKey.stringKey("db.hbase.server.name");
private TraceUtil() {
}
public static Tracer getGlobalTracer() {
return GlobalOpenTelemetry.getTracer(INSTRUMENTATION_NAME);
}
/**
* Create a {@link Kind#INTERNAL} span.
*/
public static Span createSpan(String name) {
return createSpan(name, Kind.INTERNAL);
}
/**
* Create a {@link Kind#INTERNAL} span and set table related attributes.
*/
public static Span createTableSpan(String spanName, TableName tableName) {
return createSpan(spanName).setAttribute(NAMESPACE_KEY, tableName.getNamespaceAsString())
.setAttribute(TABLE_KEY, tableName.getNameAsString());
}
/**
* Create a span with the given {@code kind}. Notice that, OpenTelemetry only expects one
* {@link Kind#CLIENT} span and one {@link Kind#SERVER} span for a traced request, so use this
* with caution when you want to create spans with kind other than {@link Kind#INTERNAL}.
*/
private static Span createSpan(String name, Kind kind) {
return getGlobalTracer().spanBuilder(name).setSpanKind(kind).startSpan();
}
/**
* Create a span which parent is from remote, i.e, passed through rpc.
* </p>
* We will set the kind of the returned span to {@link Kind#SERVER}, as this should be the top
* most span at server side.
*/
public static Span createRemoteSpan(String name, Context ctx) {
return getGlobalTracer().spanBuilder(name).setParent(ctx).setSpanKind(Kind.SERVER).startSpan();
}
/**
* Trace an asynchronous operation for a table.
*/
public static <T> CompletableFuture<T> tracedFuture(Supplier<CompletableFuture<T>> action,
String spanName, TableName tableName) {
Span span = createTableSpan(spanName, tableName);
try (Scope scope = span.makeCurrent()) {
CompletableFuture<T> future = action.get();
endSpan(future, span);
return future;
}
}
/**
* Trace an asynchronous operation.
*/
public static <T> CompletableFuture<T> tracedFuture(Supplier<CompletableFuture<T>> action,
String spanName) {
Span span = createSpan(spanName);
try (Scope scope = span.makeCurrent()) {
CompletableFuture<T> future = action.get();
endSpan(future, span);
return future;
}
}
/**
* Trace an asynchronous operation, and finish the create {@link Span} when all the given
* {@code futures} are completed.
*/
public static <T> List<CompletableFuture<T>> tracedFutures(
Supplier<List<CompletableFuture<T>>> action, String spanName, TableName tableName) {
Span span = createTableSpan(spanName, tableName);
try (Scope scope = span.makeCurrent()) {
List<CompletableFuture<T>> futures = action.get();
endSpan(CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])), span);
return futures;
}
}
/**
* Finish the {@code span} when the given {@code future} is completed.
*/
private static void endSpan(CompletableFuture<?> future, Span span) {
FutureUtils.addListener(future, (resp, error) -> {
if (error != null) {
span.recordException(error);
span.setStatus(StatusCode.ERROR);
} else {
span.setStatus(StatusCode.OK);
}
span.end();
});
}
public static void trace(Runnable action, String spanName) {
trace(action, () -> createSpan(spanName));
}
public static void trace(Runnable action, Supplier<Span> creator) {
Span span = creator.get();
try (Scope scope = span.makeCurrent()) {
action.run();
span.setStatus(StatusCode.OK);
} catch (Throwable e) {
span.recordException(e);
span.setStatus(StatusCode.ERROR);
} finally {
span.end();
}
}
}

7
hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/CallRunner.java
View File

@ -121,9 +121,10 @@ public class CallRunner {
RpcServer.CurCall.set(call);
String serviceName = getServiceName();
String methodName = getMethodName();
String traceString = serviceName + "." + methodName;
Span span = TraceUtil.getGlobalTracer().spanBuilder(traceString)
.setParent(Context.current().with(((ServerCall<?>) call).getSpan())).startSpan();
Span span = TraceUtil.getGlobalTracer().spanBuilder("RpcServer.callMethod")
.setParent(Context.current().with(((ServerCall<?>) call).getSpan())).startSpan()
.setAttribute(TraceUtil.RPC_SERVICE_KEY, serviceName)
.setAttribute(TraceUtil.RPC_METHOD_KEY, methodName);
try (Scope traceScope = span.makeCurrent()) {
if (!this.rpcServer.isStarted()) {
InetSocketAddress address = rpcServer.getListenerAddress();

3
hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/ServerRpcConnection.java
View File

@ -629,8 +629,7 @@ abstract class ServerRpcConnection implements Closeable {
};
Context traceCtx = GlobalOpenTelemetry.getPropagators().getTextMapPropagator()
.extract(Context.current(), header.getTraceInfo(), getter);
Span span =
TraceUtil.getGlobalTracer().spanBuilder("RpcServer.process").setParent(traceCtx).startSpan();
Span span = TraceUtil.createRemoteSpan("RpcServer.process", traceCtx);
try (Scope scope = span.makeCurrent()) {
int id = header.getCallId();
if (RpcServer.LOG.isTraceEnabled()) {

24
hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java
View File

@ -52,6 +52,7 @@ import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.Server;
import org.apache.hadoop.hbase.Waiter;
import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface;
import org.apache.hadoop.hbase.trace.TraceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.util.StringUtils;
@ -448,6 +449,19 @@ public abstract class AbstractTestIPC {
}
}
private SpanData waitSpan(String name) {
Waiter.waitFor(CONF, 1000,
() -> traceRule.getSpans().stream().map(SpanData::getName).anyMatch(s -> s.equals(name)));
return traceRule.getSpans().stream().filter(s -> s.getName().equals(name)).findFirst().get();
}
private void assertRpcAttribute(SpanData data, String methodName) {
assertEquals(SERVICE.getDescriptorForType().getName(),
data.getAttributes().get(TraceUtil.RPC_SERVICE_KEY));
assertEquals(methodName,
data.getAttributes().get(TraceUtil.RPC_METHOD_KEY));
}
@Test
public void testTracing() throws IOException, ServiceException {
RpcServer rpcServer = createRpcServer(null, "testRpcServer",
@ -457,9 +471,8 @@ public abstract class AbstractTestIPC {
rpcServer.start();
BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
stub.pause(null, PauseRequestProto.newBuilder().setMs(100).build());
Waiter.waitFor(CONF, 1000, () -> traceRule.getSpans().stream().map(SpanData::getName)
.anyMatch(s -> s.equals("RpcClient.callMethod.TestProtobufRpcProto.pause")));
assertRpcAttribute(waitSpan("RpcClient.callMethod"), "pause");
assertRpcAttribute(waitSpan("RpcServer.callMethod"), "pause");
assertSameTraceId();
for (SpanData data : traceRule.getSpans()) {
assertThat(
@ -471,9 +484,8 @@ public abstract class AbstractTestIPC {
traceRule.clearSpans();
assertThrows(ServiceException.class,
() -> stub.error(null, EmptyRequestProto.getDefaultInstance()));
Waiter.waitFor(CONF, 1000, () -> traceRule.getSpans().stream().map(SpanData::getName)
.anyMatch(s -> s.equals("RpcClient.callMethod.TestProtobufRpcProto.error")));
assertRpcAttribute(waitSpan("RpcClient.callMethod"), "error");
assertRpcAttribute(waitSpan("RpcServer.callMethod"), "error");
assertSameTraceId();
for (SpanData data : traceRule.getSpans()) {
assertEquals(StatusCode.ERROR, data.getStatus().getStatusCode());

8
pom.xml
View File

@ -1718,12 +1718,8 @@
<junit.version>4.13</junit.version>
<hamcrest.version>1.3</hamcrest.version>
<opentelemetry.version>0.13.1</opentelemetry.version>
<<<<<<< HEAD
<log4j2.version>2.14.1</log4j2.version>
=======
<opentelemetry-instrumentation.version>0.13.0</opentelemetry-instrumentation.version>
<log4j.version>1.2.17</log4j.version>
>>>>>>> HBASE-25424 Find a way to config OpenTelemetry tracing without direct… (#2808)
<log4j2.version>2.14.1</log4j2.version>
<mockito-core.version>2.28.2</mockito-core.version>
<protobuf.plugin.version>0.6.1</protobuf.plugin.version>
<thrift.path>thrift</thrift.path>
@ -2415,7 +2411,7 @@
<dependency>
<groupId>io.opentelemetry.javaagent</groupId>
<artifactId>opentelemetry-javaagent</artifactId>
<version>${opentelemetry-instrumentation.version}</version>
<version>${opentelemetry.version}</version>
<classifier>all</classifier>
</dependency>
<dependency>