moved compaction thread interruption to FileCompactor

Author: keith-turner

server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionInfo.java

@@ -65,7 +65,7 @@ public long getEntriesWritten() {
   }
 
   public Thread getThread() {
-    return compactor.thread;
+    return compactor.getThread();
   }
 
   public String getOutputFile() {

server/base/src/main/java/org/apache/accumulo/server/compaction/FileCompactor.java

@@ -79,6 +79,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Collections2;
 
 import io.opentelemetry.api.trace.Span;
@@ -141,13 +142,48 @@ SystemIteratorEnvironment createIteratorEnv(ServerContext context,
 
   // a unique id to identify a compactor
   private final long compactorID = nextCompactorID.getAndIncrement();
-  protected volatile Thread thread;
+  private volatile Thread thread;
   private final ServerContext context;
 
   private final AtomicBoolean interruptFlag = new AtomicBoolean(false);
 
-  public void interrupt() {
+  public synchronized void interrupt() {
     interruptFlag.set(true);
+
+    if (thread != null) {
+      // Never want to interrupt the thread after clearThread was called as the thread could have
+      // moved on to something completely different than the compaction. This method and clearThread
+      // being synchronized and clearThread setting thread to null prevent this.
+      thread.interrupt();
+    }
+  }
+
+  private class ThreadClearer implements AutoCloseable {
+    @Override
+    public void close() throws InterruptedException {
+      clearThread();
+    }
+  }
+
+  private synchronized ThreadClearer setThread() {
+    thread = Thread.currentThread();
+    return new ThreadClearer();
+  }
+
+  private synchronized void clearThread() throws InterruptedException {
+    Preconditions.checkState(thread == Thread.currentThread());
+    thread = null;
+    // If the thread was interrupted during compaction do not want to allow the thread to continue
+    // w/ the interrupt status set as this could impact code unrelated to the compaction. For
+    // internal compactions the thread will execute metadata update code after the compaction and
+    // would not want the interrupt status set for that.
+    if (Thread.interrupted()) {
+      throw new InterruptedException();
+    }
+  }
+
+  Thread getThread() {
+    return thread;
   }
 
   public long getCompactorID() {
@@ -272,7 +308,8 @@ protected Map<String,Set<ByteSequence>> getLocalityGroups(AccumuloConfiguration
   }
 
   @Override
-  public CompactionStats call() throws IOException, CompactionCanceledException {
+  public CompactionStats call()
+      throws IOException, CompactionCanceledException, InterruptedException {
 
     FileSKVWriter mfw = null;
 
@@ -290,8 +327,9 @@ public CompactionStats call() throws IOException, CompactionCanceledException {
     String newThreadName =
         "MajC compacting " + extent + " started " + threadStartDate + " file: " + outputFile;
     Thread.currentThread().setName(newThreadName);
-    thread = Thread.currentThread();
-    try {
+    // Use try w/ resources for clearing the thread instead of finally because clearing may throw an
+    // exception. Java's handling of exceptions thrown in finally blocks is not good.
+    try (var ignored = setThread()) {
       FileOperations fileFactory = FileOperations.getInstance();
       FileSystem ns = this.fs.getFileSystemByPath(outputFile.getPath());
 
@@ -374,7 +412,6 @@ public CompactionStats call() throws IOException, CompactionCanceledException {
     } finally {
       Thread.currentThread().setName(oldThreadName);
       if (remove) {
-        thread = null;
         runningCompactions.remove(this);
       }
 

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

@@ -1313,56 +1313,6 @@ 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();
-      }
-    }
-
-    @Override
-    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) {
@@ -1382,7 +1332,6 @@ 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);
@@ -1391,11 +1340,8 @@ 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)));
-      // 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);
-      }
+
+      stats = CompactableUtils.compact(tablet, job, cInfo, compactEnv, compactFiles, tmpFileName);
 
       newFile = CompactableUtils.bringOnline(tablet.getDatafileManager(), cInfo, stats,
           compactFiles, allFiles, kind, tmpFileName);
@@ -1653,8 +1599,6 @@ 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.

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

@@ -560,7 +560,7 @@ private static AccumuloConfiguration getCompactionConfig(TableConfiguration tabl
   static CompactionStats compact(Tablet tablet, CompactionJob job,
       CompactableImpl.CompactionInfo cInfo, CompactionEnv cenv,
       Map<StoredTabletFile,DataFileValue> compactFiles, TabletFile tmpFileName)
-      throws IOException, CompactionCanceledException {
+      throws IOException, CompactionCanceledException, InterruptedException {
     TableConfiguration tableConf = tablet.getTableConfiguration();
 
     AccumuloConfiguration compactionConfig = getCompactionConfig(tableConf,
@@ -576,7 +576,7 @@ static CompactionStats compact(Tablet tablet, CompactionJob job,
       }
     };
     final ScheduledFuture<?> future = tablet.getContext().getScheduledExecutor()
-        .scheduleWithFixedDelay(compactionCancellerTask, 10, 10, TimeUnit.SECONDS);
+        .scheduleWithFixedDelay(compactionCancellerTask, 3, 3, TimeUnit.SECONDS);
     try {
       return compactor.call();
     } finally {

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

@@ -137,7 +137,7 @@ public CompactionStats call() {
           log.warn("MinC failed ({}) to create {} retrying ...", e.getMessage(), outputFileName, e);
           reportedProblem = true;
           retryCounter++;
-        } catch (CompactionCanceledException e) {
+        } catch (CompactionCanceledException | InterruptedException e) {
           throw new IllegalStateException(e);
         }
 
@@ -161,7 +161,6 @@ public CompactionStats call() {
 
       } while (true);
     } finally {
-      thread = null;
       runningCompactions.remove(this);
     }
   }