removed queued jobs for a tablet if no new jobs are seen (apache#4394)

fixes apache#3528

Author: keith-turner

server/manager/src/main/java/org/apache/accumulo/manager/TabletGroupWatcher.java

@@ -678,7 +678,9 @@ public void run() {
         eventHandler.clearNeedsFullScan();
 
         iter = store.iterator(tableMgmtParams);
+        manager.getCompactionCoordinator().getJobQueues().beginFullScan(store.getLevel());
         var tabletMgmtStats = manageTablets(iter, tableMgmtParams, currentTServers, true);
+        manager.getCompactionCoordinator().getJobQueues().endFullScan(store.getLevel());
 
         // If currently looking for volume replacements, determine if the next round needs to look.
         if (lookForTabletsNeedingVolReplacement) {

server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobPriorityQueue.java

@@ -23,18 +23,23 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.metadata.schema.Ample;
 import org.apache.accumulo.core.metadata.schema.TabletMetadata;
 import org.apache.accumulo.core.spi.compaction.CompactionJob;
 import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
 import org.apache.accumulo.core.util.compaction.CompactionJobPrioritizer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.google.common.base.Preconditions;
 
@@ -49,6 +54,8 @@
  */
 public class CompactionJobPriorityQueue {
 
+  private static final Logger log = LoggerFactory.getLogger(CompactionJobPriorityQueue.class);
+
   private final CompactorGroupId groupId;
 
   private class CjpqKey implements Comparable<CjpqKey> {
@@ -99,9 +106,19 @@ public boolean equals(Object o) {
   private final AtomicLong rejectedJobs;
   private final AtomicLong dequeuedJobs;
 
+  private static class TabletJobs {
+    final long generation;
+    final HashSet<CjpqKey> jobs;
+
+    private TabletJobs(long generation, HashSet<CjpqKey> jobs) {
+      this.generation = generation;
+      this.jobs = jobs;
+    }
+  }
+
   // This map tracks what jobs a tablet currently has in the queue. Its used to efficiently remove
   // jobs in the queue when new jobs are queued for a tablet.
-  private final Map<KeyExtent,List<CjpqKey>> tabletJobs;
+  private final Map<KeyExtent,TabletJobs> tabletJobs;
 
   private final AtomicLong nextSeq = new AtomicLong(0);
 
@@ -116,24 +133,46 @@ public CompactionJobPriorityQueue(CompactorGroupId groupId, int maxSize) {
     this.dequeuedJobs = new AtomicLong(0);
   }
 
+  public synchronized void removeOlderGenerations(Ample.DataLevel level, long currGeneration) {
+    if (closed.get()) {
+      return;
+    }
+
+    List<KeyExtent> removals = new ArrayList<>();
+
+    tabletJobs.forEach((extent, jobs) -> {
+      if (Ample.DataLevel.of(extent.tableId()) == level && jobs.generation < currGeneration) {
+        removals.add(extent);
+      }
+    });
+
+    if (!removals.isEmpty()) {
+      log.trace("Removed {} queued tablets that no longer need compaction for {} {}",
+          removals.size(), groupId, level);
+    }
+
+    removals.forEach(this::removePreviousSubmissions);
+  }
+
   /**
    * @return the number of jobs added. If the queue is closed returns -1
    */
-  public synchronized int add(TabletMetadata tabletMetadata, Collection<CompactionJob> jobs) {
+  public synchronized int add(TabletMetadata tabletMetadata, Collection<CompactionJob> jobs,
+      long generation) {
     Preconditions.checkArgument(jobs.stream().allMatch(job -> job.getGroup().equals(groupId)));
     if (closed.get()) {
       return -1;
     }
 
     removePreviousSubmissions(tabletMetadata.getExtent());
 
-    List<CjpqKey> newEntries = new ArrayList<>(jobs.size());
+    HashSet<CjpqKey> newEntries = new HashSet<>(jobs.size());
 
     int jobsAdded = 0;
     for (CompactionJob job : jobs) {
       CjpqKey cjqpKey = addJobToQueue(tabletMetadata, job);
       if (cjqpKey != null) {
-        newEntries.add(cjqpKey);
+        checkState(newEntries.add(cjqpKey));
         jobsAdded++;
       } else {
         // The priority for this job was lower than all other priorities and not added
@@ -143,7 +182,8 @@ public synchronized int add(TabletMetadata tabletMetadata, Collection<Compaction
     }
 
     if (!newEntries.isEmpty()) {
-      checkState(tabletJobs.put(tabletMetadata.getExtent(), newEntries) == null);
+      checkState(tabletJobs.put(tabletMetadata.getExtent(), new TabletJobs(generation, newEntries))
+          == null);
     }
 
     return jobsAdded;
@@ -178,7 +218,7 @@ public synchronized CompactionJobQueues.MetaJob poll() {
     if (first != null) {
       dequeuedJobs.getAndIncrement();
       var extent = first.getValue().getTabletMetadata().getExtent();
-      List<CjpqKey> jobs = tabletJobs.get(extent);
+      Set<CjpqKey> jobs = tabletJobs.get(extent).jobs;
       checkState(jobs.remove(first.getKey()));
       if (jobs.isEmpty()) {
         tabletJobs.remove(extent);
@@ -207,9 +247,9 @@ public synchronized boolean closeIfEmpty() {
   }
 
   private void removePreviousSubmissions(KeyExtent extent) {
-    List<CjpqKey> prevJobs = tabletJobs.get(extent);
+    TabletJobs prevJobs = tabletJobs.get(extent);
     if (prevJobs != null) {
-      prevJobs.forEach(jobQueue::remove);
+      prevJobs.jobs.forEach(jobQueue::remove);
       tabletJobs.remove(extent);
     }
   }

server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueues.java

@@ -19,10 +19,15 @@
 package org.apache.accumulo.manager.compaction.queue;
 
 import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentHashMap.KeySetView;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
+import org.apache.accumulo.core.metadata.schema.Ample.DataLevel;
 import org.apache.accumulo.core.metadata.schema.TabletMetadata;
 import org.apache.accumulo.core.spi.compaction.CompactionJob;
 import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
@@ -46,8 +51,35 @@ public class CompactionJobQueues {
 
   private final int queueSize;
 
+  private final Map<DataLevel,AtomicLong> currentGenerations;
+
   public CompactionJobQueues(int queueSize) {
     this.queueSize = queueSize;
+    Map<DataLevel,AtomicLong> cg = new EnumMap<>(DataLevel.class);
+    for (var level : DataLevel.values()) {
+      cg.put(level, new AtomicLong());
+    }
+    currentGenerations = Collections.unmodifiableMap(cg);
+
+  }
+
+  public void beginFullScan(DataLevel level) {
+    currentGenerations.get(level).incrementAndGet();
+  }
+
+  /**
+   * The purpose of this method is to remove any tablets that were added before beginFullScan() was
+   * called. The purpose of this is to handle tablets that were queued for compaction for a while
+   * and because of some change no longer need to compact. If a full scan of the metadata table does
+   * not find any new work for tablet, then any previously queued work for that tablet should be
+   * discarded.
+   *
+   * @param level full metadata scans are done independently per DataLevel, so the tracking what
+   *        needs to be removed must be done per DataLevel
+   */
+  public void endFullScan(DataLevel level) {
+    priorityQueues.values()
+        .forEach(pq -> pq.removeOlderGenerations(level, currentGenerations.get(level).get()));
   }
 
   public void add(TabletMetadata tabletMetadata, Collection<CompactionJob> jobs) {
@@ -155,7 +187,8 @@ private void add(TabletMetadata tabletMetadata, CompactorGroupId groupId,
 
     var pq = priorityQueues.computeIfAbsent(groupId,
         gid -> new CompactionJobPriorityQueue(gid, queueSize));
-    while (pq.add(tabletMetadata, jobs) < 0) {
+    while (pq.add(tabletMetadata, jobs,
+        currentGenerations.get(DataLevel.of(tabletMetadata.getTableId())).get()) < 0) {
       // When entering this loop its expected the queue is closed
       Preconditions.checkState(pq.isClosed());
       // This loop handles race condition where poll() closes empty priority queues. The queue could

server/manager/src/test/java/org/apache/accumulo/manager/compaction/CompactionCoordinatorTest.java

@@ -294,6 +294,7 @@ public void testGetCompactionJob() throws Exception {
     TabletMetadata tm = EasyMock.createNiceMock(TabletMetadata.class);
     expect(tm.getExtent()).andReturn(ke).anyTimes();
     expect(tm.getFiles()).andReturn(Collections.emptySet()).anyTimes();
+    expect(tm.getTableId()).andReturn(ke.tableId()).anyTimes();
 
     EasyMock.replay(tconf, context, creds, tm, security);
 

server/manager/src/test/java/org/apache/accumulo/manager/compaction/queue/CompactionJobPriorityQueueTest.java

@@ -71,7 +71,7 @@ public void testTabletFileReplacement() {
     EasyMock.replay(tm, cj1, cj2);
 
     CompactionJobPriorityQueue queue = new CompactionJobPriorityQueue(GROUP, 2);
-    assertEquals(1, queue.add(tm, List.of(cj1)));
+    assertEquals(1, queue.add(tm, List.of(cj1), 1L));
 
     MetaJob job = queue.peek();
     assertEquals(cj1, job.getJob());
@@ -84,7 +84,7 @@ public void testTabletFileReplacement() {
     assertEquals(1, queue.getQueuedJobs());
 
     // replace the files for the same tablet
-    assertEquals(1, queue.add(tm, List.of(cj2)));
+    assertEquals(1, queue.add(tm, List.of(cj2), 1L));
 
     job = queue.peek();
     assertEquals(cj2, job.getJob());
@@ -126,7 +126,7 @@ public void testAddEqualToMaxSize() {
     EasyMock.replay(tm, cj1, cj2);
 
     CompactionJobPriorityQueue queue = new CompactionJobPriorityQueue(GROUP, 2);
-    assertEquals(2, queue.add(tm, List.of(cj1, cj2)));
+    assertEquals(2, queue.add(tm, List.of(cj1, cj2), 1L));
 
     EasyMock.verify(tm, cj1, cj2);
 
@@ -183,7 +183,7 @@ public void testAddMoreThanMax() {
     EasyMock.replay(tm, cj1, cj2, cj3);
 
     CompactionJobPriorityQueue queue = new CompactionJobPriorityQueue(GROUP, 2);
-    assertEquals(2, queue.add(tm, List.of(cj1, cj2, cj3)));
+    assertEquals(2, queue.add(tm, List.of(cj1, cj2, cj3), 1L));
 
     EasyMock.verify(tm, cj1, cj2, cj3);
 
@@ -234,7 +234,7 @@ public void testAddAfterClose() {
     EasyMock.replay(tm, cj1, cj2);
 
     CompactionJobPriorityQueue queue = new CompactionJobPriorityQueue(GROUP, 2);
-    assertEquals(2, queue.add(tm, List.of(cj1, cj2)));
+    assertEquals(2, queue.add(tm, List.of(cj1, cj2), 1L));
 
     assertFalse(queue.closeIfEmpty());
 
@@ -257,7 +257,7 @@ public void testAddAfterClose() {
 
     assertTrue(queue.closeIfEmpty());
 
-    assertEquals(-1, queue.add(tm, List.of(cj1, cj2)));
+    assertEquals(-1, queue.add(tm, List.of(cj1, cj2), 1L));
 
   }
 
@@ -295,7 +295,7 @@ public void test() {
     // create and add 1000 jobs
     for (int x = 0; x < 1000; x++) {
       Pair<TabletMetadata,CompactionJob> pair = createJob();
-      queue.add(pair.getFirst(), Set.of(pair.getSecond()));
+      queue.add(pair.getFirst(), Set.of(pair.getSecond()), 1L);
       expected.add(pair.getSecond());
     }