diff --git a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java index d9ad691cace..64d77d2c524 100644 --- a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java +++ b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java @@ -23,6 +23,9 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Objects; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import org.apache.accumulo.core.client.Scanner; import org.apache.accumulo.core.client.rfile.RFile.ScannerFSOptions; @@ -30,6 +33,7 @@ import org.apache.accumulo.core.data.Range; import org.apache.accumulo.core.security.Authorizations; import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -56,8 +60,29 @@ RFileSource[] getSources() throws IOException { if (sources == null) { sources = new RFileSource[paths.length]; for (int i = 0; i < paths.length; i++) { - sources[i] = new RFileSource(getFileSystem(paths[i]).open(paths[i]), - getFileSystem(paths[i]).getFileStatus(paths[i]).getLen()); + FileSystem fs = getFileSystem(paths[i]); + FileStatus status = fs.getFileStatus(paths[i]); + CompletableFuture future = + fs.openFile(paths[i]).withFileStatus(status).build(); + while (!future.isDone()) { + try { + Thread.sleep(10); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while opening file: " + paths[i], e); + } + } + try { + FSDataInputStream is = future.get(); + sources[i] = new RFileSource(is, status.getLen()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while opening file: " + paths[i], e); + } catch (CancellationException e) { + throw new IOException("Cancelled while opening file: " + paths[i], e); + } catch (ExecutionException e) { + throw new IOException("Error trying to open file: " + paths[i], e); + } } } else { for (int i = 0; i < sources.length; i++) { 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 51a23026526..575a221b8b6 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 @@ -266,8 +266,10 @@ public MLong(long i) { } public static Map estimateSizes(AccumuloConfiguration acuConf, Path mapFile, - long fileSize, Collection extents, FileSystem ns, Cache fileLenCache, - CryptoService cs) throws IOException { + FileStatus status, Collection extents, FileSystem ns, + Cache fileLenCache, CryptoService cs) throws IOException { + + final long fileSize = status.getLen(); if (extents.size() == 1) { return Collections.singletonMap(extents.iterator().next(), fileSize); @@ -282,7 +284,7 @@ public static Map estimateSizes(AccumuloConfiguration acuConf, P Text row = new Text(); FileSKVIterator index = FileOperations.getInstance().newIndexReaderBuilder() - .forFile(mapFile.toString(), ns, ns.getConf(), cs).withTableConfiguration(acuConf) + .forFile(mapFile.toString(), ns, ns.getConf(), cs, status).withTableConfiguration(acuConf) .withFileLenCache(fileLenCache).build(); try { @@ -365,9 +367,9 @@ private static Text nextRow(Text row) { public static List findOverlappingTablets(ClientContext context, KeyExtentCache keyExtentCache, Path file, FileSystem fs, Cache fileLenCache, - CryptoService cs) throws IOException { + CryptoService cs, FileStatus status) throws IOException { try (FileSKVIterator reader = FileOperations.getInstance().newReaderBuilder() - .forFile(file.toString(), fs, fs.getConf(), cs) + .forFile(file.toString(), fs, fs.getConf(), cs, status) .withTableConfiguration(context.getConfiguration()).withFileLenCache(fileLenCache) .seekToBeginning().build()) { @@ -574,12 +576,12 @@ public SortedMap computeFileToTabletMappings(FileSystem fs CompletableFuture> future = CompletableFuture.supplyAsync(() -> { try { long t1 = System.currentTimeMillis(); - List extents = - findOverlappingTablets(context, extentCache, filePath, fs, fileLensCache, cs); + List extents = findOverlappingTablets(context, extentCache, filePath, fs, + fileLensCache, cs, fileStatus); // make sure file isn't going to too many tablets checkTabletCount(maxTablets, extents.size(), filePath.toString()); Map estSizes = estimateSizes(context.getConfiguration(), filePath, - fileStatus.getLen(), extents, fs, fileLensCache, cs); + fileStatus, extents, fs, fileLensCache, cs); Map pathLocations = new HashMap<>(); for (KeyExtent ke : extents) { pathLocations.put(ke, new Bulk.FileInfo(filePath, estSizes.getOrDefault(ke, 0L))); 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 ce5c392fdcd..dd5b518d01e 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 @@ -36,6 +36,7 @@ import org.apache.accumulo.core.util.ratelimit.RateLimiter; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.mapred.FileOutputCommitter; @@ -191,12 +192,14 @@ public static class FileOptions { public final Set columnFamilies; public final boolean inclusive; public final boolean dropCacheBehind; + public final FileStatus status; public FileOptions(TableId tableId, AccumuloConfiguration tableConfiguration, String filename, FileSystem fs, Configuration fsConf, RateLimiter rateLimiter, String compression, FSDataOutputStream outputStream, boolean enableAccumuloStart, CacheProvider cacheProvider, Cache fileLenCache, boolean seekToBeginning, CryptoService cryptoService, - Range range, Set columnFamilies, boolean inclusive, boolean dropCacheBehind) { + Range range, Set columnFamilies, boolean inclusive, boolean dropCacheBehind, + FileStatus status) { this.tableId = tableId; this.tableConfiguration = tableConfiguration; this.filename = filename; @@ -214,6 +217,7 @@ public FileOptions(TableId tableId, AccumuloConfiguration tableConfiguration, St this.columnFamilies = columnFamilies; this.inclusive = inclusive; this.dropCacheBehind = dropCacheBehind; + this.status = status; } public TableId getTableId() { @@ -293,6 +297,7 @@ public static class FileHelper { private RateLimiter rateLimiter; private CryptoService cryptoService; private boolean dropCacheBehind = false; + private FileStatus status; protected FileHelper table(TableId tid) { this.tableId = tid; @@ -334,31 +339,36 @@ protected FileHelper dropCacheBehind(boolean drop) { return this; } + protected FileHelper fileStatus(FileStatus status) { + this.status = status; + return this; + } + protected FileOptions toWriterBuilderOptions(String compression, FSDataOutputStream outputStream, boolean startEnabled) { return new FileOptions(tableId, tableConfiguration, filename, fs, fsConf, rateLimiter, compression, outputStream, startEnabled, NULL_PROVIDER, null, false, cryptoService, null, - null, true, dropCacheBehind); + null, true, dropCacheBehind, status); } protected FileOptions toReaderBuilderOptions(CacheProvider cacheProvider, Cache fileLenCache, boolean seekToBeginning) { return new FileOptions(tableId, tableConfiguration, filename, fs, fsConf, rateLimiter, null, null, false, cacheProvider == null ? NULL_PROVIDER : cacheProvider, fileLenCache, - seekToBeginning, cryptoService, null, null, true, dropCacheBehind); + seekToBeginning, cryptoService, null, null, true, dropCacheBehind, status); } protected FileOptions toIndexReaderBuilderOptions(Cache fileLenCache) { return new FileOptions(tableId, tableConfiguration, filename, fs, fsConf, rateLimiter, null, null, false, NULL_PROVIDER, fileLenCache, false, cryptoService, null, null, true, - dropCacheBehind); + dropCacheBehind, status); } protected FileOptions toScanReaderBuilderOptions(Range range, Set columnFamilies, boolean inclusive) { return new FileOptions(tableId, tableConfiguration, filename, fs, fsConf, rateLimiter, null, null, false, NULL_PROVIDER, null, false, cryptoService, range, columnFamilies, inclusive, - dropCacheBehind); + dropCacheBehind, status); } protected AccumuloConfiguration getTableConfiguration() { @@ -441,6 +451,12 @@ public ReaderTableConfiguration forFile(String filename, FileSystem fs, Configur return this; } + public ReaderTableConfiguration forFile(String filename, FileSystem fs, Configuration fsConf, + CryptoService cs, FileStatus status) { + filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status); + return this; + } + @Override public ReaderBuilder withTableConfiguration(AccumuloConfiguration tableConfiguration) { tableConfiguration(tableConfiguration); @@ -509,6 +525,12 @@ public IndexReaderTableConfiguration forFile(String filename, FileSystem fs, return this; } + public IndexReaderTableConfiguration forFile(String filename, FileSystem fs, + Configuration fsConf, CryptoService cs, FileStatus status) { + filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status); + return this; + } + @Override public IndexReaderBuilder withTableConfiguration(AccumuloConfiguration tableConfiguration) { tableConfiguration(tableConfiguration); @@ -541,6 +563,12 @@ public ScanReaderTableConfiguration forFile(String filename, FileSystem fs, return this; } + public ScanReaderTableConfiguration forFile(String filename, FileSystem fs, + Configuration fsConf, CryptoService cs, FileStatus status) { + filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status); + return this; + } + @Override public ScanReaderBuilder withTableConfiguration(AccumuloConfiguration tableConfiguration) { tableConfiguration(tableConfiguration); 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 19033e4394c..2a290235811 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,6 +26,8 @@ import java.util.Collections; import java.util.Map; import java.util.Objects; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -45,7 +47,9 @@ import org.apache.accumulo.core.util.ratelimit.RateLimiter; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.FutureDataInputStreamBuilder; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.Seekable; import org.slf4j.Logger; @@ -87,29 +91,57 @@ public CachableBuilder conf(Configuration hadoopConf) { } public CachableBuilder fsPath(FileSystem fs, Path dataFile) { - return fsPath(fs, dataFile, false); + return fsPath(fs, dataFile, false, null); } - public CachableBuilder fsPath(FileSystem fs, Path dataFile, boolean dropCacheBehind) { + public CachableBuilder fsPath(FileSystem fs, Path dataFile, FileStatus status) { + return fsPath(fs, dataFile, false, status); + } + + public CachableBuilder fsPath(FileSystem fs, Path dataFile, boolean dropCacheBehind, + FileStatus status) { this.cacheId = pathToCacheId(dataFile); this.inputSupplier = () -> { - FSDataInputStream is = fs.open(dataFile); - if (dropCacheBehind) { - // Tell the DataNode that the write ahead log does not need to be cached in the OS page - // cache + FutureDataInputStreamBuilder builder = fs.openFile(dataFile); + if (status != null) { + builder.withFileStatus(status); + } + CompletableFuture future = builder.build(); + while (!future.isDone()) { try { - is.setDropBehind(Boolean.TRUE); - log.trace("Called setDropBehind(TRUE) for stream reading file {}", dataFile); - } catch (UnsupportedOperationException e) { - log.debug("setDropBehind not enabled for wal file: {}", dataFile); - } catch (IOException e) { - log.debug("IOException setting drop behind for file: {}, msg: {}", dataFile, - e.getMessage()); + Thread.sleep(10); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while opening file: " + dataFile, e); + } + } + try { + FSDataInputStream is = future.get(); + if (dropCacheBehind) { + // Tell the DataNode that the write ahead log does not need to be cached in the OS page + // cache + try { + is.setDropBehind(Boolean.TRUE); + log.trace("Called setDropBehind(TRUE) for stream reading file {}", dataFile); + } catch (UnsupportedOperationException e) { + log.debug("setDropBehind not enabled for wal file: {}", dataFile); + } catch (IOException e) { + log.debug("IOException setting drop behind for file: {}, msg: {}", dataFile, + e.getMessage()); + } } + return is; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while opening file: " + dataFile, e); + } catch (CancellationException e) { + throw new IOException("Cancelled while opening file: " + dataFile, e); + } catch (ExecutionException e) { + throw new IOException("Error trying to open file: " + dataFile, e); } - return is; }; - this.lengthSupplier = () -> fs.getFileStatus(dataFile).getLen(); + this.lengthSupplier = + () -> status == null ? fs.getFileStatus(dataFile).getLen() : status.getLen(); return this; } diff --git a/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java b/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java index c87b0716424..db930730d28 100644 --- a/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java +++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java @@ -62,7 +62,8 @@ public class RFileOperations extends FileOperations { private static RFile.Reader getReader(FileOptions options) throws IOException { CachableBuilder cb = new CachableBuilder() - .fsPath(options.getFileSystem(), new Path(options.getFilename()), options.dropCacheBehind) + .fsPath(options.getFileSystem(), new Path(options.getFilename()), options.dropCacheBehind, + options.status) .conf(options.getConfiguration()).fileLen(options.getFileLenCache()) .cacheProvider(options.cacheProvider).readLimiter(options.getRateLimiter()) .cryptoService(options.getCryptoService()); @@ -71,7 +72,11 @@ private static RFile.Reader getReader(FileOptions options) throws IOException { @Override protected long getFileSize(FileOptions options) throws IOException { - return options.getFileSystem().getFileStatus(new Path(options.getFilename())).getLen(); + if (options.status == null) { + return options.getFileSystem().getFileStatus(new Path(options.getFilename())).getLen(); + } else { + return options.status.getLen(); + } } @Override diff --git a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java index 8d224b6b450..61457d907e0 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java +++ b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java @@ -65,6 +65,7 @@ import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.conf.TableConfiguration; import org.apache.accumulo.server.fs.VolumeManager; +import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; @@ -126,6 +127,8 @@ public AssignmentStats importFiles(List files) { final Map> completeFailures = Collections.synchronizedSortedMap(new TreeMap<>()); + final Map fileStatuses = new HashMap<>(); + ClientService.Client client = null; final TabletLocator locator = TabletLocator.getLocator(context, tableId); @@ -143,8 +146,10 @@ public AssignmentStats importFiles(List files) { Runnable getAssignments = () -> { List tabletsToAssignMapFileTo = Collections.emptyList(); try { - tabletsToAssignMapFileTo = - findOverlappingTablets(context, fs, locator, mapFile, tableConf.getCryptoService()); + FileStatus status = fs.getFileStatus(mapFile); + fileStatuses.put(mapFile, status); + tabletsToAssignMapFileTo = findOverlappingTablets(context, fs, locator, mapFile, + tableConf.getCryptoService(), status); } catch (Exception ex) { log.warn("Unable to find tablets that overlap file " + mapFile, ex); } @@ -173,7 +178,7 @@ public AssignmentStats importFiles(List files) { assignmentStats.attemptingAssignments(assignments); Map> assignmentFailures = - assignMapFiles(fs, assignments, paths, numAssignThreads, numThreads); + assignMapFiles(fs, assignments, paths, numAssignThreads, numThreads, fileStatuses); assignmentStats.assignmentsFailed(assignmentFailures); Map failureCount = new TreeMap<>(); @@ -212,8 +217,9 @@ public AssignmentStats importFiles(List files) { timer.start(Timers.QUERY_METADATA); try { - tabletsToAssignMapFileTo.addAll(findOverlappingTablets(context, fs, locator, - entry.getKey(), ke, tableConf.getCryptoService())); + Path p = entry.getKey(); + tabletsToAssignMapFileTo.addAll(findOverlappingTablets(context, fs, locator, p, ke, + tableConf.getCryptoService(), fileStatuses.get(p))); keListIter.remove(); } catch (Exception ex) { log.warn("Exception finding overlapping tablets, will retry tablet " + ke, ex); @@ -228,7 +234,7 @@ public AssignmentStats importFiles(List files) { assignmentStats.attemptingAssignments(assignments); Map> assignmentFailures2 = - assignMapFiles(fs, assignments, paths, numAssignThreads, numThreads); + assignMapFiles(fs, assignments, paths, numAssignThreads, numThreads, fileStatuses); assignmentStats.assignmentsFailed(assignmentFailures2); // merge assignmentFailures2 into assignmentFailures @@ -347,15 +353,21 @@ private static List extentsOf(List locations) { } private Map> estimateSizes(final VolumeManager vm, - Map> assignments, Collection paths, int numThreads) { + Map> assignments, Collection paths, int numThreads, + Map statuses) { long t1 = System.currentTimeMillis(); final Map mapFileSizes = new TreeMap<>(); try { for (Path path : paths) { - FileSystem fs = vm.getFileSystemByPath(path); - mapFileSizes.put(path, fs.getContentSummary(path).getLength()); + FileStatus status = statuses.get(path); + if (status == null) { + FileSystem fs = vm.getFileSystemByPath(path); + mapFileSizes.put(path, fs.getContentSummary(path).getLength()); + } else { + mapFileSizes.put(path, status.getLen()); + } } } catch (IOException e) { log.error("Failed to get map files in for {}: {}", paths, e.getMessage(), e); @@ -386,9 +398,9 @@ private Map> estimateSizes(final VolumeManager vm, Path mapFile = entry.getKey(); FileSystem ns = context.getVolumeManager().getFileSystemByPath(mapFile); - estimatedSizes = BulkImport.estimateSizes(context.getConfiguration(), mapFile, - mapFileSizes.get(entry.getKey()), extentsOf(entry.getValue()), ns, null, - tableConf.getCryptoService()); + estimatedSizes = + BulkImport.estimateSizes(context.getConfiguration(), mapFile, statuses.get(mapFile), + extentsOf(entry.getValue()), ns, null, tableConf.getCryptoService()); } catch (IOException e) { log.warn("Failed to estimate map file sizes {}", e.getMessage()); } @@ -447,10 +459,10 @@ private static Map locationsOf(Map> private Map> assignMapFiles(VolumeManager fs, Map> assignments, Collection paths, int numThreads, - int numMapThreads) { + int numMapThreads, Map statuses) { timer.start(Timers.EXAMINE_MAP_FILES); Map> assignInfo = - estimateSizes(fs, assignments, paths, numMapThreads); + estimateSizes(fs, assignments, paths, numMapThreads, statuses); timer.stop(Timers.EXAMINE_MAP_FILES); Map> ret; @@ -623,15 +635,16 @@ private List assignMapFiles(ClientContext context, HostAndPort locati } public static List findOverlappingTablets(ServerContext context, VolumeManager fs, - TabletLocator locator, Path file, CryptoService cs) throws Exception { - return findOverlappingTablets(context, fs, locator, file, null, null, cs); + TabletLocator locator, Path file, CryptoService cs, FileStatus status) throws Exception { + return findOverlappingTablets(context, fs, locator, file, null, null, cs, status); } public static List findOverlappingTablets(ServerContext context, VolumeManager fs, - TabletLocator locator, Path file, KeyExtent failed, CryptoService cs) throws Exception { + TabletLocator locator, Path file, KeyExtent failed, CryptoService cs, FileStatus status) + throws Exception { locator.invalidateCache(failed); Text start = getStartRowForExtent(failed); - return findOverlappingTablets(context, fs, locator, file, start, failed.endRow(), cs); + return findOverlappingTablets(context, fs, locator, file, start, failed.endRow(), cs, status); } protected static Text getStartRowForExtent(KeyExtent extent) { @@ -648,16 +661,16 @@ protected static Text getStartRowForExtent(KeyExtent extent) { static final byte[] byte0 = {0}; public static List findOverlappingTablets(ServerContext context, VolumeManager vm, - TabletLocator locator, Path file, Text startRow, Text endRow, CryptoService cs) - throws Exception { + TabletLocator locator, Path file, Text startRow, Text endRow, CryptoService cs, + FileStatus status) throws Exception { List result = new ArrayList<>(); Collection columnFamilies = Collections.emptyList(); String filename = file.toString(); // log.debug(filename + " finding overlapping tablets " + startRow + " -> " + endRow); FileSystem fs = vm.getFileSystemByPath(file); - try (FileSKVIterator reader = - FileOperations.getInstance().newReaderBuilder().forFile(filename, fs, fs.getConf(), cs) - .withTableConfiguration(context.getConfiguration()).seekToBeginning().build()) { + try (FileSKVIterator reader = FileOperations.getInstance().newReaderBuilder() + .forFile(filename, fs, fs.getConf(), cs, status) + .withTableConfiguration(context.getConfiguration()).seekToBeginning().build()) { Text row = startRow; if (row == null) { row = new Text(); diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/FileUtil.java b/server/base/src/main/java/org/apache/accumulo/server/util/FileUtil.java index 3c189644df8..3e909b330f6 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/FileUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/FileUtil.java @@ -48,6 +48,7 @@ import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.conf.TableConfiguration; import org.apache.accumulo.server.fs.VolumeManager; +import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; @@ -434,14 +435,15 @@ private static long countIndexEntries(ServerContext context, TableConfiguration FileSKVIterator reader = null; Path path = new Path(file); FileSystem ns = context.getVolumeManager().getFileSystemByPath(path); + FileStatus status = ns.getFileStatus(path); try { if (useIndex) { reader = FileOperations.getInstance().newIndexReaderBuilder() - .forFile(path.toString(), ns, ns.getConf(), tableConf.getCryptoService()) + .forFile(path.toString(), ns, ns.getConf(), tableConf.getCryptoService(), status) .withTableConfiguration(tableConf).build(); } else { reader = FileOperations.getInstance().newScanReaderBuilder() - .forFile(path.toString(), ns, ns.getConf(), tableConf.getCryptoService()) + .forFile(path.toString(), ns, ns.getConf(), tableConf.getCryptoService(), status) .withTableConfiguration(tableConf) .overRange(new Range(prevEndRow, false, null, true), Set.of(), false).build(); } @@ -468,11 +470,11 @@ private static long countIndexEntries(ServerContext context, TableConfiguration if (useIndex) { readers.add(FileOperations.getInstance().newIndexReaderBuilder() - .forFile(path.toString(), ns, ns.getConf(), tableConf.getCryptoService()) + .forFile(path.toString(), ns, ns.getConf(), tableConf.getCryptoService(), status) .withTableConfiguration(tableConf).build()); } else { readers.add(FileOperations.getInstance().newScanReaderBuilder() - .forFile(path.toString(), ns, ns.getConf(), tableConf.getCryptoService()) + .forFile(path.toString(), ns, ns.getConf(), tableConf.getCryptoService(), status) .withTableConfiguration(tableConf) .overRange(new Range(prevEndRow, false, null, true), Set.of(), false).build()); } diff --git a/server/base/src/test/java/org/apache/accumulo/server/client/BulkImporterTest.java b/server/base/src/test/java/org/apache/accumulo/server/client/BulkImporterTest.java index 9b69efa1963..5e0647d6971 100644 --- a/server/base/src/test/java/org/apache/accumulo/server/client/BulkImporterTest.java +++ b/server/base/src/test/java/org/apache/accumulo/server/client/BulkImporterTest.java @@ -144,8 +144,8 @@ public void testFindOverlappingTablets() throws Exception { writer.append(new Key("xyzzy", "cf", "cq"), empty); writer.close(); try (var vm = VolumeManagerImpl.getLocalForTesting("file:///")) { - List overlaps = - BulkImporter.findOverlappingTablets(context, vm, locator, new Path(file), null, null, cs); + List overlaps = BulkImporter.findOverlappingTablets(context, vm, locator, + new Path(file), null, null, cs, null); assertEquals(5, overlaps.size()); Collections.sort(overlaps); assertEquals(new KeyExtent(tableId, new Text("a"), null), overlaps.get(0).tablet_extent); @@ -158,7 +158,7 @@ public void testFindOverlappingTablets() throws Exception { assertEquals(new KeyExtent(tableId, null, new Text("l")), overlaps.get(4).tablet_extent); List overlaps2 = BulkImporter.findOverlappingTablets(context, vm, locator, - new Path(file), new KeyExtent(tableId, new Text("h"), new Text("b")), cs); + new Path(file), new KeyExtent(tableId, new Text("h"), new Text("b")), cs, null); assertEquals(3, overlaps2.size()); assertEquals(new KeyExtent(tableId, new Text("d"), new Text("cm")), overlaps2.get(0).tablet_extent); 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 f2a1a5c990d..b56612314a6 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 @@ -26,8 +26,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map.Entry; -import java.util.SortedSet; -import java.util.TreeSet; +import java.util.TreeMap; import org.apache.accumulo.core.crypto.CryptoEnvironmentImpl; import org.apache.accumulo.core.data.Key; @@ -80,7 +79,7 @@ public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, L for (Path logDir : recoveryLogDirs) { LOG.debug("Opening recovery log dir {}", logDir.getName()); - SortedSet logFiles = getFiles(vm, logDir); + TreeMap logFiles = getFiles(vm, logDir); var fs = vm.getFileSystemByPath(logDir); // only check the first key once to prevent extra iterator creation and seeking @@ -88,9 +87,9 @@ public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, L validateFirstKey(context, cryptoService, fs, logFiles, logDir); } - for (Path log : logFiles) { + for (Entry entry : logFiles.entrySet()) { FileSKVIterator fileIter = FileOperations.getInstance().newReaderBuilder() - .forFile(log.toString(), fs, fs.getConf(), cryptoService) + .forFile(entry.getKey().toString(), fs, fs.getConf(), cryptoService, entry.getValue()) .withTableConfiguration(context.getConfiguration()).seekToBeginning().build(); if (range != null) { fileIter.seek(range, Collections.emptySet(), false); @@ -98,11 +97,13 @@ public RecoveryLogsIterator(ServerContext context, List recoveryLogDirs, L Iterator> scanIter = new IteratorAdapter(fileIter); if (scanIter.hasNext()) { - LOG.debug("Write ahead log {} has data in range {} {}", log.getName(), start, end); + LOG.debug("Write ahead log {} has data in range {} {}", entry.getKey().getName(), start, + end); iterators.add(scanIter); fileIters.add(fileIter); } else { - LOG.debug("Write ahead log {} has no data in range {} {}", log.getName(), start, end); + LOG.debug("Write ahead log {} has no data in range {} {}", entry.getKey().getName(), + start, end); fileIter.close(); } } @@ -137,12 +138,12 @@ public void close() throws IOException { /** * Check for sorting signal files (finished/failed) and get the logs in the provided directory. */ - private SortedSet getFiles(VolumeManager fs, Path directory) throws IOException { + private TreeMap getFiles(VolumeManager fs, Path 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. - SortedSet logFiles = new TreeSet<>(Comparator.comparing(Path::getName)); + TreeMap logFiles = new TreeMap<>(Comparator.comparing(Path::getName)); for (FileStatus child : fs.listStatus(directory)) { if (child.getPath().getName().startsWith("_")) { continue; @@ -156,7 +157,7 @@ private SortedSet getFiles(VolumeManager fs, Path directory) throws IOExce } FileSystem ns = fs.getFileSystemByPath(child.getPath()); Path fullLogPath = ns.makeQualified(child.getPath()); - logFiles.add(fullLogPath); + logFiles.put(fullLogPath, child); } if (!foundFinish) { throw new IOException( @@ -169,9 +170,10 @@ private SortedSet getFiles(VolumeManager fs, Path directory) throws IOExce * 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, - SortedSet logFiles, Path fullLogPath) throws IOException { + TreeMap logFiles, Path fullLogPath) throws IOException { + Entry first = logFiles.firstEntry(); try (FileSKVIterator fileIter = FileOperations.getInstance().newReaderBuilder() - .forFile(logFiles.first().toString(), fs, fs.getConf(), cs) + .forFile(first.getKey().toString(), fs, fs.getConf(), cs, first.getValue()) .withTableConfiguration(context.getConfiguration()).seekToBeginning().build()) { Iterator> iterator = new IteratorAdapter(fileIter);