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 6293e0cc1b0abdc8540e5fdd3817cbbc23808a40 Merge: 6a48fc8876 3d34c27538 Author: Dave Marion <[email protected]> AuthorDate: Tue Jul 7 12:20:44 2026 +0000 Merge branch '2.1' .../org/apache/accumulo/core/client/IsolatedScanner.java | 14 +++++++++----- .../accumulo/core/clientImpl/InstanceOperationsImpl.java | 11 +++++------ .../core/clientImpl/TabletServerBatchReaderIterator.java | 8 ++++---- .../main/java/org/apache/accumulo/core/data/Mutation.java | 4 ++-- .../apache/accumulo/core/file/rfile/MultiLevelIndex.java | 2 +- .../accumulo/core/iteratorsImpl/IteratorConfigUtil.java | 4 ++-- .../core/iteratorsImpl/system/SystemIteratorUtil.java | 2 +- .../apache/accumulo/core/spi/balancer/GroupBalancer.java | 3 ++- .../accumulo/core/spi/balancer/SimpleLoadBalancer.java | 6 +++--- .../java/org/apache/accumulo/core/summary/Gatherer.java | 6 +++--- .../java/org/apache/accumulo/core/util/ByteArraySet.java | 2 +- .../java/org/apache/accumulo/core/util/ByteBufferUtil.java | 2 +- .../apache/accumulo/server/compaction/FileCompactor.java | 2 +- .../java/org/apache/accumulo/server/fs/FileManager.java | 6 +++--- .../org/apache/accumulo/server/fs/VolumeManagerImpl.java | 2 +- .../org/apache/accumulo/server/util/MetadataTableUtil.java | 3 +-- .../accumulo/manager/ManagerClientServiceHandler.java | 2 +- .../java/org/apache/accumulo/tserver/log/LogSorter.java | 7 ++++--- .../apache/accumulo/tserver/session/SessionManager.java | 3 ++- .../org/apache/accumulo/tserver/tablet/ScanDataSource.java | 3 +-- .../java/org/apache/accumulo/tserver/tablet/Scanner.java | 4 ++-- .../java/org/apache/accumulo/tserver/tablet/Tablet.java | 4 ++-- 22 files changed, 52 insertions(+), 48 deletions(-) diff --cc core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java index 549a48d5ff,e21b7357a8..ec5af824aa --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/InstanceOperationsImpl.java @@@ -220,22 -209,8 +220,21 @@@ public class InstanceOperationsImpl imp } @Override + public Map<String,String> getSystemProperties() + throws AccumuloException, AccumuloSecurityException { + return ThriftClientTypes.CLIENT.execute(context, + client -> client.getSystemProperties(TraceUtil.traceInfo(), context.rpcCreds())); + } + + @Override + @Deprecated(since = "4.0.0") public List<String> getManagerLocations() { - - return context.getManagerLocations(); + Set<ServerId> managers = getServers(ServerId.Type.MANAGER); + if (managers == null || managers.isEmpty()) { + return List.of(); + } else { + return managers.stream().map(ServerId::toHostPortString).toList(); + } } @Override @@@ -267,44 -248,18 +266,45 @@@ } @Override + @Deprecated(since = "4.0.0") public List<ActiveScan> getActiveScans(String tserver) throws AccumuloException, AccumuloSecurityException { - final var parsedTserver = HostAndPort.fromString(tserver); - TabletScanClientService.Client client = null; + var si = getServerId(tserver, List.of(Type.TABLET_SERVER, Type.SCAN_SERVER)); + // getActiveScans throws exceptions so we can't use Optional.map() here + return si.isPresent() ? getActiveScans(si.orElseThrow()) : List.of(); + } + + @Override + public List<ActiveScan> getActiveScans(Collection<ServerId> servers) + throws AccumuloException, AccumuloSecurityException { + servers.forEach(InstanceOperationsImpl::checkActiveScanServer); + return queryServers(servers, this::getActiveScans, INSTANCE_OPS_SCANS_FINDER_POOL); + } + + private static void checkActiveScanServer(ServerId server) { + Objects.requireNonNull(server); + Preconditions.checkArgument( + server.getType() == ServerId.Type.SCAN_SERVER + || server.getType() == ServerId.Type.TABLET_SERVER, + "Server type %s is not %s or %s.", server.getType(), ServerId.Type.SCAN_SERVER, + ServerId.Type.TABLET_SERVER); + } + + private List<ActiveScan> getActiveScans(ServerId server) + throws AccumuloException, AccumuloSecurityException { + + checkActiveScanServer(server); + + final var parsedTserver = HostAndPort.fromParts(server.getHost(), server.getPort()); + TabletScanClientService.Client rpcClient = null; try { - client = getClient(ThriftClientTypes.TABLET_SCAN, parsedTserver, context); + rpcClient = getClient(ThriftClientTypes.TABLET_SCAN, parsedTserver, context); - List<ActiveScan> as = new ArrayList<>(); - for (var activeScan : rpcClient.getActiveScans(TraceUtil.traceInfo(), context.rpcCreds())) { - final var scans = client.getActiveScans(TraceUtil.traceInfo(), context.rpcCreds()); ++ final var scans = rpcClient.getActiveScans(TraceUtil.traceInfo(), context.rpcCreds()); + List<ActiveScan> as = new ArrayList<>(scans.size()); + for (var activeScan : scans) { try { - as.add(new ActiveScanImpl(context, activeScan)); + as.add(new ActiveScanImpl(context, activeScan, server)); } catch (TableNotFoundException e) { throw new AccumuloException(e); } @@@ -383,45 -322,33 +383,44 @@@ public List<ActiveCompaction> getActiveCompactions() throws AccumuloException, AccumuloSecurityException { - Map<String,List<HostAndPort>> compactors = ExternalCompactionUtil.getCompactorAddrs(context); - List<String> tservers = getTabletServers(); + Set<ServerId> compactionServers = new HashSet<>(); + compactionServers.addAll(getServers(ServerId.Type.COMPACTOR)); + compactionServers.addAll(getServers(ServerId.Type.TABLET_SERVER)); - int numThreads = Math.max(4, Math.min((tservers.size() + compactors.size()) / 10, 256)); - var executorService = context.threadPools().getPoolBuilder(INSTANCE_OPS_COMPACTIONS_FINDER_POOL) - .numCoreThreads(numThreads).build(); - try { - List<Future<List<ActiveCompaction>>> futures = new ArrayList<>(tservers.size()); + return getActiveCompactions(compactionServers); + } - for (String tserver : tservers) { - futures.add(executorService.submit(() -> getActiveCompactions(tserver))); - } + @Override + public List<ActiveCompaction> getActiveCompactions(Collection<ServerId> compactionServers) + throws AccumuloException, AccumuloSecurityException { + compactionServers.forEach(InstanceOperationsImpl::checkActiveCompactionServer); + return queryServers(compactionServers, this::getActiveCompactions, + INSTANCE_OPS_COMPACTIONS_FINDER_POOL); + } - compactors.values().forEach(compactorList -> { - for (HostAndPort compactorAddr : compactorList) { - Callable<List<ActiveCompaction>> task = - () -> ExternalCompactionUtil.getActiveCompaction(compactorAddr, context).stream() - .map(tac -> new ActiveCompactionImpl(context, tac, compactorAddr, - CompactionHost.Type.COMPACTOR)) - .collect(toList()); + private <T> List<T> queryServers(Collection<ServerId> servers, ServerQuery<List<T>> serverQuery, + ThreadPoolNames pool) throws AccumuloException, AccumuloSecurityException { + + final ExecutorService executorService; + // If size 0 or 1 there's no need to create a thread pool + if (servers.isEmpty()) { + return List.of(); + } else if (servers.size() == 1) { + executorService = MoreExecutors.newDirectExecutorService(); + } else { + int numThreads = Math.max(4, Math.min((servers.size()) / 10, 256)); + executorService = + context.threadPools().getPoolBuilder(pool).numCoreThreads(numThreads).build(); + } - futures.add(executorService.submit(task)); - } - }); + try { - List<Future<List<T>>> futures = new ArrayList<>(); - ++ List<Future<List<T>>> futures = new ArrayList<>(servers.size()); + for (ServerId server : servers) { + futures.add(executorService.submit(() -> serverQuery.execute(server))); + } - List<T> ret = new ArrayList<>(); - List<ActiveCompaction> ret = new ArrayList<>(futures.size()); - for (Future<List<ActiveCompaction>> future : futures) { ++ List<T> ret = new ArrayList<>(futures.size()); + for (Future<List<T>> future : futures) { try { ret.addAll(future.get()); } catch (InterruptedException | ExecutionException e) { diff --cc core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java index d26b80f761,850395d1aa..1d1e5aef0b --- a/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java +++ b/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java @@@ -358,10 -356,10 +358,10 @@@ public class Gatherer if (previousWork != null) { fileSelector = fileSelector.and(previousWork.failedFiles::contains); } - Map<String,Map<TabletFile,List<TRowRange>>> filesGBL; + Map<String,Map<StoredTabletFile,List<TRowRange>>> filesGBL; filesGBL = getFilesGroupedByLocation(fileSelector); - List<CompletableFuture<ProcessedFiles>> futures = new ArrayList<>(); + List<CompletableFuture<ProcessedFiles>> futures = new ArrayList<>(filesGBL.size() + 1); if (previousWork != null) { futures.add(CompletableFuture .completedFuture(new ProcessedFiles(previousWork.summaries, factory))); @@@ -435,15 -433,11 +435,15 @@@ public Future<SummaryCollection> processFiles(FileSystemResolver volMgr, Map<String,List<TRowRange>> files, BlockCache summaryCache, BlockCache indexCache, Cache<String,Long> fileLenCache, ExecutorService srp) { + Function<TRowRange,RowRange> fromThrift = tRowRange -> { + Text lowerBound = ByteBufferUtil.toText(tRowRange.bufferForStartRow()); + Text upperBound = ByteBufferUtil.toText(tRowRange.bufferForEndRow()); + return RowRange.range(lowerBound, false, upperBound, true); + }; - List<CompletableFuture<SummaryCollection>> futures = new ArrayList<>(); + List<CompletableFuture<SummaryCollection>> futures = new ArrayList<>(files.size()); for (Entry<String,List<TRowRange>> entry : files.entrySet()) { futures.add(CompletableFuture.supplyAsync(() -> { - List<RowRange> rrl = - entry.getValue().stream().map(RowRange::new).collect(Collectors.toList()); + List<RowRange> rrl = entry.getValue().stream().map(fromThrift).collect(Collectors.toList()); return getSummaries(volMgr, entry.getKey(), rrl, summaryCache, indexCache, fileLenCache); }, srp)); } diff --cc server/base/src/main/java/org/apache/accumulo/server/fs/FileManager.java index 1a3443fa96,4e81899ed2..d6972b90fa --- a/server/base/src/main/java/org/apache/accumulo/server/fs/FileManager.java +++ b/server/base/src/main/java/org/apache/accumulo/server/fs/FileManager.java @@@ -183,9 -187,9 +183,9 @@@ public class FileManager private List<FileSKVIterator> takeLRUOpenFiles(int numToTake) { - ArrayList<OpenReader> openReaders = new ArrayList<>(); + ArrayList<OpenReader> openReaders = new ArrayList<>(openFiles.size()); - for (Entry<String,List<OpenReader>> entry : openFiles.entrySet()) { + for (Entry<StoredTabletFile,List<OpenReader>> entry : openFiles.entrySet()) { openReaders.addAll(entry.getValue()); } @@@ -497,21 -511,19 +497,21 @@@ return newlyReservedReaders; } - public synchronized List<InterruptibleIterator> openFiles(Map<TabletFile,DataFileValue> files, - boolean detachable, SamplerConfigurationImpl samplerConfig) throws IOException { + public synchronized List<InterruptibleIterator> openFiles( + Map<StoredTabletFile,DataFileValue> files, boolean detachable, + SamplerConfigurationImpl samplerConfig) throws IOException { - Map<FileSKVIterator,String> newlyReservedReaders = openFiles( - files.keySet().stream().map(TabletFile::getPathStr).collect(Collectors.toList())); + Map<FileSKVIterator,StoredTabletFile> newlyReservedReaders = + openFiles(new ArrayList<>(files.keySet())); - ArrayList<InterruptibleIterator> iters = new ArrayList<>(); + ArrayList<InterruptibleIterator> iters = new ArrayList<>(newlyReservedReaders.size()); - boolean sawTimeSet = files.values().stream().anyMatch(DataFileValue::isTimeSet); + boolean someIteratorsWillWrap = + files.values().stream().anyMatch(DataFileValue::willWrapIterator); - for (Entry<FileSKVIterator,String> entry : newlyReservedReaders.entrySet()) { + for (Entry<FileSKVIterator,StoredTabletFile> entry : newlyReservedReaders.entrySet()) { FileSKVIterator source = entry.getKey(); - String filename = entry.getValue(); + StoredTabletFile file = entry.getValue(); InterruptibleIterator iter; if (samplerConfig != null) { diff --cc server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java index e361e4b41c,8477db3be7..828200b040 --- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java +++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java @@@ -322,9 -320,11 +322,9 @@@ public class VolumeManagerImpl implemen } @Override - public void bulkRename(Map<Path,Path> oldToNewPathMap, int poolSize, String poolName, - String transactionId) throws IOException { + public void bulkRename(Map<Path,Path> oldToNewPathMap, ExecutorService workerPool, FateId fateId) + throws IOException { - List<Future<Void>> results = new ArrayList<>(); + List<Future<Void>> results = new ArrayList<>(oldToNewPathMap.size()); - ExecutorService workerPool = ThreadPools.getServerThreadPools().getPoolBuilder(poolName) - .numCoreThreads(poolSize).build(); oldToNewPathMap.forEach((oldPath, newPath) -> results.add(workerPool.submit(() -> { boolean success; try { diff --cc server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java index 92f0c03598,c28b0a0f8f..b73eac0914 --- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java @@@ -169,10 -380,10 +168,10 @@@ public class MetadataTableUtil TabletMetadata tablet = context.getAmple().readTablet(extent, FILES, LOGS, PREV_ROW, DIR); if (tablet == null) { - throw new RuntimeException("Tablet " + extent + " not found in metadata"); + throw new IllegalStateException("Tablet " + extent + " not found in metadata"); } - result.addAll(tablet.getLogs()); + ArrayList<LogEntry> result = new ArrayList<>(tablet.getLogs()); tablet.getFilesMap().forEach(sizes::put); diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java index 56db5f2a27,d77f67301d..93a23b52e4 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/LogSorter.java @@@ -181,11 -169,12 +181,11 @@@ public class LogSorter implements Metri } @SuppressWarnings("deprecation") - Property prop = sortedLogConf.resolve(Property.TSERV_WAL_SORT_BUFFER_SIZE, - Property.TSERV_SORT_BUFFER_SIZE); - final long bufferSize = sortedLogConf.getAsBytes(prop); + final long bufferSize = sortedLogConf.getAsBytes(sortedLogConf.resolve( + Property.GENERAL_SERVER_WAL_SORT_BUFFER_SIZE, Property.TSERV_WAL_SORT_BUFFER_SIZE)); Thread.currentThread().setName("Sorting " + name + " for recovery"); while (true) { - final ArrayList<Pair<LogFileKey,LogFileValue>> buffer = new ArrayList<>(); + final ArrayList<Pair<LogFileKey,LogFileValue>> buffer = new ArrayList<>(512); try { long start = input.getPos(); while (input.getPos() - start < bufferSize) { @@@ -334,11 -303,11 +335,11 @@@ } public List<RecoveryStatus> getLogSorts() { - List<RecoveryStatus> result = new ArrayList<>(); synchronized (currentWork) { + List<RecoveryStatus> result = new ArrayList<>(currentWork.size()); for (Entry<String,LogProcessor> entries : currentWork.entrySet()) { RecoveryStatus status = new RecoveryStatus(); - status.name = entries.getKey(); + status.setName(entries.getKey()); try { double progress = entries.getValue().getBytesCopied() / walBlockSize; // to be sure progress does not exceed 100% diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/session/SessionManager.java index 38a391a0c5,830ab48749..2adb61ba7a --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/session/SessionManager.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/session/SessionManager.java @@@ -417,11 -445,23 +417,12 @@@ public class SessionManager public List<ActiveScan> getActiveScans() { - final List<ActiveScan> activeScans = new ArrayList<>(); + final List<ActiveScan> activeScans = + new ArrayList<>(sessions.size() + deferredCleanupQueue.size()); final long ct = System.currentTimeMillis(); - final Set<Entry<Long,Session>> copiedIdleSessions = new HashSet<>(); - - /** - * Add sessions that get the list returned in the active scans call - */ - for (Session session : deferredCleanupQueue) { - copiedIdleSessions.add(Maps.immutableEntry(expiredSessionMarker, session)); - } - - List.of(sessions.entrySet(), copiedIdleSessions).forEach(s -> s.forEach(entry -> { - Session session = entry.getValue(); - if (session instanceof ScanSession) { - ScanSession<?> scanSession = (ScanSession<?>) session; + Stream.concat(sessions.values().stream(), deferredCleanupQueue.stream()).forEach(session -> { + if (session instanceof ScanSession<?> scanSession) { boolean isSingle = session instanceof SingleScanSession; addActiveScan(activeScans, scanSession, diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java index 73ed820f05,54877265cc..64f7a4f302 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java @@@ -1249,12 -1997,16 +1249,12 @@@ public class Tablet extends TabletBase } } - ReentrantLock getLogLock() { - return logLock; - } - - Set<String> beginClearingUnusedLogs() { + Set<LogEntry> beginClearingUnusedLogs() { Preconditions.checkState(logLock.isHeldByCurrentThread()); - Set<String> unusedLogs = new HashSet<>(); + Set<LogEntry> unusedLogs = new HashSet<>(); - ArrayList<LogEntry> otherLogsCopy = new ArrayList<>(); - ArrayList<LogEntry> currentLogsCopy = new ArrayList<>(); - ArrayList<String> otherLogsCopy = new ArrayList<>(otherLogs.size()); - ArrayList<String> currentLogsCopy = new ArrayList<>(currentLogs.size()); ++ ArrayList<LogEntry> otherLogsCopy = new ArrayList<>(otherLogs.size()); ++ ArrayList<LogEntry> currentLogsCopy = new ArrayList<>(currentLogs.size()); synchronized (this) { if (removingLogs) {
