From f7fe624a06b5be7f3bbe6865a9b5c56af5a5c355 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Thu, 9 Jul 2026 15:21:31 +0000 Subject: [PATCH 1/3] Reduce probability of ClientContext.close / TabletLocator.getLocator deadlock Looking at the non-deterministic failures of SessionBlockVerifyIT I found a deadlock between the ClientContext.close and TabletLocator.getLocator methods where there are unclosed Scanners in the client. JStack found the deadlock at: Java stack information for the threads listed above: =================================================== "junit-timeout-thread-4": at org.apache.accumulo.core.clientImpl.TabletLocator.disable(TabletLocator.java:134) - waiting to lock <0x000000042dbf08e8> (a java.lang.Class for org.apache.accumulo.core.clientImpl.TabletLocator) at org.apache.accumulo.core.clientImpl.TabletLocator$1.disable(TabletLocator.java:245) at org.apache.accumulo.core.singletons.SingletonManager.disable(SingletonManager.java:106) at org.apache.accumulo.core.singletons.SingletonManager$$Lambda/0x00007fbd94325f88.accept(Unknown Source) at java.util.ArrayList.forEach(java.base@21.0.11/ArrayList.java:1596) at org.apache.accumulo.core.singletons.SingletonManager.transition(SingletonManager.java:196) at org.apache.accumulo.core.singletons.SingletonManager.releaseReservation(SingletonManager.java:145) - locked <0x000000042dcc4b68> (a java.lang.Class for org.apache.accumulo.core.singletons.SingletonManager) at org.apache.accumulo.core.singletons.SingletonReservation.close(SingletonReservation.java:50) at org.apache.accumulo.core.clientImpl.ClientContext.close(ClientContext.java:894) - locked <0x000000042d857668> (a org.apache.accumulo.core.clientImpl.ClientContext) at org.apache.accumulo.test.functional.SessionBlockVerifyIT.run(SessionBlockVerifyIT.java:179) "pool-2-thread-5": at org.apache.accumulo.core.clientImpl.ClientContext.tableZooHelper(ClientContext.java:626) - waiting to lock <0x000000042d857668> (a org.apache.accumulo.core.clientImpl.ClientContext) at org.apache.accumulo.core.clientImpl.ClientContext.getTableState(ClientContext.java:671) at org.apache.accumulo.core.clientImpl.TabletLocator.getLocator(TabletLocator.java:148) - locked <0x000000042dbf08e8> (a java.lang.Class for org.apache.accumulo.core.clientImpl.TabletLocator) at org.apache.accumulo.core.clientImpl.ThriftScanner.scan(ThriftScanner.java:518) at org.apache.accumulo.core.clientImpl.ScannerIterator.readBatch(ScannerIterator.java:159) - locked <0x000000042d863fb8> (a org.apache.accumulo.core.clientImpl.ThriftScanner$ScanState) at org.apache.accumulo.core.clientImpl.ScannerIterator.getNextBatch(ScannerIterator.java:177) at org.apache.accumulo.core.clientImpl.ScannerIterator.hasNext(ScannerIterator.java:109) at org.apache.accumulo.test.functional.SessionBlockVerifyIT.lambda$0(SessionBlockVerifyIT.java:122) at org.apache.accumulo.test.functional.SessionBlockVerifyIT$$Lambda/0x00007fbd943acba8.call(Unknown Source) --- .../accumulo/core/clientImpl/ClientContext.java | 12 ++++++++++++ .../accumulo/core/clientImpl/TabletLocator.java | 4 ++-- 2 files changed, 14 insertions(+), 2 deletions(-) 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..cc2c639fc27 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 @@ -631,51 +631,63 @@ private synchronized TableZooHelper tableZooHelper() { } public TableId getTableId(String tableName) throws TableNotFoundException { + ensureOpen(); return tableZooHelper().getTableId(tableName); } public TableId _getTableIdDetectNamespaceNotFound(String tableName) throws NamespaceNotFoundException, TableNotFoundException { + ensureOpen(); return tableZooHelper()._getTableIdDetectNamespaceNotFound(tableName); } public String getTableName(TableId tableId) throws TableNotFoundException { + ensureOpen(); return tableZooHelper().getTableName(tableId); } public Map getTableNameToIdMap() { + ensureOpen(); return tableZooHelper().getTableMap().getNameToIdMap(); } public Map getTableIdToNameMap() { + ensureOpen(); return tableZooHelper().getTableMap().getIdtoNameMap(); } public boolean tableNodeExists(TableId tableId) { + ensureOpen(); return tableZooHelper().tableNodeExists(tableId); } public void clearTableListCache() { + ensureOpen(); tableZooHelper().clearTableListCache(); } public String getPrintableTableInfoFromId(TableId tableId) { + ensureOpen(); return tableZooHelper().getPrintableTableInfoFromId(tableId); } public String getPrintableTableInfoFromName(String tableName) { + ensureOpen(); return tableZooHelper().getPrintableTableInfoFromName(tableName); } public TableState getTableState(TableId tableId) { + ensureOpen(); return tableZooHelper().getTableState(tableId, false); } public TableState getTableState(TableId tableId, boolean clearCachedState) { + ensureOpen(); return tableZooHelper().getTableState(tableId, clearCachedState); } public NamespaceId getNamespaceId(TableId tableId) throws TableNotFoundException { + ensureOpen(); return tableZooHelper().getNamespaceId(tableId); } diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java index 6375b9cf841..3e1c9b302b1 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java @@ -116,7 +116,7 @@ public boolean equals(LocatorKey lk) { private static final HashMap locators = new HashMap<>(); private static final HashMap offlineLocators = new HashMap<>(); - private static boolean enabled = true; + private static volatile boolean enabled = true; public static synchronized void clearLocators() { for (TabletLocator locator : locators.values()) { @@ -131,8 +131,8 @@ static synchronized boolean isEnabled() { } static synchronized void disable() { - clearLocators(); enabled = false; + clearLocators(); } static synchronized void enable() { From eccdf932832ed7bcf38c66fdc26d7f82b11e8277 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Thu, 9 Jul 2026 19:46:43 +0000 Subject: [PATCH 2/3] Implement PR suggestions --- .../core/clientImpl/ClientContext.java | 23 ++++++------------- .../core/clientImpl/TabletLocator.java | 4 ++-- 2 files changed, 9 insertions(+), 18 deletions(-) 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 cc2c639fc27..fbe75f37f6b 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 @@ -621,73 +621,64 @@ public ZooCache getZooCache() { } private TableZooHelper tableZooHelper; + private final Object tableZooMonitor = new Object(); - private synchronized TableZooHelper tableZooHelper() { + private TableZooHelper tableZooHelper() { ensureOpen(); - if (tableZooHelper == null) { - tableZooHelper = new TableZooHelper(this); + synchronized (tableZooMonitor) { + if (tableZooHelper == null) { + tableZooHelper = new TableZooHelper(this); + } + return tableZooHelper; } - return tableZooHelper; } public TableId getTableId(String tableName) throws TableNotFoundException { - ensureOpen(); return tableZooHelper().getTableId(tableName); } public TableId _getTableIdDetectNamespaceNotFound(String tableName) throws NamespaceNotFoundException, TableNotFoundException { - ensureOpen(); return tableZooHelper()._getTableIdDetectNamespaceNotFound(tableName); } public String getTableName(TableId tableId) throws TableNotFoundException { - ensureOpen(); return tableZooHelper().getTableName(tableId); } public Map getTableNameToIdMap() { - ensureOpen(); return tableZooHelper().getTableMap().getNameToIdMap(); } public Map getTableIdToNameMap() { - ensureOpen(); return tableZooHelper().getTableMap().getIdtoNameMap(); } public boolean tableNodeExists(TableId tableId) { - ensureOpen(); return tableZooHelper().tableNodeExists(tableId); } public void clearTableListCache() { - ensureOpen(); tableZooHelper().clearTableListCache(); } public String getPrintableTableInfoFromId(TableId tableId) { - ensureOpen(); return tableZooHelper().getPrintableTableInfoFromId(tableId); } public String getPrintableTableInfoFromName(String tableName) { - ensureOpen(); return tableZooHelper().getPrintableTableInfoFromName(tableName); } public TableState getTableState(TableId tableId) { - ensureOpen(); return tableZooHelper().getTableState(tableId, false); } public TableState getTableState(TableId tableId, boolean clearCachedState) { - ensureOpen(); return tableZooHelper().getTableState(tableId, clearCachedState); } public NamespaceId getNamespaceId(TableId tableId) throws TableNotFoundException { - ensureOpen(); return tableZooHelper().getNamespaceId(tableId); } diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java index 3e1c9b302b1..6375b9cf841 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocator.java @@ -116,7 +116,7 @@ public boolean equals(LocatorKey lk) { private static final HashMap locators = new HashMap<>(); private static final HashMap offlineLocators = new HashMap<>(); - private static volatile boolean enabled = true; + private static boolean enabled = true; public static synchronized void clearLocators() { for (TabletLocator locator : locators.values()) { @@ -131,8 +131,8 @@ static synchronized boolean isEnabled() { } static synchronized void disable() { - enabled = false; clearLocators(); + enabled = false; } static synchronized void enable() { From a6ecc3314dd93d2d707ea76a58e00043a74f0c60 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Fri, 10 Jul 2026 12:22:05 +0000 Subject: [PATCH 3/3] Removed most synchronization from ClientContext --- .../core/clientImpl/ClientContext.java | 148 ++++++++---------- .../core/clientImpl/ThriftTransportPool.java | 2 +- .../apache/accumulo/server/ServerContext.java | 7 +- 3 files changed, 64 insertions(+), 93 deletions(-) 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 fbe75f37f6b..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,17 +622,9 @@ public ZooCache getZooCache() { return zooCache; } - private TableZooHelper tableZooHelper; - private final Object tableZooMonitor = new Object(); - private TableZooHelper tableZooHelper() { ensureOpen(); - synchronized (tableZooMonitor) { - if (tableZooHelper == null) { - tableZooHelper = new TableZooHelper(this); - } - return tableZooHelper; - } + return tableZooHelper.get(); } public TableId getTableId(String tableName) throws TableNotFoundException { @@ -819,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 @@ -882,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(); } @@ -1121,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(); }