This is an automated email from the ASF dual-hosted git repository. dlmarion pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 2c45e264b2079ff2f572c3d4fb0d7165feb682fb Merge: 5794fbc192 30da958f67 Author: Dave Marion <[email protected]> AuthorDate: Fri Jul 10 22:17:31 2026 +0000 Merge branch '2.1' .../accumulo/core/clientImpl/ClientContext.java | 182 ++++++++++----------- .../core/clientImpl/ThriftTransportPool.java | 2 +- .../org/apache/accumulo/server/ServerContext.java | 26 ++- 3 files changed, 106 insertions(+), 104 deletions(-) diff --cc core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java index a6a880edbf,db1cd21ea3..cb8f9e5ec3 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/ClientContext.java @@@ -146,14 -124,15 +146,14 @@@ public class ClientContext implements A private static final Logger log = LoggerFactory.getLogger(ClientContext.class); private final ClientInfo info; - private InstanceId instanceId; - private final ZooReader zooReader; - private final ZooCache zooCache; + private final Supplier<ZooCache> zooCache; - private Credentials creds; - private BatchWriterConfig batchWriterConfig; - private ConditionalWriterConfig conditionalWriterConfig; + private Supplier<Credentials> creds; + private final Supplier<BatchWriterConfig> batchWriterConfig; + private final Supplier<ConditionalWriterConfig> conditionalWriterConfig; - private final AccumuloConfiguration serverConf; + private final AccumuloConfiguration accumuloConf; private final Configuration hadoopConf; + private final Map<DataLevel,ConcurrentHashMap<TableId,ClientTabletCache>> tabletLocationCache; // These fields are very frequently accessed (each time a connection is created) and expensive to // compute, so cache them. @@@ -161,32 -140,29 +161,35 @@@ private final Supplier<SaslConnectionParams> saslSupplier; private final Supplier<SslConnectionParams> sslSupplier; private final Supplier<ScanServerSelector> scanServerSelectorSupplier; + private final Supplier<ServiceLockPaths> serverPaths; + private final Supplier<NamespaceMapping> namespaceMapping; + private final Supplier<Cache<NamespaceId,TableMapping>> tableMappings; private TCredentials rpcCreds; - private ThriftTransportPool thriftTransportPool; - private ZookeeperLockChecker zkLockChecker; + protected Supplier<ThriftTransportPool> thriftTransportPool; + private final Supplier<ZookeeperLockChecker> zkLockChecker; - private final AtomicBoolean closed = new AtomicBoolean(); - private volatile boolean scannerReadAheadPoolCreated = false; - private volatile boolean cleanupThreadPoolCreated = false; - private volatile boolean thriftTransportPoolCreated = false; - private volatile boolean closed = false; ++ private final AtomicBoolean scannerReadAheadPoolCreated = new AtomicBoolean(false); ++ private final AtomicBoolean cleanupThreadPoolCreated = new AtomicBoolean(false); ++ private final AtomicBoolean thriftTransportPoolCreated = new AtomicBoolean(false); ++ protected final AtomicBoolean closed = new AtomicBoolean(); - private SecurityOperations secops = null; + private final Supplier<SecurityOperations> secops; private final TableOperationsImpl tableops; private final NamespaceOperations namespaceops; - private InstanceOperations instanceops = null; + private ResourceGroupOperations rgOps = null; + private final Supplier<InstanceOperations> instanceops; - @SuppressWarnings("deprecation") - private org.apache.accumulo.core.client.admin.ReplicationOperations replicationops = null; - private final SingletonReservation singletonReservation; private final Supplier<ThreadPools> clientThreadPools; - private ThreadPoolExecutor cleanupThreadPool; - private ThreadPoolExecutor scannerReadaheadPool; + private final Supplier<ThreadPoolExecutor> cleanupThreadPool; + private final Supplier<ThreadPoolExecutor> scannerReadaheadPool; - private final Supplier<TableZooHelper> tableZooHelper; + private MeterRegistry micrometer; + private Caches caches; + + private final AtomicBoolean zooKeeperOpened = new AtomicBoolean(false); + private final AtomicBoolean zooCacheCreated = new AtomicBoolean(false); + private final Supplier<ZooSession> zooSession; private void ensureOpen() { - if (closed) { + if (closed.get()) { throw new IllegalStateException("This client was closed."); } } @@@ -297,10 -257,21 +300,50 @@@ clientThreadPools = () -> ThreadPools.getClientThreadPools(getConfiguration(), ueh); } } + this.namespaceMapping = memoize(() -> new NamespaceMapping(this)); + this.tableMappings = + memoize(() -> getCaches().createNewBuilder(Caches.CacheName.TABLE_MAPPING_CACHE, true) + .expireAfterAccess(10, MINUTES).build()); + 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)); ++ batchWriterConfig = memoize(() -> getBatchWriterConfig(getClientProperties())); ++ conditionalWriterConfig = memoize(() -> getConditionalWriterConfig(getClientProperties())); + secops = memoize(() -> new SecurityOperationsImpl(this)); + instanceops = memoize(() -> new InstanceOperationsImpl(this)); - thriftTransportPool = - memoize(() -> ThriftTransportPool.startNew(this::getTransportPoolMaxAgeMillis, false)); - zkLockChecker = memoize(() -> new ZookeeperLockChecker(this)); ++ thriftTransportPool = memoize(() -> { ++ LongSupplier maxAgeSupplier = () -> { ++ try { ++ return getTransportPoolMaxAgeMillis(); ++ } catch (IllegalStateException e) { ++ if (closed.get()) { ++ // The transport pool has a background thread that may call this supplier in the middle ++ // of closing. This is here to avoid spurious exceptions from race conditions that ++ // happen when closing a client. ++ return ConfigurationTypeHelper ++ .getTimeInMillis(ClientProperty.RPC_TRANSPORT_IDLE_TIMEOUT.getDefaultValue()); ++ } ++ throw e; ++ } ++ }; ++ return ThriftTransportPool.startNew(maxAgeSupplier, false); ++ }); ++ ++ zkLockChecker = memoize(() -> { ++ // make this use its own ZooSession and ZooCache, because this is used by the ++ // tablet location cache, which is a static singleton reused by multiple clients ++ // so, it can't rely on being able to continue to use the same client's ZooCache, ++ // because that client could be closed, and its ZooSession also closed ++ // this needs to be fixed; TODO https://github.com/apache/accumulo/issues/2301 ++ var zk = info.getZooKeeperSupplier(ZookeeperLockChecker.class.getSimpleName(), ++ ZooUtil.getRoot(getInstanceID())).get(); ++ return new ZookeeperLockChecker(new ZooCache(zk, Set.of(Constants.ZTSERVERS))); ++ }); ++ } public Ample getAmple() { @@@ -308,24 -279,16 +351,16 @@@ return new AmpleImpl(this); } - public synchronized Future<List<KeyValue>> - submitScannerReadAheadTask(Callable<List<KeyValue>> c) { + public Future<List<KeyValue>> submitScannerReadAheadTask(Callable<List<KeyValue>> 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; ++ scannerReadAheadPoolCreated.set(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; ++ cleanupThreadPoolCreated.set(true); + this.cleanupThreadPool.get().execute(r); } /** @@@ -438,23 -399,11 +470,20 @@@ return batchWriterConfig; } - public synchronized BatchWriterConfig getBatchWriterConfig() { + public BatchWriterConfig getBatchWriterConfig() { ensureOpen(); - if (batchWriterConfig == null) { - batchWriterConfig = getBatchWriterConfig(getClientProperties()); - } - return batchWriterConfig; + return batchWriterConfig.get(); } + /** + * @return the scan server selector implementation used for determining which scan servers will be + * used when performing an eventually consistent scan + */ + public ScanServerSelector getScanServerSelector() { + ensureOpen(); + return scanServerSelectorSupplier.get(); + } + /** * @return map of live scan server addresses to lock uuids. */ @@@ -933,28 -867,18 +951,28 @@@ @Override public synchronized void close() { - closed = true; - if (thriftTransportPoolCreated) { - thriftTransportPool.get().shutdown(); - } - tableZooHelper.get().close(); - if (scannerReadAheadPoolCreated) { - scannerReadaheadPool.get().shutdownNow(); // abort all tasks, client is shutting down - } - if (cleanupThreadPoolCreated) { - cleanupThreadPool.get().shutdown(); // wait for shutdown tasks to execute + if (closed.compareAndSet(false, true)) { - if (thriftTransportPool != null) { ++ if (thriftTransportPoolCreated.get()) { + log.debug("Closing Thrift Transport Pool"); - thriftTransportPool.shutdown(); ++ thriftTransportPool.get().shutdown(); + } - if (scannerReadaheadPool != null) { ++ if (scannerReadAheadPoolCreated.get()) { + log.debug("Closing Scanner ReadAhead Pool"); - scannerReadaheadPool.shutdownNow(); // abort all tasks, client is shutting down ++ scannerReadaheadPool.get().shutdownNow(); // abort all tasks, client is shutting down + } - if (cleanupThreadPool != null) { ++ if (cleanupThreadPoolCreated.get()) { + log.debug("Closing Cleanup ThreadPool"); - cleanupThreadPool.shutdown(); // wait for shutdown tasks to execute ++ cleanupThreadPool.get().shutdown(); // wait for shutdown tasks to execute + } + if (zooCacheCreated.get()) { + log.debug("Closing ZooCache"); + zooCache.get().close(); + } + if (zooKeeperOpened.get()) { + log.debug("Closing ZooSession"); + zooSession.get().close(); + } } - singletonReservation.close(); } public static class ClientBuilderImpl<T> @@@ -1164,174 -1101,18 +1182,144 @@@ } protected long getTransportPoolMaxAgeMillis() { - ensureOpen(); - return ClientProperty.RPC_TRANSPORT_IDLE_TIMEOUT.getTimeInMillis(getProperties()); + return ClientProperty.RPC_TRANSPORT_IDLE_TIMEOUT.getTimeInMillis(getClientProperties()); } - public synchronized ThriftTransportPool getTransportPool() { - return getTransportPoolImpl(false); + public ThriftTransportPool getTransportPool() { + ensureOpen(); - thriftTransportPoolCreated = true; ++ thriftTransportPoolCreated.set(true); + return thriftTransportPool.get(); } - protected synchronized ThriftTransportPool getTransportPoolImpl(boolean shouldHalt) { + public ZookeeperLockChecker getTServerLockChecker() { ensureOpen(); - if (thriftTransportPool == null) { - LongSupplier maxAgeSupplier = () -> { - try { - return getTransportPoolMaxAgeMillis(); - } catch (IllegalStateException e) { - if (closed.get()) { - // The transport pool has a background thread that may call this supplier in the middle - // of closing. This is here to avoid spurious exceptions from race conditions that - // happen when closing a client. - return ConfigurationTypeHelper - .getTimeInMillis(ClientProperty.RPC_TRANSPORT_IDLE_TIMEOUT.getDefaultValue()); - } - throw e; - } - }; - thriftTransportPool = ThriftTransportPool.startNew(maxAgeSupplier, shouldHalt); - } - return thriftTransportPool; + return this.zkLockChecker.get(); } + + public MeterRegistry getMeterRegistry() { + ensureOpen(); + return micrometer; + } + + public void setMeterRegistry(MeterRegistry micrometer) { + ensureOpen(); + this.micrometer = micrometer; + getCaches(); + } + + public synchronized Caches getCaches() { + ensureOpen(); + if (caches == null) { + caches = Caches.getInstance(); + if (micrometer != null && getConfiguration().getBoolean(Property.GENERAL_MICROMETER_ENABLED) + && getConfiguration().getBoolean(Property.GENERAL_MICROMETER_CACHE_METRICS_ENABLED)) { + caches.registerMetrics(micrometer); + } + } + return caches; + } + - public synchronized ZookeeperLockChecker getTServerLockChecker() { - ensureOpen(); - if (this.zkLockChecker == null) { - // make this use its own ZooSession and ZooCache, because this is used by the - // tablet location cache, which is a static singleton reused by multiple clients - // so, it can't rely on being able to continue to use the same client's ZooCache, - // because that client could be closed, and its ZooSession also closed - // this needs to be fixed; TODO https://github.com/apache/accumulo/issues/2301 - var zk = info.getZooKeeperSupplier(ZookeeperLockChecker.class.getSimpleName(), - ZooUtil.getRoot(getInstanceID())).get(); - this.zkLockChecker = new ZookeeperLockChecker(new ZooCache(zk, Set.of(Constants.ZTSERVERS))); - } - return this.zkLockChecker; - } - + public ServiceLockPaths getServerPaths() { + return this.serverPaths.get(); + } + + public NamespaceMapping getNamespaceMapping() { + ensureOpen(); + NamespaceMapping namespaces = namespaceMapping.get(); + log.trace("Got namespace mapping: {}", namespaces); + return namespaces; + } + + public TableMapping getTableMapping(NamespaceId namespaceId) { + ensureOpen(); + var mapping = tableMappings.get().asMap().computeIfAbsent(requireNonNull(namespaceId), + id -> new TableMapping(this, id)); + log.trace("Got table mapping for namespaceId {}: {}", namespaceId, mapping); + return mapping; + } + + @VisibleForTesting + public boolean isTabletLocationCachePresent(TableId tableId) { + return tabletLocationCache.get(DataLevel.of(tableId)).containsKey(tableId); + } + + private volatile Duration clearFrequency = Duration.ofMinutes(10); + + /** + * Sets how often checks for unused tables are done + */ + @VisibleForTesting + public void setClearFrequency(Duration frequency) { + Preconditions.checkArgument(frequency != null && !frequency.isNegative() && !frequency.isZero(), + "frequency:%s", frequency); + clearFrequency = frequency; + } + + private final Timer lastClearTimer = Timer.startNew(); + + public ClientTabletCache getTabletLocationCache(TableId tableId) { + ensureOpen(); + if (lastClearTimer.hasElapsed(clearFrequency)) { + synchronized (lastClearTimer) { + if (lastClearTimer.hasElapsed(clearFrequency)) { + tabletLocationCache.get(DataLevel.USER).entrySet().removeIf(entry -> { + TableId tableIdToCheck = entry.getKey(); + ClientTabletCache cache = entry.getValue(); + var tableState = getTableState(tableIdToCheck); + if (tableState != TableState.ONLINE && tableState != TableState.OFFLINE) { + cache.invalidateCache(); + return true; + } + return false; + }); + lastClearTimer.restart(); + } + } + } + + return tabletLocationCache.get(DataLevel.of(tableId)).computeIfAbsent(tableId, + (TableId key) -> { + var lockChecker = getTServerLockChecker(); + if (SystemTables.ROOT.tableId().equals(tableId)) { + return new RootClientTabletCache(lockChecker); + } + var mlo = new MetadataCachedTabletObtainer(); + if (SystemTables.METADATA.tableId().equals(tableId)) { + return new ClientTabletCacheImpl(SystemTables.METADATA.tableId(), + getTabletLocationCache(SystemTables.ROOT.tableId()), mlo, lockChecker); + } else { + return new ClientTabletCacheImpl(tableId, + getTabletLocationCache(SystemTables.METADATA.tableId()), mlo, lockChecker); + } + }); + } + + /** + * Clear the currently cached tablet locations. The use of ConcurrentHashMap ensures this is + * thread-safe. However, since the ConcurrentHashMap iterator is weakly consistent, it does not + * block new locations from being cached. If new locations are added while this is executing, they + * may be immediately invalidated by this code. Multiple calls to this method in different threads + * may cause some location caches to be invalidated multiple times. That is okay, because cache + * invalidation is idempotent. + */ + public void clearTabletLocationCache() { + tabletLocationCache.forEach((dataLevel, map) -> { + // use iter.remove() instead of calling clear() on the map, to prevent clearing entries that + // may not have been invalidated + var iter = map.values().iterator(); + while (iter.hasNext()) { + iter.next().invalidate(); + iter.remove(); + } + }); + } + + private static Set<String> createPersistentWatcherPaths() { + return Set.of(Constants.ZCOMPACTORS, Constants.ZGC_LOCK, Constants.ZMANAGER_LOCK, + Constants.ZMINI_LOCK, Constants.ZMONITOR_LOCK, Constants.ZNAMESPACES, Constants.ZRECOVERY, + Constants.ZSSERVERS, Constants.ZTABLES, Constants.ZTSERVERS, Constants.ZUSERS, + RootTable.ZROOT_TABLET, Constants.ZTEST_LOCK, Constants.ZMANAGER_ASSISTANT_LOCK, + Constants.ZRESOURCEGROUPS, Constants.ZMANAGER_ASSIGNMENTS); + } + } diff --cc server/base/src/main/java/org/apache/accumulo/server/ServerContext.java index aa972444d9,12feff1371..16b20f0d98 --- a/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java +++ b/server/base/src/main/java/org/apache/accumulo/server/ServerContext.java @@@ -37,18 -38,12 +37,21 @@@ import java.util.TreeMap import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; ++import java.util.function.LongSupplier; import java.util.function.Supplier; import org.apache.accumulo.core.Constants; +import org.apache.accumulo.core.client.ConditionalWriter; +import org.apache.accumulo.core.client.ConditionalWriterConfig; +import org.apache.accumulo.core.client.TableNotFoundException; import org.apache.accumulo.core.clientImpl.ClientContext; +import org.apache.accumulo.core.clientImpl.ClientInfo; import org.apache.accumulo.core.clientImpl.ThriftTransportPool; import org.apache.accumulo.core.conf.AccumuloConfiguration; ++import org.apache.accumulo.core.conf.ClientProperty; ++import org.apache.accumulo.core.conf.ConfigurationTypeHelper; import org.apache.accumulo.core.conf.DefaultConfiguration; import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.conf.SiteConfiguration; @@@ -150,11 -131,9 +153,27 @@@ public class ServerContext extends Clie securityOperation = memoize(() -> new AuditedSecurityOperation(this, SecurityOperation.getAuthorizor(this), SecurityOperation.getAuthenticator(this), SecurityOperation.getPermHandler(this))); + lowMemoryDetector = memoize(() -> new LowMemoryDetector()); metricsInfoSupplier = memoize(() -> new MetricsInfoImpl(this)); - - thriftTransportPool = - memoize(() -> ThriftTransportPool.startNew(this::getTransportPoolMaxAgeMillis, true)); + sharedMetadataWriter = memoize(() -> createSharedConditionalWriter(DataLevel.METADATA)); + sharedUserWriter = memoize(() -> createSharedConditionalWriter(DataLevel.USER)); ++ thriftTransportPool = memoize(() -> { ++ LongSupplier maxAgeSupplier = () -> { ++ try { ++ return getTransportPoolMaxAgeMillis(); ++ } catch (IllegalStateException e) { ++ if (closed.get()) { ++ // The transport pool has a background thread that may call this supplier in the middle ++ // of closing. This is here to avoid spurious exceptions from race conditions that ++ // happen when closing a client. ++ return ConfigurationTypeHelper ++ .getTimeInMillis(ClientProperty.RPC_TRANSPORT_IDLE_TIMEOUT.getDefaultValue()); ++ } ++ throw e; ++ } ++ }; ++ return ThriftTransportPool.startNew(maxAgeSupplier, true); ++ }); } /**
