You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2016/12/01 13:18:39 UTC
hbase git commit: HBASE-16648 [JDK8] Use computeIfAbsent instead of
get and putIfAbsent
Repository: hbase
Updated Branches:
refs/heads/master fb789b340 -> 540ede376
HBASE-16648 [JDK8] Use computeIfAbsent instead of get and putIfAbsent
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/540ede37
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/540ede37
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/540ede37
Branch: refs/heads/master
Commit: 540ede376ba25db13cdb11ba86830fe0e6cec118
Parents: fb789b3
Author: zhangduo <zh...@apache.org>
Authored: Thu Dec 1 12:12:17 2016 +0800
Committer: zhangduo <zh...@apache.org>
Committed: Thu Dec 1 21:17:55 2016 +0800
----------------------------------------------------------------------
.../hadoop/hbase/client/AsyncProcess.java | 19 +--
.../hbase/client/ConnectionImplementation.java | 156 ++++++-------------
.../apache/hadoop/hbase/client/MetaCache.java | 23 +--
.../hadoop/hbase/client/MetricsConnection.java | 63 +++-----
.../client/PreemptiveFastFailInterceptor.java | 50 +++---
.../hbase/client/ServerStatisticTracker.java | 26 +---
.../hadoop/hbase/util/CollectionUtils.java | 26 +++-
.../ZKSplitLogManagerCoordination.java | 15 +-
.../hadoop/hbase/master/ServerManager.java | 32 ++--
.../hadoop/hbase/master/SplitLogManager.java | 15 +-
.../apache/hadoop/hbase/quotas/QuotaCache.java | 36 ++---
.../hadoop/hbase/regionserver/HRegion.java | 16 +-
.../regionserver/wal/SequenceIdAccounting.java | 14 +-
.../hbase/security/access/TableAuthManager.java | 28 ++--
.../hbase/wal/BoundedGroupingStrategy.java | 13 +-
.../hadoop/hbase/master/TestMasterShutdown.java | 2 +-
.../hbase/master/TestSplitLogManager.java | 20 ++-
17 files changed, 189 insertions(+), 365 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
index debb602..50a2a11 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncProcess.java
@@ -19,6 +19,8 @@
package org.apache.hadoop.hbase.client;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
@@ -645,23 +647,10 @@ class AsyncProcess {
protected void incTaskCounters(Collection<byte[]> regions, ServerName sn) {
tasksInProgress.incrementAndGet();
- AtomicInteger serverCnt = taskCounterPerServer.get(sn);
- if (serverCnt == null) {
- taskCounterPerServer.putIfAbsent(sn, new AtomicInteger());
- serverCnt = taskCounterPerServer.get(sn);
- }
- serverCnt.incrementAndGet();
+ computeIfAbsent(taskCounterPerServer, sn, AtomicInteger::new).incrementAndGet();
for (byte[] regBytes : regions) {
- AtomicInteger regionCnt = taskCounterPerRegion.get(regBytes);
- if (regionCnt == null) {
- regionCnt = new AtomicInteger();
- AtomicInteger oldCnt = taskCounterPerRegion.putIfAbsent(regBytes, regionCnt);
- if (oldCnt != null) {
- regionCnt = oldCnt;
- }
- }
- regionCnt.incrementAndGet();
+ computeIfAbsent(taskCounterPerRegion, regBytes, AtomicInteger::new).incrementAndGet();
}
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionImplementation.java
----------------------------------------------------------------------
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 96452f9..aa984b1 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
@@ -22,6 +22,8 @@ import static org.apache.hadoop.hbase.client.ConnectionUtils.NO_NONCE_GENERATOR;
import static org.apache.hadoop.hbase.client.ConnectionUtils.getStubKey;
import static org.apache.hadoop.hbase.client.ConnectionUtils.retries2Attempts;
import static org.apache.hadoop.hbase.client.MetricsConnection.CLIENT_SIDE_METRICS_ENABLED_KEY;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsentEx;
import com.google.common.annotations.VisibleForTesting;
@@ -921,11 +923,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
}
// Map keyed by service name + regionserver to service stub implementation
- private final ConcurrentHashMap<String, Object> stubs =
- new ConcurrentHashMap<String, Object>();
- // Map of locks used creating service stubs per regionserver.
- private final ConcurrentHashMap<String, String> connectionLock =
- new ConcurrentHashMap<String, String>();
+ private final ConcurrentMap<String, Object> stubs = new ConcurrentHashMap<String, Object>();
/**
* State of the MasterService connection/setup.
@@ -1008,7 +1006,7 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
long result;
ServerErrors errorStats = errorsByServer.get(server);
if (errorStats != null) {
- result = ConnectionUtils.getPauseTime(basePause, errorStats.getCount());
+ result = ConnectionUtils.getPauseTime(basePause, Math.max(0, errorStats.getCount() - 1));
} else {
result = 0; // yes, if the server is not in our list we don't wait before retrying.
}
@@ -1017,19 +1015,10 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
/**
* Reports that there was an error on the server to do whatever bean-counting necessary.
- *
* @param server The server in question.
*/
void reportServerError(ServerName server) {
- ServerErrors errors = errorsByServer.get(server);
- if (errors != null) {
- errors.addError();
- } else {
- errors = errorsByServer.putIfAbsent(server, new ServerErrors());
- if (errors != null){
- errors.addError();
- }
- }
+ computeIfAbsent(errorsByServer, server, ServerErrors::new).addError();
}
long getStartTrackingTime() {
@@ -1053,32 +1042,26 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
}
/**
- * Makes a client-side stub for master services. Sub-class to specialize.
- * Depends on hosting class so not static. Exists so we avoid duplicating a bunch of code
- * when setting up the MasterMonitorService and MasterAdminService.
+ * Class to make a MasterServiceStubMaker stub.
*/
- abstract class StubMaker {
- /**
- * Returns the name of the service stub being created.
- */
- protected abstract String getServiceName();
-
- /**
- * Make stub and cache it internal so can be used later doing the isMasterRunning call.
- */
- protected abstract Object makeStub(final BlockingRpcChannel channel);
+ private final class MasterServiceStubMaker {
- /**
- * Once setup, check it works by doing isMasterRunning check.
- */
- protected abstract void isMasterRunning() throws IOException;
+ private void isMasterRunning(MasterProtos.MasterService.BlockingInterface stub)
+ throws IOException {
+ try {
+ stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest());
+ } catch (ServiceException e) {
+ throw ProtobufUtil.handleRemoteException(e);
+ }
+ }
/**
- * 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.
+ * 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 MasterProtos.MasterService.BlockingInterface makeStubNoRetries()
+ throws IOException, KeeperException {
ZooKeeperKeepAliveConnection zkw;
try {
zkw = getKeepAliveZooKeeperWatcher();
@@ -1098,18 +1081,14 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
throw new MasterNotRunningException(sn + " is dead.");
}
// Use the security info interface name as our stub key
- String key = getStubKey(getServiceName(), sn, hostnamesCanChange);
- connectionLock.putIfAbsent(key, key);
- Object stub = null;
- synchronized (connectionLock.get(key)) {
- stub = stubs.get(key);
- if (stub == null) {
- BlockingRpcChannel channel = rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
- stub = makeStub(channel);
- isMasterRunning();
- stubs.put(key, stub);
- }
- }
+ String key = getStubKey(MasterProtos.MasterService.getDescriptor().getName(), sn,
+ hostnamesCanChange);
+ MasterProtos.MasterService.BlockingInterface stub =
+ (MasterProtos.MasterService.BlockingInterface) computeIfAbsentEx(stubs, key, () -> {
+ BlockingRpcChannel channel = rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
+ return MasterProtos.MasterService.newBlockingStub(channel);
+ });
+ isMasterRunning(stub);
return stub;
} finally {
zkw.close();
@@ -1121,9 +1100,9 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
* @return A stub to do <code>intf</code> against the master
* @throws org.apache.hadoop.hbase.MasterNotRunningException if master is not running
*/
- Object makeStub() throws IOException {
+ MasterProtos.MasterService.BlockingInterface makeStub() throws IOException {
// The lock must be at the beginning to prevent multiple master creations
- // (and leaks) in a multithread context
+ // (and leaks) in a multithread context
synchronized (masterAndZKLock) {
Exception exceptionCaught = null;
if (!closed) {
@@ -1142,80 +1121,33 @@ class ConnectionImplementation implements ClusterConnection, Closeable {
}
}
- /**
- * Class to make a MasterServiceStubMaker stub.
- */
- class MasterServiceStubMaker extends StubMaker {
- private MasterProtos.MasterService.BlockingInterface stub;
- @Override
- protected String getServiceName() {
- return MasterProtos.MasterService.getDescriptor().getName();
- }
-
- @Override
- MasterProtos.MasterService.BlockingInterface makeStub() throws IOException {
- return (MasterProtos.MasterService.BlockingInterface)super.makeStub();
- }
-
- @Override
- protected Object makeStub(BlockingRpcChannel channel) {
- this.stub = MasterProtos.MasterService.newBlockingStub(channel);
- return this.stub;
- }
-
- @Override
- protected void isMasterRunning() throws IOException {
- try {
- this.stub.isMasterRunning(null, RequestConverter.buildIsMasterRunningRequest());
- } catch (Exception e) {
- throw ProtobufUtil.handleRemoteException(e);
- }
- }
- }
-
@Override
- public AdminProtos.AdminService.BlockingInterface getAdmin(final ServerName serverName)
+ public AdminProtos.AdminService.BlockingInterface getAdmin(ServerName serverName)
throws IOException {
if (isDeadServer(serverName)) {
throw new RegionServerStoppedException(serverName + " is dead.");
}
String key = getStubKey(AdminProtos.AdminService.BlockingInterface.class.getName(), serverName,
this.hostnamesCanChange);
- this.connectionLock.putIfAbsent(key, key);
- AdminProtos.AdminService.BlockingInterface stub;
- synchronized (this.connectionLock.get(key)) {
- stub = (AdminProtos.AdminService.BlockingInterface)this.stubs.get(key);
- if (stub == null) {
- BlockingRpcChannel channel =
+ return (AdminProtos.AdminService.BlockingInterface) computeIfAbsentEx(stubs, key, () -> {
+ BlockingRpcChannel channel =
this.rpcClient.createBlockingRpcChannel(serverName, user, rpcTimeout);
- stub = AdminProtos.AdminService.newBlockingStub(channel);
- this.stubs.put(key, stub);
- }
- }
- return stub;
+ return AdminProtos.AdminService.newBlockingStub(channel);
+ });
}
@Override
- public BlockingInterface getClient(final ServerName sn)
- throws IOException {
- if (isDeadServer(sn)) {
- throw new RegionServerStoppedException(sn + " is dead.");
+ public BlockingInterface getClient(ServerName serverName) throws IOException {
+ if (isDeadServer(serverName)) {
+ throw new RegionServerStoppedException(serverName + " is dead.");
}
- String key = getStubKey(ClientProtos.ClientService.BlockingInterface.class.getName(), sn,
- this.hostnamesCanChange);
- this.connectionLock.putIfAbsent(key, key);
- ClientProtos.ClientService.BlockingInterface stub = null;
- synchronized (this.connectionLock.get(key)) {
- stub = (ClientProtos.ClientService.BlockingInterface)this.stubs.get(key);
- if (stub == null) {
- BlockingRpcChannel channel = this.rpcClient.createBlockingRpcChannel(sn, user, rpcTimeout);
- stub = ClientProtos.ClientService.newBlockingStub(channel);
- // In old days, after getting stub/proxy, we'd make a call. We are not doing that here.
- // Just fail on first actual call rather than in here on setup.
- this.stubs.put(key, stub);
- }
- }
- return stub;
+ String key = getStubKey(ClientProtos.ClientService.BlockingInterface.class.getName(),
+ serverName, this.hostnamesCanChange);
+ return (ClientProtos.ClientService.BlockingInterface) computeIfAbsentEx(stubs, key, () -> {
+ BlockingRpcChannel channel =
+ this.rpcClient.createBlockingRpcChannel(serverName, user, rpcTimeout);
+ return ClientProtos.ClientService.newBlockingStub(channel);
+ });
}
private ZooKeeperKeepAliveConnection keepAliveZookeeper;
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java
index 3914df5..14e0afd 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetaCache.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hbase.client;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
@@ -49,8 +51,7 @@ public class MetaCache {
* Map of table to table {@link HRegionLocation}s.
*/
private final ConcurrentMap<TableName, ConcurrentNavigableMap<byte[], RegionLocations>>
- cachedRegionLocations =
- new CopyOnWriteArrayMap<>();
+ cachedRegionLocations = new CopyOnWriteArrayMap<>();
// The presence of a server in the map implies it's likely that there is an
// entry in cachedRegionLocations that map to this server; but the absence
@@ -191,21 +192,11 @@ public class MetaCache {
* @param tableName
* @return Map of cached locations for passed <code>tableName</code>
*/
- private ConcurrentNavigableMap<byte[], RegionLocations>
- getTableLocations(final TableName tableName) {
+ private ConcurrentNavigableMap<byte[], RegionLocations> getTableLocations(
+ final TableName tableName) {
// find the map of cached locations for this table
- ConcurrentNavigableMap<byte[], RegionLocations> result;
- result = this.cachedRegionLocations.get(tableName);
- // if tableLocations for this table isn't built yet, make one
- if (result == null) {
- result = new CopyOnWriteArrayMap<>(Bytes.BYTES_COMPARATOR);
- ConcurrentNavigableMap<byte[], RegionLocations> old =
- this.cachedRegionLocations.putIfAbsent(tableName, result);
- if (old != null) {
- return old;
- }
- }
- return result;
+ return computeIfAbsent(cachedRegionLocations, tableName,
+ () -> new CopyOnWriteArrayMap<>(Bytes.BYTES_COMPARATOR));
}
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java
index 36627bd..64b1661 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java
@@ -17,29 +17,31 @@
*/
package org.apache.hadoop.hbase.client;
-import com.google.common.annotations.VisibleForTesting;
-import org.apache.hadoop.hbase.shaded.com.google.protobuf.Descriptors.MethodDescriptor;
-import org.apache.hadoop.hbase.shaded.com.google.protobuf.Message;
+import static com.codahale.metrics.MetricRegistry.name;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
import com.codahale.metrics.Counter;
import com.codahale.metrics.Histogram;
-import com.codahale.metrics.MetricRegistry;
-import com.codahale.metrics.Timer;
import com.codahale.metrics.JmxReporter;
+import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.RatioGauge;
-import org.apache.hadoop.hbase.ServerName;
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
-import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ClientService;
-import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutateRequest;
-import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutationProto.MutationType;
-import org.apache.hadoop.hbase.util.Bytes;
+import com.codahale.metrics.Timer;
+import com.google.common.annotations.VisibleForTesting;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import static com.codahale.metrics.MetricRegistry.name;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.shaded.com.google.protobuf.Descriptors.MethodDescriptor;
+import org.apache.hadoop.hbase.shaded.com.google.protobuf.Message;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ClientService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutateRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutationProto.MutationType;
+import org.apache.hadoop.hbase.util.Bytes;
/**
* This class is for maintaining the various connection statistics and publishing them through
@@ -207,32 +209,15 @@ public class MetricsConnection implements StatisticTrackable {
}
@Override
- public void updateRegionStats(ServerName serverName, byte[] regionName,
- RegionLoadStats stats) {
+ public void updateRegionStats(ServerName serverName, byte[] regionName, RegionLoadStats stats) {
String name = serverName.getServerName() + "," + Bytes.toStringBinary(regionName);
- ConcurrentMap<byte[], RegionStats> rsStats = null;
- if (serverStats.containsKey(serverName)) {
- rsStats = serverStats.get(serverName);
- } else {
- rsStats = serverStats.putIfAbsent(serverName,
- new ConcurrentSkipListMap<byte[], RegionStats>(Bytes.BYTES_COMPARATOR));
- if (rsStats == null) {
- rsStats = serverStats.get(serverName);
- }
- }
- RegionStats regionStats = null;
- if (rsStats.containsKey(regionName)) {
- regionStats = rsStats.get(regionName);
- } else {
- regionStats = rsStats.putIfAbsent(regionName, new RegionStats(this.registry, name));
- if (regionStats == null) {
- regionStats = rsStats.get(regionName);
- }
- }
+ ConcurrentMap<byte[], RegionStats> rsStats = computeIfAbsent(serverStats, serverName,
+ () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));
+ RegionStats regionStats =
+ computeIfAbsent(rsStats, regionName, () -> new RegionStats(this.registry, name));
regionStats.update(stats);
}
-
/** A lambda for dispatching to the appropriate metric factory method */
private static interface NewMetric<T> {
T newMetric(Class<?> clazz, String name, String scope);
@@ -407,13 +392,7 @@ public class MetricsConnection implements StatisticTrackable {
* Get a metric for {@code key} from {@code map}, or create it with {@code factory}.
*/
private <T> T getMetric(String key, ConcurrentMap<String, T> map, NewMetric<T> factory) {
- T t = map.get(key);
- if (t == null) {
- t = factory.newMetric(this.getClass(), key, scope);
- T tmp = map.putIfAbsent(key, t);
- t = (tmp == null) ? t : tmp;
- }
- return t;
+ return computeIfAbsent(map, key, () -> factory.newMetric(getClass(), key, scope));
}
/** Update call stats for non-critical-path methods */
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-client/src/main/java/org/apache/hadoop/hbase/client/PreemptiveFastFailInterceptor.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/PreemptiveFastFailInterceptor.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/PreemptiveFastFailInterceptor.java
index 7ac5c45..448e5b1 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/PreemptiveFastFailInterceptor.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/PreemptiveFastFailInterceptor.java
@@ -17,6 +17,10 @@
*/
package org.apache.hadoop.hbase.client;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
+import com.google.common.annotations.VisibleForTesting;
+
import java.io.IOException;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
@@ -35,34 +39,29 @@ import org.apache.hadoop.hbase.ipc.CallTimeoutException;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.ipc.RemoteException;
-import com.google.common.annotations.VisibleForTesting;
-
/**
- *
* The concrete {@link RetryingCallerInterceptor} class that implements the preemptive fast fail
* feature.
- *
- * The motivation is as follows :
- * In case where a large number of clients try and talk to a particular region server in hbase, if
- * the region server goes down due to network problems, we might end up in a scenario where
- * the clients would go into a state where they all start to retry.
+ * <p>
+ * The motivation is as follows : In case where a large number of clients try and talk to a
+ * particular region server in hbase, if the region server goes down due to network problems, we
+ * might end up in a scenario where the clients would go into a state where they all start to retry.
* This behavior will set off many of the threads in pretty much the same path and they all would be
* sleeping giving rise to a state where the client either needs to create more threads to send new
* requests to other hbase machines or block because the client cannot create anymore threads.
- *
+ * <p>
* In most cases the clients might prefer to have a bound on the number of threads that are created
* in order to send requests to hbase. This would mostly result in the client thread starvation.
- *
- * To circumvent this problem, the approach that is being taken here under is to let 1 of the many
- * threads who are trying to contact the regionserver with connection problems and let the other
- * threads get a {@link PreemptiveFastFailException} so that they can move on and take other
- * requests.
- *
- * This would give the client more flexibility on the kind of action he would want to take in cases
- * where the regionserver is down. He can either discard the requests and send a nack upstream
- * faster or have an application level retry or buffer the requests up so as to send them down to
- * hbase later.
- *
+ * <p>
+ * To circumvent this problem, the approach that is being taken here under is to let 1 of the many
+ * threads who are trying to contact the regionserver with connection problems and let the other
+ * threads get a {@link PreemptiveFastFailException} so that they can move on and take other
+ * requests.
+ * <p>
+ * This would give the client more flexibility on the kind of action he would want to take in cases
+ * where the regionserver is down. He can either discard the requests and send a nack upstream
+ * faster or have an application level retry or buffer the requests up so as to send them down to
+ * hbase later.
*/
@InterfaceAudience.Private
class PreemptiveFastFailInterceptor extends RetryingCallerInterceptor {
@@ -155,15 +154,8 @@ class PreemptiveFastFailInterceptor extends RetryingCallerInterceptor {
return;
}
long currentTime = EnvironmentEdgeManager.currentTime();
- FailureInfo fInfo = repeatedFailuresMap.get(serverName);
- if (fInfo == null) {
- fInfo = new FailureInfo(currentTime);
- FailureInfo oldfInfo = repeatedFailuresMap.putIfAbsent(serverName, fInfo);
-
- if (oldfInfo != null) {
- fInfo = oldfInfo;
- }
- }
+ FailureInfo fInfo =
+ computeIfAbsent(repeatedFailuresMap, serverName, () -> new FailureInfo(currentTime));
fInfo.timeOfLatestAttemptMilliSec = currentTime;
fInfo.numConsecutiveFailures.incrementAndGet();
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ServerStatisticTracker.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ServerStatisticTracker.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ServerStatisticTracker.java
index cb21e8b..f66e7fc 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ServerStatisticTracker.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ServerStatisticTracker.java
@@ -17,15 +17,18 @@
*/
package org.apache.hadoop.hbase.client;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
import com.google.common.annotations.VisibleForTesting;
+
+import java.util.concurrent.ConcurrentHashMap;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.client.backoff.ServerStatistics;
-import java.util.concurrent.ConcurrentHashMap;
-
/**
* Tracks the statistics for multiple regions
*/
@@ -36,23 +39,8 @@ public class ServerStatisticTracker implements StatisticTrackable {
new ConcurrentHashMap<ServerName, ServerStatistics>();
@Override
- public void updateRegionStats(ServerName server, byte[] region, RegionLoadStats
- currentStats) {
- ServerStatistics stat = stats.get(server);
-
- if (stat == null) {
- stat = stats.get(server);
- // We don't have stats for that server yet, so we need to make an entry.
- // If we race with another thread it's a harmless unnecessary allocation.
- if (stat == null) {
- stat = new ServerStatistics();
- ServerStatistics old = stats.putIfAbsent(server, stat);
- if (old != null) {
- stat = old;
- }
- }
- }
- stat.update(region, currentStats);
+ public void updateRegionStats(ServerName server, byte[] region, RegionLoadStats currentStats) {
+ computeIfAbsent(stats, server, ServerStatistics::new).update(region, currentStats);
}
public ServerStatistics getStats(ServerName server) {
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-common/src/main/java/org/apache/hadoop/hbase/util/CollectionUtils.java
----------------------------------------------------------------------
diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/CollectionUtils.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/CollectionUtils.java
index 4e19b77..8cc71a3 100644
--- a/hbase-common/src/main/java/org/apache/hadoop/hbase/util/CollectionUtils.java
+++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/util/CollectionUtils.java
@@ -110,15 +110,12 @@ public class CollectionUtils {
/**
* In HBASE-16648 we found that ConcurrentHashMap.get is much faster than computeIfAbsent if the
- * value already exists. So here we copy the implementation of
- * {@link ConcurrentMap#computeIfAbsent(Object, java.util.function.Function)}. It uses get and
- * putIfAbsent to implement computeIfAbsent. And notice that the implementation does not guarantee
- * that the supplier will only be executed once.
+ * value already exists. Notice that the implementation does not guarantee that the supplier will
+ * only be executed once.
*/
public static <K, V> V computeIfAbsent(ConcurrentMap<K, V> map, K key, Supplier<V> supplier) {
- V v, newValue;
- return ((v = map.get(key)) == null && (newValue = supplier.get()) != null
- && (v = map.putIfAbsent(key, newValue)) == null) ? newValue : v;
+ return computeIfAbsent(map, key, supplier, () -> {
+ });
}
/**
@@ -142,4 +139,19 @@ public class CollectionUtils {
return ((v = map.get(key)) == null && (newValue = supplier.get()) != null
&& (v = map.putIfAbsent(key, newValue)) == null) ? newValue : v;
}
+
+ public static <K, V> V computeIfAbsent(ConcurrentMap<K, V> map, K key, Supplier<V> supplier,
+ Runnable actionIfAbsent) {
+ V v = map.get(key);
+ if (v != null) {
+ return v;
+ }
+ V newValue = supplier.get();
+ v = map.putIfAbsent(key, newValue);
+ if (v != null) {
+ return v;
+ }
+ actionIfAbsent.run();
+ return newValue;
+ }
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java
index cb5df75..a226eb6 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/coordination/ZKSplitLogManagerCoordination.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.coordination;
+import static org.apache.hadoop.hbase.util.CollectionUtils.*;
import static org.apache.hadoop.hbase.master.SplitLogManager.ResubmitDirective.CHECK;
import static org.apache.hadoop.hbase.master.SplitLogManager.ResubmitDirective.FORCE;
import static org.apache.hadoop.hbase.master.SplitLogManager.TerminationStatus.DELETED;
@@ -52,6 +53,7 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.ZooKeeperProtos.SplitLo
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
import org.apache.hadoop.hbase.wal.WALSplitter;
+import org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper;
import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperListener;
@@ -449,7 +451,7 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements
setDone(path, FAILURE);
return;
}
- data = this.watcher.getRecoverableZooKeeper().removeMetaData(data);
+ data = RecoverableZooKeeper.removeMetaData(data);
SplitLogTask slt = SplitLogTask.parseFrom(data);
if (slt.isUnassigned()) {
LOG.debug("task not yet acquired " + path + " ver = " + version);
@@ -531,16 +533,11 @@ public class ZKSplitLogManagerCoordination extends ZooKeeperListener implements
return;
}
- Task findOrCreateOrphanTask(String path) {
- Task orphanTask = new Task();
- Task task;
- task = details.getTasks().putIfAbsent(path, orphanTask);
- if (task == null) {
+ private Task findOrCreateOrphanTask(String path) {
+ return computeIfAbsent(details.getTasks(), path, Task::new, () -> {
LOG.info("creating orphan task " + path);
SplitLogCounters.tot_mgr_orphan_task_acquired.incrementAndGet();
- task = orphanTask;
- }
- return task;
+ });
}
private void heartbeat(String path, int new_version, ServerName workerName) {
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/main/java/org/apache/hadoop/hbase/master/ServerManager.java
----------------------------------------------------------------------
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 278030f..a567e1d 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
@@ -18,6 +18,10 @@
*/
package org.apache.hadoop.hbase.master;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
+import com.google.common.annotations.VisibleForTesting;
+
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
@@ -56,6 +60,11 @@ import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.regionserver.RegionOpeningState;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.shaded.com.google.protobuf.ServiceException;
+import org.apache.hadoop.hbase.shaded.com.google.protobuf.UnsafeByteOperations;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
import org.apache.hadoop.hbase.shaded.protobuf.ResponseConverter;
@@ -67,9 +76,6 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.ClusterStatusProtos.Reg
import org.apache.hadoop.hbase.shaded.protobuf.generated.ClusterStatusProtos.StoreSequenceId;
import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.RegionServerStartupRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
-import org.apache.hadoop.hbase.regionserver.HRegionServer;
-import org.apache.hadoop.hbase.regionserver.RegionOpeningState;
-import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.RetryCounter;
@@ -78,11 +84,6 @@ import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
import org.apache.zookeeper.KeeperException;
-import com.google.common.annotations.VisibleForTesting;
-
-import org.apache.hadoop.hbase.shaded.com.google.protobuf.ServiceException;
-import org.apache.hadoop.hbase.shaded.com.google.protobuf.UnsafeByteOperations;
-
/**
* The ServerManager class manages info about region servers.
* <p>
@@ -273,18 +274,6 @@ public class ServerManager {
return sn;
}
- private ConcurrentNavigableMap<byte[], Long> getOrCreateStoreFlushedSequenceId(
- byte[] regionName) {
- ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
- storeFlushedSequenceIdsByRegion.get(regionName);
- if (storeFlushedSequenceId != null) {
- return storeFlushedSequenceId;
- }
- storeFlushedSequenceId = new ConcurrentSkipListMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
- ConcurrentNavigableMap<byte[], Long> alreadyPut =
- storeFlushedSequenceIdsByRegion.putIfAbsent(regionName, storeFlushedSequenceId);
- return alreadyPut == null ? storeFlushedSequenceId : alreadyPut;
- }
/**
* Updates last flushed sequence Ids for the regions on server sn
* @param sn
@@ -309,7 +298,8 @@ public class ServerManager {
+ existingValue + ") for region " + Bytes.toString(entry.getKey()) + " Ignoring.");
}
ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
- getOrCreateStoreFlushedSequenceId(encodedRegionName);
+ computeIfAbsent(storeFlushedSequenceIdsByRegion, encodedRegionName,
+ () -> new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR));
for (StoreSequenceId storeSeqId : entry.getValue().getStoreCompleteSequenceId()) {
byte[] family = storeSeqId.getFamilyName().toByteArray();
existingValue = storeFlushedSequenceId.get(family);
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/main/java/org/apache/hadoop/hbase/master/SplitLogManager.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/SplitLogManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/SplitLogManager.java
index 9328687..589da14 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/SplitLogManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/SplitLogManager.java
@@ -117,7 +117,8 @@ public class SplitLogManager {
*/
protected final ReentrantLock recoveringRegionLock = new ReentrantLock();
- private final ConcurrentMap<String, Task> tasks = new ConcurrentHashMap<String, Task>();
+ @VisibleForTesting
+ final ConcurrentMap<String, Task> tasks = new ConcurrentHashMap<String, Task>();
private TimeoutMonitor timeoutMonitor;
private volatile Set<ServerName> deadWorkers = null;
@@ -504,18 +505,6 @@ public class SplitLogManager {
}
}
- Task findOrCreateOrphanTask(String path) {
- Task orphanTask = new Task();
- Task task;
- task = tasks.putIfAbsent(path, orphanTask);
- if (task == null) {
- LOG.info("creating orphan task " + path);
- SplitLogCounters.tot_mgr_orphan_task_acquired.incrementAndGet();
- task = orphanTask;
- }
- return task;
- }
-
public void stop() {
if (choreService != null) {
choreService.shutdown();
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaCache.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaCache.java
index 15962d2..1451052 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaCache.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/QuotaCache.java
@@ -18,6 +18,10 @@
package org.apache.hadoop.hbase.quotas;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
+import com.google.common.annotations.VisibleForTesting;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -38,8 +42,6 @@ import org.apache.hadoop.hbase.regionserver.RegionServerServices;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.security.UserGroupInformation;
-import com.google.common.annotations.VisibleForTesting;
-
/**
* Cache that keeps track of the quota settings for the users and tables that
* are interacting with it.
@@ -114,20 +116,12 @@ public class QuotaCache implements Stoppable {
/**
* Returns the QuotaState associated to the specified user.
- *
* @param ugi the user
* @return the quota info associated to specified user
*/
public UserQuotaState getUserQuotaState(final UserGroupInformation ugi) {
- String key = ugi.getShortUserName();
- UserQuotaState quotaInfo = userQuotaCache.get(key);
- if (quotaInfo == null) {
- quotaInfo = new UserQuotaState();
- if (userQuotaCache.putIfAbsent(key, quotaInfo) == null) {
- triggerCacheRefresh();
- }
- }
- return quotaInfo;
+ return computeIfAbsent(userQuotaCache, ugi.getShortUserName(), UserQuotaState::new,
+ this::triggerCacheRefresh);
}
/**
@@ -151,24 +145,12 @@ public class QuotaCache implements Stoppable {
}
/**
- * Returns the QuotaState requested.
- * If the quota info is not in cache an empty one will be returned
- * and the quota request will be enqueued for the next cache refresh.
+ * Returns the QuotaState requested. If the quota info is not in cache an empty one will be
+ * returned and the quota request will be enqueued for the next cache refresh.
*/
private <K> QuotaState getQuotaState(final ConcurrentHashMap<K, QuotaState> quotasMap,
final K key) {
- QuotaState quotaInfo = quotasMap.get(key);
- if (quotaInfo == null) {
- quotaInfo = new QuotaState();
- if (quotasMap.putIfAbsent(key, quotaInfo) == null) {
- triggerCacheRefresh();
- }
- }
- return quotaInfo;
- }
-
- private Configuration getConfiguration() {
- return rsServices.getConfiguration();
+ return computeIfAbsent(quotasMap, key, QuotaState::new, this::triggerCacheRefresh);
}
@VisibleForTesting
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index 19e1235..d2a1838 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -1,4 +1,4 @@
-/*
+/**
* 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
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.regionserver;
import static org.apache.hadoop.hbase.HConstants.REPLICATION_SCOPE_LOCAL;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Optional;
@@ -5314,19 +5315,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
// Keep trying until we have a lock or error out.
// TODO: do we need to add a time component here?
while (result == null) {
-
- // Try adding a RowLockContext to the lockedRows.
- // If we can add it then there's no other transactions currently running.
- rowLockContext = new RowLockContext(rowKey);
- RowLockContext existingContext = lockedRows.putIfAbsent(rowKey, rowLockContext);
-
- // if there was a running transaction then there's already a context.
- if (existingContext != null) {
- rowLockContext = existingContext;
- }
-
+ rowLockContext = computeIfAbsent(lockedRows, rowKey, () -> new RowLockContext(rowKey));
// Now try an get the lock.
- //
// This can fail as
if (readLock) {
result = rowLockContext.newReadLock();
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java
index 62dea53..6e7ad9b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java
@@ -17,6 +17,8 @@
*/
package org.apache.hadoop.hbase.regionserver.wal;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
@@ -215,16 +217,8 @@ class SequenceIdAccounting {
@VisibleForTesting
ConcurrentMap<ImmutableByteArray, Long> getOrCreateLowestSequenceIds(byte[] encodedRegionName) {
// Intentionally, this access is done outside of this.regionSequenceIdLock. Done per append.
- ConcurrentMap<ImmutableByteArray, Long> m = this.lowestUnflushedSequenceIds
- .get(encodedRegionName);
- if (m != null) {
- return m;
- }
- m = new ConcurrentHashMap<>();
- // Another thread may have added it ahead of us.
- ConcurrentMap<ImmutableByteArray, Long> alreadyPut = this.lowestUnflushedSequenceIds
- .putIfAbsent(encodedRegionName, m);
- return alreadyPut == null ? m : alreadyPut;
+ return computeIfAbsent(this.lowestUnflushedSequenceIds, encodedRegionName,
+ ConcurrentHashMap::new);
}
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/TableAuthManager.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/TableAuthManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/TableAuthManager.java
index 25cfc8b..5032d96 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/TableAuthManager.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/TableAuthManager.java
@@ -1,4 +1,4 @@
-/*
+/**
* 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
@@ -18,6 +18,13 @@
package org.apache.hadoop.hbase.security.access;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ListMultimap;
+import com.google.common.collect.Lists;
+
import java.io.Closeable;
import java.io.IOException;
import java.util.HashMap;
@@ -28,11 +35,11 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.hbase.AuthUtil;
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.AuthUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.security.Superusers;
import org.apache.hadoop.hbase.security.User;
@@ -41,11 +48,6 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
import org.apache.zookeeper.KeeperException;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.ListMultimap;
-import com.google.common.collect.Lists;
-
/**
* Performs authorization checks for a given user's assigned permissions
*/
@@ -276,17 +278,11 @@ public class TableAuthManager implements Closeable {
}
private PermissionCache<TablePermission> getTablePermissions(TableName table) {
- if (!tableCache.containsKey(table)) {
- tableCache.putIfAbsent(table, new PermissionCache<TablePermission>());
- }
- return tableCache.get(table);
+ return computeIfAbsent(tableCache, table, PermissionCache::new);
}
private PermissionCache<TablePermission> getNamespacePermissions(String namespace) {
- if (!nsCache.containsKey(namespace)) {
- nsCache.putIfAbsent(namespace, new PermissionCache<TablePermission>());
- }
- return nsCache.get(namespace);
+ return computeIfAbsent(nsCache, namespace, PermissionCache::new);
}
/**
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/BoundedGroupingStrategy.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/BoundedGroupingStrategy.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/BoundedGroupingStrategy.java
index 06f8792..5b32347 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/BoundedGroupingStrategy.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/BoundedGroupingStrategy.java
@@ -18,6 +18,8 @@
*/
package org.apache.hadoop.hbase.wal;
+import static org.apache.hadoop.hbase.util.CollectionUtils.computeIfAbsent;
+
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@@ -44,15 +46,8 @@ public class BoundedGroupingStrategy implements RegionGroupingStrategy{
@Override
public String group(byte[] identifier, byte[] namespace) {
String idStr = Bytes.toString(identifier);
- String groupName = groupNameCache.get(idStr);
- if (null == groupName) {
- groupName = groupNames[getAndIncrAtomicInteger(counter, groupNames.length)];
- String extantName = groupNameCache.putIfAbsent(idStr, groupName);
- if (extantName != null) {
- return extantName;
- }
- }
- return groupName;
+ return computeIfAbsent(groupNameCache, idStr,
+ () -> groupNames[getAndIncrAtomicInteger(counter, groupNames.length)]);
}
// Non-blocking incrementing & resetting of AtomicInteger.
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterShutdown.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterShutdown.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterShutdown.java
index 7f95e75..ebf16b7 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterShutdown.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterShutdown.java
@@ -122,7 +122,7 @@ public class TestMasterShutdown {
final MasterThread master = cluster.getMasters().get(MASTER_INDEX);
master.start();
LOG.info("Called master start on " + master.getName());
- Thread shutdownThread = new Thread() {
+ Thread shutdownThread = new Thread("Shutdown-Thread") {
public void run() {
LOG.info("Before call to shutdown master");
try {
http://git-wip-us.apache.org/repos/asf/hbase/blob/540ede37/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestSplitLogManager.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestSplitLogManager.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestSplitLogManager.java
index 8791e3e..022e7b6 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestSplitLogManager.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestSplitLogManager.java
@@ -58,8 +58,8 @@ import org.apache.hadoop.hbase.Waiter;
import org.apache.hadoop.hbase.coordination.ZKSplitLogManagerCoordination;
import org.apache.hadoop.hbase.master.SplitLogManager.Task;
import org.apache.hadoop.hbase.master.SplitLogManager.TaskBatch;
-import org.apache.hadoop.hbase.shaded.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
import org.apache.hadoop.hbase.regionserver.TestMasterAddressTracker.NodeCreationListener;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
import org.apache.hadoop.hbase.testclassification.MasterTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
@@ -196,6 +196,14 @@ public class TestSplitLogManager {
assertEquals(newval, e.eval());
}
+ private Task findOrCreateOrphanTask(String path) {
+ return slm.tasks.computeIfAbsent(path, k -> {
+ LOG.info("creating orphan task " + k);
+ SplitLogCounters.tot_mgr_orphan_task_acquired.incrementAndGet();
+ return new Task();
+ });
+ }
+
private String submitTaskAndWait(TaskBatch batch, String name) throws KeeperException,
InterruptedException {
String tasknode = ZKSplitLog.getEncodedNodeName(zkw, name);
@@ -205,7 +213,7 @@ public class TestSplitLogManager {
slm.enqueueSplitTask(name, batch);
assertEquals(1, batch.installed);
- assertTrue(slm.findOrCreateOrphanTask(tasknode).batch == batch);
+ assertTrue(findOrCreateOrphanTask(tasknode).batch == batch);
assertEquals(1L, tot_mgr_node_create_queued.get());
LOG.debug("waiting for task node creation");
@@ -244,7 +252,7 @@ public class TestSplitLogManager {
slm = new SplitLogManager(master, conf);
waitForCounter(tot_mgr_orphan_task_acquired, 0, 1, to/2);
- Task task = slm.findOrCreateOrphanTask(tasknode);
+ Task task = findOrCreateOrphanTask(tasknode);
assertTrue(task.isOrphan());
waitForCounter(tot_mgr_heartbeat, 0, 1, to/2);
assertFalse(task.isUnassigned());
@@ -270,12 +278,12 @@ public class TestSplitLogManager {
slm = new SplitLogManager(master, conf);
waitForCounter(tot_mgr_orphan_task_acquired, 0, 1, to/2);
- Task task = slm.findOrCreateOrphanTask(tasknode);
+ Task task = findOrCreateOrphanTask(tasknode);
assertTrue(task.isOrphan());
assertTrue(task.isUnassigned());
// wait for RESCAN node to be created
- waitForCounter(tot_mgr_rescan, 0, 1, to/2);
- Task task2 = slm.findOrCreateOrphanTask(tasknode);
+ waitForCounter(tot_mgr_rescan, 0, 1, to / 2);
+ Task task2 = findOrCreateOrphanTask(tasknode);
assertTrue(task == task2);
LOG.debug("task = " + task);
assertEquals(1L, tot_mgr_resubmit.get());