dlmarion commented on code in PR #5811:
URL: https://github.com/apache/accumulo/pull/5811#discussion_r2304216637


##########
server/compactor/src/main/java/org/apache/accumulo/compactor/Compactor.java:
##########
@@ -829,237 +829,241 @@ public void run() {
         
getConfiguration().getTimeInMillis(Property.COMPACTOR_CANCEL_CHECK_INTERVAL));
 
     LOG.info("Compactor started, waiting for work");
-    try {
-
-      final AtomicReference<Throwable> err = new AtomicReference<>();
-      final LogSorter logSorter = new LogSorter(this);
-      long nextSortLogsCheckTime = System.currentTimeMillis();
 
-      while (!isShutdownRequested()) {
-        if (Thread.currentThread().isInterrupted()) {
-          LOG.info("Server process thread has been interrupted, shutting 
down");
-          break;
-        }
-        try {
-          // mark compactor as idle while not in the compaction loop
-          updateIdleStatus(true);
+    final AtomicReference<Throwable> err = new AtomicReference<>();
+    final LogSorter logSorter = new LogSorter(this);
+    long nextSortLogsCheckTime = System.currentTimeMillis();
 
-          currentCompactionId.set(null);
-          err.set(null);
-          JOB_HOLDER.reset();
-
-          if (System.currentTimeMillis() > nextSortLogsCheckTime) {
-            // Attempt to process all existing log sorting work serially in 
this thread.
-            // When no work remains, this call will return so that we can look 
for compaction
-            // work.
-            LOG.debug("Checking to see if any recovery logs need sorting");
+    while (!isShutdownRequested()) {
+      if (Thread.currentThread().isInterrupted()) {
+        LOG.info("Server process thread has been interrupted, shutting down");
+        break;
+      }
+      try {
+        // mark compactor as idle while not in the compaction loop
+        updateIdleStatus(true);
+
+        currentCompactionId.set(null);
+        err.set(null);
+        JOB_HOLDER.reset();
+
+        if (System.currentTimeMillis() > nextSortLogsCheckTime) {
+          // Attempt to process all existing log sorting work serially in this 
thread.
+          // When no work remains, this call will return so that we can look 
for compaction
+          // work.
+          LOG.debug("Checking to see if any recovery logs need sorting");
+          try {
             nextSortLogsCheckTime = logSorter.sortLogsIfNeeded();
+          } catch (KeeperException e) {
+            LOG.error("Error sorting logs", e);
           }
+        }
 
-          performFailureProcessing(errorHistory);
+        performFailureProcessing(errorHistory);
 
-          TExternalCompactionJob job;
-          try {
-            TNextCompactionJob next = getNextJob(getNextId());
-            job = next.getJob();
-            if (!job.isSetExternalCompactionId()) {
-              LOG.trace("No external compactions in queue {}", 
this.getResourceGroup());
-              
UtilWaitThread.sleep(getWaitTimeBetweenCompactionChecks(next.getCompactorCount()));
-              continue;
-            }
-            if 
(!job.getExternalCompactionId().equals(currentCompactionId.get().toString())) {
-              throw new IllegalStateException("Returned eci " + 
job.getExternalCompactionId()
-                  + " does not match supplied eci " + 
currentCompactionId.get());
-            }
-          } catch (RetriesExceededException e2) {
-            LOG.warn("Retries exceeded getting next job. Retrying...");
+        TExternalCompactionJob job;
+        try {
+          TNextCompactionJob next = getNextJob(getNextId());
+          job = next.getJob();
+          if (!job.isSetExternalCompactionId()) {
+            LOG.trace("No external compactions in queue {}", 
this.getResourceGroup());
+            
UtilWaitThread.sleep(getWaitTimeBetweenCompactionChecks(next.getCompactorCount()));
             continue;
           }
-          LOG.debug("Received next compaction job: {}", job);
+          if 
(!job.getExternalCompactionId().equals(currentCompactionId.get().toString())) {
+            throw new IllegalStateException("Returned eci " + 
job.getExternalCompactionId()
+                + " does not match supplied eci " + currentCompactionId.get());
+          }
+        } catch (RetriesExceededException e2) {
+          LOG.warn("Retries exceeded getting next job. Retrying...");
+          continue;
+        }
+        LOG.debug("Received next compaction job: {}", job);
 
-          final LongAdder totalInputEntries = new LongAdder();
-          final LongAdder totalInputBytes = new LongAdder();
-          final CountDownLatch started = new CountDownLatch(1);
-          final CountDownLatch stopped = new CountDownLatch(1);
+        final LongAdder totalInputEntries = new LongAdder();
+        final LongAdder totalInputBytes = new LongAdder();
+        final CountDownLatch started = new CountDownLatch(1);
+        final CountDownLatch stopped = new CountDownLatch(1);
 
-          final FileCompactorRunnable fcr =
-              createCompactionJob(job, totalInputEntries, totalInputBytes, 
started, stopped, err);
+        final FileCompactorRunnable fcr =
+            createCompactionJob(job, totalInputEntries, totalInputBytes, 
started, stopped, err);
 
-          final Thread compactionThread = Threads.createNonCriticalThread(
-              "Compaction job for tablet " + job.getExtent().toString(), fcr);
+        final Thread compactionThread = Threads.createNonCriticalThread(
+            "Compaction job for tablet " + job.getExtent().toString(), fcr);
 
-          JOB_HOLDER.set(job, compactionThread, fcr.getFileCompactor());
+        JOB_HOLDER.set(job, compactionThread, fcr.getFileCompactor());
 
-          try {
-            // mark compactor as busy while compacting
-            updateIdleStatus(false);
+        try {
+          // mark compactor as busy while compacting
+          updateIdleStatus(false);
 
+          try {
             // Need to call FileCompactorRunnable.initialize after calling 
JOB_HOLDER.set
             fcr.initialize();
-
-            compactionThread.start(); // start the compactionThread
-            started.await(); // wait until the compactor is started
-            final long inputEntries = totalInputEntries.sum();
-            final long waitTime = 
calculateProgressCheckTime(totalInputBytes.sum());
-            LOG.debug("Progress checks will occur every {} seconds", waitTime);
-            String percentComplete = "unknown";
-
-            while (!stopped.await(waitTime, TimeUnit.SECONDS)) {
-              List<CompactionInfo> running =
-                  
org.apache.accumulo.server.compaction.FileCompactor.getRunningCompactions();
-              if (!running.isEmpty()) {
-                // Compaction has started. There should only be one in the list
-                CompactionInfo info = running.get(0);
-                if (info != null) {
-                  final long entriesRead = info.getEntriesRead();
-                  final long entriesWritten = info.getEntriesWritten();
-                  if (inputEntries > 0) {
-                    percentComplete = Float.toString((entriesRead / (float) 
inputEntries) * 100);
-                  }
-                  String message = String.format(
-                      "Compaction in progress, read %d of %d input entries ( 
%s %s ), written %d entries",
-                      entriesRead, inputEntries, percentComplete, "%", 
entriesWritten);
-                  watcher.run();
-                  try {
-                    LOG.debug("Updating coordinator with compaction progress: 
{}.", message);
-                    TCompactionStatusUpdate update = new 
TCompactionStatusUpdate(
-                        TCompactionState.IN_PROGRESS, message, inputEntries, 
entriesRead,
-                        entriesWritten, fcr.getCompactionAge().toNanos());
-                    updateCompactionState(job, update);
-                  } catch (RetriesExceededException e) {
-                    LOG.warn("Error updating coordinator with compaction 
progress, error: {}",
-                        e.getMessage());
-                  }
-                }
-              } else {
-                LOG.debug("Waiting on compaction thread to finish, but no 
RUNNING compaction");
-              }
-            }
-            compactionThread.join();
-            LOG.trace("Compaction thread finished.");
-            // Run the watcher again to clear out the finished compaction and 
set the
-            // stuck count to zero.
-            watcher.run();
-
-            if (err.get() != null) {
-              // maybe the error occured because the table was deleted or 
something like that, so
-              // force a cancel check to possibly reduce noise in the logs
-              checkIfCanceled();
+          } catch (RetriesExceededException e) {
+            LOG.error(
+                "Error starting FileCompactableRunnable, cancelling compaction 
and moving to next job.",
+                e);
+            try {
+              cancel(job.getExternalCompactionId());
+            } catch (TException e1) {
+              LOG.error("Error cancelling compaction.", e1);
             }
+            continue;
+          } finally {
+            currentCompactionId.set(null);

Review Comment:
   Yep, good catch.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscr...@accumulo.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Reply via email to