implements interuption of compactions and adds an IT to verify compaction cancels on tablet migration

server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/CompactableImpl.java

@@ -1313,6 +1313,55 @@ private void completeCompaction(CompactionJob job, Set<StoredTabletFile> jobFile
     selectFiles();
   }
 
+  private final Set<CompactionInterrupter> compactionInterrupters = ConcurrentHashMap.newKeySet();
+
+  private class CompactionInterrupter implements AutoCloseable {
+    private Thread thread;
+
+    CompactionInterrupter() throws InterruptedException {
+
+      this.thread = Thread.currentThread();
+
+      if (Thread.currentThread().isInterrupted() || closed) {
+        throw new InterruptedException();
+      }
+
+      compactionInterrupters.add(this);
+      log.trace("Registered compaction interrupter for {} {}", thread.getId(), getExtent());
+    }
+
+    public synchronized void interrupt() {
+      if (thread != null) {
+        log.debug("Interrupting compaction thread {} for {}", thread.getId(), getExtent());
+        thread.interrupt();
+      }
+    }
+
+    public void close() throws InterruptedException {
+      synchronized (this) {
+        Preconditions.checkState(thread == Thread.currentThread());
+
+        // The goal of this class is not to interrupt the compaction thread after this close()
+        // method returns. Two things help achieve this goal, first we set the thread instance var
+        // null here. Second this code block and the interrupt() method are synchronized.
+        thread = null;
+      }
+
+      // call this outside of synchronized block to avoid deadlock with the locks inside the
+      // concurrent hash set
+      compactionInterrupters.remove(this);
+
+      log.trace("Unregistered compaction interrupter for {} {}", Thread.currentThread().getId(),
+          getExtent());
+
+      // Its possible the threads interrupt status was set but nothing ever checked it. For example
+      // interrupt() could have been called immediately before this method was called.
+      if (Thread.interrupted()) {
+        throw new InterruptedException();
+      }
+    }
+  }
+
   @Override
   public void compact(CompactionServiceId service, CompactionJob job, BooleanSupplier keepRunning,
       RateLimiter readLimiter, RateLimiter writeLimiter, long queuedTime) {
@@ -1332,6 +1381,7 @@ public void compact(CompactionServiceId service, CompactionJob job, BooleanSuppl
     boolean successful = false;
     try {
       TabletLogger.compacting(getExtent(), job, cInfo.localCompactionCfg);
+
       tablet.incrementStatusMajor();
       var check = new CompactionCheck(service, kind, keepRunning, cInfo.checkCompactionId);
       TabletFile tmpFileName = tablet.getNextMapFilenameForMajc(cInfo.propagateDeletes);
@@ -1340,8 +1390,11 @@ public void compact(CompactionServiceId service, CompactionJob job, BooleanSuppl
       SortedMap<StoredTabletFile,DataFileValue> allFiles = tablet.getDatafiles();
       HashMap<StoredTabletFile,DataFileValue> compactFiles = new HashMap<>();
       cInfo.jobFiles.forEach(file -> compactFiles.put(file, allFiles.get(file)));
-
-      stats = CompactableUtils.compact(tablet, job, cInfo, compactEnv, compactFiles, tmpFileName);
+      // Limit interrupting compactions to the part that reads and writes files and avoid
+      // interrupting the metadata table updates.
+      try (var ignored = new CompactionInterrupter()) {
+        stats = CompactableUtils.compact(tablet, job, cInfo, compactEnv, compactFiles, tmpFileName);
+      }
 
       newFile = CompactableUtils.bringOnline(tablet.getDatafileManager(), cInfo, stats,
           compactFiles, allFiles, kind, tmpFileName);
@@ -1599,6 +1652,8 @@ public void close() {
 
       closed = true;
 
+      compactionInterrupters.forEach(CompactionInterrupter::interrupt);
+
       // Wait while internal jobs are running or external compactions are committing. When
       // chopStatus is MARKING or selectStatus is SELECTING, there may be metadata table writes so
       // wait on those. Do not wait on external compactions that are running.

test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java

@@ -37,11 +37,15 @@
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
 import java.util.Objects;
+import java.util.SortedSet;
+import java.util.TreeSet;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import org.apache.accumulo.core.client.Accumulo;
 import org.apache.accumulo.core.client.AccumuloClient;
@@ -86,11 +90,14 @@
 import org.apache.accumulo.core.spi.compaction.SimpleCompactionDispatcher;
 import org.apache.accumulo.core.util.HostAndPort;
 import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.minicluster.ServerType;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
 import org.apache.accumulo.test.VerifyIngest;
 import org.apache.accumulo.test.VerifyIngest.VerifyParams;
 import org.apache.accumulo.test.compaction.CompactionExecutorIT;
 import org.apache.accumulo.test.compaction.ExternalCompaction_1_IT.FSelector;
+import org.apache.accumulo.test.util.Wait;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -797,6 +804,108 @@ public void testGetActiveCompactions() throws Exception {
     }
   }
 
+  @Test
+  public void testMigrationCancelCompaction() throws Exception {
+
+    // This test creates 40 tablets w/ slow iterator, causes 40 compactions to start, and then
+    // starts a new tablet server. Some of the tablets should migrate to the new tserver and cancel
+    // their compaction. Because the test uses a slow iterator, if close blocks on compaction then
+    // the test should timeout. Two tables are used to have different iterator settings inorder to
+    // test the two different way compactions can be canceled. Compactions can be canceled by thread
+    // interrupt or by a check that is done after a compaction iterator returns a key value.
+
+    final String[] tables = this.getUniqueNames(2);
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+      client.instanceOperations().setProperty(
+          Property.TSERV_COMPACTION_SERVICE_DEFAULT_EXECUTORS.getKey(),
+          "[{'name':'any','numThreads':20}]".replaceAll("'", "\""));
+
+      SortedSet<Text> splits = IntStream.range(1, 20).mapToObj(i -> String.format("%06d", i * 1000))
+          .map(Text::new).collect(Collectors.toCollection(TreeSet::new));
+
+      // This iterator is intended to cover the case of a compaction being canceled by thread
+      // interrupt.
+      IteratorSetting setting1 = new IteratorSetting(50, "sleepy", SlowIterator.class);
+      setting1.addOption("sleepTime", "300000");
+      setting1.addOption("seekSleepTime", "3000");
+      SlowIterator.sleepUninterruptibly(setting1, false);
+
+      client.tableOperations().create(tables[0], new NewTableConfiguration().withSplits(splits)
+          .attachIterator(setting1, EnumSet.of(IteratorScope.majc)));
+
+      // This iterator is intended to cover the case of compaction being canceled by the check after
+      // a key value is returned. The iterator is configured to ignore interrupts.
+      IteratorSetting setting2 = new IteratorSetting(50, "sleepy", SlowIterator.class);
+      setting2.addOption("sleepTime", "2000");
+      setting2.addOption("seekSleepTime", "2000");
+      SlowIterator.sleepUninterruptibly(setting2, true);
+
+      client.tableOperations().create(tables[1], new NewTableConfiguration().withSplits(splits)
+          .attachIterator(setting2, EnumSet.of(IteratorScope.majc)));
+
+      // write files to each tablet, should cause compactions to start
+      for (var table : tables) {
+        for (int round = 0; round < 5; round++) {
+          try (var writer = client.createBatchWriter(table)) {
+            for (int i = 0; i < 20_000; i++) {
+              Mutation m = new Mutation(String.format("%06d", i));
+              m.put("f", "q", "v");
+              writer.addMutation(m);
+            }
+          }
+          client.tableOperations().flush(table, null, null, true);
+        }
+      }
+
+      assertEquals(2, client.instanceOperations().getTabletServers().size());
+
+      var ctx = (ClientContext) client;
+      var tableId1 = ctx.getTableId(tables[0]);
+      var tableId2 = ctx.getTableId(tables[1]);
+
+      Wait.waitFor(() -> {
+        var runningCompactions = client.instanceOperations().getActiveCompactions().stream()
+            .map(ac -> ac.getTablet().getTable())
+            .filter(tid -> tid.equals(tableId1) || tid.equals(tableId2)).count();
+        log.debug("Running compactions {}", runningCompactions);
+        return runningCompactions == 40;
+      });
+
+      ((MiniAccumuloClusterImpl) getCluster()).getConfig().setNumTservers(3);
+      getCluster().getClusterControl().start(ServerType.TABLET_SERVER, "localhost");
+
+      Wait.waitFor(() -> {
+        var servers = client.instanceOperations().getTabletServers().size();
+        log.debug("Server count {}", servers);
+        return 3 == servers;
+      });
+
+      Wait.waitFor(() -> {
+        try (var tablets = ctx.getAmple().readTablets().forLevel(Ample.DataLevel.USER)
+            .fetch(ColumnType.LOCATION, ColumnType.PREV_ROW).build()) {
+          Map<String,Long> counts = new HashMap<>();
+          for (var tablet : tablets) {
+            if (!tablet.getTableId().equals(tableId1) && !tablet.getTableId().equals(tableId2)) {
+              continue;
+            }
+
+            if (tablet.getLocation() != null
+                && tablet.getLocation().getType() == TabletMetadata.LocationType.CURRENT) {
+              counts.merge(tablet.getLocation().getHostPort(), 1L, Long::sum);
+            }
+          }
+
+          var total = counts.values().stream().mapToLong(l -> l).sum();
+          var min = counts.values().stream().mapToLong(l -> l).min().orElse(0);
+          var max = counts.values().stream().mapToLong(l -> l).max().orElse(100);
+          var serversSeen = counts.keySet();
+          log.debug("total:{} min:{} max:{} serversSeen:{}", total, min, max, serversSeen);
+          return total == 40 && min == 12 && max == 14 && serversSeen.size() == 3;
+        }
+      });
+    }
+  }
+
   /**
    * Counts the number of tablets and files in a table.
    */

test/src/main/java/org/apache/accumulo/test/functional/SlowIterator.java

@@ -18,8 +18,6 @@
  */
 package org.apache.accumulo.test.functional;
 
-import static org.apache.accumulo.core.util.UtilWaitThread.sleepUninterruptibly;
-
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Map;
@@ -33,14 +31,17 @@
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.core.iterators.WrappingIterator;
+import org.apache.accumulo.core.util.UtilWaitThread;
 
 public class SlowIterator extends WrappingIterator {
 
   private static final String SLEEP_TIME = "sleepTime";
   private static final String SEEK_SLEEP_TIME = "seekSleepTime";
+  private static final String SLEEP_UNINTERRUPTIBLY = "sleepUninterruptibly";
 
   private long sleepTime = 0;
   private long seekSleepTime = 0;
+  private boolean sleepUninterruptibly = true;
 
   public static void setSleepTime(IteratorSetting is, long millis) {
     is.addOption(SLEEP_TIME, Long.toString(millis));
@@ -50,21 +51,37 @@ public static void setSeekSleepTime(IteratorSetting is, long t) {
     is.addOption(SEEK_SLEEP_TIME, Long.toString(t));
   }
 
+  public static void sleepUninterruptibly(IteratorSetting is, boolean b) {
+    is.addOption(SLEEP_UNINTERRUPTIBLY, Boolean.toString(b));
+  }
+
+  private void sleep(long time) throws IOException {
+    if (sleepUninterruptibly) {
+      UtilWaitThread.sleepUninterruptibly(time, TimeUnit.MILLISECONDS);
+    } else {
+      try {
+        Thread.sleep(sleepTime);
+      } catch (InterruptedException e) {
+        throw new IOException(e);
+      }
+    }
+  }
+
   @Override
   public SortedKeyValueIterator<Key,Value> deepCopy(IteratorEnvironment env) {
     throw new UnsupportedOperationException();
   }
 
   @Override
   public void next() throws IOException {
-    sleepUninterruptibly(sleepTime, TimeUnit.MILLISECONDS);
+    sleep(sleepTime);
     super.next();
   }
 
   @Override
   public void seek(Range range, Collection<ByteSequence> columnFamilies, boolean inclusive)
       throws IOException {
-    sleepUninterruptibly(seekSleepTime, TimeUnit.MILLISECONDS);
+    sleep(seekSleepTime);
     super.seek(range, columnFamilies, inclusive);
   }
 
@@ -79,6 +96,10 @@ public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> op
     if (options.containsKey(SEEK_SLEEP_TIME)) {
       seekSleepTime = Long.parseLong(options.get(SEEK_SLEEP_TIME));
     }
+
+    if (options.containsKey(SLEEP_UNINTERRUPTIBLY)) {
+      sleepUninterruptibly = Boolean.parseBoolean(options.get(SLEEP_UNINTERRUPTIBLY));
+    }
   }
 
 }