From ff5a2b3e0c13319820ac345e3df04f74fd505d2e Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Mon, 24 Aug 2026 15:24:00 +0000 Subject: [PATCH 1/2] Fail when configuring or compacting table with bad compaction svc Throw an exception when trying to configure a table to use an unknown compaction service or when trying to compact a table with a compaction service that does not exist in the configuration. Closes #4999 --- .../apache/accumulo/server/util/PropUtil.java | 19 +++++- .../accumulo/server/util/PropUtilTest.java | 61 +++++++++++++++++ .../accumulo/manager/FateServiceHandler.java | 10 +++ .../compaction/CompactionConfigChangeIT.java | 66 +++++++++++++++++++ 4 files changed, 155 insertions(+), 1 deletion(-) diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/PropUtil.java b/server/base/src/main/java/org/apache/accumulo/server/util/PropUtil.java index 27ef417486a..eec7b0526c5 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/PropUtil.java +++ b/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; + } + } } diff --git a/server/base/src/test/java/org/apache/accumulo/server/util/PropUtilTest.java b/server/base/src/test/java/org/apache/accumulo/server/util/PropUtilTest.java index 3baad4305e5..68f9f7e7105 100644 --- a/server/base/src/test/java/org/apache/accumulo/server/util/PropUtilTest.java +++ b/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); + } + } diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java b/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java index 28c681f4742..8eb7da1c11a 100644 --- a/server/manager/src/main/java/org/apache/accumulo/manager/FateServiceHandler.java +++ b/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, diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionConfigChangeIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/CompactionConfigChangeIT.java index 8a3f24ec0ac..5d21853856f 100644 --- a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionConfigChangeIT.java +++ b/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()); + + } + } + } From 7589a25a0c7a7cbe029d944f0f62e3391532eaa2 Mon Sep 17 00:00:00 2001 From: Dave Marion Date: Thu, 27 Aug 2026 15:47:54 +0000 Subject: [PATCH 2/2] Fix ITs --- .../server/conf/util/ImportConfigCommand.java | 36 +++++++++++++++++-- .../BadCompactionServiceConfigIT.java | 28 ++++++++++++--- 2 files changed, 57 insertions(+), 7 deletions(-) diff --git a/server/base/src/main/java/org/apache/accumulo/server/conf/util/ImportConfigCommand.java b/server/base/src/main/java/org/apache/accumulo/server/conf/util/ImportConfigCommand.java index 5c926cf761c..bfead80261d 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/conf/util/ImportConfigCommand.java +++ b/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 } 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 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); } } diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/BadCompactionServiceConfigIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/BadCompactionServiceConfigIT.java index 0f79d83b79a..2c784528029 100644 --- a/test/src/main/java/org/apache/accumulo/test/compaction/BadCompactionServiceConfigIT.java +++ b/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.