From 3070ec45aa2edd93810d787fe309d7eb95655b17 Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Mon, 4 May 2026 10:30:03 -0500 Subject: [PATCH 1/4] 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 b6fc4695f4f..df61275c26c 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 7c7dc3e6706..f70e69cbd81 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; @@ -556,10 +563,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 a209b60431043a701fbdd63d13ee9bcbe1a51dc8 Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Tue, 5 May 2026 09:17:34 -0500 Subject: [PATCH 2/4] 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. {"fundingSource": "41201", "team": "FED.ICGSA.OPS.MOE", "fshGit": "dummy-lo", "fshDocker": "sha256:20cf0045"} --- .../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 df61275c26c..794418c23d6 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 db9fc6f8f3134622cd3e8e01df1aac04a5e8689f Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Tue, 5 May 2026 10:31:55 -0500 Subject: [PATCH 3/4] 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 794418c23d6..f639b3ddafc 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 f70e69cbd81..76352329d78 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 { @@ -563,10 +571,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; } @@ -590,14 +605,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 26b1bd2a639d0199148aecf93d7bfb455fcd7143 Mon Sep 17 00:00:00 2001 From: andrew mcdonald Date: Thu, 11 Jun 2026 10:21:12 -0500 Subject: [PATCH 4/4] use com.github.benmanes.caffeine.cache.Cache added var unused = in Combiner sawDelete and CompactionHelper --- .../core/clientImpl/bulk/BulkImport.java | 6 ++-- .../apache/accumulo/core/data/InstanceId.java | 14 +++------ .../accumulo/core/data/NamespaceId.java | 15 +++------- .../apache/accumulo/core/data/TableId.java | 15 +++------- .../accumulo/core/file/FileOperations.java | 2 +- .../blockfile/impl/CachableBlockFile.java | 15 ++++++---- .../accumulo/core/iterators/Combiner.java | 29 ++++++++----------- .../accumulo/core/summary/Gatherer.java | 2 +- .../accumulo/core/summary/SummaryReader.java | 2 +- .../core/util/tables/TableZooHelper.java | 13 +++------ .../core/iterators/CombinerTestUtil.java | 2 +- .../accumulo/server/fs/FileManager.java | 2 +- .../apache/accumulo/server/fs/VolumeUtil.java | 6 +--- .../manager/recovery/RecoveryManager.java | 20 ++++++++----- .../accumulo/tserver/TabletClientHandler.java | 2 +- .../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 | 15 ++++++---- .../tserver/tablet/CompactableUtils.java | 27 +++++++---------- .../tserver/log/SortedLogRecoveryTest.java | 6 ++-- 25 files changed, 98 insertions(+), 122 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 575a221b8b6..5acb797511e 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 { @@ -401,7 +401,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/data/InstanceId.java b/core/src/main/java/org/apache/accumulo/core/data/InstanceId.java index 598d35937d3..70e90192ea0 100644 --- a/core/src/main/java/org/apache/accumulo/core/data/InstanceId.java +++ b/core/src/main/java/org/apache/accumulo/core/data/InstanceId.java @@ -20,10 +20,9 @@ import java.util.Objects; import java.util.UUID; -import java.util.concurrent.ExecutionException; -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; /** * A strongly typed representation of an Accumulo instance ID. The constructor for this class will @@ -36,7 +35,7 @@ public class InstanceId extends AbstractId { // cache is for canonicalization/deduplication of created objects, // to limit the number of InstanceId objects in the JVM at any given moment // WeakReferences are used because we don't need them to stick around any longer than they need to - static final Cache cache = CacheBuilder.newBuilder().weakValues().build(); + static final Cache cache = Caffeine.newBuilder().weakValues().build(); private InstanceId(String canonical) { super(canonical); @@ -49,12 +48,7 @@ private InstanceId(String canonical) { * @return InstanceId object */ public static InstanceId of(final String canonical) { - try { - return cache.get(canonical, () -> new InstanceId(canonical)); - } catch (ExecutionException e) { - throw new AssertionError( - "This should never happen: ID constructor should never return null."); - } + return cache.get(canonical, k -> new InstanceId(canonical)); } /** diff --git a/core/src/main/java/org/apache/accumulo/core/data/NamespaceId.java b/core/src/main/java/org/apache/accumulo/core/data/NamespaceId.java index 90a8c74e1be..c201113ce30 100644 --- a/core/src/main/java/org/apache/accumulo/core/data/NamespaceId.java +++ b/core/src/main/java/org/apache/accumulo/core/data/NamespaceId.java @@ -18,10 +18,8 @@ */ package org.apache.accumulo.core.data; -import java.util.concurrent.ExecutionException; - -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; /** * A strongly typed representation of a namespace ID. This class cannot be used to get a namespace @@ -35,7 +33,7 @@ public class NamespaceId extends AbstractId { // cache is for canonicalization/deduplication of created objects, // to limit the number of NamespaceId objects in the JVM at any given moment // WeakReferences are used because we don't need them to stick around any longer than they need to - static final Cache cache = CacheBuilder.newBuilder().weakValues().build(); + static final Cache cache = Caffeine.newBuilder().weakValues().build(); private NamespaceId(String canonical) { super(canonical); @@ -48,11 +46,6 @@ private NamespaceId(String canonical) { * @return NamespaceId object */ public static NamespaceId of(final String canonical) { - try { - return cache.get(canonical, () -> new NamespaceId(canonical)); - } catch (ExecutionException e) { - throw new AssertionError( - "This should never happen: ID constructor should never return null."); - } + return cache.get(canonical, k -> new NamespaceId(canonical)); } } diff --git a/core/src/main/java/org/apache/accumulo/core/data/TableId.java b/core/src/main/java/org/apache/accumulo/core/data/TableId.java index 9493d8dc160..a97c9d2f623 100644 --- a/core/src/main/java/org/apache/accumulo/core/data/TableId.java +++ b/core/src/main/java/org/apache/accumulo/core/data/TableId.java @@ -18,10 +18,8 @@ */ package org.apache.accumulo.core.data; -import java.util.concurrent.ExecutionException; - -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; /** * A strongly typed representation of a table ID. This class cannot be used to get a table ID from a @@ -35,7 +33,7 @@ public class TableId extends AbstractId { // cache is for canonicalization/deduplication of created objects, // to limit the number of TableId objects in the JVM at any given moment // WeakReferences are used because we don't need them to stick around any longer than they need to - static final Cache cache = CacheBuilder.newBuilder().weakValues().build(); + static final Cache cache = Caffeine.newBuilder().weakValues().build(); private TableId(final String canonical) { super(canonical); @@ -48,11 +46,6 @@ private TableId(final String canonical) { * @return TableId object */ public static TableId of(final String canonical) { - try { - return cache.get(canonical, () -> new TableId(canonical)); - } catch (ExecutionException e) { - throw new AssertionError( - "This should never happen: ID constructor should never return null."); - } + return cache.get(canonical, k -> new TableId(canonical)); } } 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..c3d25054f24 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,9 +189,15 @@ public static class Reader implements Closeable { private long getCachedFileLen() throws IOException { try { - return fileLenCache.get(cacheId, lengthSupplier::get); - } catch (ExecutionException e) { - throw new IOException("Failed to get " + cacheId + " len from cache ", 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.getCause()); } } diff --git a/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java b/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java index a67bcf0c27e..a0559bf7791 100644 --- a/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java +++ b/core/src/main/java/org/apache/accumulo/core/iterators/Combiner.java @@ -27,7 +27,6 @@ import java.util.List; import java.util.Map; import java.util.NoSuchElementException; -import java.util.concurrent.ExecutionException; import org.apache.accumulo.core.client.IteratorSetting; import org.apache.accumulo.core.client.IteratorSetting.Column; @@ -43,10 +42,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.Splitter; -import com.google.common.cache.Cache; -import com.google.common.cache.CacheBuilder; import com.google.common.collect.Lists; /** @@ -187,23 +186,19 @@ public void next() throws IOException { @VisibleForTesting static final Cache loggedMsgCache = - CacheBuilder.newBuilder().expireAfterWrite(1, HOURS).maximumSize(10000).build(); + Caffeine.newBuilder().expireAfterWrite(1, HOURS).maximumSize(10000).build(); private void sawDelete() { if (isMajorCompaction && !reduceOnFullCompactionOnly) { - try { - loggedMsgCache.get(this.getClass().getName(), () -> { - sawDeleteLog.error( - "Combiner of type {} saw a delete during a" - + " partial compaction. This could cause undesired results. See" - + " ACCUMULO-2232. Will not log subsequent occurrences for at least 1 hour.", - Combiner.this.getClass().getSimpleName()); - // the value is not used and does not matter - return Boolean.TRUE; - }); - } catch (ExecutionException e) { - throw new RuntimeException(e); - } + var unused = loggedMsgCache.get(this.getClass().getName(), k -> { + sawDeleteLog.error( + "Combiner of type {} saw a delete during a" + + " partial compaction. This could cause undesired results. See" + + " ACCUMULO-2232. Will not log subsequent occurrences for at least 1 hour.", + Combiner.this.getClass().getSimpleName()); + // the value is not used and does not matter + return Boolean.TRUE; + }); } } 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/core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java b/core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java index 7487ed57637..39e041c3684 100644 --- a/core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java +++ b/core/src/main/java/org/apache/accumulo/core/util/tables/TableZooHelper.java @@ -25,7 +25,6 @@ import java.util.List; import java.util.Objects; -import java.util.concurrent.ExecutionException; import org.apache.accumulo.core.Constants; import org.apache.accumulo.core.client.NamespaceNotFoundException; @@ -40,8 +39,8 @@ import org.apache.accumulo.core.metadata.MetadataTable; import org.apache.accumulo.core.metadata.RootTable; -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; public class TableZooHelper implements AutoCloseable { @@ -49,7 +48,7 @@ public class TableZooHelper implements AutoCloseable { // Per instance cache will expire after 10 minutes in case we // encounter an instance not used frequently private final Cache instanceToMapCache = - CacheBuilder.newBuilder().expireAfterAccess(10, MINUTES).build(); + Caffeine.newBuilder().expireAfterAccess(10, MINUTES).build(); public TableZooHelper(ClientContext context) { this.context = Objects.requireNonNull(context); @@ -127,11 +126,7 @@ public TableMap getTableMap() { } private TableMap getCachedTableMap() { - try { - return instanceToMapCache.get(this, () -> new TableMap(context)); - } catch (ExecutionException e) { - throw new RuntimeException(e); - } + return instanceToMapCache.get(this, k -> new TableMap(context)); } public boolean tableNodeExists(TableId tableId) { diff --git a/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTestUtil.java b/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTestUtil.java index 18904cd501e..39567964b50 100644 --- a/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTestUtil.java +++ b/core/src/test/java/org/apache/accumulo/core/iterators/CombinerTestUtil.java @@ -24,6 +24,6 @@ public static void clearLogCache() { } public static long cacheSize() { - return Combiner.loggedMsgCache.size(); + return Combiner.loggedMsgCache.estimatedSize(); } } 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 4e81899ed2f..6c748806a1e 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 @@ -62,7 +62,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..6babbac8eae 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,11 +89,7 @@ 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) { + 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/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java index 666970cc09f..cfc16eb6fad 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java +++ b/server/manager/src/main/java/org/apache/accumulo/manager/recovery/RecoveryManager.java @@ -22,13 +22,13 @@ import java.io.FileNotFoundException; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -51,8 +51,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; public class RecoveryManager { @@ -69,7 +69,7 @@ public class RecoveryManager { public RecoveryManager(Manager manager, long timeToCacheExistsInMillis) { this.manager = manager; existenceCache = - CacheBuilder.newBuilder().expireAfterWrite(timeToCacheExistsInMillis, TimeUnit.MILLISECONDS) + Caffeine.newBuilder().expireAfterWrite(timeToCacheExistsInMillis, TimeUnit.MILLISECONDS) .maximumWeight(10_000_000).weigher((path, exist) -> path.toString().length()).build(); executor = @@ -143,9 +143,15 @@ private void initiateSort(String sortId, String source, final String destination private boolean exists(final Path path) throws IOException { try { - return existenceCache.get(path, () -> manager.getVolumeManager().exists(path)); - } catch (ExecutionException e) { - throw new IOException(e); + return existenceCache.get(path, k -> { + try { + return manager.getVolumeManager().exists(path); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + }); + } catch (UncheckedIOException e) { + throw e.getCause(); // Unwrap and rethrow the original IOException } } 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 608291db7af..1f3a2c31ba2 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 @@ -131,7 +131,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.cache.Cache; +import com.github.benmanes.caffeine.cache.Cache; import io.opentelemetry.api.trace.Span; import io.opentelemetry.context.Scope; 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 f639b3ddafc..5d1c264640b 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 7b3ec10e86a..dc21329b73a 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 02ea01cae58..4526f5ebe11 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 @@ -50,7 +50,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 76352329d78..6eaa5fbfff8 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. @@ -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(3, TimeUnit.SECONDS).maximumSize(1000).build(); } private DfsLogger initializeLoggers(final AtomicInteger logIdOut) throws IOException { @@ -576,8 +576,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 592c9c0f946..c09a2506842 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)) {