Merge branch '3.1'

Author: ddanielr

server/gc/src/main/java/org/apache/accumulo/gc/GCRun.java

@@ -26,6 +26,7 @@
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -37,8 +38,9 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.SortedMap;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Stream;
 
 import org.apache.accumulo.core.Constants;
@@ -63,6 +65,7 @@
 import org.apache.accumulo.core.metadata.schema.TabletMetadata;
 import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
 import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.Timer;
 import org.apache.accumulo.core.util.threads.ThreadPools;
 import org.apache.accumulo.core.volume.Volume;
 import org.apache.accumulo.server.ServerContext;
@@ -88,16 +91,19 @@ public class GCRun implements GarbageCollectionEnvironment {
   private final Ample.DataLevel level;
   private final ServerContext context;
   private final AccumuloConfiguration config;
+  private final Duration loggingInterval = Duration.ofMinutes(1);
   private long candidates = 0;
   private long inUse = 0;
   private long deleted = 0;
   private long errors = 0;
+  private AtomicInteger batchCount;
 
   public GCRun(Ample.DataLevel level, ServerContext context) {
     this.log = LoggerFactory.getLogger(GCRun.class.getName() + "." + level.name());
     this.level = level;
     this.context = context;
     this.config = context.getConfiguration();
+    this.batchCount = new AtomicInteger(0);
   }
 
   @Override
@@ -125,7 +131,8 @@ public void deleteGcCandidates(Collection<GcCandidate> gcCandidates, GcCandidate
       return;
     }
 
-    log.info("Attempting to delete gcCandidates of type {} from metadata", type);
+    log.info("Batch {} attempting to delete {} gcCandidates of type {} from metadata",
+        batchCount.get(), gcCandidates.size(), type);
     context.getAmple().deleteGcCandidates(level, gcCandidates, type);
   }
 
