handles failure in compaction planner

core/src/main/java/org/apache/accumulo/core/util/cache/Caches.java

@@ -42,6 +42,8 @@ public enum CacheName {
     COMPACTION_DIR_CACHE,
     COMPACTION_DISPATCHERS,
     COMPACTION_SERVICE_UNKNOWN,
+    COMPACTION_PLANNER_INIT_FAILED,
+    COMPACTION_PLANNER_FAILED,
     COMPACTOR_GROUP_ID,
     COMPRESSION_ALGORITHM,
     CRYPT_PASSWORDS,

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

@@ -55,6 +55,7 @@
 import org.apache.accumulo.core.util.time.SteadyTime;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
 
 import com.github.benmanes.caffeine.cache.Cache;
 
@@ -67,6 +68,8 @@ public class CompactionJobGenerator {
   private final PluginEnvironment env;
   private final Map<FateId,Map<String,String>> allExecutionHints;
   private final Cache<Pair<TableId,CompactionServiceId>,Long> unknownCompactionServiceErrorCache;
+  private final Cache<Pair<TableId,CompactionServiceId>,Long> plannerInitErrorCache;
+  private final Cache<Pair<TableId,CompactionServiceId>,Long> planningErrorCache;
   private final SteadyTime steadyTime;
 
   public CompactionJobGenerator(PluginEnvironment env,
@@ -90,19 +93,20 @@ public CompactionJobGenerator(PluginEnvironment env,
     unknownCompactionServiceErrorCache =
         Caches.getInstance().createNewBuilder(CacheName.COMPACTION_SERVICE_UNKNOWN, false)
             .expireAfterWrite(5, TimeUnit.MINUTES).build();
+    plannerInitErrorCache =
+        Caches.getInstance().createNewBuilder(CacheName.COMPACTION_PLANNER_INIT_FAILED, false)
+            .expireAfterWrite(5, TimeUnit.MINUTES).build();
+    planningErrorCache =
+        Caches.getInstance().createNewBuilder(CacheName.COMPACTION_PLANNER_FAILED, false)
+            .expireAfterWrite(5, TimeUnit.MINUTES).build();
     this.steadyTime = steadyTime;
   }
 
   public Collection<CompactionJob> generateJobs(TabletMetadata tablet, Set<CompactionKind> kinds) {
-
-    // ELASTICITY_TODO do not want user configured plugins to cause exceptions that prevents tablets
-    // from being
-    // assigned. So probably want to catch exceptions and log, but not too spammily OR some how
-    // report something
-    // back to the manager so it can log.
-
     Collection<CompactionJob> systemJobs = Set.of();
 
+    log.debug("Planning for {} {} {}", tablet.getExtent(), kinds, this.hashCode());
+
     if (kinds.contains(CompactionKind.SYSTEM)) {
       CompactionServiceId serviceId = dispatch(CompactionKind.SYSTEM, tablet, Map.of());
       systemJobs = planCompactions(serviceId, CompactionKind.SYSTEM, tablet, Map.of());
@@ -162,23 +166,29 @@ public Map<String,String> getExecutionHints() {
     return dispatcher.dispatch(dispatchParams).getService();
   }
 
+  private Level getLevel(TableId tableId, CompactionServiceId serviceId,
+      Cache<Pair<TableId,CompactionServiceId>,Long> lastErrorCache) {
+    var cacheKey = new Pair<>(tableId, serviceId);
+    var last = lastErrorCache.getIfPresent(cacheKey);
+    if (last == null) {
+      lastErrorCache.put(cacheKey, System.currentTimeMillis());
+      return Level.ERROR;
+    } else {
+      return Level.TRACE;
+    }
+  }
+
   private Collection<CompactionJob> planCompactions(CompactionServiceId serviceId,
       CompactionKind kind, TabletMetadata tablet, Map<String,String> executionHints) {
 
     if (!servicesConfig.getPlanners().containsKey(serviceId.canonical())) {
-      var cacheKey = new Pair<>(tablet.getTableId(), serviceId);
-      var last = unknownCompactionServiceErrorCache.getIfPresent(cacheKey);
-      if (last == null) {
-        // have not logged an error recently for this, so lets log one
-        log.error(
-            "Tablet {} returned non-existent compaction service {} for compaction type {}.  Check"
-                + " the table compaction dispatcher configuration. No compactions will happen"
-                + " until the configuration is fixed. This log message is temporarily suppressed for the"
-                + " entire table.",
-            tablet.getExtent(), serviceId, kind);
-        unknownCompactionServiceErrorCache.put(cacheKey, System.currentTimeMillis());
-      }
-
+      Level level = getLevel(tablet.getTableId(), serviceId, unknownCompactionServiceErrorCache);
+      log.atLevel(level).log(
+          "Tablet {} returned non-existent compaction service {} for compaction type {}.  Check"
+              + " the table compaction dispatcher configuration. No compactions will happen"
+              + " until the configuration is fixed. This log message is temporarily suppressed for the"
+              + " entire table.",
+          tablet.getExtent(), serviceId, kind);
       return Set.of();
     }
 
@@ -299,8 +309,7 @@ public CompactionPlan.Builder createPlanBuilder() {
         return new CompactionPlanImpl.BuilderImpl(kind, allFiles, candidates);
       }
     };
-
-    return planner.makePlan(params).getJobs();
+    return planCompactions(planner, params, serviceId);
   }
 
   private CompactionPlanner createPlanner(TableId tableId, CompactionServiceId serviceId) {
@@ -317,11 +326,28 @@ private CompactionPlanner createPlanner(TableId tableId, CompactionServiceId ser
           servicesConfig.getOptions().get(serviceId.canonical()), (ServiceEnvironment) env);
       planner.init(initParameters);
     } catch (Exception e) {
-      log.error(
-          "Failed to create compaction planner for {} using class:{} options:{}.  Compaction service will not start any new compactions until its configuration is fixed.",
-          serviceId, plannerClassName, options, e);
+      Level level = getLevel(tableId, serviceId, plannerInitErrorCache);
+      log.atLevel(level).setCause(e).log(
+          "Failed to create compaction planner for {} using class:{} options:{}.  Compaction service will not "
+              + "start any new compactions until its configuration is fixed. This log message is temporarily "
+              + "suppressed for the entire table {}.",
+          serviceId, plannerClassName, options, tableId);
       planner = new ProvisionalCompactionPlanner(serviceId);
     }
     return planner;
   }
+
+  private Collection<CompactionJob> planCompactions(CompactionPlanner planner,
+      CompactionPlanner.PlanningParameters params, CompactionServiceId serviceId) {
+    try {
+      return planner.makePlan(params).getJobs();
+    } catch (Exception e) {
+      Level level = getLevel(params.getTableId(), serviceId, planningErrorCache);
+      log.atLevel(level).setCause(e).log(
+          "Failed to plan compactions for {} {} hints:{}.  Compaction service may not start any new compactions"
+              + " until this issue is resolved. This log message is temporarily suppressed for the entire table. {}",
+          serviceId, params.getKind(), params.getExecutionHints(), params.getTableId());
+      return Set.of();
+    }
+  }
 }

test/src/main/java/org/apache/accumulo/test/compaction/CompactionExecutorIT.java

@@ -152,6 +152,21 @@ public CompactionPlan makePlan(PlanningParameters params) {
     }
   }
 
+  public static class ErroringPlanner implements CompactionPlanner {
+    @Override
+    public void init(InitParameters params) {
+      if (Boolean.parseBoolean(params.getOptions().getOrDefault("failInInit", "false"))) {
+        throw new IllegalStateException("error initializing");
+      }
+
+    }
+
+    @Override
+    public CompactionPlan makePlan(PlanningParameters params) {
+      throw new IllegalStateException("error planning");
+    }
+  }
+
   public static class CompactionExecutorITConfig implements MiniClusterConfigurationCallback {
     @Override
     public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration conf) {
@@ -177,6 +192,16 @@ public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration conf)
       cfg.setProperty(csp + "cs4.planner.opts.filesPerCompaction", "11");
       cfg.setProperty(csp + "cs4.planner.opts.process", "USER");
 
+      // Setup three planner that fail to initialize or plan, these planners should not impede
+      // tablet assignment.
+      cfg.setProperty(csp + "cse1.planner", ErroringPlanner.class.getName());
+      cfg.setProperty(csp + "cse1.planner.opts.failInInit", "true");
+
+      cfg.setProperty(csp + "cse2.planner", ErroringPlanner.class.getName());
+      cfg.setProperty(csp + "cse2.planner.opts.failInInit", "false");
+
+      cfg.setProperty(csp + "cse3.planner", "NonExistentPlanner20240522");
+
       // this is meant to be dynamically reconfigured
       cfg.setProperty(csp + "recfg.planner", TestPlanner.class.getName());
       cfg.setProperty(csp + "recfg.planner.opts.groups", "[{'group':'i1'},{'group':'i2'}]");
@@ -230,6 +255,35 @@ public void cleanup() {
     }
   }
 
+  @Test
+  public void testFailingPlanners() throws Exception {
+    // This test ensures that a table w/ failing compaction planner can still be read and written.
+
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+      createTable(client, "fail1", "cse1");
+      createTable(client, "fail2", "cse2");
+      createTable(client, "fail3", "cse3");
+
+      // ensure tablets can still be assigned and written w/ failing compaction services
+      addFiles(client, "fail1", 30);
+      addFiles(client, "fail2", 30);
+      addFiles(client, "fail3", 30);
+
+      // ensure tablets can still be assigned and scanned w/ failing compaction services
+      assertEquals(30, scanTable(client, "fail1").size());
+      assertEquals(30, scanTable(client, "fail2").size());
+      assertEquals(30, scanTable(client, "fail3").size());
+
+      // compactions should never run on these tables, but sleep a bit to be sure
+      Thread.sleep(2000);
+
+      // do no expect any compactions to run
+      assertEquals(30, getFiles(client, "fail1").size());
+      assertEquals(30, getFiles(client, "fail2").size());
+      assertEquals(30, getFiles(client, "fail3").size());
+    }
+  }
+
   @Test
   public void testReconfigureCompactionService() throws Exception {
     Stream.of("i1", "i2").forEach(g -> {