This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/main by this push:
new ac76a1b6f5 Added VolumeManager.deleteBulk method to use
FileSystem.createBulkDelete (#6443)
ac76a1b6f5 is described below
commit ac76a1b6f5bb317f6a00e0309f1e5e5f2ca85ce1
Author: Dave Marion <[email protected]>
AuthorDate: Mon Jul 20 11:49:15 2026 -0400
Added VolumeManager.deleteBulk method to use FileSystem.createBulkDelete
(#6443)
Added method to VolumeManager that accepts a collection of Paths
to delete. If the underlying FileSystem supports bulk deletion,
then that API method is called. Otherwise the Paths are deleted
using an ExecutorService.
Reviewed the existing code that used the other delete methods in
the VolumeManager. Only two places seemed like candidates for using
the new deleteBulk method. The other locations either only deleted
one file aperiodically, or deleted many files but using some complex
logic. The former didn't make sense to modify and the latter could
potentially be modified later with some refactoring.
Closes #5131
---
.../apache/accumulo/server/fs/VolumeManager.java | 15 ++
.../accumulo/server/fs/VolumeManagerImpl.java | 158 +++++++++++++++++++++
.../server/util/FindCompactionTmpFiles.java | 85 ++++-------
.../tableOps/bulkVer2/CleanUpBulkImport.java | 4 +-
4 files changed, 199 insertions(+), 63 deletions(-)
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java
b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java
index c1ef3df0de..caa8b9a8bc 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManager.java
@@ -51,6 +51,10 @@ import org.slf4j.LoggerFactory;
*/
public interface VolumeManager extends AutoCloseable {
+ enum DeleteStatus {
+ ERROR, FALSE, TRUE
+ };
+
enum FileType {
TABLE(Constants.TABLE_DIR), WAL(Constants.WAL_DIR),
RECOVERY(Constants.RECOVERY_DIR);
@@ -132,6 +136,17 @@ public interface VolumeManager extends AutoCloseable {
// delete a directory and anything under it
boolean deleteRecursively(Path path) throws IOException;
+ /**
+ * Deletes a collection of files by grouping the files by FileSystem, then
either calling
+ * {@code FileSystem#createBulkDelete(Path)} on FileSystem implementations
that support it or by
+ * calling {@code #delete(Path)}.
+ *
+ * @param paths paths of files to delete
+ * @return paths map of input path to delete state
+ * @throws IOException on any exception deleting files
+ */
+ Map<Path,DeleteStatus> deleteBulk(Collection<Path> paths) throws IOException;
+
// forward to the appropriate FileSystem object
boolean exists(Path path) throws IOException;
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
index 828200b040..636a52b6c5 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeManagerImpl.java
@@ -29,13 +29,17 @@ import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -53,6 +57,7 @@ import org.apache.accumulo.core.volume.Volume;
import org.apache.accumulo.core.volume.VolumeConfiguration;
import org.apache.accumulo.core.volume.VolumeImpl;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BulkDelete;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.FSDataInputStream;
@@ -73,6 +78,7 @@ import org.slf4j.LoggerFactory;
import com.github.benmanes.caffeine.cache.Cache;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
+import com.google.common.util.concurrent.MoreExecutors;
public class VolumeManagerImpl implements VolumeManager {
@@ -86,6 +92,7 @@ public class VolumeManagerImpl implements VolumeManager {
private final Map<String,Volume> volumesByName;
private final Multimap<URI,Volume> volumesByFileSystemUri;
+ private final Set<Volume> bulkDeleteVolumes;
private final VolumeChooser chooser;
private final AccumuloConfiguration conf;
private final Configuration hadoopConf;
@@ -95,6 +102,29 @@ public class VolumeManagerImpl implements VolumeManager {
this.volumesByName = volumes;
// We may have multiple directories used in a single FileSystem (e.g.
testing)
this.volumesByFileSystemUri = invertVolumesByFileSystem(volumesByName);
+
+ this.bulkDeleteVolumes = new HashSet<>();
+ for (Volume v : volumes.values()) {
+ FileSystem fs = v.getFileSystem();
+ String base = v.getBasePath().isBlank() ? "/" : v.getBasePath();
+ Path basePath = fs.makeQualified(new Path(base));
+ try {
+ if (fs.hasPathCapability(basePath, "fs.capability.bulk.delete")) {
+ try (BulkDelete bulk = fs.createBulkDelete(basePath)) {
+ // Don't use the BulkDelete API if the page size is only 1.
+ // The DefaultBulkDeleteOperation implementation has a page size
of 1.
+ // The S3A FileSystem implementation has a default of 250 and
+ // is configured by the property `fs.s3a.bulk.delete.page.size`.
+ if (bulk.pageSize() > 1) {
+ this.bulkDeleteVolumes.add(v);
+ }
+ }
+ }
+ } catch (IllegalArgumentException | IOException e) {
+ log.warn("Error determining bulk delete capability for volume {}",
v.getBasePath(), e);
+ }
+ }
+
ensureSyncIsEnabled();
// if they supplied a property and we cannot load it, then fail hard
VolumeChooser chooser1;
@@ -213,6 +243,109 @@ public class VolumeManagerImpl implements VolumeManager {
return getFileSystemByPath(path).delete(path, true);
}
+ @Override
+ public Map<Path,DeleteStatus> deleteBulk(Collection<Path> paths) throws
IOException {
+
+ requireNonNull(paths);
+
+ if (paths.isEmpty()) {
+ return Map.of();
+ }
+
+ if (paths.size() == 1) {
+ Path p = paths.iterator().next();
+ try {
+ if (delete(p)) {
+ return Map.of(p, DeleteStatus.TRUE);
+ } else {
+ return Map.of(p, DeleteStatus.FALSE);
+ }
+ } catch (IOException e) {
+ return Map.of(p, DeleteStatus.ERROR);
+ }
+ }
+
+ final Map<Volume,ArrayList<Path>> pathsSupportingBulkDelete = new
HashMap<>();
+ final List<Path> pathsNotSupportingBulkDelete = new ArrayList<>();
+ final Map<Path,DeleteStatus> results = new ConcurrentHashMap<>();
+ final List<Future<Void>> futures = new LinkedList<>();
+
+ for (Path p : paths) {
+ Volume v = getVolumeForPath(p);
+ if (v != null && this.bulkDeleteVolumes.contains(v)) {
+ pathsSupportingBulkDelete.computeIfAbsent(v, (k) -> new
ArrayList<Path>()).add(p);
+ } else {
+ pathsNotSupportingBulkDelete.add(p);
+ }
+ }
+
+ if (!pathsNotSupportingBulkDelete.isEmpty()) {
+ final ExecutorService delSvc;
+ if (pathsNotSupportingBulkDelete.size() < 4) {
+ // Do not bother creating a thread pool and threads for a few files.
+ delSvc = MoreExecutors.newDirectExecutorService();
+ } else {
+ delSvc = Executors.newFixedThreadPool(8);
+ }
+ pathsNotSupportingBulkDelete.forEach(p -> {
+ futures.add(delSvc.submit(() -> {
+ try {
+ if (delete(p)) {
+ results.put(p, DeleteStatus.TRUE);
+ } else {
+ results.put(p, DeleteStatus.FALSE);
+ }
+ return null;
+ } catch (IOException e) {
+ log.error("Error deleting file at {}", p, e);
+ results.put(p, DeleteStatus.ERROR);
+ return null;
+ }
+ }));
+ });
+ delSvc.shutdown();
+ }
+
+ for (Entry<Volume,ArrayList<Path>> e :
pathsSupportingBulkDelete.entrySet()) {
+ Volume v = e.getKey();
+ List<Path> deletes = e.getValue();
+ FileSystem fs = v.getFileSystem();
+ Path basePath = fs.makeQualified(new Path(v.getBasePath()));
+ try (BulkDelete bulk = fs.createBulkDelete(basePath)) {
+ int batchSize = bulk.pageSize();
+ for (int i = 0; i <= deletes.size(); i += batchSize) {
+ List<Path> subset = deletes.subList(i, Math.min(i + batchSize,
deletes.size()));
+ List<Entry<Path,String>> errors = bulk.bulkDelete(subset);
+ errors.forEach((entry) -> {
+ log.error("Failed to delete file at {}, reason: {}",
entry.getKey(), entry.getValue());
+ results.put(entry.getKey(), DeleteStatus.ERROR);
+ if (!subset.remove(entry.getKey())) {
+ log.error("Did not find error path {} in input set {}",
entry.getKey(), subset);
+ }
+ });
+ subset.forEach(success -> results.put(success, DeleteStatus.TRUE));
+ }
+ }
+ }
+
+ while (!futures.isEmpty()) {
+ Iterator<Future<Void>> iter = futures.iterator();
+ while (iter.hasNext()) {
+ Future<Void> f = iter.next();
+ if (f.isDone()) {
+ iter.remove();
+ }
+ }
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ return results;
+ }
+
protected void ensureSyncIsEnabled() {
for (Entry<String,Volume> entry : volumesByName.entrySet()) {
FileSystem fs = entry.getValue().getFileSystem();
@@ -284,6 +417,31 @@ public class VolumeManagerImpl implements VolumeManager {
}
}
+ private Volume getVolumeForPath(Path path) {
+ FileSystem desiredFs;
+ try {
+ Configuration volumeConfig = hadoopConf;
+ for (String vol : volumesByName.keySet()) {
+ if (path.toString().startsWith(vol)) {
+ volumeConfig = getVolumeManagerConfiguration(conf, hadoopConf, vol);
+ break;
+ }
+ }
+ desiredFs = requireNonNull(path).getFileSystem(volumeConfig);
+ } catch (IOException ex) {
+ throw new UncheckedIOException(ex);
+ }
+ URI desiredFsUri = desiredFs.getUri();
+ Collection<Volume> candidateVolumes =
volumesByFileSystemUri.get(desiredFsUri);
+ if (candidateVolumes != null) {
+ return candidateVolumes.stream().filter(volume ->
volume.containsPath(path)).findFirst()
+ .orElse(null);
+ } else {
+ log.debug("Could not determine volume for Path: {}", path);
+ return null;
+ }
+ }
+
@Override
public RemoteIterator<LocatedFileStatus> listFiles(final Path path, final
boolean recursive)
throws IOException {
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/util/FindCompactionTmpFiles.java
b/server/base/src/main/java/org/apache/accumulo/server/util/FindCompactionTmpFiles.java
index b4ccf99425..3237cd9f8b 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/util/FindCompactionTmpFiles.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/util/FindCompactionTmpFiles.java
@@ -20,12 +20,13 @@ package org.apache.accumulo.server.util;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
-import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.ExecutionException;
@@ -46,6 +47,7 @@ import org.apache.accumulo.core.trace.TraceUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.volume.Volume;
import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.fs.VolumeManager.DeleteStatus;
import org.apache.accumulo.server.util.FindCompactionTmpFiles.FindOpts;
import org.apache.accumulo.start.spi.CommandGroup;
import org.apache.accumulo.start.spi.CommandGroups;
@@ -59,7 +61,6 @@ import org.slf4j.LoggerFactory;
import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import com.google.auto.service.AutoService;
-import com.google.common.util.concurrent.MoreExecutors;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.context.Scope;
@@ -176,71 +177,33 @@ public class FindCompactionTmpFiles extends
ServerKeywordExecutable<FindOpts> {
}
}
- private static boolean deleteTmpFile(ServerContext context, Path p) throws
IOException {
- if (context.getVolumeManager().exists(p)) {
- boolean result = context.getVolumeManager().delete(p);
- if (result) {
- LOG.debug("Removed old temp file {}", p);
- } else {
- LOG.error("Unable to remove old temp file {}, operation returned false
with no exception",
- p);
- }
- return result;
- }
- return true;
- }
-
public static DeleteStats deleteTempFiles(ServerContext context, Set<Path>
filesToDelete) {
- final ExecutorService delSvc;
- if (filesToDelete.size() < 4) {
- // Do not bother creating a thread pool and threads for a few files.
- delSvc = MoreExecutors.newDirectExecutorService();
- } else {
- delSvc = Executors.newFixedThreadPool(8);
- }
-
final DeleteStats stats = new DeleteStats();
-
- // use a linked list to make removal from the middle of the list quick
- final List<Future<Boolean>> futures = new LinkedList<>();
-
- filesToDelete.forEach(p -> {
- futures.add(delSvc.submit(() -> deleteTmpFile(context, p)));
- });
- delSvc.shutdown();
-
try {
- int expectedResponses = filesToDelete.size();
- while (expectedResponses > 0) {
- Iterator<Future<Boolean>> iter = futures.iterator();
- while (iter.hasNext()) {
- Future<Boolean> future = iter.next();
- if (future.isDone()) {
- expectedResponses--;
- iter.remove();
- try {
- if (future.get()) {
- stats.success++;
- } else {
- stats.failure++;
- }
- } catch (ExecutionException e) {
- stats.error++;
- LOG.error("Error deleting a compaction tmp file", e);
- }
- }
- }
- if (expectedResponses > 0) {
- LOG.debug("Waiting on {} background delete operations",
expectedResponses);
- UtilWaitThread.sleep(1_000);
+ Map<Path,DeleteStatus> results =
context.getVolumeManager().deleteBulk(filesToDelete);
+ results.forEach((k, v) -> {
+ switch (v) {
+ case ERROR:
+ LOG.error("Error deleting a compaction tmp file {}", k);
+ stats.error++;
+ break;
+ case FALSE:
+ LOG.error(
+ "Unable to remove old temp file {}, operation returned false
with no exception", k);
+ stats.failure++;
+ break;
+ case TRUE:
+ LOG.debug("Removed old temp file {}", k);
+ stats.success++;
+ break;
+ default:
+ break;
}
- }
- delSvc.awaitTermination(10, TimeUnit.MINUTES);
+ });
return stats;
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new IllegalStateException(e);
+ } catch (IOException e) {
+ throw new UncheckedIOException("Error in VolumeManager.deleteBulk", e);
}
}
diff --git
a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/CleanUpBulkImport.java
b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/CleanUpBulkImport.java
index d1ca90e67b..ea3fb97eff 100644
---
a/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/CleanUpBulkImport.java
+++
b/server/manager/src/main/java/org/apache/accumulo/manager/tableOps/bulkVer2/CleanUpBulkImport.java
@@ -23,6 +23,7 @@ import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.data.AbstractId;
@@ -78,8 +79,7 @@ public class CleanUpBulkImport extends
AbstractBulkFateOperation {
Path renamingFile = new Path(bulkDir, Constants.BULK_RENAME_FILE);
Path mappingFile = new Path(bulkDir, Constants.BULK_LOAD_MAPPING);
try {
- env.getVolumeManager().delete(renamingFile);
- env.getVolumeManager().delete(mappingFile);
+ env.getVolumeManager().deleteBulk(Set.of(renamingFile, mappingFile));
} catch (IOException ioe) {
log.debug("{} Failed to delete renames and/or loadmap", fateId, ioe);
}
