From f621fccd710b8bbdaeab4f1b04a0b444fb8497b6 Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Mon, 4 May 2026 10:30:03 -0500 Subject: [PATCH 1/5] Back porting improves tablet load times for tablet w/ walogs and no data #4873 This commit makes two major changes. First it changed log recovery to use block caches. Second it checks if a tablet has any data in walogs before acquiring the recovery lock. These two changes together really speed up loading tablets that have no data in walogs. These changes introduce an extra opening of the walogs to see if the recovery lock needs to be acquired. Using the block caches for this extra opening should avoid any extra cost. The block caches also help in the case where many tablets with the same walogs are assigned to a tablet server. In some simple test saw an 8x speedup in tablet load times. Anytime a tablet has an unclean shutdown it will have the walogs of the dead tserver assigned to it even if had no data in those walogs. These change make loading tablets in that situation much faster. --- .../accumulo/tserver/AssignmentHandler.java | 73 +++++++++---------- .../apache/accumulo/tserver/TabletServer.java | 25 ++++++- .../tserver/log/RecoveryLogsIterator.java | 42 ++++++++--- .../tserver/log/SortedLogRecovery.java | 27 +++++-- .../tserver/log/TabletServerLogger.java | 39 +++++++++- .../tserver/log/SortedLogRecoveryTest.java | 19 ++++- 6 files changed, 170 insertions(+), 55 deletions(-) diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java index e6154e1392e..a0612b8891b 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/AssignmentHandler.java @@ -157,44 +157,45 @@ public void run() { boolean successful = false; try { - server.acquireRecoveryMemory(extent); - - TabletResourceManager trm = server.resourceManager.createTabletResourceManager(extent, - server.getTableConfiguration(extent)); - TabletData data = new TabletData(tabletMetadata); - - tablet = new Tablet(server, extent, trm, data); - // If a minor compaction starts after a tablet opens, this indicates a log recovery - // occurred. This recovered data must be minor compacted. - // There are three reasons to wait for this minor compaction to finish before placing the - // tablet in online tablets. - // - // 1) The log recovery code does not handle data written to the tablet on multiple tablet - // servers. - // 2) The log recovery code does not block if memory is full. Therefore recovering lots of - // tablets that use a lot of memory could run out of memory. - // 3) The minor compaction finish event did not make it to the logs (the file will be in - // metadata, preventing replay of compacted data)... but do not - // want a majc to wipe the file out from metadata and then have another process failure... - // this could cause duplicate data to replay. - if (tablet.getNumEntriesInMemory() > 0 - && !tablet.minorCompactNow(MinorCompactionReason.RECOVERY)) { - throw new RuntimeException("Minor compaction after recovery fails for " + extent); - } - Assignment assignment = - new Assignment(extent, server.getTabletSession(), tabletMetadata.getLast()); - TabletStateStore.setLocation(server.getContext(), assignment); + try (var recoveryLock = server.acquireRecoveryMemory(tabletMetadata)) { + + TabletResourceManager trm = server.resourceManager.createTabletResourceManager(extent, + server.getTableConfiguration(extent)); + TabletData data = new TabletData(tabletMetadata); + + tablet = new Tablet(server, extent, trm, data); + // If a minor compaction starts after a tablet opens, this indicates a log recovery + // occurred. This recovered data must be minor compacted. + // There are three reasons to wait for this minor compaction to finish before placing the + // tablet in online tablets. + // + // 1) The log recovery code does not handle data written to the tablet on multiple tablet + // servers. + // 2) The log recovery code does not block if memory is full. Therefore recovering lots of + // tablets that use a lot of memory could run out of memory. + // 3) The minor compaction finish event did not make it to the logs (the file will be in + // metadata, preventing replay of compacted data)... but do not + // want a majc to wipe the file out from metadata and then have another process failure... + // this could cause duplicate data to replay. + if (tablet.getNumEntriesInMemory() > 0 + && !tablet.minorCompactNow(MinorCompactionReason.RECOVERY)) { + throw new RuntimeException("Minor compaction after recovery fails for " + extent); + } + Assignment assignment = + new Assignment(extent, server.getTabletSession(), tabletMetadata.getLast()); + TabletStateStore.setLocation(server.getContext(), assignment); - synchronized (server.openingTablets) { - synchronized (server.onlineTablets) { - server.openingTablets.remove(extent); - server.onlineTablets.put(extent, tablet); - server.openingTablets.notifyAll(); - server.recentlyUnloadedCache.remove(tablet.getExtent()); + synchronized (server.openingTablets) { + synchronized (server.onlineTablets) { + server.openingTablets.remove(extent); + server.onlineTablets.put(extent, tablet); + server.openingTablets.notifyAll(); + server.recentlyUnloadedCache.remove(tablet.getExtent()); + } } + tablet = null; // release this reference + successful = true; } - tablet = null; // release this reference - successful = true; } catch (Exception e) { log.warn("exception trying to assign tablet {} {}", extent, locationToOpen, e); @@ -205,8 +206,6 @@ public void run() { TableId tableId = extent.tableId(); ProblemReports.getInstance(server.getContext()).report(new ProblemReport(tableId, TABLET_LOAD, extent.getUUID().toString(), server.getClientAddressString(), e)); - } finally { - server.releaseRecoveryMemory(extent); } if (successful) { diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java index ecb26871809..8e0fc6f09d1 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java @@ -29,6 +29,7 @@ import static org.apache.accumulo.core.util.threads.ThreadPools.watchNonCriticalScheduledTask; import java.io.IOException; +import java.io.UncheckedIOException; import java.lang.management.ManagementFactory; import java.net.UnknownHostException; import java.security.SecureRandom; @@ -89,6 +90,7 @@ import org.apache.accumulo.core.metadata.MetadataTable; import org.apache.accumulo.core.metadata.RootTable; import org.apache.accumulo.core.metadata.TServerInstance; +import org.apache.accumulo.core.metadata.schema.TabletMetadata; import org.apache.accumulo.core.metadata.schema.TabletsMetadata; import org.apache.accumulo.core.metrics.MetricsInfo; import org.apache.accumulo.core.process.thrift.ServerProcessService; @@ -547,9 +549,14 @@ public void enqueueManagerMessage(ManagerMessage m) { managerMessages.addLast(m); } - void acquireRecoveryMemory(KeyExtent extent) { - if (!extent.isMeta()) { + private static final AutoCloseable NOOP_CLOSEABLE = () -> {}; + + AutoCloseable acquireRecoveryMemory(TabletMetadata tabletMetadata) { + if (tabletMetadata.getExtent().isMeta() || !needsRecovery(tabletMetadata)) { + return NOOP_CLOSEABLE; + } else { recoveryLock.lock(); + return () -> recoveryLock.unlock(); } } @@ -559,6 +566,20 @@ void releaseRecoveryMemory(KeyExtent extent) { } } + public boolean needsRecovery(TabletMetadata tabletMetadata) { + var logEntries = tabletMetadata.getLogs(); + + if (logEntries.isEmpty()) { + return false; + } + + try { + return logger.needsRecovery(getContext(), tabletMetadata.getExtent(), logEntries); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + private ServerAddress startServer(String address, TProcessor processor) throws UnknownHostException { @SuppressWarnings("deprecation") diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java index b56612314a6..45cd336305c 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java @@ -34,6 +34,7 @@ import org.apache.accumulo.core.data.Value; import org.apache.accumulo.core.file.FileOperations; import org.apache.accumulo.core.file.FileSKVIterator; +import org.apache.accumulo.core.file.blockfile.impl.CacheProvider; import org.apache.accumulo.core.iterators.IteratorAdapter; import org.apache.accumulo.core.spi.crypto.CryptoEnvironment; import org.apache.accumulo.core.spi.crypto.CryptoService; @@ -49,6 +50,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.common.cache.Cache; import com.google.common.collect.Iterators; /** @@ -63,11 +65,17 @@ public class RecoveryLogsIterator private final Iterator> iter; private final CryptoEnvironment env = new CryptoEnvironmentImpl(CryptoEnvironment.Scope.RECOVERY); + public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, LogFileKey start, + LogFileKey end, boolean checkFirstKey) throws IOException { + this(context, recoveryLogDirs, start, end, checkFirstKey, null, null); + } + /** * Scans the files in each recoveryLogDir over the range [start,end]. */ public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, LogFileKey start, - LogFileKey end, boolean checkFirstKey) throws IOException { + LogFileKey end, boolean checkFirstKey, Cache fileLenCache, + CacheProvider cacheProvider) throws IOException { List>> iterators = new ArrayList<>(recoveryLogDirs.size()); fileIters = new ArrayList<>(); @@ -84,13 +92,12 @@ public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, L // only check the first key once to prevent extra iterator creation and seeking if (checkFirstKey && !logFiles.isEmpty()) { - validateFirstKey(context, cryptoService, fs, logFiles, logDir); + validateFirstKey(context, cryptoService, fs, logFiles, logDir, fileLenCache, cacheProvider); } for (Entry entry : logFiles.entrySet()) { - FileSKVIterator fileIter = FileOperations.getInstance().newReaderBuilder() - .forFile(entry.getKey().toString(), fs, fs.getConf(), cryptoService, entry.getValue()) - .withTableConfiguration(context.getConfiguration()).seekToBeginning().build(); + FileSKVIterator fileIter = + openLogFile(context, entry.getKey(), cryptoService, fs, fileLenCache, cacheProvider); if (range != null) { fileIter.seek(range, Collections.emptySet(), false); } @@ -135,6 +142,23 @@ public void close() throws IOException { } } + FileSKVIterator openLogFile(ServerContext context, Path logFile, CryptoService cs, FileSystem fs, + Cache fileLenCache, CacheProvider cacheProvider) throws IOException { + var builder = FileOperations.getInstance().newReaderBuilder() + .forFile(logFile.toString(), fs, fs.getConf(), cs) + .withTableConfiguration(context.getConfiguration()); + + if (fileLenCache != null) { + builder = builder.withFileLenCache(fileLenCache); + } + + if (cacheProvider != null) { + builder = builder.withCacheProvider(cacheProvider); + } + + return builder.seekToBeginning().build(); + } + /** * Check for sorting signal files (finished/failed) and get the logs in the provided directory. */ @@ -170,11 +194,11 @@ private TreeMap getFiles(VolumeManager fs, Path directory) thro * Check that the first entry in the WAL is OPEN. Only need to do this once. */ private void validateFirstKey(ServerContext context, CryptoService cs, FileSystem fs, - TreeMap logFiles, Path fullLogPath) throws IOException { + TreeMap logFiles, Path fullLogPath, Cache fileLenCache, + CacheProvider cacheProvider) throws IOException { Entry first = logFiles.firstEntry(); - try (FileSKVIterator fileIter = FileOperations.getInstance().newReaderBuilder() - .forFile(first.getKey().toString(), fs, fs.getConf(), cs, first.getValue()) - .withTableConfiguration(context.getConfiguration()).seekToBeginning().build()) { + try (FileSKVIterator fileIter = + openLogFile(context, first.getKey(), cs, fs, fileLenCache, cacheProvider)) { Iterator> iterator = new IteratorAdapter(fileIter); if (iterator.hasNext()) { diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java index 40e8496ba79..99470c21565 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java @@ -42,6 +42,7 @@ import org.apache.accumulo.core.data.Mutation; import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.file.blockfile.impl.CacheProvider; import org.apache.accumulo.core.metadata.RootTable; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.tserver.logger.LogEvents; @@ -51,6 +52,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.common.cache.Cache; import com.google.common.collect.Collections2; import com.google.common.collect.Iterators; import com.google.common.collect.PeekingIterator; @@ -65,8 +67,21 @@ public class SortedLogRecovery { private final ServerContext context; - public SortedLogRecovery(ServerContext context) { + private final CacheProvider cacheProvider; + + private final Cache fileLenCache; + + public SortedLogRecovery(ServerContext context, Cache fileLenCache, + CacheProvider cacheProvider) { this.context = context; + this.cacheProvider = cacheProvider; + this.fileLenCache = fileLenCache; + } + + public boolean needsRecovery(KeyExtent extent, List recoveryDirs) throws IOException { + Entry> maxEntry = findLogsThatDefineTablet(extent, recoveryDirs); + int tabletId = maxEntry.getKey(); + return tabletId != -1; } static LogFileKey maxKey(LogEvents event) { @@ -104,7 +119,7 @@ private int findMaxTabletId(KeyExtent extent, List recoveryLogDirs) throws int tabletId = -1; try (var rli = new RecoveryLogsIterator(context, recoveryLogDirs, minKey(DEFINE_TABLET), - maxKey(DEFINE_TABLET), true)) { + maxKey(DEFINE_TABLET), true, fileLenCache, cacheProvider)) { KeyExtent alternative = extent; if (extent.isRootTablet()) { @@ -207,8 +222,9 @@ private long findRecoverySeq(List recoveryLogs, Set tabletFiles, i long lastFinish = 0; long recoverySeq = 0; - try (RecoveryLogsIterator rli = new RecoveryLogsIterator(context, recoveryLogs, - minKey(COMPACTION_START, tabletId), maxKey(COMPACTION_START, tabletId), false)) { + try (RecoveryLogsIterator rli = + new RecoveryLogsIterator(context, recoveryLogs, minKey(COMPACTION_START, tabletId), + maxKey(COMPACTION_START, tabletId), false, fileLenCache, cacheProvider)) { DeduplicatingIterator ddi = new DeduplicatingIterator(rli); @@ -265,7 +281,8 @@ private void playbackMutations(List recoveryLogs, MutationReceiver mr, int LogFileKey end = maxKey(MUTATION, tabletId); - try (var rli = new RecoveryLogsIterator(context, recoveryLogs, start, end, false)) { + try (var rli = new RecoveryLogsIterator(context, recoveryLogs, start, end, false, fileLenCache, + cacheProvider)) { while (rli.hasNext()) { Entry entry = rli.next(); LogFileKey logFileKey = entry.getKey(); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java index caa6ace170c..3b1eca42bc3 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.nio.channels.ClosedChannelException; +import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; @@ -40,7 +41,12 @@ import org.apache.accumulo.core.data.Mutation; import org.apache.accumulo.core.dataImpl.KeyExtent; import org.apache.accumulo.core.fate.zookeeper.ServiceLock; +import org.apache.accumulo.core.file.blockfile.impl.BasicCacheProvider; +import org.apache.accumulo.core.file.blockfile.impl.CacheProvider; +import org.apache.accumulo.core.logging.LoggingBlockCache; import org.apache.accumulo.core.protobuf.ProtobufUtil; +import org.apache.accumulo.core.spi.cache.CacheType; +import org.apache.accumulo.core.tabletserver.log.LogEntry; import org.apache.accumulo.core.util.Halt; import org.apache.accumulo.core.util.Retry; import org.apache.accumulo.core.util.Retry.RetryFactory; @@ -51,6 +57,7 @@ import org.apache.accumulo.server.util.ReplicationTableUtil; import org.apache.accumulo.tserver.TabletMutations; import org.apache.accumulo.tserver.TabletServer; +import org.apache.accumulo.tserver.TabletServerResourceManager; import org.apache.accumulo.tserver.log.DfsLogger.LoggerOperation; import org.apache.accumulo.tserver.log.DfsLogger.ServerResources; import org.apache.accumulo.tserver.tablet.CommitSession; @@ -558,10 +565,40 @@ public long minorCompactionStarted(final CommitSession commitSession, final long return seq; } + private List resolve(Collection walogs) { + List sortedLogs = new ArrayList<>(walogs.size()); + for (var logEntry : walogs) { + sortedLogs.add(new Path(logEntry.filename)); + } + return sortedLogs; + } + + private CacheProvider createCacheProvider(TabletServerResourceManager resourceMgr) { + return new BasicCacheProvider( + LoggingBlockCache.wrap(CacheType.INDEX, resourceMgr.getIndexCache()), + LoggingBlockCache.wrap(CacheType.DATA, resourceMgr.getDataCache())); + } + + public boolean needsRecovery(ServerContext context, KeyExtent extent, Collection walogs) + throws IOException { + try { + var resourceMgr = tserver.getResourceManager(); + var cacheProvider = createCacheProvider(resourceMgr); + SortedLogRecovery recovery = + new SortedLogRecovery(context, resourceMgr.getFileLenCache(), cacheProvider); + return recovery.needsRecovery(extent, resolve(walogs)); + } catch (Exception e) { + throw new IOException(e); + } + } + public void recover(ServerContext context, KeyExtent extent, List recoveryDirs, Set tabletFiles, MutationReceiver mr) throws IOException { try { - SortedLogRecovery recovery = new SortedLogRecovery(context); + var resourceMgr = tserver.getResourceManager(); + var cacheProvider = createCacheProvider(resourceMgr); + SortedLogRecovery recovery = + new SortedLogRecovery(context, resourceMgr.getFileLenCache(), cacheProvider); recovery.recover(extent, recoveryDirs, tabletFiles, mr); } catch (Exception e) { throw new IOException(e); diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java index 1e4a79c71c6..9450e9018e9 100644 --- a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java @@ -51,10 +51,15 @@ import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.core.data.Value; import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.file.blockfile.cache.impl.BlockCacheConfiguration; +import org.apache.accumulo.core.file.blockfile.cache.tinylfu.TinyLfuBlockCache; +import org.apache.accumulo.core.file.blockfile.impl.BasicCacheProvider; +import org.apache.accumulo.core.file.blockfile.impl.CacheProvider; import org.apache.accumulo.core.file.rfile.bcfile.Compression; import org.apache.accumulo.core.file.rfile.bcfile.CompressionAlgorithm; import org.apache.accumulo.core.file.rfile.bcfile.Utils; import org.apache.accumulo.core.file.streams.SeekableDataInputStream; +import org.apache.accumulo.core.spi.cache.CacheType; import org.apache.accumulo.core.spi.crypto.CryptoServiceFactory; import org.apache.accumulo.core.spi.crypto.GenericCryptoServiceFactory; import org.apache.accumulo.core.util.Pair; @@ -75,6 +80,9 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; + import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; @SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "paths not set by user input") @@ -165,6 +173,15 @@ private List recover(Map logs, KeyExtent extent) th private List recover(Map logs, Set files, KeyExtent extent, int bufferSize) throws IOException { + CacheProvider cacheProvider = new BasicCacheProvider( + new TinyLfuBlockCache( + BlockCacheConfiguration.forTabletServer(DefaultConfiguration.getInstance()), + CacheType.INDEX), + new TinyLfuBlockCache( + BlockCacheConfiguration.forTabletServer(DefaultConfiguration.getInstance()), + CacheType.DATA)); + Cache fileLenCache = CacheBuilder.newBuilder().build(); + final String workdir = new File(tempDir, testName()).getAbsolutePath(); try (var fs = VolumeManagerImpl.getLocalForTesting(workdir)) { CryptoServiceFactory cryptoFactory = new GenericCryptoServiceFactory(); @@ -198,7 +215,7 @@ private List recover(Map logs, Set files, K dirs.add(new Path(destPath)); } // Recover - SortedLogRecovery recovery = new SortedLogRecovery(context); + SortedLogRecovery recovery = new SortedLogRecovery(context, fileLenCache, cacheProvider); CaptureMutations capture = new CaptureMutations(); recovery.recover(extent, dirs, files, capture); verify(context); From 6970262c28ec5b4e22faadbe736046f7a3556c2c Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Tue, 5 May 2026 09:17:34 -0500 Subject: [PATCH 2/5] Back porting switches volumes prior to recovery check (#4889) In #4873 a check was added to inspect walogs during tablet load to see if they had any data for the tablet. This check happens prior to volume replacement that also runs during tablet load. Therefore if volume replacement is needed for the walogs then this check will fail because it can not find the files and the tablet will fail to load. To fix this problem modified the new check to switch volumes if needed prior to running the check. --- .../apache/accumulo/server/fs/VolumeUtil.java | 4 ++++ .../apache/accumulo/tserver/TabletServer.java | 19 ++++++++++++++++++- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java index 35e1fbe1bc1..7fd7b98bf19 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java @@ -89,6 +89,10 @@ public static Path switchVolume(String path, FileType ft, List> return null; } + public static LogEntry switchVolume(LogEntry le, List> replacements) { + return switchVolumes(le, replacements); + } + protected static LogEntry switchVolumes(LogEntry le, List> replacements) { Path switchedPath = switchVolume(le.filename, FileType.WAL, replacements); String switchedString; diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java index 8e0fc6f09d1..eeecc56573a 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java @@ -121,6 +121,7 @@ import org.apache.accumulo.server.conf.TableConfiguration; import org.apache.accumulo.server.fs.VolumeChooserEnvironmentImpl; import org.apache.accumulo.server.fs.VolumeManager; +import org.apache.accumulo.server.fs.VolumeUtil; import org.apache.accumulo.server.log.SortedLogState; import org.apache.accumulo.server.log.WalStateManager; import org.apache.accumulo.server.log.WalStateManager.WalMarkerException; @@ -573,8 +574,24 @@ public boolean needsRecovery(TabletMetadata tabletMetadata) { return false; } + // This method is called prior to volumes being switched for a tablet during the load process, + // so switch volumes before calling needsRecovery() + var switchedLogEntries = new ArrayList(logEntries.size()); + + for (LogEntry logEntry : logEntries) { + var switchedWalog = VolumeUtil.switchVolume(logEntry, context.getVolumeReplacements()); + LogEntry walog; + if (switchedWalog != null) { + log.debug("Volume switched for needsRecovery {} -> {}", logEntry, switchedWalog); + walog = switchedWalog; + } else { + walog = logEntry; + } + switchedLogEntries.add(walog); + } + try { - return logger.needsRecovery(getContext(), tabletMetadata.getExtent(), logEntries); + return logger.needsRecovery(getContext(), tabletMetadata.getExtent(), switchedLogEntries); } catch (IOException e) { throw new UncheckedIOException(e); } From 6a547c2bc0295af865ac0d7729c33c09a709010d Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Tue, 5 May 2026 10:31:55 -0500 Subject: [PATCH 3/5] Back porting Avoids listing the sorted logs dir multiple times during log recovery. (#4874) The log recovery code would list the sorted walog files multiple times during recovery. These changes modify the code to only list the files once. Also the listing is cached for a short period of time to improve the case of multiple tablet referencing the same walogs. This along with #4873 should result in much less traffic to the namenode when an entire accumulo cluster shutsdown and needs to recover. --- .../apache/accumulo/tserver/TabletServer.java | 13 +- .../tserver/log/RecoveryLogsIterator.java | 22 +-- .../tserver/log/ResolvedSortedLog.java | 151 ++++++++++++++++++ .../tserver/log/SortedLogRecovery.java | 42 ++--- .../tserver/log/TabletServerLogger.java | 25 ++- .../accumulo/tserver/logger/LogReader.java | 6 +- .../tserver/log/RecoveryLogsIteratorTest.java | 25 ++- .../tserver/log/SortedLogRecoveryTest.java | 7 +- 8 files changed, 239 insertions(+), 52 deletions(-) create mode 100644 server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java index eeecc56573a..2110570d40a 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java @@ -1288,24 +1288,21 @@ public void minorCompactionStarted(CommitSession tablet, long lastUpdateSequence public void recover(VolumeManager fs, KeyExtent extent, List logEntries, Set tabletFiles, MutationReceiver mutationReceiver) throws IOException { - List recoveryDirs = new ArrayList<>(); List sorted = new ArrayList<>(logEntries); sorted.sort((e1, e2) -> (int) (e1.timestamp - e2.timestamp)); + + // Validate that recovery logs exist before attempting recovery for (LogEntry entry : sorted) { - Path recovery = null; Path finished = RecoveryPath.getRecoveryPath(new Path(entry.filename)); finished = SortedLogState.getFinishedMarkerPath(finished); TabletServer.log.debug("Looking for " + finished); - if (fs.exists(finished)) { - recovery = finished.getParent(); - } - if (recovery == null) { + if (!fs.exists(finished)) { throw new IOException( "Unable to find recovery files for extent " + extent + " logEntry: " + entry); } - recoveryDirs.add(recovery); } - logger.recover(getContext(), extent, recoveryDirs, tabletFiles, mutationReceiver); + + logger.recover(getContext(), extent, sorted, tabletFiles, mutationReceiver); } public int createLogId() { diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java index 45cd336305c..02ea01cae58 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java @@ -65,16 +65,16 @@ public class RecoveryLogsIterator private final Iterator> iter; private final CryptoEnvironment env = new CryptoEnvironmentImpl(CryptoEnvironment.Scope.RECOVERY); - public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, LogFileKey start, - LogFileKey end, boolean checkFirstKey) throws IOException { + public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, + LogFileKey start, LogFileKey end, boolean checkFirstKey) throws IOException { this(context, recoveryLogDirs, start, end, checkFirstKey, null, null); } /** * Scans the files in each recoveryLogDir over the range [start,end]. */ - public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, LogFileKey start, - LogFileKey end, boolean checkFirstKey, Cache fileLenCache, + public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, + LogFileKey start, LogFileKey end, boolean checkFirstKey, Cache fileLenCache, CacheProvider cacheProvider) throws IOException { List>> iterators = new ArrayList<>(recoveryLogDirs.size()); @@ -85,14 +85,15 @@ public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, L final CryptoService cryptoService = context.getCryptoFactory().getService(env, context.getConfiguration().getAllCryptoProperties()); - for (Path logDir : recoveryLogDirs) { - LOG.debug("Opening recovery log dir {}", logDir.getName()); + for (ResolvedSortedLog logDir : recoveryLogDirs) { + LOG.debug("Opening recovery log dir {}", logDir.getDir().getName()); TreeMap logFiles = getFiles(vm, logDir); - var fs = vm.getFileSystemByPath(logDir); + var fs = vm.getFileSystemByPath(logDir.getDir()); // only check the first key once to prevent extra iterator creation and seeking if (checkFirstKey && !logFiles.isEmpty()) { - validateFirstKey(context, cryptoService, fs, logFiles, logDir, fileLenCache, cacheProvider); + validateFirstKey(context, cryptoService, fs, logFiles, logDir.getDir(), fileLenCache, + cacheProvider); } for (Entry entry : logFiles.entrySet()) { @@ -162,13 +163,14 @@ FileSKVIterator openLogFile(ServerContext context, Path logFile, CryptoService c /** * Check for sorting signal files (finished/failed) and get the logs in the provided directory. */ - private TreeMap getFiles(VolumeManager fs, Path directory) throws IOException { + private TreeMap getFiles(VolumeManager fs, ResolvedSortedLog directory) + throws IOException { boolean foundFinish = false; // Path::getName compares the last component of each Path value. In this case, the last // component should // always have the format 'part-r-XXXXX.rf', where XXXXX are one-up values. TreeMap logFiles = new TreeMap<>(Comparator.comparing(Path::getName)); - for (FileStatus child : fs.listStatus(directory)) { + for (FileStatus child : fs.listStatus(directory.getDir())) { if (child.getPath().getName().startsWith("_")) { continue; } diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java new file mode 100644 index 00000000000..06f23cd120f --- /dev/null +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java @@ -0,0 +1,151 @@ +/* + * 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 + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.accumulo.tserver.log; + +import java.io.IOException; +import java.util.Collections; +import java.util.Comparator; +import java.util.SortedSet; +import java.util.TreeSet; + +import org.apache.accumulo.core.tabletserver.log.LogEntry; +import org.apache.accumulo.server.fs.VolumeManager; +import org.apache.accumulo.server.log.SortedLogState; +import org.apache.accumulo.server.manager.recovery.RecoveryPath; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; + +/** + * Write ahead logs have two paths in DFS. There is the path of the original unsorted walog and the + * path of the sorted walog. The purpose of this class is to convert the unsorted wal path to a + * sorted wal path and validate the sorted dir exists and is finished. + */ +public class ResolvedSortedLog { + + private final SortedSet children; + private final LogEntry origin; + private final Path sortedLogDir; + + private ResolvedSortedLog(LogEntry origin, Path sortedLogDir, SortedSet children) { + this.origin = origin; + this.sortedLogDir = sortedLogDir; + this.children = Collections.unmodifiableSortedSet(children); + } + + /** + * @return the unsorted walog path from which this was created. + */ + public LogEntry getOrigin() { + return origin; + } + + /** + * @return the path of the directory in which sorted logs are stored + */ + public Path getDir() { + return sortedLogDir; + } + + /** + * @return When an unsorted walog is sorted the sorted data is stored in one or more rfiles, this + * returns the paths of those rfiles. + */ + public SortedSet getChildren() { + return children; + } + + @Override + public String toString() { + return sortedLogDir.toString(); + } + + /** + * For a given path of an unsorted walog check to see if the corresponding sorted log dir exists + * and is finished. If it is return an immutable object containing information about the sorted + * walogs. + */ + public static ResolvedSortedLog resolve(LogEntry logEntry, VolumeManager fs) throws IOException { + + // convert the path of an unsorted log to the expected path for the corresponding sorted log + // dir + Path sortedLogPath = RecoveryPath.getRecoveryPath(new Path(logEntry.filename)); + + boolean foundFinish = false; + // Path::getName compares the last component of each Path value. In this case, the last + // component should + // always have the format 'part-r-XXXXX.rf', where XXXXX are one-up values. + SortedSet logFiles = new TreeSet<>(Comparator.comparing(Path::getName)); + for (FileStatus child : fs.listStatus(sortedLogPath)) { + if (child.getPath().getName().startsWith("_")) { + continue; + } + if (SortedLogState.isFinished(child.getPath().getName())) { + foundFinish = true; + continue; + } + if (SortedLogState.FAILED.getMarker().equals(child.getPath().getName())) { + continue; + } + FileSystem ns = fs.getFileSystemByPath(child.getPath()); + Path fullLogPath = ns.makeQualified(child.getPath()); + logFiles.add(fullLogPath); + } + if (!foundFinish) { + throw new IOException("Sort '" + SortedLogState.FINISHED.getMarker() + "' flag not found in " + + sortedLogPath + " for walog " + logEntry.filename); + } + + return new ResolvedSortedLog(logEntry, sortedLogPath, logFiles); + } + + /** + * Create a ResolvedSortedLog directly from a sorted log directory path. This is useful for + * diagnostic tools that operate directly on sorted recovery logs without going through the normal + * recovery flow with LogEntry objects. + */ + public static ResolvedSortedLog fromSortedLogDir(Path sortedLogDir, VolumeManager fs) + throws IOException { + boolean foundFinish = false; + SortedSet logFiles = new TreeSet<>(Comparator.comparing(Path::getName)); + for (FileStatus child : fs.listStatus(sortedLogDir)) { + if (child.getPath().getName().startsWith("_")) { + continue; + } + if (SortedLogState.isFinished(child.getPath().getName())) { + foundFinish = true; + continue; + } + if (SortedLogState.FAILED.getMarker().equals(child.getPath().getName())) { + continue; + } + FileSystem ns = fs.getFileSystemByPath(child.getPath()); + Path fullLogPath = ns.makeQualified(child.getPath()); + logFiles.add(fullLogPath); + } + if (!foundFinish) { + throw new IOException( + "Sort '" + SortedLogState.FINISHED.getMarker() + "' flag not found in " + sortedLogDir); + } + + // Create a dummy LogEntry for the origin (used only for diagnostics) + LogEntry dummyOrigin = new LogEntry(null, 0, sortedLogDir.toString()); + return new ResolvedSortedLog(dummyOrigin, sortedLogDir, logFiles); + } +} diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java index 99470c21565..71210de1203 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java @@ -78,8 +78,10 @@ public SortedLogRecovery(ServerContext context, Cache fileLenCache, this.fileLenCache = fileLenCache; } - public boolean needsRecovery(KeyExtent extent, List recoveryDirs) throws IOException { - Entry> maxEntry = findLogsThatDefineTablet(extent, recoveryDirs); + public boolean needsRecovery(KeyExtent extent, List recoveryDirs) + throws IOException { + Entry> maxEntry = + findLogsThatDefineTablet(extent, recoveryDirs); int tabletId = maxEntry.getKey(); return tabletId != -1; } @@ -115,7 +117,8 @@ static LogFileKey minKey(LogEvents event, int tabletId) { return key; } - private int findMaxTabletId(KeyExtent extent, List recoveryLogDirs) throws IOException { + private int findMaxTabletId(KeyExtent extent, List recoveryLogDirs) + throws IOException { int tabletId = -1; try (var rli = new RecoveryLogsIterator(context, recoveryLogDirs, minKey(DEFINE_TABLET), @@ -155,18 +158,18 @@ private int findMaxTabletId(KeyExtent extent, List recoveryLogDirs) throws * @return The maximum tablet ID observed AND the list of logs that contained the maximum tablet * ID. */ - private Entry> findLogsThatDefineTablet(KeyExtent extent, - List recoveryDirs) throws IOException { - Map> logsThatDefineTablet = new HashMap<>(); + private Entry> findLogsThatDefineTablet(KeyExtent extent, + List recoveryDirs) throws IOException { + Map> logsThatDefineTablet = new HashMap<>(); - for (Path walDir : recoveryDirs) { + for (ResolvedSortedLog walDir : recoveryDirs) { int tabletId = findMaxTabletId(extent, Collections.singletonList(walDir)); if (tabletId == -1) { - log.debug("Did not find tablet {} in recovery log {}", extent, walDir.getName()); + log.debug("Did not find tablet {} in recovery log {}", extent, walDir.getDir().getName()); } else { logsThatDefineTablet.computeIfAbsent(tabletId, k -> new ArrayList<>()).add(walDir); log.debug("Found tablet {} with id {} in recovery log {}", extent, tabletId, - walDir.getName()); + walDir.getDir().getName()); } } @@ -211,8 +214,8 @@ public Entry next() { } - private long findRecoverySeq(List recoveryLogs, Set tabletFiles, int tabletId) - throws IOException { + private long findRecoverySeq(List recoveryLogs, Set tabletFiles, + int tabletId) throws IOException { HashSet suffixes = new HashSet<>(); for (String path : tabletFiles) { suffixes.add(getPathSuffix(path)); @@ -274,8 +277,8 @@ private long findRecoverySeq(List recoveryLogs, Set tabletFiles, i return recoverySeq; } - private void playbackMutations(List recoveryLogs, MutationReceiver mr, int tabletId, - long recoverySeq) throws IOException { + private void playbackMutations(List recoveryLogs, MutationReceiver mr, + int tabletId, long recoverySeq) throws IOException { LogFileKey start = minKey(MUTATION, tabletId); start.setSeq(recoverySeq); @@ -303,20 +306,21 @@ private void playbackMutations(List recoveryLogs, MutationReceiver mr, int } } - Collection asNames(List recoveryLogs) { - return Collections2.transform(recoveryLogs, Path::getName); + Collection asNames(List recoveryLogs) { + return Collections2.transform(recoveryLogs, rl -> rl.getDir().getName()); } - public void recover(KeyExtent extent, List recoveryDirs, Set tabletFiles, - MutationReceiver mr) throws IOException { + public void recover(KeyExtent extent, List recoveryDirs, + Set tabletFiles, MutationReceiver mr) throws IOException { - Entry> maxEntry = findLogsThatDefineTablet(extent, recoveryDirs); + Entry> maxEntry = + findLogsThatDefineTablet(extent, recoveryDirs); // A tablet may leave a tserver and then come back, in which case it would have a different and // higher tablet id. Only want to consider events in the log related to the last time the tablet // was loaded. int tabletId = maxEntry.getKey(); - List logsThatDefineTablet = maxEntry.getValue(); + List logsThatDefineTablet = maxEntry.getValue(); if (tabletId == -1) { log.info("Tablet {} is not defined in recovery logs {} ", extent, asNames(recoveryDirs)); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java index 3b1eca42bc3..3d4a9a14c4e 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java @@ -65,6 +65,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; + /** * Central logging facility for the TServerInfo. * @@ -83,6 +86,9 @@ public class TabletServerLogger { private final TabletServer tserver; + // Cache for resolved sorted logs to avoid repeated expensive I/O operations + private final Cache sortedLogCache; + // The current logger private DfsLogger currentLog = null; private final SynchronousQueue nextLog = new SynchronousQueue<>(); @@ -164,6 +170,8 @@ public TabletServerLogger(TabletServer tserver, long maxSize, AtomicLong syncCou this.createRetry = null; this.writeRetryFactory = writeRetryFactory; this.maxAge = maxAge; + this.sortedLogCache = + CacheBuilder.newBuilder().expireAfterWrite(3, TimeUnit.SECONDS).maximumSize(1000).build(); } private DfsLogger initializeLoggers(final AtomicInteger logIdOut) throws IOException { @@ -565,10 +573,17 @@ public long minorCompactionStarted(final CommitSession commitSession, final long return seq; } - private List resolve(Collection walogs) { - List sortedLogs = new ArrayList<>(walogs.size()); + private List resolve(Collection walogs) throws IOException { + List sortedLogs = new ArrayList<>(walogs.size()); + VolumeManager fs = tserver.getContext().getVolumeManager(); for (var logEntry : walogs) { - sortedLogs.add(new Path(logEntry.filename)); + try { + ResolvedSortedLog resolvedLog = + sortedLogCache.get(logEntry, () -> ResolvedSortedLog.resolve(logEntry, fs)); + sortedLogs.add(resolvedLog); + } catch (Exception e) { + throw new IOException("Failed to resolve sorted log for " + logEntry.filename, e); + } } return sortedLogs; } @@ -592,14 +607,14 @@ public boolean needsRecovery(ServerContext context, KeyExtent extent, Collection } } - public void recover(ServerContext context, KeyExtent extent, List recoveryDirs, + public void recover(ServerContext context, KeyExtent extent, Collection walogs, Set tabletFiles, MutationReceiver mr) throws IOException { try { var resourceMgr = tserver.getResourceManager(); var cacheProvider = createCacheProvider(resourceMgr); SortedLogRecovery recovery = new SortedLogRecovery(context, resourceMgr.getFileLenCache(), cacheProvider); - recovery.recover(extent, recoveryDirs, tabletFiles, mr); + recovery.recover(extent, resolve(walogs), tabletFiles, mr); } catch (Exception e) { throw new IOException(e); } diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java index 704a57804c6..326497834ab 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java @@ -48,6 +48,7 @@ import org.apache.accumulo.tserver.log.DfsLogger; import org.apache.accumulo.tserver.log.DfsLogger.LogHeaderIncompleteException; import org.apache.accumulo.tserver.log.RecoveryLogsIterator; +import org.apache.accumulo.tserver.log.ResolvedSortedLog; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.Path; @@ -175,8 +176,9 @@ public void execute(String[] args) throws Exception { } else { // read the log entries in a sorted RFile. This has to be a directory that contains the // finished file. - try (var rli = new RecoveryLogsIterator(context, Collections.singletonList(path), null, - null, false)) { + try (var rli = new RecoveryLogsIterator(context, + Collections.singletonList(ResolvedSortedLog.fromSortedLogDir(path, fs)), null, null, + false)) { while (rli.hasNext()) { Entry entry = rli.next(); printLogEvent(entry.getKey(), entry.getValue(), row, rowMatcher, ke, tabletIds, diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java index 75d933db372..b6473fd494f 100644 --- a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java @@ -132,7 +132,8 @@ public void testSimpleRLI() throws IOException { createRecoveryDir(logs, dirs, true); - try (RecoveryLogsIterator rli = new RecoveryLogsIterator(context, dirs, null, null, false)) { + try (RecoveryLogsIterator rli = + new RecoveryLogsIterator(context, pathsToResolvedLogs(dirs), null, null, false)) { while (rli.hasNext()) { Entry entry = rli.next(); assertEquals(1, entry.getKey().getTabletId(), "TabletId does not match"); @@ -159,7 +160,7 @@ public void testFinishMarker() throws IOException { createRecoveryDir(logs, dirs, false); assertThrows(IOException.class, - () -> new RecoveryLogsIterator(context, dirs, null, null, false), + () -> new RecoveryLogsIterator(context, pathsToResolvedLogs(dirs), null, null, false), "Finish marker should not be found"); } @@ -168,9 +169,10 @@ public void testSingleFile() throws IOException { String destPath = workDir + "/test.rf"; fs.create(new Path(destPath)); - assertThrows( - IOException.class, () -> new RecoveryLogsIterator(context, - Collections.singletonList(new Path(destPath)), null, null, false), + assertThrows(IOException.class, + () -> new RecoveryLogsIterator(context, + Collections.singletonList(ResolvedSortedLog.fromSortedLogDir(new Path(destPath), fs)), + null, null, false), "Finish marker should not be found for a single file."); } @@ -192,7 +194,7 @@ public void testCheckFirstKeyFailed() throws IOException { createRecoveryDir(logs, dirs, true); assertThrows(IllegalStateException.class, - () -> new RecoveryLogsIterator(context, dirs, null, null, true), + () -> new RecoveryLogsIterator(context, pathsToResolvedLogs(dirs), null, null, true), "First log entry is not OPEN so exception should be thrown."); } @@ -219,7 +221,8 @@ public void testCheckFirstKeyPass() throws IOException { createRecoveryDir(logs, dirs, true); - try (RecoveryLogsIterator rli = new RecoveryLogsIterator(context, dirs, null, null, true)) { + try (RecoveryLogsIterator rli = + new RecoveryLogsIterator(context, pathsToResolvedLogs(dirs), null, null, true)) { while (rli.hasNext()) { Entry entry = rli.next(); assertNotNull(entry.getKey()); @@ -248,4 +251,12 @@ private void createRecoveryDir(Map logs, ArrayList dirs dirs.add(new Path(destPath)); } } + + private List pathsToResolvedLogs(List dirs) throws IOException { + List resolvedLogs = new ArrayList<>(); + for (Path dir : dirs) { + resolvedLogs.add(ResolvedSortedLog.fromSortedLogDir(dir, fs)); + } + return resolvedLogs; + } } diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java index 9450e9018e9..592c9c0f946 100644 --- a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java @@ -217,7 +217,12 @@ private List recover(Map logs, Set files, K // Recover SortedLogRecovery recovery = new SortedLogRecovery(context, fileLenCache, cacheProvider); CaptureMutations capture = new CaptureMutations(); - recovery.recover(extent, dirs, files, capture); + // Convert Path objects to ResolvedSortedLog objects for recovery + List resolvedLogs = new ArrayList<>(); + for (Path dir : dirs) { + resolvedLogs.add(ResolvedSortedLog.fromSortedLogDir(dir, fs)); + } + recovery.recover(extent, resolvedLogs, files, capture); verify(context); return capture.result; } From 8721c1369bb4b830a9a95f56c102de95ee6368b3 Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Thu, 22 Oct 2026 11:27:59 -0500 Subject: [PATCH 4/5] replaced TreeMap logFiles = getFiles(vm, logDir); with SortedSet logFiles = logDir.getChildren(); deleted unused getFiles replaced ResolvedSortedLog.fromSortedLogDir with ResolvedSortedLog.resolve renaming Path entry : logFiles -> Path log : logFiles increase sortedLogCache expiration from 3s to 60s and maxsize from 1000 to 10_000 --- .../tserver/log/RecoveryLogsIterator.java | 56 +++---------------- .../tserver/log/ResolvedSortedLog.java | 47 +++++----------- .../tserver/log/TabletServerLogger.java | 2 +- .../accumulo/tserver/logger/LogReader.java | 3 +- .../tserver/log/RecoveryLogsIteratorTest.java | 6 +- .../tserver/log/SortedLogRecoveryTest.java | 2 +- 6 files changed, 28 insertions(+), 88 deletions(-) diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java index 02ea01cae58..af13b11ffb9 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java @@ -22,11 +22,10 @@ import java.util.AbstractMap; import java.util.ArrayList; import java.util.Collections; -import java.util.Comparator; import java.util.Iterator; import java.util.List; import java.util.Map.Entry; -import java.util.TreeMap; +import java.util.SortedSet; import org.apache.accumulo.core.crypto.CryptoEnvironmentImpl; import org.apache.accumulo.core.data.Key; @@ -39,12 +38,9 @@ import org.apache.accumulo.core.spi.crypto.CryptoEnvironment; import org.apache.accumulo.core.spi.crypto.CryptoService; import org.apache.accumulo.server.ServerContext; -import org.apache.accumulo.server.fs.VolumeManager; -import org.apache.accumulo.server.log.SortedLogState; import org.apache.accumulo.tserver.logger.LogEvents; import org.apache.accumulo.tserver.logger.LogFileKey; import org.apache.accumulo.tserver.logger.LogFileValue; -import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.slf4j.Logger; @@ -87,7 +83,7 @@ public RecoveryLogsIterator(ServerContext context, List recov for (ResolvedSortedLog logDir : recoveryLogDirs) { LOG.debug("Opening recovery log dir {}", logDir.getDir().getName()); - TreeMap logFiles = getFiles(vm, logDir); + SortedSet logFiles = logDir.getChildren(); var fs = vm.getFileSystemByPath(logDir.getDir()); // only check the first key once to prevent extra iterator creation and seeking @@ -96,22 +92,20 @@ public RecoveryLogsIterator(ServerContext context, List recov cacheProvider); } - for (Entry entry : logFiles.entrySet()) { + for (Path log : logFiles) { FileSKVIterator fileIter = - openLogFile(context, entry.getKey(), cryptoService, fs, fileLenCache, cacheProvider); + openLogFile(context, log, cryptoService, fs, fileLenCache, cacheProvider); if (range != null) { fileIter.seek(range, Collections.emptySet(), false); } Iterator> scanIter = new IteratorAdapter(fileIter); if (scanIter.hasNext()) { - LOG.debug("Write ahead log {} has data in range {} {}", entry.getKey().getName(), start, - end); + LOG.debug("Write ahead log {} has data in range {} {}", log.getName(), start, end); iterators.add(scanIter); fileIters.add(fileIter); } else { - LOG.debug("Write ahead log {} has no data in range {} {}", entry.getKey().getName(), - start, end); + LOG.debug("Write ahead log {} has no data in range {} {}", log.getName(), start, end); fileIter.close(); } } @@ -160,47 +154,15 @@ FileSKVIterator openLogFile(ServerContext context, Path logFile, CryptoService c return builder.seekToBeginning().build(); } - /** - * Check for sorting signal files (finished/failed) and get the logs in the provided directory. - */ - private TreeMap getFiles(VolumeManager fs, ResolvedSortedLog directory) - throws IOException { - boolean foundFinish = false; - // Path::getName compares the last component of each Path value. In this case, the last - // component should - // always have the format 'part-r-XXXXX.rf', where XXXXX are one-up values. - TreeMap logFiles = new TreeMap<>(Comparator.comparing(Path::getName)); - for (FileStatus child : fs.listStatus(directory.getDir())) { - if (child.getPath().getName().startsWith("_")) { - continue; - } - if (SortedLogState.isFinished(child.getPath().getName())) { - foundFinish = true; - continue; - } - if (SortedLogState.FAILED.getMarker().equals(child.getPath().getName())) { - continue; - } - FileSystem ns = fs.getFileSystemByPath(child.getPath()); - Path fullLogPath = ns.makeQualified(child.getPath()); - logFiles.put(fullLogPath, child); - } - if (!foundFinish) { - throw new IOException( - "Sort '" + SortedLogState.FINISHED.getMarker() + "' flag not found in " + directory); - } - return logFiles; - } - /** * Check that the first entry in the WAL is OPEN. Only need to do this once. */ private void validateFirstKey(ServerContext context, CryptoService cs, FileSystem fs, - TreeMap logFiles, Path fullLogPath, Cache fileLenCache, + SortedSet logFiles, Path fullLogPath, Cache fileLenCache, CacheProvider cacheProvider) throws IOException { - Entry first = logFiles.firstEntry(); + Path first = logFiles.first(); try (FileSKVIterator fileIter = - openLogFile(context, first.getKey(), cs, fs, fileLenCache, cacheProvider)) { + openLogFile(context, first, cs, fs, fileLenCache, cacheProvider)) { Iterator> iterator = new IteratorAdapter(fileIter); if (iterator.hasNext()) { diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java index 06f23cd120f..91041f93969 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/ResolvedSortedLog.java @@ -87,6 +87,19 @@ public static ResolvedSortedLog resolve(LogEntry logEntry, VolumeManager fs) thr // dir Path sortedLogPath = RecoveryPath.getRecoveryPath(new Path(logEntry.filename)); + return resolve(logEntry, sortedLogPath, fs); + } + + /** + * Resolve a sorted log directory path. This is useful for diagnostic tools that operate directly + * on sorted recovery logs without going through the normal recovery flow with LogEntry objects. + */ + public static ResolvedSortedLog resolve(Path sortedLogDir, VolumeManager fs) throws IOException { + return resolve(new LogEntry(null, 0, sortedLogDir.toString()), sortedLogDir, fs); + } + + private static ResolvedSortedLog resolve(LogEntry logEntry, Path sortedLogPath, VolumeManager fs) + throws IOException { boolean foundFinish = false; // Path::getName compares the last component of each Path value. In this case, the last // component should @@ -114,38 +127,4 @@ public static ResolvedSortedLog resolve(LogEntry logEntry, VolumeManager fs) thr return new ResolvedSortedLog(logEntry, sortedLogPath, logFiles); } - - /** - * Create a ResolvedSortedLog directly from a sorted log directory path. This is useful for - * diagnostic tools that operate directly on sorted recovery logs without going through the normal - * recovery flow with LogEntry objects. - */ - public static ResolvedSortedLog fromSortedLogDir(Path sortedLogDir, VolumeManager fs) - throws IOException { - boolean foundFinish = false; - SortedSet logFiles = new TreeSet<>(Comparator.comparing(Path::getName)); - for (FileStatus child : fs.listStatus(sortedLogDir)) { - if (child.getPath().getName().startsWith("_")) { - continue; - } - if (SortedLogState.isFinished(child.getPath().getName())) { - foundFinish = true; - continue; - } - if (SortedLogState.FAILED.getMarker().equals(child.getPath().getName())) { - continue; - } - FileSystem ns = fs.getFileSystemByPath(child.getPath()); - Path fullLogPath = ns.makeQualified(child.getPath()); - logFiles.add(fullLogPath); - } - if (!foundFinish) { - throw new IOException( - "Sort '" + SortedLogState.FINISHED.getMarker() + "' flag not found in " + sortedLogDir); - } - - // Create a dummy LogEntry for the origin (used only for diagnostics) - LogEntry dummyOrigin = new LogEntry(null, 0, sortedLogDir.toString()); - return new ResolvedSortedLog(dummyOrigin, sortedLogDir, logFiles); - } } diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java index 3d4a9a14c4e..90a7603d0ed 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java @@ -171,7 +171,7 @@ public TabletServerLogger(TabletServer tserver, long maxSize, AtomicLong syncCou this.writeRetryFactory = writeRetryFactory; this.maxAge = maxAge; this.sortedLogCache = - CacheBuilder.newBuilder().expireAfterWrite(3, TimeUnit.SECONDS).maximumSize(1000).build(); + Caffeine.newBuilder().expireAfterWrite(1, TimeUnit.MINUTES).maximumSize(10_000).build(); } private DfsLogger initializeLoggers(final AtomicInteger logIdOut) throws IOException { diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java index 326497834ab..efbaa0c4c71 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/logger/LogReader.java @@ -177,8 +177,7 @@ public void execute(String[] args) throws Exception { // read the log entries in a sorted RFile. This has to be a directory that contains the // finished file. try (var rli = new RecoveryLogsIterator(context, - Collections.singletonList(ResolvedSortedLog.fromSortedLogDir(path, fs)), null, null, - false)) { + Collections.singletonList(ResolvedSortedLog.resolve(path, fs)), null, null, false)) { while (rli.hasNext()) { Entry entry = rli.next(); printLogEvent(entry.getKey(), entry.getValue(), row, rowMatcher, ke, tabletIds, diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java index b6473fd494f..13595ad16b1 100644 --- a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/RecoveryLogsIteratorTest.java @@ -171,8 +171,8 @@ public void testSingleFile() throws IOException { assertThrows(IOException.class, () -> new RecoveryLogsIterator(context, - Collections.singletonList(ResolvedSortedLog.fromSortedLogDir(new Path(destPath), fs)), - null, null, false), + Collections.singletonList(ResolvedSortedLog.resolve(new Path(destPath), fs)), null, + null, false), "Finish marker should not be found for a single file."); } @@ -255,7 +255,7 @@ private void createRecoveryDir(Map logs, ArrayList dirs private List pathsToResolvedLogs(List dirs) throws IOException { List resolvedLogs = new ArrayList<>(); for (Path dir : dirs) { - resolvedLogs.add(ResolvedSortedLog.fromSortedLogDir(dir, fs)); + resolvedLogs.add(ResolvedSortedLog.resolve(dir, fs)); } return resolvedLogs; } diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java index 592c9c0f946..801644d67c8 100644 --- a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java @@ -220,7 +220,7 @@ private List recover(Map logs, Set files, K // Convert Path objects to ResolvedSortedLog objects for recovery List resolvedLogs = new ArrayList<>(); for (Path dir : dirs) { - resolvedLogs.add(ResolvedSortedLog.fromSortedLogDir(dir, fs)); + resolvedLogs.add(ResolvedSortedLog.resolve(dir, fs)); } recovery.recover(extent, resolvedLogs, files, capture); verify(context); From 24ca6745e0c4bcfe371fbe8216e7b3c8d73619bf Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Mon, 2 Nov 2026 10:52:05 -0500 Subject: [PATCH 5/5] com.github.benmanes.caffeine.cache.Cache - changes for RecoveryLogIterator --- .../core/clientImpl/bulk/BulkImport.java | 6 ++--- .../accumulo/core/file/FileOperations.java | 2 +- .../blockfile/impl/CachableBlockFile.java | 13 ++++++--- .../accumulo/core/summary/Gatherer.java | 2 +- .../accumulo/core/summary/SummaryReader.java | 2 +- .../accumulo/server/fs/FileManager.java | 2 +- .../apache/accumulo/server/fs/VolumeUtil.java | 2 +- .../accumulo/tserver/TabletClientHandler.java | 18 +++++-------- .../apache/accumulo/tserver/TabletServer.java | 2 +- .../tserver/TabletServerResourceManager.java | 6 ++--- .../compaction/MajorCompactionRequest.java | 2 +- .../compactions/CompactionManager.java | 7 +++-- .../compactions/CompactionService.java | 6 ++--- .../tserver/log/RecoveryLogsIterator.java | 2 +- .../tserver/log/SortedLogRecovery.java | 2 +- .../tserver/log/TabletServerLogger.java | 13 ++++++--- .../tserver/tablet/CompactableUtils.java | 27 ++++++++----------- .../tserver/log/SortedLogRecoveryTest.java | 6 ++--- 18 files changed, 59 insertions(+), 61 deletions(-) diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java index fc993bbe648..1487c4a0b23 100644 --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java @@ -88,9 +88,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import com.google.common.base.Preconditions; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; import com.google.common.collect.Sets; public class BulkImport implements ImportDestinationArguments, ImportMappingOptions { @@ -399,7 +399,7 @@ private static Cache getPopulatedFileLenCache(Path dir, List absFileLens = new HashMap<>(); fileLens.forEach((k, v) -> absFileLens.put(pathToCacheId(new Path(dir, k)), v)); - Cache fileLenCache = CacheBuilder.newBuilder().build(); + Cache fileLenCache = Caffeine.newBuilder().build(); fileLenCache.putAll(absFileLens); diff --git a/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java b/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java index 578e6ca0db0..f423e19c7f5 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java +++ b/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java @@ -46,7 +46,7 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.mapred.FileOutputCommitter; -import com.google.common.cache.Cache; +import com.github.benmanes.caffeine.cache.Cache; public abstract class FileOperations { diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java index 191c260c0b5..d3ce8eacc29 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java +++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java @@ -26,7 +26,6 @@ import java.util.Collections; import java.util.Map; import java.util.Objects; -import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -53,8 +52,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.github.benmanes.caffeine.cache.Cache; import com.google.common.base.Preconditions; -import com.google.common.cache.Cache; /** * This is a wrapper class for BCFile that includes a cache for independent caches for datablocks @@ -190,8 +189,14 @@ public static class Reader implements Closeable { private long getCachedFileLen() throws IOException { try { - return fileLenCache.get(cacheId, lengthSupplier::get); - } catch (ExecutionException e) { + return fileLenCache.get(cacheId, k -> { + try { + return lengthSupplier.get(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }); + } catch (UncheckedIOException e) { throw new IOException("Failed to get " + cacheId + " len from cache ", e); } } diff --git a/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java b/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java index 850395d1aae..569f42e3cb9 100644 --- a/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java +++ b/core/src/main/java/org/apache/accumulo/core/summary/Gatherer.java @@ -84,8 +84,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.github.benmanes.caffeine.cache.Cache; import com.google.common.base.Preconditions; -import com.google.common.cache.Cache; import com.google.common.hash.Hashing; /** diff --git a/core/src/main/java/org/apache/accumulo/core/summary/SummaryReader.java b/core/src/main/java/org/apache/accumulo/core/summary/SummaryReader.java index 83c06339747..b14ff3caaf1 100644 --- a/core/src/main/java/org/apache/accumulo/core/summary/SummaryReader.java +++ b/core/src/main/java/org/apache/accumulo/core/summary/SummaryReader.java @@ -48,7 +48,7 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.WritableUtils; -import com.google.common.cache.Cache; +import com.github.benmanes.caffeine.cache.Cache; public class SummaryReader { diff --git 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 index 24ff167019a..6ad25ba6e4b 100644 --- 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 @@ -64,7 +64,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.cache.Cache; +import com.github.benmanes.caffeine.cache.Cache; public class FileManager { diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java index 7fd7b98bf19..d1ac0b543dc 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java @@ -93,7 +93,7 @@ public static LogEntry switchVolume(LogEntry le, List> replaceme return switchVolumes(le, replacements); } - protected static LogEntry switchVolumes(LogEntry le, List> replacements) { + public static LogEntry switchVolumes(LogEntry le, List> replacements) { Path switchedPath = switchVolume(le.filename, FileType.WAL, replacements); String switchedString; int numSwitched = 0; diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java index eac87f25f3d..98a3d11eaf0 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java @@ -24,7 +24,6 @@ import java.io.IOException; import java.nio.ByteBuffer; -import java.time.Duration; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -132,9 +131,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.base.Supplier; -import com.google.common.base.Suppliers; -import com.google.common.cache.Cache; +import com.github.benmanes.caffeine.cache.Cache; import io.opentelemetry.api.trace.Span; import io.opentelemetry.context.Scope; @@ -142,7 +139,7 @@ public class TabletClientHandler implements TabletClientService.Iface { private static final Logger log = LoggerFactory.getLogger(TabletClientHandler.class); - private final Supplier scanResultWaitTime; + private final long MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS; private static final long RECENTLY_SPLIT_MILLIES = MINUTES.toMillis(1); private final TabletServer server; protected final TransactionWatcher watcher; @@ -158,12 +155,8 @@ public TabletClientHandler(TabletServer server, TransactionWatcher watcher, this.writeTracker = writeTracker; this.security = context.getSecurityOperation(); this.server = server; - scanResultWaitTime = - Suppliers - .memoizeWithExpiration( - () -> server.getContext().getConfiguration() - .getTimeInMillis(Property.TSERV_SCAN_RESULTS_MAX_TIMEOUT), - Duration.ofSeconds(10)); + MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS = server.getContext().getConfiguration() + .getTimeInMillis(Property.TSERV_SCAN_RESULTS_MAX_TIMEOUT); log.debug("{} created", TabletClientHandler.class.getName()); } @@ -1541,7 +1534,8 @@ public void removeLogs(TInfo tinfo, TCredentials credentials, List filen private TSummaries getSummaries(Future future) throws TimeoutException { try { - SummaryCollection sc = future.get(scanResultWaitTime.get(), TimeUnit.MILLISECONDS); + SummaryCollection sc = + future.get(MAX_TIME_TO_WAIT_FOR_SCAN_RESULT_MILLIS, TimeUnit.MILLISECONDS); return sc.toThrift(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java index 2110570d40a..09fae0df4c9 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java @@ -579,7 +579,7 @@ public boolean needsRecovery(TabletMetadata tabletMetadata) { var switchedLogEntries = new ArrayList(logEntries.size()); for (LogEntry logEntry : logEntries) { - var switchedWalog = VolumeUtil.switchVolume(logEntry, context.getVolumeReplacements()); + var switchedWalog = VolumeUtil.switchVolumes(logEntry, context.getVolumeReplacements()); LogEntry walog; if (switchedWalog != null) { log.debug("Volume switched for needsRecovery {} -> {}", logEntry, switchedWalog); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java index be327e0f461..11f0d777273 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java @@ -92,10 +92,10 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Suppliers; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; @@ -385,7 +385,7 @@ public TabletServerResourceManager(ServerContext context, TabletHostingServer ts int maxOpenFiles = acuConf.getCount(Property.TSERV_SCAN_MAX_OPENFILES); fileLenCache = - CacheBuilder.newBuilder().maximumSize(Math.min(maxOpenFiles * 1000L, 100_000)).build(); + Caffeine.newBuilder().maximumSize(Math.min(maxOpenFiles * 1000L, 100_000)).build(); fileManager = new FileManager(context, maxOpenFiles, fileLenCache); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/compaction/MajorCompactionRequest.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/compaction/MajorCompactionRequest.java index 41153d654d4..8858b2b9fd1 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/compaction/MajorCompactionRequest.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/compaction/MajorCompactionRequest.java @@ -51,8 +51,8 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; +import com.github.benmanes.caffeine.cache.Cache; import com.google.common.annotations.VisibleForTesting; -import com.google.common.cache.Cache; /** * Information that can be used to determine how a tablet is to be major compacted, if needed. diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/compactions/CompactionManager.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/compactions/CompactionManager.java index 6e14324e47f..09f5ac65dcf 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/compactions/CompactionManager.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/compactions/CompactionManager.java @@ -56,9 +56,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import com.google.common.base.Preconditions; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; import com.google.common.collect.Sets; public class CompactionManager { @@ -227,8 +227,7 @@ public CompactionManager(Supplier compactables, Map tmpServices = new HashMap<>(); - unknownCompactionServiceErrorCache = - CacheBuilder.newBuilder().expireAfterWrite(5, MINUTES).build(); + unknownCompactionServiceErrorCache = Caffeine.newBuilder().expireAfterWrite(5, MINUTES).build(); currentCfg.getPlanners().forEach((serviceName, plannerClassName) -> { try { diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/compactions/CompactionService.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/compactions/CompactionService.java index 2d609182533..b836f1b8933 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/compactions/CompactionService.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/compactions/CompactionService.java @@ -72,9 +72,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import com.google.common.base.Preconditions; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; import com.google.common.collect.Sets; public class CompactionService { @@ -146,7 +146,7 @@ public CompactionService(String serviceName, String plannerClass, Long maxRate, queuedForPlanning.put(kind, new ConcurrentHashMap()); } - maxScanFilesExceededErrorCache = CacheBuilder.newBuilder().expireAfterWrite(5, MINUTES).build(); + maxScanFilesExceededErrorCache = Caffeine.newBuilder().expireAfterWrite(5, MINUTES).build(); log.debug("Created new compaction service id:{} rate limit:{} planner:{} planner options:{}", myId, maxRate, plannerClass, plannerOptions); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java index af13b11ffb9..faae3c473ce 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java @@ -46,7 +46,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.cache.Cache; +import com.github.benmanes.caffeine.cache.Cache; import com.google.common.collect.Iterators; /** diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java index 71210de1203..7d57b141f5a 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/SortedLogRecovery.java @@ -52,7 +52,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.cache.Cache; +import com.github.benmanes.caffeine.cache.Cache; import com.google.common.collect.Collections2; import com.google.common.collect.Iterators; import com.google.common.collect.PeekingIterator; diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java index 90a7603d0ed..c443624834c 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/TabletServerLogger.java @@ -65,8 +65,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; /** * Central logging facility for the TServerInfo. @@ -578,8 +578,13 @@ private List resolve(Collection walogs) throws IOEx VolumeManager fs = tserver.getContext().getVolumeManager(); for (var logEntry : walogs) { try { - ResolvedSortedLog resolvedLog = - sortedLogCache.get(logEntry, () -> ResolvedSortedLog.resolve(logEntry, fs)); + ResolvedSortedLog resolvedLog = sortedLogCache.get(logEntry, k -> { + try { + return ResolvedSortedLog.resolve(logEntry, fs); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); sortedLogs.add(resolvedLog); } catch (Exception e) { throw new IOException("Failed to resolve sorted log for " + logEntry.filename, e); diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableUtils.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableUtils.java index 0ef01a18838..383abec2305 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableUtils.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableUtils.java @@ -29,7 +29,6 @@ import java.util.Optional; import java.util.Set; import java.util.SortedMap; -import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.function.Predicate; @@ -97,9 +96,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import com.google.common.base.Preconditions; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; import com.google.common.collect.Collections2; @SuppressWarnings("removal") @@ -108,7 +107,7 @@ public class CompactableUtils { private static final Logger log = LoggerFactory.getLogger(CompactableUtils.class); private final static Cache strategyWarningsCache = - CacheBuilder.newBuilder().maximumSize(1000).build(); + Caffeine.newBuilder().maximumSize(1000).build(); public static Map> getFirstAndLastKeys(Tablet tablet, Set allFiles) throws IOException { @@ -510,18 +509,14 @@ public static CompactionHelper getHelper(CompactionKind kind, Tablet tablet, Lon var stratClassName = tconf.get(Property.TABLE_COMPACTION_STRATEGY); if (cselCfg == null && tconf.isPropertySet(Property.TABLE_COMPACTION_STRATEGY) && stratClassName != null && !stratClassName.isBlank()) { - try { - strategyWarningsCache.get(tablet.getExtent().tableId(), () -> { - log.warn( - "Table id {} set {} to {}. Compaction strategies are deprecated. See the Javadoc" - + " for class {} for more details.", - tablet.getExtent().tableId(), Property.TABLE_COMPACTION_STRATEGY.getKey(), - stratClassName, CompactionStrategyConfig.class.getName()); - return true; - }); - } catch (ExecutionException e) { - throw new RuntimeException(e); - } + var unused = strategyWarningsCache.get(tablet.getExtent().tableId(), k -> { + log.warn( + "Table id {} set {} to {}. Compaction strategies are deprecated. See the Javadoc" + + " for class {} for more details.", + tablet.getExtent().tableId(), Property.TABLE_COMPACTION_STRATEGY.getKey(), + stratClassName, CompactionStrategyConfig.class.getName()); + return true; + }); var opts = tconf.getAllPropertiesWithPrefixStripped(Property.TABLE_COMPACTION_STRATEGY_PREFIX); diff --git a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java index 801644d67c8..d4b0ccb6a1d 100644 --- a/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java +++ b/server/tserver/src/test/java/org/apache/accumulo/tserver/log/SortedLogRecoveryTest.java @@ -80,8 +80,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; @@ -180,7 +180,7 @@ private List recover(Map logs, Set files, K new TinyLfuBlockCache( BlockCacheConfiguration.forTabletServer(DefaultConfiguration.getInstance()), CacheType.DATA)); - Cache fileLenCache = CacheBuilder.newBuilder().build(); + Cache fileLenCache = Caffeine.newBuilder().build(); final String workdir = new File(tempDir, testName()).getAbsolutePath(); try (var fs = VolumeManagerImpl.getLocalForTesting(workdir)) {