Fail when configuring or compacting table with bad compaction svc

server/base/src/main/java/org/apache/accumulo/server/util/PropUtil.java

@@ -26,6 +26,7 @@
 
 import org.apache.accumulo.core.classloader.ClassLoaderUtil;
 import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.core.util.compaction.CompactionServicesConfig;
 import org.apache.accumulo.server.ServerContext;
 import org.apache.accumulo.server.conf.store.IdBasedPropStoreKey;
 import org.apache.accumulo.server.conf.store.NamespacePropKey;
@@ -96,7 +97,14 @@ public static void validateProperties(final ServerContext context,
         ResourceGroupPropUtil.validateResourceGroupProperty(prop.getKey(), prop.getValue());
       }
 
-      if (prop.getKey().equals(Property.TABLE_ERASURE_CODE_POLICY.getKey())
+      if (prop.getKey().equals(Property.TABLE_COMPACTION_DISPATCHER_OPTS.getKey() + "service")) {
+        // Validate the compaction service exists
+        var compactionService = prop.getValue();
+        if (!isTableCompactionServiceValid(context, compactionService)) {
+          throw new IllegalArgumentException(
+              "Compaction Service " + compactionService + " has not been configured.");
+        }
+      } else if (prop.getKey().equals(Property.TABLE_ERASURE_CODE_POLICY.getKey())
           && !prop.getValue().isEmpty()) {
         var volumes = context.getVolumeManager().getVolumes();
         for (var volume : volumes) {
@@ -190,4 +198,13 @@ private static String target(PropStoreKey propStoreKey) {
     }
   }
 
+  public static boolean isTableCompactionServiceValid(ServerContext context, String serviceName) {
+    if (serviceName == null) {
+      return true; // no compaction service set on table
+    } else {
+      var servicesConfig = new CompactionServicesConfig(context.getConfiguration());
+      var plannerClass = servicesConfig.getPlanners().get(serviceName);
+      return plannerClass != null;
+    }
+  }
 }

server/base/src/test/java/org/apache/accumulo/server/util/PropUtilTest.java

@@ -18,21 +18,26 @@
  */
 package org.apache.accumulo.server.util;
 
+import static org.apache.accumulo.core.Constants.DEFAULT_COMPACTION_SERVICE_NAME;
 import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.Map;
 
 import org.apache.accumulo.core.classloader.ClassLoaderUtil;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.ConfigurationCopy;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.InstanceId;
 import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.spi.common.ContextClassLoaderFactory;
+import org.apache.accumulo.core.spi.compaction.RatioBasedCompactionPlanner;
 import org.apache.accumulo.server.ServerContext;
 import org.apache.accumulo.server.conf.store.TablePropKey;
 import org.junit.jupiter.api.BeforeEach;
@@ -99,4 +104,60 @@ public void testSetClasspathContext() {
     verify(ctx, conf, iid, tid);
   }
 
+  @Test
+  public void testSetCompactionService() {
+    ConfigurationCopy conf = new ConfigurationCopy();
+
+    conf.set(
+        Property.COMPACTION_SERVICE_PREFIX.getKey() + DEFAULT_COMPACTION_SERVICE_NAME + ".planner",
+        RatioBasedCompactionPlanner.class.getName());
+    conf.set(Property.COMPACTION_SERVICE_PREFIX.getKey() + DEFAULT_COMPACTION_SERVICE_NAME
+        + ".planner.opts.maxOpen", "10");
+    conf.set(
+        Property.COMPACTION_SERVICE_PREFIX.getKey() + DEFAULT_COMPACTION_SERVICE_NAME
+            + ".planner.opts.groups",
+        "[{'group':'small','maxSize':'32M'},{'group':'medium','maxSize':'128M'},{'group':'large'}]");
+    conf.set(Property.COMPACTION_SERVICE_PREFIX.getKey() + DEFAULT_COMPACTION_SERVICE_NAME
+        + ".planner.opts.validProp", "1");
+
+    conf.set(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner",
+        RatioBasedCompactionPlanner.class.getName());
+    conf.set(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.maxOpen", "10");
+    conf.set(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.groups",
+        "[{'group':'small','maxSize':'32M'},{'group':'medium','maxSize':'128M'},{'group':'large'}]");
+    conf.set(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.validProp", "1");
+
+    ServerContext ctx = createMock(ServerContext.class);
+    expect(ctx.getConfiguration()).andReturn(conf).once();
+
+    replay(ctx);
+    assertTrue(PropUtil.isTableCompactionServiceValid(ctx, "cs1"));
+    assertTrue(PropUtil.isTableCompactionServiceValid(ctx, null));
+    verify(ctx);
+  }
+
+  @Test
+  public void testSetCompactionServiceFails() {
+    ConfigurationCopy conf = new ConfigurationCopy();
+
+    conf.set(
+        Property.COMPACTION_SERVICE_PREFIX.getKey() + DEFAULT_COMPACTION_SERVICE_NAME + ".planner",
+        RatioBasedCompactionPlanner.class.getName());
+    conf.set(Property.COMPACTION_SERVICE_PREFIX.getKey() + DEFAULT_COMPACTION_SERVICE_NAME
+        + ".planner.opts.maxOpen", "10");
+    conf.set(
+        Property.COMPACTION_SERVICE_PREFIX.getKey() + DEFAULT_COMPACTION_SERVICE_NAME
+            + ".planner.opts.groups",
+        "[{'group':'small','maxSize':'32M'},{'group':'medium','maxSize':'128M'},{'group':'large'}]");
+    conf.set(Property.COMPACTION_SERVICE_PREFIX.getKey() + DEFAULT_COMPACTION_SERVICE_NAME
+        + ".planner.opts.validProp", "1");
+
+    ServerContext ctx = createMock(ServerContext.class);
+    expect(ctx.getConfiguration()).andReturn(conf).once();
+
+    replay(ctx);
+    assertFalse(PropUtil.isTableCompactionServiceValid(ctx, "cs1"));
+    verify(ctx);
+  }
+
 }

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

@@ -106,6 +106,7 @@
 import org.apache.accumulo.server.client.ClientServiceHandler;
 import org.apache.accumulo.server.conf.store.TablePropKey;
 import org.apache.accumulo.server.security.AuditedSecurityOperation;
+import org.apache.accumulo.server.util.PropUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
@@ -548,6 +549,15 @@ public void executeFateOperation(TInfo tinfo, TCredentials c, TFateId opid, TFat
           throw new ThriftSecurityException(c.getPrincipal(), SecurityErrorCode.PERMISSION_DENIED);
         }
 
+        var tableConf = manager.getContext().getTableConfiguration(tableId);
+        var compactionService =
+            tableConf.get(Property.TABLE_COMPACTION_DISPATCHER_OPTS.getKey() + "service");
+        if (!PropUtil.isTableCompactionServiceValid(manager.getContext(), compactionService)) {
+          throw new ThriftTableOperationException(tableId.canonical(), null, tableOp,
+              TableOperationExceptionType.OTHER,
+              "Compaction Service " + compactionService + " has not been configured.");
+        }
+
         goalMessage += "Compact table (" + tableId + ") with config " + compactionConfig;
         manager.fateClient(type).seedTransaction(op, fateId,
             new TraceRepo<>(new CompactRange(namespaceId, tableId, compactionConfig)), autoCleanup,

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

@@ -22,16 +22,19 @@
 import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.createTable;
 import static org.apache.accumulo.test.compaction.ExternalCompactionTestUtils.verify;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import java.util.List;
 
 import org.apache.accumulo.core.client.Accumulo;
 import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.client.admin.CompactionConfig;
 import org.apache.accumulo.core.clientImpl.ClientContext;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.spi.compaction.RatioBasedCompactionPlanner;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloClusterImpl;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
 import org.apache.accumulo.test.functional.SlowIterator;
 import org.apache.accumulo.test.harness.AccumuloClusterHarness;
@@ -109,4 +112,67 @@ public void testRemovingCompactionExecutor() throws Exception {
       verify(client, table, 1);
     }
   }
+
+  @Test
+  public void testCreateTableBadCompactionService() throws Exception {
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+      final String table = getUniqueNames(1)[0];
+      // Creating a table with an incorrect compaction service should fail
+      AccumuloException ae =
+          assertThrows(AccumuloException.class, () -> createTable(client, table, "cs3", 50));
+      assertEquals("Internal error processing waitForFateOperation", ae.getMessage());
+    }
+  }
+
+  @Test
+  public void testConfigureBadCompactionService() throws Exception {
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+      final String table = getUniqueNames(1)[0];
+      createTable(client, table, "cs1", 50);
+      // Setting an incorrect compaction service should fail
+      AccumuloException ae =
+          assertThrows(AccumuloException.class, () -> client.tableOperations().setProperty(table,
+              Property.TABLE_COMPACTION_DISPATCHER_OPTS.getKey() + "service", "cs4"));
+      assertEquals(
+          "ThriftPropertyException(property:table.compaction.dispatcher.opts.service, value:cs4, description:Compaction Service cs4 has not been configured.)",
+          ae.getMessage());
+    }
+  }
+
+  @Test
+  public void testCompactionFailsBadCompactionService() throws Exception {
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+
+      client.instanceOperations().setProperty(
+          Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs2.planner",
+          RatioBasedCompactionPlanner.class.getName());
+      client.instanceOperations().setProperty(
+          Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs2.planner.opts.groups",
+          "[{'group':'" + ExternalCompactionTestUtils.GROUP2 + "'}]");
+
+      ((MiniAccumuloClusterImpl) getCluster()).getConfig().getClusterServerConfiguration()
+          .addCompactorResourceGroup(ExternalCompactionTestUtils.GROUP2, 1);
+      getCluster().start();
+
+      final String table = getUniqueNames(1)[0];
+      createTable(client, table, "cs2", 2);
+      ExternalCompactionTestUtils.writeData(client, table, MAX_DATA);
+      client.tableOperations().flush(table, null, null, true);
+      client.tableOperations().compact(table, new CompactionConfig().setWait(true));
+
+      // Remove the compaction service configuration, next compaction should fail
+      // because the configuration is invalid.
+      client.instanceOperations()
+          .removeProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs2.planner.opts.groups");
+      client.instanceOperations()
+          .removeProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs2.planner");
+
+      // Wait for property change propagation
+      AccumuloException ae = assertThrows(AccumuloException.class,
+          () -> client.tableOperations().compact(table, new CompactionConfig().setWait(true)));
+      assertEquals("Compaction Service cs2 has not been configured.", ae.getMessage());
+
+    }
+  }
+
 }