@@ -136,6 +143,7 @@ public List<GcCandidate> readCandidatesThatFitInMemory(Iterator<GcCandidate> can
     long candidateBatchSize = getCandidateBatchSize() / 2;
 
     List<GcCandidate> candidatesBatch = new ArrayList<>();
+    batchCount.incrementAndGet();
 
     while (candidates.hasNext()) {
       GcCandidate candidate = candidates.next();
@@ -279,15 +287,17 @@ public void deleteConfirmedCandidates(SortedMap<String,GcCandidate> confirmedDel
 
     List<GcCandidate> processedDeletes = Collections.synchronizedList(new ArrayList<>());
 
-    minimizeDeletes(confirmedDeletes, processedDeletes, fs, log);
+    minimizeDeletes(confirmedDeletes, processedDeletes, fs, log, loggingInterval);
 
-    ExecutorService deleteThreadPool = ThreadPools.getServerThreadPools()
+    ThreadPoolExecutor deleteThreadPool = ThreadPools.getServerThreadPools()
         .createExecutorService(config, Property.GC_DELETE_THREADS);
 
     final Map<Path,Path> replacements = context.getVolumeReplacements();
 
+    log.info("Batch {} attempting to delete {} gcCandidate files", batchCount.get(),
+        confirmedDeletes.size());
+    Timer timer = Timer.startNew();
     for (final GcCandidate delete : confirmedDeletes.values()) {
-
       Runnable deleteTask = () -> {
         boolean removeFlag = false;
 
@@ -311,7 +321,7 @@ public void deleteConfirmedCandidates(SortedMap<String,GcCandidate> confirmedDel
           }
 
           for (Path pathToDel : GcVolumeUtil.expandAllVolumesUri(fs, fullPath)) {
-            log.debug("{} Deleting {}", fileActionPrefix, pathToDel);
+            log.debug("Batch {} {} Deleting {}", batchCount.get(), fileActionPrefix, pathToDel);
 
             if (moveToTrash(pathToDel) || fs.deleteRecursively(pathToDel)) {
               // delete succeeded, still want to delete
@@ -364,7 +374,12 @@ public void deleteConfirmedCandidates(SortedMap<String,GcCandidate> confirmedDel
     deleteThreadPool.shutdown();
 
     try {
-      while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) { // empty
+      while (!deleteThreadPool.awaitTermination(1000, TimeUnit.MILLISECONDS)) {
+        if (timer.hasElapsed(loggingInterval)) {
+          log.info("Batch {} deleting file {} of {}", batchCount.get(),
+              deleteThreadPool.getCompletedTaskCount(), confirmedDeletes.size());
+          timer.restart();
+        }
       }
     } catch (InterruptedException e1) {
       log.error("{}", e1.getMessage(), e1);
@@ -408,7 +423,8 @@ public void incrementInUseStat(long i) {
 
   @VisibleForTesting
   static void minimizeDeletes(SortedMap<String,GcCandidate> confirmedDeletes,
-      List<GcCandidate> processedDeletes, VolumeManager fs, Logger logger) {
+      List<GcCandidate> processedDeletes, VolumeManager fs, Logger logger,
+      Duration loggingInterval) {
     Set<Path> seenVolumes = new HashSet<>();
 
     // when deleting a dir and all files in that dir, only need to delete the dir.
@@ -418,7 +434,11 @@ static void minimizeDeletes(SortedMap<String,GcCandidate> confirmedDeletes,
 
     String lastDirRel = null;
     Path lastDirAbs = null;
+    Timer progressTimer = Timer.startNew();
+    int progressCount = 0;
+    int totalDeletes = confirmedDeletes.size();
     while (cdIter.hasNext()) {
+      progressCount++;
       Map.Entry<String,GcCandidate> entry = cdIter.next();
       String relPath = entry.getKey();
       Path absPath = new Path(entry.getValue().getPath());
@@ -458,6 +478,10 @@ static void minimizeDeletes(SortedMap<String,GcCandidate> confirmedDeletes,
           lastDirAbs = null;
         }
       }
+      if (progressTimer.hasElapsed(loggingInterval)) {
+        logger.debug("Minimizing delete {} of {}", progressCount, totalDeletes);
+        progressTimer.restart();
+      }
     }
   }
 

server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java

@@ -344,6 +344,7 @@ public long collect(GarbageCollectionEnvironment gce)
 
     Iterator<GcCandidate> candidatesIter = gce.getCandidates();
     long totalBlips = 0;
+    int batchCount = 0;
 
     while (candidatesIter.hasNext()) {
       List<GcCandidate> batchOfCandidates;
@@ -356,19 +357,21 @@ public long collect(GarbageCollectionEnvironment gce)
       } finally {
         candidatesSpan.end();
       }
-      totalBlips = deleteBatch(gce, batchOfCandidates);
+      batchCount++;
+      totalBlips = deleteBatch(gce, batchOfCandidates, batchCount);
     }
     return totalBlips;
   }
 
   /**
    * Given a sub-list of possible deletion candidates, process and remove valid deletion candidates.
    */
-  private long deleteBatch(GarbageCollectionEnvironment gce, List<GcCandidate> currentBatch)
-      throws InterruptedException, TableNotFoundException, IOException {
+  private long deleteBatch(GarbageCollectionEnvironment gce, List<GcCandidate> currentBatch,
+      int batchCount) throws InterruptedException, TableNotFoundException, IOException {
 
     long origSize = currentBatch.size();
     gce.incrementCandidatesStat(origSize);
+    log.info("Batch {} total deletion candidates: {}", batchCount, origSize);
 
     SortedMap<String,GcCandidate> candidateMap = makeRelative(currentBatch);
 

server/gc/src/test/java/org/apache/accumulo/gc/SimpleGarbageCollectorTest.java

@@ -31,6 +31,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.FileNotFoundException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -181,7 +182,7 @@ public void testMinimizeDeletes() {
 
     List<GcCandidate> processedDeletes = new ArrayList<>();
 
-    GCRun.minimizeDeletes(confirmed, processedDeletes, volMgr2, log);
+    GCRun.minimizeDeletes(confirmed, processedDeletes, volMgr2, log, Duration.ofMinutes(1));
 
     TreeMap<String,GcCandidate> expected = new TreeMap<>();
     expected.put("5a/t-0001", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001", 0L));