Fail when configuring or compacting table with bad compaction svc

server/base/src/main/java/org/apache/accumulo/server/conf/util/ImportConfigCommand.java

@@ -20,14 +20,17 @@
 
 import java.io.InputStream;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.accumulo.core.cli.ServerOpts;
 import org.apache.accumulo.core.client.NamespaceNotFoundException;
 import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.ResourceGroupId;
 import org.apache.accumulo.server.ServerContext;
 import org.apache.accumulo.server.conf.store.NamespacePropKey;
@@ -145,14 +148,41 @@ private static void validate(ServerContext serverContext, List<ScopedProperties>
       }
       if (fail) {
         throw new IllegalArgumentException(
-            "Yaml and Accumulo do not have the same tables,namespaces, and/or resource groups");
+            "Yaml and Accumulo do not have the same tables, namespaces, and/or resource groups");
       }
     }
 
-    // validate all scope+name before attempting to update any scope+name
+    // Namespace or Table level compaction service properties will fail in
+    // PropUtil.validateProperties because they require the compaction service to be
+    // defined at the system level. Validate these manually and then remove them
+    // before calling PropUtil.validateProperties
+    final String tableCompactionServiceKey =
+        Property.TABLE_COMPACTION_DISPATCHER_OPTS.getKey() + "service";
     for (var scopedProps : allProps) {
+      Map<String,String> scopeProperties = new HashMap<>(scopedProps.props());
+      if ((scopedProps.scope() == Scope.NAMESPACE || scopedProps.scope() == Scope.TABLE)
+          && scopeProperties.containsKey(tableCompactionServiceKey)) {
+        String compactionSvcName = scopeProperties.get(tableCompactionServiceKey);
+        String compactionSvcPlannerKey =
+            Property.COMPACTION_SERVICE_PREFIX.getKey() + compactionSvcName + ".planner";
+        AtomicBoolean compactionSvcPlannerKeyFound = new AtomicBoolean(false);
+        // validate that the compaction service is defined at the system level
+        allProps.forEach(sp -> {
+          if (sp.scope() == Scope.SYSTEM && sp.props().containsKey(compactionSvcPlannerKey)) {
+            compactionSvcPlannerKeyFound.set(true);
+          }
+        });
+        if (!compactionSvcPlannerKeyFound.get()) {
+          throw new IllegalArgumentException("Compaction service property "
+              + tableCompactionServiceKey + " defined in " + scopedProps.name()
+              + "scope, but corresponding compaction service not defined in SYSTEM scope");
+        }
+        // We have validated, remove for next level validation so it doesn't fail
+        scopeProperties.remove(tableCompactionServiceKey);
+      }
+      // validate all scope+name before attempting to update any scope+name
       var propStoreKey = getKey(scopedProps.scope(), scopedProps.name(), serverContext);
-      PropUtil.validateProperties(serverContext, propStoreKey, scopedProps.props());
+      PropUtil.validateProperties(serverContext, propStoreKey, scopeProperties);
     }
   }
 

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/BadCompactionServiceConfigIT.java

@@ -21,6 +21,8 @@
 import static org.apache.accumulo.core.Constants.DEFAULT_COMPACTION_SERVICE_NAME;
 import static org.apache.accumulo.core.metrics.Metric.COMPACTION_USER_SVC_ERRORS;
 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 java.util.Collections;
 import java.util.HashMap;
@@ -36,7 +38,10 @@
 
 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.AccumuloSecurityException;
 import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.TableExistsException;
 import org.apache.accumulo.core.client.admin.CompactionConfig;
 import org.apache.accumulo.core.client.admin.NewTableConfiguration;
 import org.apache.accumulo.core.conf.Property;
@@ -286,17 +291,32 @@ public void testUsingNonExistentService() throws Exception {
     // Create a table that is configured to use a compaction service that does not exist
     try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
 
+      // cs5 does not exist, should fail
       NewTableConfiguration ntc = new NewTableConfiguration().setProperties(
           Map.of(Property.TABLE_COMPACTION_DISPATCHER_OPTS.getKey() + "service", "cs5"));
-      client.tableOperations().create(table, ntc);
-
       Wait.waitFor(() -> serviceMisconfigured.get() == true);
 
-      // The setup of this test creates an invalid configuration, fix this first thing.
+      assertThrows(AccumuloException.class, () -> client.tableOperations().create(table, ntc));
+      assertFalse(client.tableOperations().exists(table));
+
+      // change compaction service to one that is defined
+      NewTableConfiguration ntc2 = new NewTableConfiguration().setProperties(
+          Map.of(Property.TABLE_COMPACTION_DISPATCHER_OPTS.getKey() + "service", "cs1"));
+
+      // The setup of this test creates an invalid configuration, fix this and recreate the table.
       var value = "[{'group':'cs1q1'}]".replaceAll("'", "\"");
       client.instanceOperations().setProperty(CSP + "cs1.planner.opts.groups", value);
-
       Wait.waitFor(() -> serviceMisconfigured.get() == false);
+      Wait.waitFor(() -> {
+        try {
+          client.tableOperations().create(table, ntc2);
+          return client.tableOperations().exists(table);
+        } catch (AccumuloSecurityException | AccumuloException e) {
+          return false;
+        } catch (TableExistsException e) {
+          return true;
+        }
+      });
 
       // Add splits so that the tserver logs can manually be inspected to ensure they are not
       // spammed. Not sure how to check this automatically.

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());
+
+    }
+  }
+
 }