diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java index d267d4b7d41..db1cd21ea33 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java @@ -128,9 +128,9 @@ public class ClientContext implements AccumuloClient { private final ZooReader zooReader; private final ZooCache zooCache; - private Credentials creds; - private BatchWriterConfig batchWriterConfig; - private ConditionalWriterConfig conditionalWriterConfig; + private Supplier creds; + private final Supplier batchWriterConfig; + private final Supplier conditionalWriterConfig; private final AccumuloConfiguration serverConf; private final Configuration hadoopConf; @@ -141,21 +141,25 @@ public class ClientContext implements AccumuloClient { private final Supplier sslSupplier; private final Supplier scanServerSelectorSupplier; private TCredentials rpcCreds; - private ThriftTransportPool thriftTransportPool; - private ZookeeperLockChecker zkLockChecker; + protected Supplier thriftTransportPool; + private final Supplier zkLockChecker; + private volatile boolean scannerReadAheadPoolCreated = false; + private volatile boolean cleanupThreadPoolCreated = false; + private volatile boolean thriftTransportPoolCreated = false; private volatile boolean closed = false; - private SecurityOperations secops = null; + private final Supplier secops; private final TableOperationsImpl tableops; private final NamespaceOperations namespaceops; - private InstanceOperations instanceops = null; + private final Supplier instanceops; @SuppressWarnings("deprecation") private org.apache.accumulo.core.client.admin.ReplicationOperations replicationops = null; private final SingletonReservation singletonReservation; private final Supplier clientThreadPools; - private ThreadPoolExecutor cleanupThreadPool; - private ThreadPoolExecutor scannerReadaheadPool; + private final Supplier cleanupThreadPool; + private final Supplier scannerReadaheadPool; + private final Supplier tableZooHelper; private void ensureOpen() { if (closed) { @@ -253,6 +257,21 @@ public ClientContext(SingletonReservation reservation, ClientInfo info, clientThreadPools = () -> ThreadPools.getClientThreadPools(getConfiguration(), ueh); } } + scannerReadaheadPool = memoize(() -> clientThreadPools.get() + .getPoolBuilder(SCANNER_READ_AHEAD_POOL).numCoreThreads(0).numMaxThreads(Integer.MAX_VALUE) + .withTimeOut(3L, SECONDS).withQueue(new SynchronousQueue<>()).build()); + cleanupThreadPool = + memoize(() -> clientThreadPools.get().getPoolBuilder(CONDITIONAL_WRITER_CLEANUP_POOL) + .numCoreThreads(1).withTimeOut(3L, SECONDS).build()); + creds = memoize(() -> new Credentials(info.getPrincipal(), info.getAuthenticationToken())); + batchWriterConfig = memoize(() -> getBatchWriterConfig(info.getProperties())); + conditionalWriterConfig = memoize(() -> getConditionalWriterConfig(info.getProperties())); + tableZooHelper = memoize(() -> new TableZooHelper(this)); + secops = memoize(() -> new SecurityOperationsImpl(this)); + instanceops = memoize(() -> new InstanceOperationsImpl(this)); + thriftTransportPool = + memoize(() -> ThriftTransportPool.startNew(this::getTransportPoolMaxAgeMillis, false)); + zkLockChecker = memoize(() -> new ZookeeperLockChecker(this)); } public Ample getAmple() { @@ -260,24 +279,16 @@ public Ample getAmple() { return new AmpleImpl(this); } - public synchronized Future> - submitScannerReadAheadTask(Callable> c) { + public Future> submitScannerReadAheadTask(Callable> c) { ensureOpen(); - if (scannerReadaheadPool == null) { - scannerReadaheadPool = clientThreadPools.get().getPoolBuilder(SCANNER_READ_AHEAD_POOL) - .numCoreThreads(0).numMaxThreads(Integer.MAX_VALUE).withTimeOut(3L, SECONDS) - .withQueue(new SynchronousQueue<>()).build(); - } - return scannerReadaheadPool.submit(c); + scannerReadAheadPoolCreated = true; + return scannerReadaheadPool.get().submit(c); } - public synchronized void executeCleanupTask(Runnable r) { + public void executeCleanupTask(Runnable r) { ensureOpen(); - if (cleanupThreadPool == null) { - cleanupThreadPool = clientThreadPools.get().getPoolBuilder(CONDITIONAL_WRITER_CLEANUP_POOL) - .numCoreThreads(1).withTimeOut(3L, SECONDS).build(); - } - this.cleanupThreadPool.execute(r); + cleanupThreadPoolCreated = true; + this.cleanupThreadPool.get().execute(r); } /** @@ -291,12 +302,9 @@ public ThreadPools threadPools() { /** * Retrieve the credentials used to construct this context */ - public synchronized Credentials getCredentials() { + public Credentials getCredentials() { ensureOpen(); - if (creds == null) { - creds = new Credentials(info.getPrincipal(), info.getAuthenticationToken()); - } - return creds; + return creds.get(); } public String getPrincipal() { @@ -318,10 +326,10 @@ public Properties getProperties() { * Update the credentials in the current context after changing the current user's password or * other auth token */ - public synchronized void setCredentials(Credentials newCredentials) { + public void setCredentials(Credentials newCredentials) { ensureOpen(); checkArgument(newCredentials != null, "newCredentials is null"); - creds = newCredentials; + creds = memoize(() -> newCredentials); rpcCreds = null; } @@ -391,12 +399,9 @@ static BatchWriterConfig getBatchWriterConfig(Properties props) { return batchWriterConfig; } - public synchronized BatchWriterConfig getBatchWriterConfig() { + public BatchWriterConfig getBatchWriterConfig() { ensureOpen(); - if (batchWriterConfig == null) { - batchWriterConfig = getBatchWriterConfig(info.getProperties()); - } - return batchWriterConfig; + return batchWriterConfig.get(); } /** @@ -451,12 +456,9 @@ static ConditionalWriterConfig getConditionalWriterConfig(Properties props) { return conditionalWriterConfig; } - public synchronized ConditionalWriterConfig getConditionalWriterConfig() { + public ConditionalWriterConfig getConditionalWriterConfig() { ensureOpen(); - if (conditionalWriterConfig == null) { - conditionalWriterConfig = getConditionalWriterConfig(info.getProperties()); - } - return conditionalWriterConfig; + return conditionalWriterConfig.get(); } /** @@ -620,14 +622,9 @@ public ZooCache getZooCache() { return zooCache; } - private TableZooHelper tableZooHelper; - - private synchronized TableZooHelper tableZooHelper() { + private TableZooHelper tableZooHelper() { ensureOpen(); - if (tableZooHelper == null) { - tableZooHelper = new TableZooHelper(this); - } - return tableZooHelper; + return tableZooHelper.get(); } public TableId getTableId(String tableName) throws TableNotFoundException { @@ -816,35 +813,27 @@ public String whoami() { } @Override - public synchronized TableOperations tableOperations() { + public TableOperations tableOperations() { ensureOpen(); return tableops; } @Override - public synchronized NamespaceOperations namespaceOperations() { + public NamespaceOperations namespaceOperations() { ensureOpen(); return namespaceops; } @Override - public synchronized SecurityOperations securityOperations() { + public SecurityOperations securityOperations() { ensureOpen(); - if (secops == null) { - secops = new SecurityOperationsImpl(this); - } - - return secops; + return secops.get(); } @Override - public synchronized InstanceOperations instanceOperations() { + public InstanceOperations instanceOperations() { ensureOpen(); - if (instanceops == null) { - instanceops = new InstanceOperationsImpl(this); - } - - return instanceops; + return instanceops.get(); } @Override @@ -879,17 +868,15 @@ public AuthenticationToken token() { @Override public synchronized void close() { closed = true; - if (thriftTransportPool != null) { - thriftTransportPool.shutdown(); - } - if (tableZooHelper != null) { - tableZooHelper.close(); + if (thriftTransportPoolCreated) { + thriftTransportPool.get().shutdown(); } - if (scannerReadaheadPool != null) { - scannerReadaheadPool.shutdownNow(); // abort all tasks, client is shutting down + tableZooHelper.get().close(); + if (scannerReadAheadPoolCreated) { + scannerReadaheadPool.get().shutdownNow(); // abort all tasks, client is shutting down } - if (cleanupThreadPool != null) { - cleanupThreadPool.shutdown(); // wait for shutdown tasks to execute + if (cleanupThreadPoolCreated) { + cleanupThreadPool.get().shutdown(); // wait for shutdown tasks to execute } singletonReservation.close(); } @@ -1118,24 +1105,14 @@ protected long getTransportPoolMaxAgeMillis() { return ClientProperty.RPC_TRANSPORT_IDLE_TIMEOUT.getTimeInMillis(getProperties()); } - public synchronized ThriftTransportPool getTransportPool() { - return getTransportPoolImpl(false); - } - - protected synchronized ThriftTransportPool getTransportPoolImpl(boolean shouldHalt) { + public ThriftTransportPool getTransportPool() { ensureOpen(); - if (thriftTransportPool == null) { - thriftTransportPool = - ThriftTransportPool.startNew(this::getTransportPoolMaxAgeMillis, shouldHalt); - } - return thriftTransportPool; + thriftTransportPoolCreated = true; + return thriftTransportPool.get(); } - public synchronized ZookeeperLockChecker getTServerLockChecker() { + public ZookeeperLockChecker getTServerLockChecker() { ensureOpen(); - if (this.zkLockChecker == null) { - this.zkLockChecker = new ZookeeperLockChecker(this); - } - return this.zkLockChecker; + return this.zkLockChecker.get(); } } diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java index 59a3b9d8c2a..2f543eed316 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ThriftTransportPool.java @@ -117,7 +117,7 @@ private Runnable thriftConnectionPoolChecker() { * and false if the thread is running within a client * @return a new instance with its checker thread started to clean up idle transports */ - static ThriftTransportPool startNew(LongSupplier maxAgeMillis, boolean shouldHalt) { + public static ThriftTransportPool startNew(LongSupplier maxAgeMillis, boolean shouldHalt) { var pool = new ThriftTransportPool(maxAgeMillis, shouldHalt); log.debug("Set thrift transport pool idle time to {}ms", maxAgeMillis.getAsLong()); pool.checkThread.start(); diff --git a/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java b/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java index 9f1a0f607df..12feff13710 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java +++ b/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java @@ -132,6 +132,8 @@ private ServerContext(ServerInfo info) { memoize(() -> new AuditedSecurityOperation(this, SecurityOperation.getAuthorizor(this), SecurityOperation.getAuthenticator(this), SecurityOperation.getPermHandler(this))); metricsInfoSupplier = memoize(() -> new MetricsInfoImpl(this)); + thriftTransportPool = + memoize(() -> ThriftTransportPool.startNew(this::getTransportPoolMaxAgeMillis, true)); } /** @@ -451,11 +453,6 @@ protected long getTransportPoolMaxAgeMillis() { return getClientTimeoutInMillis(); } - @Override - public synchronized ThriftTransportPool getTransportPool() { - return getTransportPoolImpl(true); - } - public AuditedSecurityOperation getSecurityOperation() { return securityOperation.get(); }