You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dd...@apache.org on 2023/12/27 22:28:52 UTC
(accumulo) branch main updated (829634653b -> b8869fb72e)
This is an automated email from the ASF dual-hosted git repository.
ddanielr pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from 829634653b Merge branch '2.1'
add fdd73fb8b2 Fix bug in compaction props (#4117)
new b8869fb72e Merge branch '2.1'
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../spi/compaction/DefaultCompactionPlanner.java | 6 +-
.../compaction/CompactionPlannerInitParams.java | 2 +-
.../compaction/DefaultCompactionPlannerTest.java | 343 ++++++---------------
3 files changed, 106 insertions(+), 245 deletions(-)
(accumulo) 01/01: Merge branch '2.1'
Posted by dd...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ddanielr pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit b8869fb72e2aa58736250c52f3366f196ac86c79
Merge: 829634653b fdd73fb8b2
Author: Daniel Roberts <dd...@gmail.com>
AuthorDate: Wed Dec 27 20:30:47 2023 +0000
Merge branch '2.1'
.../spi/compaction/DefaultCompactionPlanner.java | 6 +-
.../compaction/CompactionPlannerInitParams.java | 2 +-
.../compaction/DefaultCompactionPlannerTest.java | 343 ++++++---------------
3 files changed, 106 insertions(+), 245 deletions(-)
diff --cc core/src/main/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlanner.java
index b73544143e,9385806831..8f7969519a
--- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlanner.java
+++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlanner.java
@@@ -279,106 -210,105 +280,107 @@@ public class DefaultCompactionPlanner i
determineMaxFilesToCompact(params);
}
- @SuppressWarnings("removal")
++ @SuppressWarnings("deprecation")
private void determineMaxFilesToCompact(InitParameters params) {
- String fqo = params.getFullyQualifiedOption("maxOpen");
- if (!params.getServiceEnvironment().getConfiguration().isSet(fqo)
- && params.getServiceEnvironment().getConfiguration()
- .isSet(Property.TSERV_MAJC_THREAD_MAXOPEN.getKey())) {
- log.warn("The property " + Property.TSERV_MAJC_THREAD_MAXOPEN.getKey()
- + " was set, it is deprecated. Set the " + fqo + " option instead.");
- this.maxFilesToCompact = Integer.parseInt(params.getServiceEnvironment().getConfiguration()
- .get(Property.TSERV_MAJC_THREAD_MAXOPEN.getKey()));
- } else {
- this.maxFilesToCompact = Integer.parseInt(params.getOptions().getOrDefault("maxOpen",
- Property.TSERV_COMPACTION_SERVICE_DEFAULT_MAX_OPEN.getDefaultValue()));
+
+ String maxOpen = params.getOptions().get("maxOpen");
+ if (maxOpen == null) {
- maxOpen = "10";
- log.trace("default maxOpen not set, defaulting to 10");
++ maxOpen = Property.TSERV_COMPACTION_SERVICE_DEFAULT_MAX_OPEN.getDefaultValue();
++ log.trace("default maxOpen not set, defaulting to {}", maxOpen);
}
+ this.maxFilesToCompact = Integer.parseInt(maxOpen);
}
- @Override
- public CompactionPlan makePlan(PlanningParameters params) {
- try {
+ private void validateConfig(JsonElement json, List<String> fields, String className) {
- if (params.getCandidates().isEmpty()) {
- return params.createPlanBuilder().build();
- }
+ JsonObject jsonObject = GSON.get().fromJson(json, JsonObject.class);
+
+ List<String> objectProperties = new ArrayList<>(jsonObject.keySet());
+ HashSet<String> classFieldNames = new HashSet<>(fields);
+
+ if (!classFieldNames.containsAll(objectProperties)) {
+ objectProperties.removeAll(classFieldNames);
+ throw new JsonParseException(
+ "Invalid fields: " + objectProperties + " provided for class: " + className);
+ }
+ }
- Set<CompactableFile> filesCopy = new HashSet<>(params.getCandidates());
+ @Override
+ public CompactionPlan makePlan(PlanningParameters params) {
+ if (params.getCandidates().isEmpty()) {
+ return params.createPlanBuilder().build();
+ }
- long maxSizeToCompact = getMaxSizeToCompact(params.getKind());
+ Set<CompactableFile> filesCopy = new HashSet<>(params.getCandidates());
- Collection<CompactableFile> group;
- if (params.getRunningCompactions().isEmpty()) {
- group = findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact,
- maxSizeToCompact);
+ long maxSizeToCompact = getMaxSizeToCompact(params.getKind());
- if (!group.isEmpty() && group.size() < params.getCandidates().size()
- && params.getCandidates().size() <= maxFilesToCompact
- && (params.getKind() == CompactionKind.USER
- || params.getKind() == CompactionKind.SELECTOR)) {
- // USER and SELECTOR compactions must eventually compact all files. When a subset of files
- // that meets the compaction ratio is selected, look ahead and see if the next compaction
- // would also meet the compaction ratio. If not then compact everything to avoid doing
- // more than logarithmic work across multiple comapctions.
+ Collection<CompactableFile> group;
+ if (params.getRunningCompactions().isEmpty()) {
+ group =
+ findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact, maxSizeToCompact);
- filesCopy.removeAll(group);
- filesCopy.add(getExpected(group, 0));
+ if (!group.isEmpty() && group.size() < params.getCandidates().size()
+ && params.getCandidates().size() <= maxFilesToCompact
+ && (params.getKind() == CompactionKind.USER
+ || params.getKind() == CompactionKind.SELECTOR)) {
+ // USER and SELECTOR compactions must eventually compact all files. When a subset of files
+ // that meets the compaction ratio is selected, look ahead and see if the next compaction
+ // would also meet the compaction ratio. If not then compact everything to avoid doing
+ // more than logarithmic work across multiple comapctions.
- if (findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact,
- maxSizeToCompact).isEmpty()) {
- // The next possible compaction does not meet the compaction ratio, so compact
- // everything.
- group = Set.copyOf(params.getCandidates());
- }
+ filesCopy.removeAll(group);
+ filesCopy.add(getExpected(group, 0));
+ if (findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact,
+ maxSizeToCompact).isEmpty()) {
+ // The next possible compaction does not meet the compaction ratio, so compact
+ // everything.
+ group = Set.copyOf(params.getCandidates());
}
- } else if (params.getKind() == CompactionKind.SYSTEM) {
- // This code determines if once the files compacting finish would they be included in a
- // compaction with the files smaller than them? If so, then wait for the running compaction
- // to complete.
+ }
- // The set of files running compactions may produce
- var expectedFiles = getExpected(params.getRunningCompactions());
+ } else if (params.getKind() == CompactionKind.SYSTEM) {
+ // This code determines if once the files compacting finish would they be included in a
+ // compaction with the files smaller than them? If so, then wait for the running compaction
+ // to complete.
- if (!Collections.disjoint(filesCopy, expectedFiles)) {
- throw new AssertionError();
- }
+ // The set of files running compactions may produce
+ var expectedFiles = getExpected(params.getRunningCompactions());
- filesCopy.addAll(expectedFiles);
+ if (!Collections.disjoint(filesCopy, expectedFiles)) {
+ throw new AssertionError();
+ }
- group = findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact,
- maxSizeToCompact);
+ filesCopy.addAll(expectedFiles);
- if (!Collections.disjoint(group, expectedFiles)) {
- // file produced by running compaction will eventually compact with existing files, so
- // wait.
- group = Set.of();
- }
- } else {
+ group =
+ findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact, maxSizeToCompact);
+
+ if (!Collections.disjoint(group, expectedFiles)) {
+ // file produced by running compaction will eventually compact with existing files, so
+ // wait.
group = Set.of();
}
+ } else {
+ group = Set.of();
+ }
- if (group.isEmpty()
- && (params.getKind() == CompactionKind.USER || params.getKind() == CompactionKind.SELECTOR
- || params.getKind() == CompactionKind.CHOP)
- && params.getRunningCompactions().stream()
- .noneMatch(job -> job.getKind() == params.getKind())) {
- group = findMaximalRequiredSetToCompact(params.getCandidates(), maxFilesToCompact);
- }
+ if (group.isEmpty()
+ && (params.getKind() == CompactionKind.USER || params.getKind() == CompactionKind.SELECTOR)
+ && params.getRunningCompactions().stream()
+ .noneMatch(job -> job.getKind() == params.getKind())) {
+ group = findMaximalRequiredSetToCompact(params.getCandidates(), maxFilesToCompact);
+ }
- if (group.isEmpty()) {
- return params.createPlanBuilder().build();
- } else {
- // determine which executor to use based on the size of the files
- var ceid = getExecutor(group);
+ if (group.isEmpty()) {
+ return params.createPlanBuilder().build();
+ } else {
+ // determine which executor to use based on the size of the files
+ var ceid = getExecutor(group);
- return params.createPlanBuilder().addJob(createPriority(params, group), ceid, group)
- .build();
- }
- } catch (RuntimeException e) {
- throw e;
+ return params.createPlanBuilder().addJob(createPriority(params, group), ceid, group).build();
}
}
diff --cc core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionPlannerInitParams.java
index 3a3f72c699,eb287153d6..86fd8c478a
--- a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionPlannerInitParams.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionPlannerInitParams.java
@@@ -61,7 -60,7 +61,7 @@@ public class CompactionPlannerInitParam
@Override
public String getFullyQualifiedOption(String key) {
- return prefix + serviceId + ".opts." + key;
- return Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + serviceId + ".planner.opts." + key;
++ return prefix + serviceId + ".planner.opts." + key;
}
@Override
diff --cc core/src/test/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlannerTest.java
index c043eac841,ab2003841b..bafda93e3e
--- a/core/src/test/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlannerTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/spi/compaction/DefaultCompactionPlannerTest.java
@@@ -20,10 -20,8 +20,9 @@@ package org.apache.accumulo.core.spi.co
import static com.google.common.collect.MoreCollectors.onlyElement;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
- import static org.junit.jupiter.api.Assertions.fail;
import java.net.URI;
import java.net.URISyntaxException;
@@@ -34,19 -33,24 +34,23 @@@ import java.util.Set
import java.util.stream.Collectors;
import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
+ import org.apache.accumulo.core.conf.ConfigurationCopy;
import org.apache.accumulo.core.conf.ConfigurationTypeHelper;
+ import org.apache.accumulo.core.conf.DefaultConfiguration;
import org.apache.accumulo.core.conf.Property;
-import org.apache.accumulo.core.conf.SiteConfiguration;
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.spi.common.ServiceEnvironment;
import org.apache.accumulo.core.spi.common.ServiceEnvironment.Configuration;
import org.apache.accumulo.core.spi.compaction.CompactionPlan.Builder;
+import org.apache.accumulo.core.spi.compaction.CompactionPlanner.InitParameters;
+ import org.apache.accumulo.core.util.ConfigurationImpl;
import org.apache.accumulo.core.util.compaction.CompactionExecutorIdImpl;
import org.apache.accumulo.core.util.compaction.CompactionPlanImpl;
+ import org.apache.accumulo.core.util.compaction.CompactionPlannerInitParams;
-import org.apache.accumulo.core.util.compaction.CompactionServicesConfig;
import org.easymock.EasyMock;
import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
+import com.google.gson.JsonParseException;
public class DefaultCompactionPlannerTest {
@@@ -54,8 -58,14 +58,13 @@@
return c.stream().collect(onlyElement());
}
+ private static final Configuration defaultConf =
+ new ConfigurationImpl(DefaultConfiguration.getInstance());
+ private static final CompactionServiceId csid = CompactionServiceId.of("cs1");
-
- private static final Logger log = LoggerFactory.getLogger(DefaultCompactionPlannerTest.class);
++ private static final String prefix = Property.COMPACTION_SERVICE_PREFIX.getKey();
+
@Test
- public void testFindFilesToCompact() {
+ public void testFindFilesToCompact() throws Exception {
testFFtC(createCFs("F4", "1M", "F5", "1M", "F6", "1M"),
createCFs("F1", "100M", "F2", "100M", "F3", "100M", "F4", "1M", "F5", "1M", "F6", "1M"),
@@@ -133,8 -143,14 +142,14 @@@
}
@Test
- public void testRunningCompaction() {
+ public void testRunningCompaction() throws Exception {
- var planner = createPlanner(true);
+ String executors = "[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
+ + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2},"
+ + "{'name':'large','type': 'internal','maxSize':'512M','numThreads':3},"
+ + "{'name':'huge','type': 'internal','numThreads':4}]";
+
+ var planner = createPlanner(defaultConf, executors);
+
var all = createCFs("F1", "3M", "F2", "3M", "F3", "11M", "F4", "12M", "F5", "13M");
var candidates = createCFs("F3", "11M", "F4", "12M", "F5", "13M");
var compacting =
@@@ -156,12 -172,138 +171,21 @@@
// planner should compact.
var job = getOnlyElement(plan.getJobs());
assertEquals(candidates, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.externalId("medium"), job.getExecutor());
+ assertEquals(CompactionExecutorIdImpl.internalId(csid, "medium"), job.getExecutor());
}
- /**
- * Tests that the maxOpen property overrides the deprecated open.max property with the default
- * service
- */
@Test
- @SuppressWarnings("removal")
- public void testOverrideMaxOpenDefaultService() {
- Map<String,String> overrides = new HashMap<>();
- // Set old property and use that for max open files.
- overrides.put(Property.TSERV_MAJC_THREAD_MAXOPEN.getKey(), "17");
- SiteConfiguration aconf = SiteConfiguration.empty().withOverrides(overrides).build();
- ConfigurationImpl config = new ConfigurationImpl(aconf);
-
- ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
- EasyMock.expect(senv.getConfiguration()).andReturn(config).anyTimes();
- EasyMock.replay(senv);
-
- // Use the CompactionServicesConfig to create options based on default property values
- var compactionServices = new CompactionServicesConfig(aconf, log::warn);
- var options = compactionServices.getOptions().get("default");
-
- var initParams =
- new CompactionPlannerInitParams(CompactionServiceId.of("default"), options, senv);
-
- var planner = new DefaultCompactionPlanner();
- planner.init(initParams);
-
- var all = createCFs("F1", "10M", "F2", "11M", "F3", "12M", "F4", "13M", "F5", "14M", "F6",
- "15M", "F7", "16M", "F8", "17M", "F9", "18M", "FA", "19M", "FB", "20M", "FC", "21M", "FD",
- "22M", "FE", "23M", "FF", "24M", "FG", "25M", "FH", "26M");
- Set<CompactionJob> compacting = Set.of();
- var params = createPlanningParams(all, all, compacting, 2, CompactionKind.USER);
- var plan = planner.makePlan(params);
- var job = getOnlyElement(plan.getJobs());
- assertEquals(all, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(CompactionServiceId.of("default"), "large"),
- job.getExecutor());
-
- overrides.put(Property.TSERV_MAJC_THREAD_MAXOPEN.getKey(), "5");
- aconf = SiteConfiguration.empty().withOverrides(overrides).build();
- config = new ConfigurationImpl(aconf);
- senv = EasyMock.createMock(ServiceEnvironment.class);
- EasyMock.expect(senv.getConfiguration()).andReturn(config).anyTimes();
- EasyMock.replay(senv);
-
- // Create new initParams so executor IDs can be reused
- initParams = new CompactionPlannerInitParams(CompactionServiceId.of("default"), options, senv);
- planner = new DefaultCompactionPlanner();
- planner.init(initParams);
-
- params = createPlanningParams(all, all, compacting, 2, CompactionKind.USER);
- plan = planner.makePlan(params);
- job = getOnlyElement(plan.getJobs());
- assertEquals(createCFs("F1", "10M", "F2", "11M", "F3", "12M", "F4", "13M", "F5", "14M"),
- job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(CompactionServiceId.of("default"), "medium"),
- job.getExecutor());
- }
-
- /**
- * Tests that the maxOpen property overrides the deprecated open.max property
- */
- @Test
- @SuppressWarnings("removal")
- public void testOverrideMaxOpen() {
- Map<String,String> overrides = new HashMap<>();
- // Set old property and use that for max open files.
- overrides.put(Property.TSERV_MAJC_THREAD_MAXOPEN.getKey(), "17");
- SiteConfiguration aconf = SiteConfiguration.empty().withOverrides(overrides).build();
- ConfigurationImpl config = new ConfigurationImpl(aconf);
-
- String executors = "[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
- + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2},"
- + "{'name':'large','type': 'internal','maxSize':'512M','numThreads':3},"
- + "{'name':'huge','type': 'internal','numThreads':4}]";
-
- var planner = createPlanner(config, executors);
- var all = createCFs("F1", "1M", "F2", "2M", "F3", "4M", "F4", "8M", "F5", "16M", "F6", "32M",
- "F7", "64M", "F8", "128M", "F9", "256M", "FA", "512M", "FB", "1G", "FC", "2G", "FD", "4G",
- "FE", "8G", "FF", "16G", "FG", "32G", "FH", "64G");
- Set<CompactionJob> compacting = Set.of();
- var params = createPlanningParams(all, all, compacting, 2, CompactionKind.USER);
- var plan = planner.makePlan(params);
- var job = getOnlyElement(plan.getJobs());
- assertEquals(all, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "huge"), job.getExecutor());
-
- // Set new property that overrides the old property.
- overrides.put(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.maxOpen",
- "15");
- aconf = SiteConfiguration.empty().withOverrides(overrides).build();
- config = new ConfigurationImpl(aconf);
- planner = createPlanner(config, executors);
- params = createPlanningParams(all, all, compacting, 2, CompactionKind.USER);
- plan = planner.makePlan(params);
-
- // 17 files that do not meet the compaction ratio. When max files to compact is 15,
- // the plan should do 3 files then 15
- job = getOnlyElement(plan.getJobs());
- assertEquals(createCFs("F1", "1M", "F2", "2M", "F3", "4M"), job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "small"), job.getExecutor());
-
- overrides.put(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.maxOpen",
- "5");
- aconf = SiteConfiguration.empty().withOverrides(overrides).build();
- // 17 files that do not meet the compaction ratio. When max files to compact is 5 should do 5,
- // files then another 5, then the final 5.
- config = new ConfigurationImpl(aconf);
- planner = createPlanner(config, executors);
- params = createPlanningParams(all, all, compacting, 2, CompactionKind.USER);
- plan = planner.makePlan(params);
- job = getOnlyElement(plan.getJobs());
- assertEquals(createCFs("F4", "8M", "F3", "4M", "F2", "2M", "F1", "1M", "F5", "16M"),
- job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "small"), job.getExecutor());
- }
-
- @Test
- public void testUserCompaction() {
+ public void testUserCompaction() throws Exception {
- var planner = createPlanner(true);
+ ConfigurationCopy aconf = new ConfigurationCopy(DefaultConfiguration.getInstance());
- aconf.set(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.maxOpen", "15");
++ aconf.set(prefix + "cs1.planner.opts.maxOpen", "15");
+ ConfigurationImpl config = new ConfigurationImpl(aconf);
+
+ String executors = "[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
+ + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2},"
+ + "{'name':'large','type': 'internal','maxSize':'512M','numThreads':3},"
+ + "{'name':'huge','type': 'internal','numThreads':4}]";
+
+ var planner = createPlanner(config, executors);
var all = createCFs("F1", "3M", "F2", "3M", "F3", "11M", "F4", "12M", "F5", "13M");
var candidates = createCFs("F3", "11M", "F4", "12M", "F5", "13M");
var compacting =
@@@ -224,8 -366,12 +248,12 @@@
}
@Test
- public void testMaxSize() {
+ public void testMaxSize() throws Exception {
- var planner = createPlanner(false);
+ String executors = "[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
+ + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2},"
+ + "{'name':'large','type': 'internal','maxSize':'512M','numThreads':3}]";
+
+ var planner = createPlanner(defaultConf, executors);
var all = createCFs("F1", "128M", "F2", "129M", "F3", "130M", "F4", "131M", "F5", "132M");
var params = createPlanningParams(all, all, Set.of(), 2, CompactionKind.SYSTEM);
var plan = planner.makePlan(params);
@@@ -240,77 -386,9 +268,64 @@@
plan = planner.makePlan(params);
job = getOnlyElement(plan.getJobs());
assertEquals(all, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.externalId("large"), job.getExecutor());
+ assertEquals(CompactionExecutorIdImpl.internalId(csid, "large"), job.getExecutor());
}
+ @Test
+ public void testQueueCreation() throws Exception {
+ DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- Configuration conf = EasyMock.createMock(Configuration.class);
- EasyMock.expect(conf.isSet(EasyMock.anyString())).andReturn(false).anyTimes();
-
- ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
- EasyMock.expect(senv.getConfiguration()).andReturn(conf).anyTimes();
- EasyMock.replay(conf, senv);
+
+ String queues = "[{\"name\": \"small\", \"maxSize\":\"32M\"},{\"name\":\"midsize\"}]";
- planner.init(getInitParamQueues(senv, queues));
++ planner.init(getInitParamQueues(defaultConf, queues));
+
+ var all = createCFs("F1", "1M", "F2", "1M", "F3", "1M", "F4", "1M");
+ var params = createPlanningParams(all, all, Set.of(), 2, CompactionKind.SYSTEM);
+ var plan = planner.makePlan(params);
+
+ var job = getOnlyElement(plan.getJobs());
+ assertEquals(all, job.getFiles());
+ assertEquals(CompactionExecutorIdImpl.externalId("small"), job.getExecutor());
+
+ all = createCFs("F1", "100M", "F2", "100M", "F3", "100M", "F4", "100M");
+ params = createPlanningParams(all, all, Set.of(), 2, CompactionKind.SYSTEM);
+ plan = planner.makePlan(params);
+
+ job = getOnlyElement(plan.getJobs());
+ assertEquals(all, job.getFiles());
+ assertEquals(CompactionExecutorIdImpl.externalId("midsize"), job.getExecutor());
+ }
+
+ /**
+ * Tests that additional fields in the JSON objects cause errors to be thrown.
+ */
+ @Test
+ public void testErrorAdditionalConfigFields() {
- Configuration conf = EasyMock.createMock(Configuration.class);
- EasyMock.expect(conf.isSet(EasyMock.anyString())).andReturn(false).anyTimes();
-
- ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
- EasyMock.expect(senv.getConfiguration()).andReturn(conf).anyTimes();
- EasyMock.replay(conf, senv);
-
+ DefaultCompactionPlanner QueuePlanner = new DefaultCompactionPlanner();
+
+ String queues =
+ "[{\"name\":\"smallQueue\", \"maxSize\":\"32M\"}, {\"name\":\"largeQueue\", \"type\":\"internal\", \"foo\":\"bar\", \"queue\":\"broken\"}]";
+
- final InitParameters queueParams = getInitParamQueues(senv, queues);
++ final InitParameters queueParams = getInitParamQueues(defaultConf, queues);
+ assertNotNull(queueParams);
+ var e = assertThrows(JsonParseException.class, () -> QueuePlanner.init(queueParams),
+ "Failed to throw error");
+ assertTrue(e.getMessage().contains("[type, foo, queue]"),
+ "Error message didn't contain '[type, foo, queue]'");
+
- String executors = getExecutors("'type': 'internal','maxSize':'32M','numThreads':1",
- "'type': 'internal','maxSize':'128M','numThreads':2, 'foo':'bar'",
- "'type': 'internal','numThreads':1, 'unexpectedField':'foo'");
++ String executors = "[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
++ + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2, 'foo':'bar'},"
++ + "{'name':'large','type': 'internal','numThreads':1, 'unexpectedField':'foo'}]";
+
- final InitParameters execParams = getInitParams(senv, executors);
++ final InitParameters execParams = getInitParams(defaultConf, executors);
+ assertNotNull(execParams);
+
+ DefaultCompactionPlanner ExecPlanner = new DefaultCompactionPlanner();
+ var err = assertThrows(JsonParseException.class, () -> ExecPlanner.init(execParams),
+ "Failed to throw error");
+ assertTrue(err.getMessage().contains("Invalid fields: [foo]"),
+ "Error message didn't contain '[foo]'");
+ }
+
/**
* Tests internal type executor with no numThreads set throws error
*/
@@@ -374,59 -434,6 +371,45 @@@
assertTrue(e.getMessage().contains("queue"), "Error message didn't contain queue");
}
+ /**
+ * Tests queue with missing name throws error
+ */
+ @Test
+ public void testErrorQueueNoName() {
+ DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- Configuration conf = EasyMock.createMock(Configuration.class);
- EasyMock.expect(conf.isSet(EasyMock.anyString())).andReturn(false).anyTimes();
-
- ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
- EasyMock.expect(senv.getConfiguration()).andReturn(conf).anyTimes();
- EasyMock.replay(conf, senv);
-
+ String queues = "[{\"name\":\"smallQueue\", \"maxSize\":\"32M\"}, {\"maxSize\":\"120M\"}]";
+
- final InitParameters params = getInitParamQueues(senv, queues);
++ final InitParameters params = getInitParamQueues(defaultConf, queues);
+ assertNotNull(params);
+
+ var e = assertThrows(NullPointerException.class, () -> planner.init(params),
+ "Failed to throw error");
+ assertEquals(e.getMessage(), "'name' must be specified", "Error message didn't contain 'name'");
+ }
+
+ /**
+ * Tests not having executors or queues throws errors
+ */
+ @Test
+ public void testErrorNoExecutors() {
+ DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- Configuration conf = EasyMock.createMock(Configuration.class);
- EasyMock.expect(conf.isSet(EasyMock.anyString())).andReturn(false).anyTimes();
-
- ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
- EasyMock.expect(senv.getConfiguration()).andReturn(conf).anyTimes();
- EasyMock.replay(conf, senv);
-
- var execParams = getInitParams(senv, "");
++ var execParams = getInitParams(defaultConf, "");
+ assertNotNull(execParams);
+
+ var e = assertThrows(IllegalStateException.class, () -> planner.init(execParams),
+ "Failed to throw error");
+ assertEquals("No defined executors or queues for this planner", e.getMessage(),
+ "Error message was not equal");
+
- var params = getInitParamQueues(senv, "");
++ var params = getInitParamQueues(defaultConf, "");
+ assertNotNull(params);
+
+ var e2 = assertThrows(IllegalStateException.class, () -> planner.init(params),
+ "Failed to throw error");
+ assertEquals("No defined executors or queues for this planner", e2.getMessage(),
+ "Error message was not equal");
+ }
+
/**
* Tests executors can only have one without a max size.
*/
@@@ -471,92 -467,11 +443,12 @@@
private CompactionJob createJob(CompactionKind kind, Set<CompactableFile> all,
Set<CompactableFile> files) {
return new CompactionPlanImpl.BuilderImpl(kind, all, all)
- .addJob((short) all.size(), CompactionExecutorIdImpl.externalId("small"), files).build()
- .getJobs().iterator().next();
+ .addJob((short) all.size(), CompactionExecutorIdImpl.internalId(csid, "small"), files)
+ .build().getJobs().iterator().next();
}
- private static Set<CompactableFile> createCFs(String... namesSizePairs) {
+ private static Set<CompactableFile> createCFs(String... namesSizePairs)
+ throws URISyntaxException {
Set<CompactableFile> files = new HashSet<>();
for (int i = 0; i < namesSizePairs.length; i += 2) {
@@@ -645,80 -564,30 +537,47 @@@
};
}
- private static DefaultCompactionPlanner createPlanner(boolean withHugeExecutor) {
- DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- Configuration conf = EasyMock.createMock(Configuration.class);
- EasyMock.expect(conf.isSet(EasyMock.anyString())).andReturn(false).anyTimes();
++ private static CompactionPlanner.InitParameters getInitParamQueues(Configuration conf,
++ String queues) {
+
- ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
- EasyMock.expect(senv.getConfiguration()).andReturn(conf).anyTimes();
++ String maxOpen = conf.get(prefix + "cs1.planner.opts.maxOpen");
++ Map<String,String> options = new HashMap<>();
++ options.put("queues", queues.replaceAll("'", "\""));
+
- EasyMock.replay(conf, senv);
++ if (maxOpen != null) {
++ options.put("maxOpen", maxOpen);
++ }
+
- StringBuilder execBldr =
- new StringBuilder("[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
- + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2},"
- + "{'name':'large','type': 'internal','maxSize':'512M','numThreads':3}");
++ ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
++ EasyMock.expect(senv.getConfiguration()).andReturn(conf).anyTimes();
++ EasyMock.replay(senv);
+
- if (withHugeExecutor) {
- execBldr.append(",{'name':'huge','type': 'internal','numThreads':4}]");
- } else {
- execBldr.append("]");
- }
++ return new CompactionPlannerInitParams(csid, prefix, options, senv);
++ }
+
- String executors = execBldr.toString().replaceAll("'", "\"");
+ private static CompactionPlanner.InitParameters getInitParams(Configuration conf,
+ String executors) {
- planner.init(new CompactionPlanner.InitParameters() {
- String maxOpen =
- conf.get(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.maxOpen");
++ String maxOpen = conf.get(prefix + "cs1.planner.opts.maxOpen");
+ Map<String,String> options = new HashMap<>();
+ options.put("executors", executors.replaceAll("'", "\""));
- @Override
- public ServiceEnvironment getServiceEnvironment() {
- return senv;
- }
+ if (maxOpen != null) {
+ options.put("maxOpen", maxOpen);
+ }
- @Override
- public Map<String,String> getOptions() {
- return Map.of("executors", executors, "maxOpen", "15");
- }
+ ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
+ EasyMock.expect(senv.getConfiguration()).andReturn(conf).anyTimes();
+ EasyMock.replay(senv);
- @Override
- public String getFullyQualifiedOption(String key) {
- assertEquals("maxOpen", key);
- return Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts." + key;
- }
- return new CompactionPlannerInitParams(csid, options, senv);
++ return new CompactionPlannerInitParams(csid, prefix, options, senv);
+ }
- @Override
- public ExecutorManager getExecutorManager() {
- return new ExecutorManager() {
- @Override
- public CompactionExecutorId createExecutor(String name, int threads) {
- switch (name) {
- case "small":
- assertEquals(1, threads);
- break;
- case "medium":
- assertEquals(2, threads);
- break;
- case "large":
- assertEquals(3, threads);
- break;
- case "huge":
- assertEquals(4, threads);
- break;
- default:
- fail("Unexpected name " + name);
- break;
- }
- return CompactionExecutorIdImpl.externalId(name);
- }
-
- @Override
- public CompactionExecutorId getExternalExecutor(String name) {
- throw new UnsupportedOperationException();
- }
- };
- }
- });
+ private static DefaultCompactionPlanner createPlanner(Configuration conf, String executors) {
+ DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
+ var initParams = getInitParams(conf, executors);
+ planner.init(initParams);
return planner;
}
}