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 2024/01/07 00:59:12 UTC
(accumulo) branch elasticity updated: Remove old compaction code (#4083)
This is an automated email from the ASF dual-hosted git repository.
ddanielr pushed a commit to branch elasticity
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/elasticity by this push:
new e15757f68b Remove old compaction code (#4083)
e15757f68b is described below
commit e15757f68b292fdadaf91f7f8572d0f0b9642f70
Author: Daniel Roberts <dd...@gmail.com>
AuthorDate: Sat Jan 6 19:59:06 2024 -0500
Remove old compaction code (#4083)
* SPI class modifications
* Metadata Changes and GroupID conversion
* Changes compaction groups to compactor groups
* Renames CompactionGroupID to CompactorGroupID
---
.../org/apache/accumulo/core/conf/Property.java | 85 +++----
.../apache/accumulo/core/logging/TabletLogger.java | 6 +-
.../accumulo/core/metadata/schema/Ample.java | 2 +-
...actionMetadata.java => CompactionMetadata.java} | 31 ++-
.../core/metadata/schema/TabletMetadata.java | 10 +-
.../metadata/schema/TabletMetadataBuilder.java | 2 +-
.../core/metadata/schema/TabletMutatorBase.java | 2 +-
.../core/spi/compaction/CompactionJob.java | 4 +-
.../core/spi/compaction/CompactionPlan.java | 7 +-
.../core/spi/compaction/CompactionPlanner.java | 9 +-
...actionExecutorId.java => CompactorGroupId.java} | 8 +-
.../spi/compaction/DefaultCompactionPlanner.java | 253 +++++--------------
.../{ExecutorManager.java => GroupManager.java} | 16 +-
.../accumulo/core/spi/compaction/package-info.java | 8 +-
.../core/util/compaction/CompactionJobImpl.java | 22 +-
.../core/util/compaction/CompactionPlanImpl.java | 6 +-
.../compaction/CompactionPlannerInitParams.java | 45 ++--
.../util/compaction/CompactionServicesConfig.java | 70 +-----
...ecutorIdImpl.java => CompactorGroupIdImpl.java} | 27 +-
.../core/metadata/schema/TabletMetadataTest.java | 6 +-
.../compaction/DefaultCompactionPlannerTest.java | 275 +++++++--------------
.../util/compaction/CompactionPrioritizerTest.java | 2 +-
.../compaction/CompactionServicesConfigTest.java | 64 +----
.../miniclusterImpl/MiniAccumuloClusterImpl.java | 11 +-
.../miniclusterImpl/MiniAccumuloConfigImpl.java | 24 +-
.../server/compaction/CompactionJobGenerator.java | 2 +-
.../server/conf/CheckCompactionConfig.java | 13 +-
.../server/conf/CheckCompactionConfigTest.java | 136 +++-------
.../coordinator/CompactionCoordinator.java | 34 ++-
.../compaction/coordinator/QueueMetrics.java | 14 +-
.../queue/CompactionJobPriorityQueue.java | 11 +-
.../compaction/queue/CompactionJobQueues.java | 54 ++--
.../compaction/BadCompactionServiceConfigIT.java | 14 +-
.../test/compaction/CompactionConfigChangeIT.java | 20 +-
.../test/compaction/CompactionExecutorIT.java | 54 ++--
.../CompactionPriorityQueueMetricsIT.java | 5 +-
.../compaction/ExternalCompactionTestUtils.java | 32 +--
.../test/functional/IdleProcessMetricsIT.java | 12 +-
.../apache/accumulo/test/functional/MergeIT.java | 12 +-
.../apache/accumulo/test/shell/ConfigSetIT.java | 9 +-
40 files changed, 463 insertions(+), 954 deletions(-)
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index e568dd44d4..07c651fc08 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
@@ -53,12 +53,40 @@ public enum Property {
+ "A new external compaction service would be defined like the following:\n"
+ "`compaction.service.newService.planner="
+ "\"org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner\".`\n"
- + "`compaction.service.newService.opts.queues=\""
+ + "`compaction.service.newService.opts.groups=\""
+ "[{\"name\": \"small\", \"maxSize\":\"32M\"},"
+ "{ \"name\":\"medium\", \"maxSize\":\"512M\"},{\"name\":\"large\"}]`\n"
+ "`compaction.service.newService.opts.maxOpen=50`.\n"
+ "Additional options can be defined using the `compaction.service.<service>.opts.<option>` property.",
"3.1.0"),
+ COMPACTION_SERVICE_ROOT_PLANNER(COMPACTION_SERVICE_PREFIX + "root.planner",
+ DefaultCompactionPlanner.class.getName(), PropertyType.CLASSNAME,
+ "Compaction planner for root tablet service.", "4.0.0"),
+ COMPACTION_SERVICE_ROOT_MAX_OPEN(COMPACTION_SERVICE_PREFIX + "root.planner.opts.maxOpen", "30",
+ PropertyType.COUNT, "The maximum number of files a compaction will open.", "4.0.0"),
+ COMPACTION_SERVICE_ROOT_GROUPS(COMPACTION_SERVICE_PREFIX + "root.planner.opts.groups",
+ "[{'name':'accumulo_meta'}]".replaceAll("'", "\""), PropertyType.STRING,
+ "See {% jlink -f org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner %}.",
+ "4.0.0"),
+ COMPACTION_SERVICE_META_PLANNER(COMPACTION_SERVICE_PREFIX + "meta.planner",
+ DefaultCompactionPlanner.class.getName(), PropertyType.CLASSNAME,
+ "Compaction planner for metadata table.", "4.0.0"),
+ COMPACTION_SERVICE_META_MAX_OPEN(COMPACTION_SERVICE_PREFIX + "meta.planner.opts.maxOpen", "30",
+ PropertyType.COUNT, "The maximum number of files a compaction will open.", "4.0.0"),
+ COMPACTION_SERVICE_META_GROUPS(COMPACTION_SERVICE_PREFIX + "meta.planner.opts.groups",
+ "[{'name':'accumulo_meta'}]".replaceAll("'", "\""), PropertyType.JSON,
+ "See {% jlink -f org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner %}.",
+ "4.0.0"),
+ COMPACTION_SERVICE_DEFAULT_PLANNER(COMPACTION_SERVICE_PREFIX + "default.planner",
+ DefaultCompactionPlanner.class.getName(), PropertyType.CLASSNAME,
+ "Planner for default compaction service.", "4.0.0"),
+ COMPACTION_SERVICE_DEFAULT_MAX_OPEN(COMPACTION_SERVICE_PREFIX + "default.planner.opts.maxOpen",
+ "10", PropertyType.COUNT, "The maximum number of files a compaction will open.", "4.0.0"),
+ COMPACTION_SERVICE_DEFAULT_GROUPS(COMPACTION_SERVICE_PREFIX + "default.planner.opts.groups",
+ ("[{'name':'user_small','maxSize':'128M'}, {'name':'user_large'}]").replaceAll("'", "\""),
+ PropertyType.STRING,
+ "See {% jlink -f org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner %}.",
+ "4.0.0"),
COMPACTION_WARN_TIME(COMPACTION_PREFIX + "warn.time", "10m", PropertyType.TIMEDURATION,
"When a compaction has not made progress for this time period, a warning will be logged.",
"3.1.0"),
@@ -605,63 +633,8 @@ public enum Property {
"2.1.0"),
TSERV_MIGRATE_MAXCONCURRENT("tserver.migrations.concurrent.max", "1", PropertyType.COUNT,
"The maximum number of concurrent tablet migrations for a tablet server.", "1.3.5"),
- @Deprecated(since = "3.1")
- @ReplacedBy(property = COMPACTION_SERVICE_PREFIX)
- TSERV_COMPACTION_SERVICE_PREFIX("tserver.compaction.major.service.", null, PropertyType.PREFIX,
- "Prefix for compaction services.", "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_ROOT_PLANNER("tserver.compaction.major.service.root.planner",
- DefaultCompactionPlanner.class.getName(), PropertyType.CLASSNAME,
- "Compaction planner for root tablet service.", "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_ROOT_MAX_OPEN(
- "tserver.compaction.major.service.root.planner.opts.maxOpen", "30", PropertyType.COUNT,
- "The maximum number of files a compaction will open.", "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_ROOT_EXECUTORS(
- "tserver.compaction.major.service.root.planner.opts.executors",
- "[{'name':'all','type':'external','group':'accumulo_meta'}]".replaceAll("'", "\""),
- PropertyType.STRING,
- "See {% jlink -f org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner %}.",
- "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_META_PLANNER("tserver.compaction.major.service.meta.planner",
- DefaultCompactionPlanner.class.getName(), PropertyType.CLASSNAME,
- "Compaction planner for metadata table.", "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_META_MAX_OPEN(
- "tserver.compaction.major.service.meta.planner.opts.maxOpen", "30", PropertyType.COUNT,
- "The maximum number of files a compaction will open.", "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_META_EXECUTORS(
- "tserver.compaction.major.service.meta.planner.opts.executors",
- "[{'name':'all','type':'external','group':'accumulo_meta'}]".replaceAll("'", "\""),
- PropertyType.JSON,
- "See {% jlink -f org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner %}.",
- "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_DEFAULT_PLANNER("tserver.compaction.major.service.default.planner",
- DefaultCompactionPlanner.class.getName(), PropertyType.CLASSNAME,
- "Planner for default compaction service.", "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_DEFAULT_MAX_OPEN(
- "tserver.compaction.major.service.default.planner.opts.maxOpen", "10", PropertyType.COUNT,
- "The maximum number of files a compaction will open.", "2.1.0"),
- @Deprecated(since = "3.1")
- TSERV_COMPACTION_SERVICE_DEFAULT_EXECUTORS(
- "tserver.compaction.major.service.default.planner.opts.executors",
- ("[{'name':'small','type':'external','maxSize':'128M','group':'user_small'}, {'name':'large','type':'external','group':'user_large'}]")
- .replaceAll("'", "\""),
- PropertyType.STRING,
- "See {% jlink -f org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner %}.",
- "2.1.0"),
TSERV_MINC_MAXCONCURRENT("tserver.compaction.minor.concurrent.max", "4", PropertyType.COUNT,
"The maximum number of concurrent minor compactions for a tablet server.", "1.3.5"),
- @Deprecated(since = "3.1")
- @ReplacedBy(property = COMPACTION_WARN_TIME)
- TSERV_COMPACTION_WARN_TIME("tserver.compaction.warn.time", "10m", PropertyType.TIMEDURATION,
- "When a compaction has not made progress for this time period, a warning will be logged.",
- "1.6.0"),
TSERV_BLOOM_LOAD_MAXCONCURRENT("tserver.bloom.load.concurrent.max", "4", PropertyType.COUNT,
"The number of concurrent threads that will load bloom filters in the background. "
+ "Setting this to zero will make bloom filters load in the foreground.",
diff --git a/core/src/main/java/org/apache/accumulo/core/logging/TabletLogger.java b/core/src/main/java/org/apache/accumulo/core/logging/TabletLogger.java
index 0bdf7e2055..1950aa5c80 100644
--- a/core/src/main/java/org/apache/accumulo/core/logging/TabletLogger.java
+++ b/core/src/main/java/org/apache/accumulo/core/logging/TabletLogger.java
@@ -135,12 +135,12 @@ public class TabletLogger {
public static void compacting(KeyExtent extent, CompactionJob job, CompactionConfig config) {
if (fileLog.isDebugEnabled()) {
if (config == null) {
- fileLog.debug("Compacting {} on {} for {} from {} size {}", extent, job.getExecutor(),
+ fileLog.debug("Compacting {} on {} for {} from {} size {}", extent, job.getGroup(),
job.getKind(), asMinimalString(job.getFiles()), getSize(job.getFiles()));
} else {
fileLog.debug("Compacting {} on {} for {} from {} size {} config {}", extent,
- job.getExecutor(), job.getKind(), asMinimalString(job.getFiles()),
- getSize(job.getFiles()), config);
+ job.getGroup(), job.getKind(), asMinimalString(job.getFiles()), getSize(job.getFiles()),
+ config);
}
}
}
diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java
index ac98db3f16..98666a74c8 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/Ample.java
@@ -381,7 +381,7 @@ public interface Ample {
T deleteSuspension();
- T putExternalCompaction(ExternalCompactionId ecid, ExternalCompactionMetadata ecMeta);
+ T putExternalCompaction(ExternalCompactionId ecid, CompactionMetadata ecMeta);
T deleteExternalCompaction(ExternalCompactionId ecid);
diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/ExternalCompactionMetadata.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/CompactionMetadata.java
similarity index 77%
rename from core/src/main/java/org/apache/accumulo/core/metadata/schema/ExternalCompactionMetadata.java
rename to core/src/main/java/org/apache/accumulo/core/metadata/schema/CompactionMetadata.java
index c6df9c8673..c790f3698b 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/ExternalCompactionMetadata.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/CompactionMetadata.java
@@ -28,30 +28,30 @@ import java.util.Set;
import org.apache.accumulo.core.metadata.ReferencedTabletFile;
import org.apache.accumulo.core.metadata.StoredTabletFile;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionKind;
-import org.apache.accumulo.core.util.compaction.CompactionExecutorIdImpl;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
+import org.apache.accumulo.core.util.compaction.CompactorGroupIdImpl;
-public class ExternalCompactionMetadata {
+public class CompactionMetadata {
private final Set<StoredTabletFile> jobFiles;
private final ReferencedTabletFile compactTmpName;
private final String compactorId;
private final CompactionKind kind;
private final short priority;
- private final CompactionExecutorId ceid;
+ private final CompactorGroupId cgid;
private final boolean propagateDeletes;
private final Long fateTxId;
- public ExternalCompactionMetadata(Set<StoredTabletFile> jobFiles,
- ReferencedTabletFile compactTmpName, String compactorId, CompactionKind kind, short priority,
- CompactionExecutorId ceid, boolean propagateDeletes, Long fateTxId) {
+ public CompactionMetadata(Set<StoredTabletFile> jobFiles, ReferencedTabletFile compactTmpName,
+ String compactorId, CompactionKind kind, short priority, CompactorGroupId ceid,
+ boolean propagateDeletes, Long fateTxId) {
this.jobFiles = Objects.requireNonNull(jobFiles);
this.compactTmpName = Objects.requireNonNull(compactTmpName);
this.compactorId = Objects.requireNonNull(compactorId);
this.kind = Objects.requireNonNull(kind);
this.priority = priority;
- this.ceid = Objects.requireNonNull(ceid);
+ this.cgid = Objects.requireNonNull(ceid);
this.propagateDeletes = propagateDeletes;
this.fateTxId = fateTxId;
}
@@ -76,8 +76,8 @@ public class ExternalCompactionMetadata {
return priority;
}
- public CompactionExecutorId getCompactionExecutorId() {
- return ceid;
+ public CompactorGroupId getCompactionGroupId() {
+ return cgid;
}
public boolean getPropagateDeletes() {
@@ -95,7 +95,7 @@ public class ExternalCompactionMetadata {
String tmp;
String compactor;
String kind;
- String executorId;
+ String groupId;
short priority;
boolean propDels;
Long fateTxId;
@@ -108,21 +108,20 @@ public class ExternalCompactionMetadata {
jData.tmp = compactTmpName.insert().getMetadata();
jData.compactor = compactorId;
jData.kind = kind.name();
- jData.executorId = ((CompactionExecutorIdImpl) ceid).getExternalName();
+ jData.groupId = cgid.toString();
jData.priority = priority;
jData.propDels = propagateDeletes;
jData.fateTxId = fateTxId;
return GSON.get().toJson(jData);
}
- public static ExternalCompactionMetadata fromJson(String json) {
+ public static CompactionMetadata fromJson(String json) {
GSonData jData = GSON.get().fromJson(json, GSonData.class);
- return new ExternalCompactionMetadata(
- jData.inputs.stream().map(StoredTabletFile::new).collect(toSet()),
+ return new CompactionMetadata(jData.inputs.stream().map(StoredTabletFile::new).collect(toSet()),
StoredTabletFile.of(jData.tmp).getTabletFile(), jData.compactor,
CompactionKind.valueOf(jData.kind), jData.priority,
- CompactionExecutorIdImpl.externalId(jData.executorId), jData.propDels, jData.fateTxId);
+ CompactorGroupIdImpl.groupId(jData.groupId), jData.propDels, jData.fateTxId);
}
@Override
diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
index 8576d4c459..df37c05566 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadata.java
@@ -116,7 +116,7 @@ public class TabletMetadata {
private OptionalLong flush = OptionalLong.empty();
private List<LogEntry> logs;
private Double splitRatio = null;
- private Map<ExternalCompactionId,ExternalCompactionMetadata> extCompactions;
+ private Map<ExternalCompactionId,CompactionMetadata> extCompactions;
private boolean merged;
private TabletHostingGoal goal = TabletHostingGoal.ONDEMAND;
private boolean onDemandHostingRequested = false;
@@ -414,7 +414,7 @@ public class TabletMetadata {
return keyValues;
}
- public Map<ExternalCompactionId,ExternalCompactionMetadata> getExternalCompactions() {
+ public Map<ExternalCompactionId,CompactionMetadata> getExternalCompactions() {
ensureFetched(ColumnType.ECOMP);
return extCompactions;
}
@@ -449,8 +449,7 @@ public class TabletMetadata {
final var filesBuilder = ImmutableMap.<StoredTabletFile,DataFileValue>builder();
final var scansBuilder = ImmutableList.<StoredTabletFile>builder();
final var logsBuilder = ImmutableList.<LogEntry>builder();
- final var extCompBuilder =
- ImmutableMap.<ExternalCompactionId,ExternalCompactionMetadata>builder();
+ final var extCompBuilder = ImmutableMap.<ExternalCompactionId,CompactionMetadata>builder();
final var loadedFilesBuilder = ImmutableMap.<StoredTabletFile,Long>builder();
final var compactedBuilder = ImmutableSet.<Long>builder();
ByteSequence row = null;
@@ -542,8 +541,7 @@ public class TabletMetadata {
logsBuilder.add(LogEntry.fromMetaWalEntry(kv));
break;
case ExternalCompactionColumnFamily.STR_NAME:
- extCompBuilder.put(ExternalCompactionId.of(qual),
- ExternalCompactionMetadata.fromJson(val));
+ extCompBuilder.put(ExternalCompactionId.of(qual), CompactionMetadata.fromJson(val));
break;
case MergedColumnFamily.STR_NAME:
te.merged = true;
diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataBuilder.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataBuilder.java
index 7d102ea7ef..d490cd3f50 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataBuilder.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMetadataBuilder.java
@@ -190,7 +190,7 @@ public class TabletMetadataBuilder implements Ample.TabletUpdates<TabletMetadata
@Override
public TabletMetadataBuilder putExternalCompaction(ExternalCompactionId ecid,
- ExternalCompactionMetadata ecMeta) {
+ CompactionMetadata ecMeta) {
fetched.add(ECOMP);
internalBuilder.putExternalCompaction(ecid, ecMeta);
return this;
diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMutatorBase.java b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMutatorBase.java
index a0b83c00a8..b7cdbb03f8 100644
--- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMutatorBase.java
+++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletMutatorBase.java
@@ -241,7 +241,7 @@ public abstract class TabletMutatorBase<T extends Ample.TabletUpdates<T>>
}
@Override
- public T putExternalCompaction(ExternalCompactionId ecid, ExternalCompactionMetadata ecMeta) {
+ public T putExternalCompaction(ExternalCompactionId ecid, CompactionMetadata ecMeta) {
mutation.put(ExternalCompactionColumnFamily.STR_NAME, ecid.canonical(), ecMeta.toJson());
return getThis();
}
diff --git a/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionJob.java b/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionJob.java
index 3b51f06418..6f22f33d2d 100644
--- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionJob.java
+++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionJob.java
@@ -33,9 +33,9 @@ public interface CompactionJob {
short getPriority();
/**
- * @return The executor to run the job.
+ * @return The group to run the job.
*/
- CompactionExecutorId getExecutor();
+ CompactorGroupId getGroup();
/**
* @return The files to compact
diff --git a/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionPlan.java b/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionPlan.java
index f39752f2e4..33f277b6c4 100644
--- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionPlan.java
+++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionPlan.java
@@ -42,12 +42,11 @@ public interface CompactionPlan {
* @param priority This determines the order in which the job is taken off the execution queue.
* Larger numbers are taken off the queue first. If two jobs are on the queue, one with a
* priority of 4 and another with 5, then the one with 5 will be taken first.
- * @param executor Where the job should run.
- * @param group The files to compact.
+ * @param group Where the job should run.
+ * @param files The files to compact.
* @return this
*/
- Builder addJob(short priority, CompactionExecutorId executor,
- Collection<CompactableFile> group);
+ Builder addJob(short priority, CompactorGroupId group, Collection<CompactableFile> files);
CompactionPlan build();
}
diff --git a/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionPlanner.java b/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionPlanner.java
index 222ee651ef..302f14dde0 100644
--- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionPlanner.java
+++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionPlanner.java
@@ -61,10 +61,9 @@ public interface CompactionPlanner {
String getFullyQualifiedOption(String key);
/**
- * @return an execution manager that can be used to created thread pools within a compaction
- * service.
+ * @return a group manager that can be used to create groups for a compaction service.
*/
- ExecutorManager getExecutorManager();
+ GroupManager getGroupManager();
}
public void init(InitParameters params);
@@ -156,8 +155,8 @@ public interface CompactionPlanner {
* {@code [F3,F4,F5]} and it must eventually compact those three files to one.
*
* <p>
- * When a planner returns a compactions plan, task will be queued on executors. Previously queued
- * task that do not match the latest plan are removed. The planner is called periodically,
+ * When a planner returns a compactions plan, task will be queued on a compactor group. Previously
+ * queued task that do not match the latest plan are removed. The planner is called periodically,
* whenever a new file is added, and whenever a compaction finishes.
*
* <p>
diff --git a/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionExecutorId.java b/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactorGroupId.java
similarity index 83%
rename from core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionExecutorId.java
rename to core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactorGroupId.java
index 5a456f6e99..f9fd032d75 100644
--- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactionExecutorId.java
+++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/CompactorGroupId.java
@@ -21,17 +21,17 @@ package org.apache.accumulo.core.spi.compaction;
import org.apache.accumulo.core.data.AbstractId;
/**
- * A unique identifier for a a compaction executor that a {@link CompactionPlanner} can schedule
+ * A unique identifier for a compactor group that a {@link CompactionPlanner} can schedule
* compactions on using a {@link CompactionJob}.
*
- * @since 2.1.0
+ * @since 3.1.0
* @see org.apache.accumulo.core.spi.compaction
*/
-public class CompactionExecutorId extends AbstractId<CompactionExecutorId> {
+public class CompactorGroupId extends AbstractId<CompactorGroupId> {
// ELASTICITY_TODO make this cache ids like TableId. This will help save manager memory.
private static final long serialVersionUID = 1L;
- protected CompactionExecutorId(String canonical) {
+ protected CompactorGroupId(String canonical) {
super(canonical);
}
}
diff --git 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
index 6960042e92..4cb1f3a478 100644
--- 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
@@ -59,85 +59,49 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
* compaction service you are configuring.
*
* <ul>
- * <li>{@code compaction.service.<service>.opts.executors} This is a json array of objects where
- * each object has the fields:
+ * <li>Note that the CompactionCoordinator and at least one running Compactor must be assigned to
+ * the "large" compactor group.
+ * <li>{@code compaction.service.<service>.opts.maxOpen} This determines the maximum number of files
+ * that will be included in a single compaction.
+ * <li>{@code compaction.service.<service>.opts.groups} This is a json array of compactor group
+ * objects which have the following fields:
* <table>
- * <caption>Default Compaction Planner Executor options</caption>
+ * <caption>Default Compaction Planner Group options</caption>
* <tr>
* <th>Field Name</th>
* <th>Description</th>
* </tr>
* <tr>
* <td>name</td>
- * <td>name or alias of the executor (required)</td>
- * </tr>
- * <tr>
- * <td>type</td>
- * <td>valid values 'internal' or 'external' (required)</td>
+ * <td>name of the compactor group (required)</td>
* </tr>
* <tr>
* <td>maxSize</td>
* <td>threshold sum of the input files (required for all but one of the configs)</td>
* </tr>
- * <tr>
- * <td>numThreads</td>
- * <td>number of threads for this executor configuration (required for 'internal', cannot be
- * specified for 'external')</td>
- * </tr>
- * <tr>
- * <td>group</td>
- * <td>name of the external compaction group (required for 'external', cannot be specified for
- * 'internal')</td>
- * </tr>
* </table>
* <br>
- * Note: The "executors" option has been deprecated in 3.1 and will be removed in a future release.
- * This example uses the new `compaction.service` prefix. The property prefix
- * "tserver.compaction.major.service" has also been deprecated in 3.1 and will be removed in a
- * future release. The maxSize field determines the maximum size of compaction that will run on an
- * executor. The maxSize field can have a suffix of K,M,G for kilobytes, megabytes, or gigabytes and
- * represents the sum of the input files for a given compaction. One executor can have no max size
- * and it will run everything that is too large for the other executors. If all executors have a max
- * size, then system compactions will only run for compactions smaller than the largest max size.
- * User, chop, and selector compactions will always run, even if there is no executor for their
- * size. These compactions will run on the executor with the largest max size. The following example
- * value for this property will create 3 threads to run compactions of files whose file size sum is
- * less than 100M, 3 threads to run compactions of files whose file size sum is less than 500M, and
- * run all other compactions on Compactors configured to run compactions for Queue1:
+ * This 'groups' object provides information that is used for mapping a compaction job to a
+ * compactor group. The maxSize field determines the maximum size of compaction that will run in a
+ * group. The maxSize field can have a suffix of K,M,G for kilobytes, megabytes, or gigabytes and
+ * represents the sum of the input files for a given compaction. One group can have no max size and
+ * it will run everything that is too large for the other groups. If all groups have a max size,
+ * then system compactions will only run for compactions smaller than the largest max size. User and
+ * selector compactions will always run, even if there is no group for their size. These compactions
+ * will run on the group with the largest max size. The following example value for this property
+ * will create three separate compactor groups. "small" will run compactions of files whose file
+ * size sum is less than 100M, "medium" will run compactions of files whose file size sum is less
+ * than 500M, and "large" will run all other compactions on Compactors configured to pull jobs from
+ * the large group.
*
* <pre>
* {@code
* [
- * {"name":"small", "type": "internal", "maxSize":"100M","numThreads":3},
- * {"name":"medium", "type": "internal", "maxSize":"500M","numThreads":3},
- * {"name": "large", "type": "external", "group", "Queue1"}
+ * {"name":"small", "maxSize":"100M"},
+ * {"name":"medium", "maxSize":"500M"},
+ * {"name": "large"}
* ]}
* </pre>
- *
- * Note that the use of 'external' requires that the CompactionCoordinator and at least one
- * Compactor for Queue1 is running.
- * <li>{@code compaction.service.<service>.opts.maxOpen} This determines the maximum number of files
- * that will be included in a single compaction.
- * <li>{@code compaction.service.<service>.opts.queues} This is a json array of queue objects which
- * have the following fields:
- * <table>
- * <caption>Default Compaction Planner Queue options</caption>
- * <tr>
- * <th>Field Name</th>
- * <th>Description</th>
- * </tr>
- * <tr>
- * <td>name</td>
- * <td>name or alias of the queue (required)</td>
- * </tr>
- * <tr>
- * <td>maxSize</td>
- * <td>threshold sum of the input files (required for all but one of the configs)</td>
- * </tr>
- * </table>
- * <br>
- * This 'queues' object is used for defining external compaction queues without needing to use the
- * thread-based 'executors' property.
* </ul>
*
* <p>
@@ -154,7 +118,6 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
* planned. If the compaction ratio is set to 3, then this plugin will find the largest compaction
* ratio less than 3 that results in a compaction.
*
- *
* @since 3.1.0
* @see org.apache.accumulo.core.spi.compaction
*/
@@ -163,67 +126,18 @@ public class DefaultCompactionPlanner implements CompactionPlanner {
private final static Logger log = LoggerFactory.getLogger(DefaultCompactionPlanner.class);
- private static class ExecutorConfig {
- String type;
- String name;
- String maxSize;
- Integer numThreads;
- String group;
-
- public String getType() {
- return type;
- }
-
- public void setType(String type) {
- this.type = type;
- }
-
- public String getName() {
- return name;
- }
-
- public void setName(String name) {
- this.name = name;
- }
-
- public String getMaxSize() {
- return maxSize;
- }
-
- public void setMaxSize(String maxSize) {
- this.maxSize = maxSize;
- }
-
- public Integer getNumThreads() {
- return numThreads;
- }
-
- public void setNumThreads(Integer numThreads) {
- this.numThreads = numThreads;
- }
-
- public String getQueue() {
- return group;
- }
-
- public void setGroup(String group) {
- this.group = group;
- }
-
- }
-
- private static class QueueConfig {
+ private static class GroupConfig {
String name;
String maxSize;
}
- private static class Executor {
- final CompactionExecutorId ceid;
+ private static class CompactionGroup {
+ final CompactorGroupId cgid;
final Long maxSize;
- public Executor(CompactionExecutorId ceid, Long maxSize) {
+ public CompactionGroup(CompactorGroupId cgid, Long maxSize) {
Preconditions.checkArgument(maxSize == null || maxSize > 0, "Invalid value for maxSize");
- this.ceid = Objects.requireNonNull(ceid, "Compaction ID is null");
+ this.cgid = Objects.requireNonNull(cgid, "Compaction ID is null");
this.maxSize = maxSize;
}
@@ -233,7 +147,7 @@ public class DefaultCompactionPlanner implements CompactionPlanner {
@Override
public String toString() {
- return "[ceid=" + ceid + ", maxSize=" + maxSize + "]";
+ return "[cgid=" + cgid + ", maxSize=" + maxSize + "]";
}
}
@@ -253,112 +167,68 @@ public class DefaultCompactionPlanner implements CompactionPlanner {
}
}
- private List<Executor> executors;
+ private List<CompactionGroup> groups;
private int maxFilesToCompact;
@SuppressFBWarnings(value = {"UWF_UNWRITTEN_FIELD", "NP_UNWRITTEN_FIELD"},
justification = "Field is written by Gson")
@Override
public void init(InitParameters params) {
- List<Executor> tmpExec = new ArrayList<>();
+ List<CompactionGroup> tmpGroups = new ArrayList<>();
String values;
- if (params.getOptions().containsKey("executors")
- && !params.getOptions().get("executors").isBlank()) {
- values = params.getOptions().get("executors");
-
- // Generate a list of fields from the desired object.
- final List<String> execFields = Arrays.stream(ExecutorConfig.class.getDeclaredFields())
- .map(Field::getName).collect(Collectors.toList());
-
- for (JsonElement element : GSON.get().fromJson(values, JsonArray.class)) {
- validateConfig(element, execFields, ExecutorConfig.class.getName());
- ExecutorConfig executorConfig = GSON.get().fromJson(element, ExecutorConfig.class);
-
- Long maxSize = executorConfig.maxSize == null ? null
- : ConfigurationTypeHelper.getFixedMemoryAsBytes(executorConfig.maxSize);
- CompactionExecutorId ceid;
-
- // If not supplied, GSON will leave type null. Default to internal
- if (executorConfig.type == null) {
- executorConfig.type = "internal";
- }
-
- switch (executorConfig.type) {
- case "internal":
- Preconditions.checkArgument(null == executorConfig.group,
- "'group' should not be specified for internal compactions");
- int numThreads = Objects.requireNonNull(executorConfig.numThreads,
- "'numThreads' must be specified for internal type");
- ceid = params.getExecutorManager().createExecutor(executorConfig.name, numThreads);
- break;
- case "external":
- Preconditions.checkArgument(null == executorConfig.numThreads,
- "'numThreads' should not be specified for external compactions");
- String group = Objects.requireNonNull(executorConfig.group,
- "'group' must be specified for external type");
- ceid = params.getExecutorManager().getExternalExecutor(group);
- break;
- default:
- throw new IllegalArgumentException("type must be 'internal' or 'external'");
- }
- tmpExec.add(new Executor(ceid, maxSize));
- }
- }
-
- if (params.getOptions().containsKey("queues") && !params.getOptions().get("queues").isBlank()) {
- values = params.getOptions().get("queues");
+ if (params.getOptions().containsKey("groups") && !params.getOptions().get("groups").isBlank()) {
+ values = params.getOptions().get("groups");
// Generate a list of fields from the desired object.
- final List<String> queueFields = Arrays.stream(QueueConfig.class.getDeclaredFields())
+ final List<String> groupFields = Arrays.stream(GroupConfig.class.getDeclaredFields())
.map(Field::getName).collect(Collectors.toList());
for (JsonElement element : GSON.get().fromJson(values, JsonArray.class)) {
- validateConfig(element, queueFields, QueueConfig.class.getName());
- QueueConfig queueConfig = GSON.get().fromJson(element, QueueConfig.class);
+ validateConfig(element, groupFields, GroupConfig.class.getName());
+ GroupConfig groupConfig = GSON.get().fromJson(element, GroupConfig.class);
- Long maxSize = queueConfig.maxSize == null ? null
- : ConfigurationTypeHelper.getFixedMemoryAsBytes(queueConfig.maxSize);
+ Long maxSize = groupConfig.maxSize == null ? null
+ : ConfigurationTypeHelper.getFixedMemoryAsBytes(groupConfig.maxSize);
- CompactionExecutorId ceid;
- String queue = Objects.requireNonNull(queueConfig.name, "'name' must be specified");
- ceid = params.getExecutorManager().getExternalExecutor(queue);
- tmpExec.add(new Executor(ceid, maxSize));
+ CompactorGroupId cgid;
+ String group = Objects.requireNonNull(groupConfig.name, "'name' must be specified");
+ cgid = params.getGroupManager().getGroup(group);
+ tmpGroups.add(new CompactionGroup(cgid, maxSize));
}
}
- if (tmpExec.size() < 1) {
- throw new IllegalStateException("No defined executors or queues for this planner");
+ if (tmpGroups.size() < 1) {
+ throw new IllegalStateException("No defined compactor groups for this planner");
}
- tmpExec.sort(Comparator.comparing(Executor::getMaxSize,
+ tmpGroups.sort(Comparator.comparing(CompactionGroup::getMaxSize,
Comparator.nullsLast(Comparator.naturalOrder())));
- executors = List.copyOf(tmpExec);
+ groups = List.copyOf(tmpGroups);
- if (executors.stream().filter(e -> e.getMaxSize() == null).count() > 1) {
+ if (groups.stream().filter(g -> g.getMaxSize() == null).count() > 1) {
throw new IllegalArgumentException(
- "Can only have one executor w/o a maxSize. " + params.getOptions().get("executors"));
+ "Can only have one group w/o a maxSize. " + params.getOptions().get("groups"));
}
// use the add method on the Set interface to check for duplicate maxSizes
Set<Long> maxSizes = new HashSet<>();
- executors.forEach(e -> {
- if (!maxSizes.add(e.getMaxSize())) {
+ groups.forEach(g -> {
+ if (!maxSizes.add(g.getMaxSize())) {
throw new IllegalArgumentException(
- "Duplicate maxSize set in executors. " + params.getOptions().get("executors"));
+ "Duplicate maxSize set in groups. " + params.getOptions().get("groups"));
}
});
determineMaxFilesToCompact(params);
}
- @SuppressWarnings("deprecation")
private void determineMaxFilesToCompact(InitParameters params) {
String maxOpen = params.getOptions().get("maxOpen");
if (maxOpen == null) {
- maxOpen = Property.TSERV_COMPACTION_SERVICE_DEFAULT_MAX_OPEN.getDefaultValue();
+ maxOpen = Property.COMPACTION_SERVICE_DEFAULT_MAX_OPEN.getDefaultValue();
log.trace("default maxOpen not set, defaulting to {}", maxOpen);
}
this.maxFilesToCompact = Integer.parseInt(maxOpen);
@@ -391,7 +261,8 @@ public class DefaultCompactionPlanner implements CompactionPlanner {
long maxSizeToCompact = getMaxSizeToCompact(params.getKind());
// This set represents future files that will be produced by running compactions. If the optimal
- // set of files to compact is computed and contains one of these files, then its optimal to wait
+ // set of files to compact is computed and contains one of these files, then it's optimal to
+ // wait
// for this compaction to finish.
Set<CompactableFile> expectedFiles = new HashSet<>();
params.getRunningCompactions().stream().filter(job -> job.getKind() == params.getKind())
@@ -482,8 +353,8 @@ public class DefaultCompactionPlanner implements CompactionPlanner {
}
var builder = params.createPlanBuilder();
- compactionJobs.forEach(jobFiles -> builder.addJob(createPriority(params, jobFiles),
- getExecutor(jobFiles), jobFiles));
+ compactionJobs.forEach(
+ jobFiles -> builder.addJob(createPriority(params, jobFiles), getGroup(jobFiles), jobFiles));
return builder.build();
}
@@ -567,7 +438,7 @@ public class DefaultCompactionPlanner implements CompactionPlanner {
private long getMaxSizeToCompact(CompactionKind kind) {
if (kind == CompactionKind.SYSTEM) {
- Long max = executors.get(executors.size() - 1).maxSize;
+ Long max = groups.get(groups.size() - 1).maxSize;
if (max != null) {
return max;
}
@@ -732,17 +603,17 @@ public class DefaultCompactionPlanner implements CompactionPlanner {
return sortedFiles.subList(0, larsmaIndex + 1);
}
- CompactionExecutorId getExecutor(Collection<CompactableFile> files) {
+ CompactorGroupId getGroup(Collection<CompactableFile> files) {
long size = files.stream().mapToLong(CompactableFile::getEstimatedSize).sum();
- for (Executor executor : executors) {
- if (executor.maxSize == null || size < executor.maxSize) {
- return executor.ceid;
+ for (CompactionGroup group : groups) {
+ if (group.maxSize == null || size < group.maxSize) {
+ return group.cgid;
}
}
- return executors.get(executors.size() - 1).ceid;
+ return groups.get(groups.size() - 1).cgid;
}
private static List<CompactableFile> sortByFileSize(Collection<CompactableFile> files) {
diff --git a/core/src/main/java/org/apache/accumulo/core/spi/compaction/ExecutorManager.java b/core/src/main/java/org/apache/accumulo/core/spi/compaction/GroupManager.java
similarity index 71%
rename from core/src/main/java/org/apache/accumulo/core/spi/compaction/ExecutorManager.java
rename to core/src/main/java/org/apache/accumulo/core/spi/compaction/GroupManager.java
index f362a50a59..9c2685b861 100644
--- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/ExecutorManager.java
+++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/GroupManager.java
@@ -19,21 +19,15 @@
package org.apache.accumulo.core.spi.compaction;
/**
- * Offered to a Compaction Planner at initialization time so it can create executors.
+ * Offered to a Compaction Planner at initialization time, so it can create compactor groups.
*
- *
- * @since 2.1.0
+ * @since 3.1.0
* @see CompactionPlanner#init(org.apache.accumulo.core.spi.compaction.CompactionPlanner.InitParameters)
* @see org.apache.accumulo.core.spi.compaction
*/
-public interface ExecutorManager {
- /**
- * Create a thread pool executor within a compaction service.
- */
- public CompactionExecutorId createExecutor(String name, int threads);
-
+public interface GroupManager {
/**
- * @return an id for a configured external execution queue.
+ * @return an id for a configured compactor group.
*/
- public CompactionExecutorId getExternalExecutor(String name);
+ CompactorGroupId getGroup(String name);
}
diff --git a/core/src/main/java/org/apache/accumulo/core/spi/compaction/package-info.java b/core/src/main/java/org/apache/accumulo/core/spi/compaction/package-info.java
index e1bea0f602..cb22c0dfa4 100644
--- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/package-info.java
+++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/package-info.java
@@ -37,13 +37,13 @@
* <ul>
* <li><b>Compaction Service</b> A non pluggable component that compacts tablets. One or more of
* these are created based on user configuration. Users can assign a table to a compaction service.
- * Has a single compaction planner and one ore more compaction executors.
+ * Has a single compaction planner and one or more compactor groups.
* <ul>
- * <li><b>Compaction Executor</b> A non pluggable component that executes compactions using multiple
- * threads and has a priority queue.
+ * <li><b>Compactor Group</b> A non pluggable component that executes compactions using multiple
+ * compactor instances and has a priority queue.
* <li><b>Compaction Planner</b> A pluggable component that can be configured by users when they
* configure a compaction service. It makes decisions about which files to compact on which
- * executors. See {@link org.apache.accumulo.core.spi.compaction.CompactionPlanner},
+ * compactor groups. See {@link org.apache.accumulo.core.spi.compaction.CompactionPlanner},
* {@link org.apache.accumulo.core.spi.compaction.CompactionPlanner#makePlan(org.apache.accumulo.core.spi.compaction.CompactionPlanner.PlanningParameters)},
* and {@link org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner}
* </ul>
diff --git a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionJobImpl.java b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionJobImpl.java
index 1f7354e333..95d1491700 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionJobImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionJobImpl.java
@@ -24,9 +24,9 @@ import java.util.Optional;
import java.util.Set;
import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionJob;
import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
/**
* An immutable object that describes what files to compact and where to compact them.
@@ -37,10 +37,10 @@ import org.apache.accumulo.core.spi.compaction.CompactionKind;
public class CompactionJobImpl implements CompactionJob {
private final short priority;
- private final CompactionExecutorId executor;
+ private final CompactorGroupId group;
private final Set<CompactableFile> files;
private final CompactionKind kind;
- // Tracks if a job selected all of the tablets files that existed at the time the job was created.
+ // Tracks if a job selected all the tablet's files that existed at the time the job was created.
private final Optional<Boolean> jobSelectedAll;
/**
@@ -49,10 +49,10 @@ public class CompactionJobImpl implements CompactionJob {
* to start compaction. After a job is running, its not used. So when a job object is
* recreated for a running external compaction this parameter can be empty.
*/
- public CompactionJobImpl(short priority, CompactionExecutorId executor,
+ public CompactionJobImpl(short priority, CompactorGroupId group,
Collection<CompactableFile> files, CompactionKind kind, Optional<Boolean> jobSelectedAll) {
this.priority = priority;
- this.executor = Objects.requireNonNull(executor);
+ this.group = Objects.requireNonNull(group);
this.files = Set.copyOf(files);
this.kind = Objects.requireNonNull(kind);
this.jobSelectedAll = Objects.requireNonNull(jobSelectedAll);
@@ -64,11 +64,11 @@ public class CompactionJobImpl implements CompactionJob {
}
/**
- * @return The executor to run the job.
+ * @return The group to run the job.
*/
@Override
- public CompactionExecutorId getExecutor() {
- return executor;
+ public CompactorGroupId getGroup() {
+ return group;
}
/**
@@ -89,7 +89,7 @@ public class CompactionJobImpl implements CompactionJob {
@Override
public int hashCode() {
- return Objects.hash(priority, executor, files, kind);
+ return Objects.hash(priority, group, files, kind);
}
public boolean selectedAll() {
@@ -101,7 +101,7 @@ public class CompactionJobImpl implements CompactionJob {
if (o instanceof CompactionJobImpl) {
CompactionJobImpl ocj = (CompactionJobImpl) o;
- return priority == ocj.priority && executor.equals(ocj.executor) && files.equals(ocj.files)
+ return priority == ocj.priority && group.equals(ocj.group) && files.equals(ocj.files)
&& kind == ocj.kind;
}
@@ -110,7 +110,7 @@ public class CompactionJobImpl implements CompactionJob {
@Override
public String toString() {
- return "CompactionJob [priority=" + priority + ", executor=" + executor + ", files=" + files
+ return "CompactionJob [priority=" + priority + ", group=" + group + ", files=" + files
+ ", kind=" + kind + "]";
}
diff --git a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionPlanImpl.java b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionPlanImpl.java
index 18f14b08ee..2eff876b7d 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionPlanImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionPlanImpl.java
@@ -27,10 +27,10 @@ import java.util.Optional;
import java.util.Set;
import org.apache.accumulo.core.client.admin.compaction.CompactableFile;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionJob;
import org.apache.accumulo.core.spi.compaction.CompactionKind;
import org.apache.accumulo.core.spi.compaction.CompactionPlan;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
import com.google.common.base.Preconditions;
@@ -68,7 +68,7 @@ public class CompactionPlanImpl implements CompactionPlan {
}
@Override
- public Builder addJob(short priority, CompactionExecutorId executor,
+ public Builder addJob(short priority, CompactorGroupId group,
Collection<CompactableFile> files) {
Set<CompactableFile> filesSet =
files instanceof Set ? (Set<CompactableFile>) files : Set.copyOf(files);
@@ -80,7 +80,7 @@ public class CompactionPlanImpl implements CompactionPlan {
seenFiles.addAll(filesSet);
- jobs.add(new CompactionJobImpl(priority, executor, filesSet, kind,
+ jobs.add(new CompactionJobImpl(priority, group, filesSet, kind,
Optional.of(filesSet.equals(allFiles))));
return this;
}
diff --git 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
index bb83243420..6966815439 100644
--- 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
@@ -18,23 +18,21 @@
*/
package org.apache.accumulo.core.util.compaction;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.apache.accumulo.core.spi.common.ServiceEnvironment;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionPlanner;
import org.apache.accumulo.core.spi.compaction.CompactionServiceId;
-import org.apache.accumulo.core.spi.compaction.ExecutorManager;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
+import org.apache.accumulo.core.spi.compaction.GroupManager;
import com.google.common.base.Preconditions;
public class CompactionPlannerInitParams implements CompactionPlanner.InitParameters {
private final Map<String,String> plannerOpts;
- private final Map<CompactionExecutorId,Integer> requestedExecutors;
- private final Set<CompactionExecutorId> requestedExternalExecutors;
+ private final Set<CompactorGroupId> requestedGroups;
private final ServiceEnvironment senv;
private final CompactionServiceId serviceId;
private final String prefix;
@@ -43,8 +41,7 @@ public class CompactionPlannerInitParams implements CompactionPlanner.InitParame
Map<String,String> plannerOpts, ServiceEnvironment senv) {
this.serviceId = serviceId;
this.plannerOpts = plannerOpts;
- this.requestedExecutors = new HashMap<>();
- this.requestedExternalExecutors = new HashSet<>();
+ this.requestedGroups = new HashSet<>();
this.senv = senv;
this.prefix = prefix;
}
@@ -65,35 +62,21 @@ public class CompactionPlannerInitParams implements CompactionPlanner.InitParame
}
@Override
- public ExecutorManager getExecutorManager() {
- return new ExecutorManager() {
- @Override
- public CompactionExecutorId createExecutor(String executorName, int threads) {
- Preconditions.checkArgument(threads > 0, "Positive number of threads required : %s",
- threads);
- var ceid = CompactionExecutorIdImpl.internalId(serviceId, executorName);
- Preconditions.checkState(!getRequestedExecutors().containsKey(ceid),
- "Duplicate Compaction Executor ID found");
- getRequestedExecutors().put(ceid, threads);
- return ceid;
- }
+ public GroupManager getGroupManager() {
+ return new GroupManager() {
@Override
- public CompactionExecutorId getExternalExecutor(String name) {
- var ceid = CompactionExecutorIdImpl.externalId(name);
- Preconditions.checkArgument(!getRequestedExternalExecutors().contains(ceid),
- "Duplicate external executor for group " + name);
- getRequestedExternalExecutors().add(ceid);
- return ceid;
+ public CompactorGroupId getGroup(String name) {
+ var cgid = CompactorGroupIdImpl.groupId(name);
+ Preconditions.checkArgument(!getRequestedGroups().contains(cgid),
+ "Duplicate compactor group for group: " + name);
+ getRequestedGroups().add(cgid);
+ return cgid;
}
};
}
- public Map<CompactionExecutorId,Integer> getRequestedExecutors() {
- return requestedExecutors;
- }
-
- public Set<CompactionExecutorId> getRequestedExternalExecutors() {
- return requestedExternalExecutors;
+ public Set<CompactorGroupId> getRequestedGroups() {
+ return requestedGroups;
}
}
diff --git a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionServicesConfig.java b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionServicesConfig.java
index 7b3df2e279..3fa3ca8682 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionServicesConfig.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionServicesConfig.java
@@ -22,34 +22,26 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.function.Predicate;
-import java.util.stream.Collectors;
import org.apache.accumulo.core.client.PluginEnvironment;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.conf.ConfigurationTypeHelper;
import org.apache.accumulo.core.conf.Property;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.google.common.collect.Sets;
/**
* This class serves to configure compaction services from an {@link AccumuloConfiguration} object.
- *
- * Specifically, compaction service properties (those prefixed by "tserver.compaction.major
- * .service" or "compaction.service") are used.
+ * Specifically, compaction service properties (those prefixed by "compaction.service") are used.
*/
public class CompactionServicesConfig {
- private static final Logger log = LoggerFactory.getLogger(CompactionServicesConfig.class);
private final Map<String,String> planners = new HashMap<>();
private final Map<String,String> plannerPrefixes = new HashMap<>();
private final Map<String,Long> rateLimits = new HashMap<>();
private final Map<String,Map<String,String>> options = new HashMap<>();
- @SuppressWarnings("deprecation")
- private static final Property oldPrefix = Property.TSERV_COMPACTION_SERVICE_PREFIX;
- private static final Property newPrefix = Property.COMPACTION_SERVICE_PREFIX;
+ private static final Property prefix = Property.COMPACTION_SERVICE_PREFIX;
private interface ConfigIndirection {
Map<String,String> getAllPropertiesWithPrefixStripped(Property p);
@@ -58,27 +50,8 @@ public class CompactionServicesConfig {
private static Map<String,Map<String,String>> getConfiguration(ConfigIndirection aconf) {
Map<String,Map<String,String>> properties = new HashMap<>();
- var newProps = aconf.getAllPropertiesWithPrefixStripped(newPrefix);
- properties.put(newPrefix.getKey(), newProps);
-
- // get all the services under the new prefix
- var newServices =
- newProps.keySet().stream().map(prop -> prop.split("\\.")[0]).collect(Collectors.toSet());
-
- Map<String,String> oldServices = new HashMap<>();
-
- for (Map.Entry<String,String> entry : aconf.getAllPropertiesWithPrefixStripped(oldPrefix)
- .entrySet()) {
- // Discard duplicate service definitions
- var service = entry.getKey().split("\\.")[0];
- if (newServices.contains(service)) {
- log.warn("Duplicate compaction service '{}' definition exists. Ignoring property : '{}'",
- service, entry.getKey());
- } else {
- oldServices.put(entry.getKey(), entry.getValue());
- }
- }
- properties.put(oldPrefix.getKey(), oldServices);
+ var newProps = aconf.getAllPropertiesWithPrefixStripped(prefix);
+ properties.put(prefix.getKey(), newProps);
// Return unmodifiable map
return Map.copyOf(properties);
}
@@ -97,54 +70,21 @@ public class CompactionServicesConfig {
this(getConfiguration(aconf::getAllPropertiesWithPrefixStripped), aconf::isPropertySet);
}
- @SuppressWarnings("removal")
private CompactionServicesConfig(Map<String,Map<String,String>> configs,
Predicate<Property> isSetPredicate) {
configs.forEach((prefix, props) -> {
props.forEach((prop, val) -> {
String[] tokens = prop.split("\\.");
if (tokens.length == 2 && tokens[1].equals("planner")) {
- if (prefix.equals(oldPrefix.getKey())) {
- // Log a warning if the old prefix planner is defined by a user.
- Property userDefined = null;
- try {
- userDefined = Property.valueOf(prefix + prop);
- } catch (IllegalArgumentException e) {
- log.trace("Property: {} is not set by default configuration", prefix + prop);
- }
- boolean isPropSet = true;
- if (userDefined != null) {
- isPropSet = isSetPredicate.test(userDefined);
- }
- if (isPropSet) {
- log.warn(
- "Found compaction planner '{}' using a deprecated prefix. Please update property to use the '{}' prefix",
- tokens[0], newPrefix);
- }
- }
plannerPrefixes.put(tokens[0], prefix);
planners.put(tokens[0], val);
- }
- });
- });
-
- // Now find all compaction planner options.
- configs.forEach((prefix, props) -> {
- props.forEach((prop, val) -> {
- String[] tokens = prop.split("\\.");
- if (!plannerPrefixes.containsKey(tokens[0])) {
- throw new IllegalArgumentException(
- "Incomplete compaction service definition, missing planner class: " + prop);
- }
- if (tokens.length == 4 && tokens[1].equals("planner") && tokens[2].equals("opts")) {
+ } else if (tokens.length == 4 && tokens[1].equals("planner") && tokens[2].equals("opts")) {
options.computeIfAbsent(tokens[0], k -> new HashMap<>()).put(tokens[3], val);
} else if (tokens.length == 3 && tokens[1].equals("rate") && tokens[2].equals("limit")) {
var eprop = Property.getPropertyByKey(prop);
if (eprop == null || isSetPredicate.test(eprop)) {
rateLimits.put(tokens[0], ConfigurationTypeHelper.getFixedMemoryAsBytes(val));
}
- } else if (tokens.length == 2 && tokens[1].equals("planner")) {
- return; // moves to next opt
} else {
throw new IllegalArgumentException(
"Malformed compaction service property " + prefix + prop);
diff --git a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionExecutorIdImpl.java b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactorGroupIdImpl.java
similarity index 53%
rename from core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionExecutorIdImpl.java
rename to core/src/main/java/org/apache/accumulo/core/util/compaction/CompactorGroupIdImpl.java
index 0b0d8b6224..f3b15de965 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactionExecutorIdImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/compaction/CompactorGroupIdImpl.java
@@ -18,34 +18,17 @@
*/
package org.apache.accumulo.core.util.compaction;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
-import org.apache.accumulo.core.spi.compaction.CompactionServiceId;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
-import com.google.common.base.Preconditions;
+public class CompactorGroupIdImpl extends CompactorGroupId {
-public class CompactionExecutorIdImpl extends CompactionExecutorId {
-
- protected CompactionExecutorIdImpl(String canonical) {
+ protected CompactorGroupIdImpl(String canonical) {
super(canonical);
}
private static final long serialVersionUID = 1L;
- public boolean isExternalId() {
- return canonical().startsWith("e.");
- }
-
- public String getExternalName() {
- Preconditions.checkState(isExternalId());
- return canonical().substring("e.".length());
- }
-
- public static CompactionExecutorId internalId(CompactionServiceId csid, String executorName) {
- return new CompactionExecutorIdImpl("i." + csid + "." + executorName);
- }
-
- public static CompactionExecutorId externalId(String executorName) {
- return new CompactionExecutorIdImpl("e." + executorName);
+ public static CompactorGroupId groupId(String groupName) {
+ return new CompactorGroupIdImpl(groupName);
}
-
}
diff --git a/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java b/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
index a5f47e6a20..59f27a908e 100644
--- a/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/metadata/schema/TabletMetadataTest.java
@@ -74,7 +74,7 @@ import org.apache.accumulo.core.metadata.schema.TabletMetadata.Location;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.LocationType;
import org.apache.accumulo.core.spi.compaction.CompactionKind;
import org.apache.accumulo.core.tabletserver.log.LogEntry;
-import org.apache.accumulo.core.util.compaction.CompactionExecutorIdImpl;
+import org.apache.accumulo.core.util.compaction.CompactorGroupIdImpl;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.junit.jupiter.api.Test;
@@ -383,8 +383,8 @@ public class TabletMetadataTest {
assertThrows(IllegalStateException.class, tm2::getCompacted);
var ecid1 = ExternalCompactionId.generate(UUID.randomUUID());
- ExternalCompactionMetadata ecm = new ExternalCompactionMetadata(Set.of(sf1, sf2), rf1, "cid1",
- CompactionKind.USER, (short) 3, CompactionExecutorIdImpl.externalId("Q1"), true, 99L);
+ CompactionMetadata ecm = new CompactionMetadata(Set.of(sf1, sf2), rf1, "cid1",
+ CompactionKind.USER, (short) 3, CompactorGroupIdImpl.groupId("Q1"), true, 99L);
LogEntry le1 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
LogEntry le2 = LogEntry.fromPath("localhost+8020/" + UUID.randomUUID());
diff --git 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
index ecc95d6e72..265778f1af 100644
--- 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
@@ -50,10 +50,10 @@ 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.CompactionJobImpl;
import org.apache.accumulo.core.util.compaction.CompactionPlanImpl;
import org.apache.accumulo.core.util.compaction.CompactionPlannerInitParams;
+import org.apache.accumulo.core.util.compaction.CompactorGroupIdImpl;
import org.easymock.EasyMock;
import org.junit.jupiter.api.Test;
@@ -72,7 +72,7 @@ public class DefaultCompactionPlannerTest {
private static final String prefix = Property.COMPACTION_SERVICE_PREFIX.getKey();
@Test
- public void testFindFilesToCompact() throws Exception {
+ public void testFindFilesToCompact() {
testFFtC(createCFs("F4", "1M", "F5", "1M", "F6", "1M"),
createCFs("F1", "100M", "F2", "100M", "F3", "100M", "F4", "1M", "F5", "1M", "F6", "1M"),
@@ -150,13 +150,11 @@ public class DefaultCompactionPlannerTest {
}
@Test
- public void testRunningCompaction() throws Exception {
- 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}]";
+ public void testRunningCompaction() {
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large','maxSize':'512M'}, {'name':'huge'}]";
- var planner = createPlanner(defaultConf, executors);
+ var planner = createPlanner(defaultConf, groups);
var all = createCFs("F1", "3M", "F2", "3M", "F3", "11M", "F4", "12M", "F5", "13M");
var candidates = createCFs("F3", "11M", "F4", "12M", "F5", "13M");
@@ -179,21 +177,19 @@ public class DefaultCompactionPlannerTest {
// planner should compact.
var job = getOnlyElement(plan.getJobs());
assertEquals(candidates, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "medium"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("medium"), job.getGroup());
}
@Test
- public void testUserCompaction() throws Exception {
+ public void testUserCompaction() {
ConfigurationCopy aconf = new ConfigurationCopy(DefaultConfiguration.getInstance());
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}]";
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large','maxSize':'512M'}, {'name':'huge'}]";
- var planner = createPlanner(config, executors);
+ var planner = createPlanner(config, groups);
var all = createCFs("F1", "3M", "F2", "3M", "F3", "11M", "F4", "12M", "F5", "13M");
var candidates = createCFs("F3", "11M", "F4", "12M", "F5", "13M");
var compacting =
@@ -204,7 +200,7 @@ public class DefaultCompactionPlannerTest {
// a running non-user compaction should not prevent a user compaction
var job = getOnlyElement(plan.getJobs());
assertEquals(candidates, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "medium"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("medium"), job.getGroup());
// should only run one user compaction at a time
compacting = Set.of(createJob(CompactionKind.USER, all, createCFs("F1", "3M", "F2", "3M")));
@@ -222,7 +218,7 @@ public class DefaultCompactionPlannerTest {
plan = planner.makePlan(params);
job = getOnlyElement(plan.getJobs());
assertEquals(createCFs("F1", "1M", "F2", "2M", "F3", "4M"), job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "small"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("small"), job.getGroup());
// should compact all 15
all = createCFs("FI", "7M", "F4", "8M", "F5", "16M", "F6", "32M", "F7", "64M", "F8", "128M",
@@ -232,17 +228,17 @@ public class DefaultCompactionPlannerTest {
plan = planner.makePlan(params);
job = getOnlyElement(plan.getJobs());
assertEquals(all, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "huge"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("huge"), job.getGroup());
// For user compaction, can compact a subset that meets the compaction ratio if there is also a
- // larger set of files the meets the compaction ratio
+ // larger set of files that meets the compaction ratio
all = createCFs("F1", "3M", "F2", "4M", "F3", "5M", "F4", "6M", "F5", "50M", "F6", "51M", "F7",
"52M");
params = createPlanningParams(all, all, compacting, 2, CompactionKind.USER);
plan = planner.makePlan(params);
job = getOnlyElement(plan.getJobs());
assertEquals(createCFs("F1", "3M", "F2", "4M", "F3", "5M", "F4", "6M"), job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "small"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("small"), job.getGroup());
// There is a subset of small files that meets the compaction ratio, but the larger set does not
// so compact everything to avoid doing more than logarithmic work
@@ -251,17 +247,16 @@ public class DefaultCompactionPlannerTest {
plan = planner.makePlan(params);
job = getOnlyElement(plan.getJobs());
assertEquals(all, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "medium"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("medium"), job.getGroup());
}
@Test
- public void testMaxSize() throws Exception {
- 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}]";
+ public void testMaxSize() {
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large','maxSize':'512M'}]";
- var planner = createPlanner(defaultConf, executors);
+ var planner = createPlanner(defaultConf, groups);
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);
@@ -269,26 +264,25 @@ public class DefaultCompactionPlannerTest {
// should only compact files less than max size
var job = getOnlyElement(plan.getJobs());
assertEquals(createCFs("F1", "128M", "F2", "129M", "F3", "130M"), job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "large"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("large"), job.getGroup());
// user compaction can exceed the max size
params = createPlanningParams(all, all, Set.of(), 2, CompactionKind.USER);
plan = planner.makePlan(params);
job = getOnlyElement(plan.getJobs());
assertEquals(all, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "large"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("large"), job.getGroup());
}
@Test
public void testMultipleCompactions() {
// This test validates that when a tablet has many files that multiple compaction jobs can be
// issued at the same time.
- 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}]";
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large','maxSize':'512M'}]";
for (var kind : List.of(CompactionKind.USER, CompactionKind.SYSTEM)) {
- var planner = createPlanner(defaultConf, executors);
+ var planner = createPlanner(defaultConf, groups);
var all = IntStream.range(0, 990).mapToObj(i -> createCF("F" + i, 1000)).collect(toSet());
// simulate 10 larger files, these should not compact at the same time as the smaller files.
// Its more optimal to wait for all of the smaller files to compact and them compact the
@@ -303,7 +297,7 @@ public class DefaultCompactionPlannerTest {
plan.getJobs().forEach(job -> {
assertEquals(15, job.getFiles().size());
assertEquals(kind, job.getKind());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "small"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("small"), job.getGroup());
// ensure the files across all of the jobs are disjoint
job.getFiles().forEach(cf -> assertTrue(filesSeen.add(cf)));
});
@@ -318,10 +312,9 @@ public class DefaultCompactionPlannerTest {
@Test
public void testMultipleCompactionsAndLargeCompactionRatio() {
- 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);
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large','maxSize':'512M'}]";
+ var planner = createPlanner(defaultConf, groups);
var all = IntStream.range(0, 65).mapToObj(i -> createCF("F" + i, i + 1)).collect(toSet());
// This compaction ratio would not cause a system compaction, how a user compaction must compact
// all of the files so it should generate some compactions.
@@ -353,11 +346,10 @@ public class DefaultCompactionPlannerTest {
// This test validates that when a tablet has many files that multiple compaction jobs can be
// issued at the same time even if there are running compaction as long everything meets the
// compaction ratio.
- 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}]";
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large','maxSize':'512M'}]";
for (var kind : List.of(CompactionKind.USER, CompactionKind.SYSTEM)) {
- var planner = createPlanner(defaultConf, executors);
+ var planner = createPlanner(defaultConf, groups);
var all = IntStream.range(0, 990).mapToObj(i -> createCF("F" + i, 1000)).collect(toSet());
// simulate 10 larger files, these should not compact at the same time as the smaller files.
// Its more optimal to wait for all of the smaller files to compact and them compact the
@@ -380,7 +372,7 @@ public class DefaultCompactionPlannerTest {
plan.getJobs().forEach(job -> {
assertEquals(15, job.getFiles().size());
assertEquals(kind, job.getKind());
- assertEquals(CompactionExecutorIdImpl.internalId(csid, "small"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("small"), job.getGroup());
// ensure the files across all of the jobs are disjoint
job.getFiles().forEach(cf -> assertTrue(filesSeen.add(cf)));
});
@@ -394,12 +386,11 @@ public class DefaultCompactionPlannerTest {
@Test
public void testUserCompactionDoesNotWaitOnSystemCompaction() {
- // this test ensure user compactions do not wait on system compactions to complete
+ // this test ensures user compactions do not wait on system compactions to complete
- 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);
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large','maxSize':'512M'}]";
+ var planner = createPlanner(defaultConf, groups);
var all = createCFs("F1", "1M", "F2", "1M", "F3", "1M", "F4", "3M", "F5", "3M", "F6", "3M",
"F7", "20M");
var candidates = createCFs("F4", "3M", "F5", "3M", "F6", "3M", "F7", "20M");
@@ -438,11 +429,11 @@ public class DefaultCompactionPlannerTest {
}
@Test
- public void testQueueCreation() throws Exception {
+ public void testQueueCreation() {
DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- String queues = "[{\"name\": \"small\", \"maxSize\":\"32M\"},{\"name\":\"midsize\"}]";
- planner.init(getInitParamQueues(defaultConf, queues));
+ String groups = "[{\"name\": \"small\", \"maxSize\":\"32M\"},{\"name\":\"midsize\"}]";
+ planner.init(getInitParams(defaultConf, groups));
var all = createCFs("F1", "1M", "F2", "1M", "F3", "1M", "F4", "1M");
var params = createPlanningParams(all, all, Set.of(), 2, CompactionKind.SYSTEM);
@@ -450,7 +441,7 @@ public class DefaultCompactionPlannerTest {
var job = getOnlyElement(plan.getJobs());
assertEquals(all, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.externalId("small"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("small"), job.getGroup());
all = createCFs("F1", "100M", "F2", "100M", "F3", "100M", "F4", "100M");
params = createPlanningParams(all, all, Set.of(), 2, CompactionKind.SYSTEM);
@@ -458,7 +449,7 @@ public class DefaultCompactionPlannerTest {
job = getOnlyElement(plan.getJobs());
assertEquals(all, job.getFiles());
- assertEquals(CompactionExecutorIdImpl.externalId("midsize"), job.getExecutor());
+ assertEquals(CompactorGroupIdImpl.groupId("midsize"), job.getGroup());
}
/**
@@ -466,86 +457,29 @@ public class DefaultCompactionPlannerTest {
*/
@Test
public void testErrorAdditionalConfigFields() {
- DefaultCompactionPlanner QueuePlanner = new DefaultCompactionPlanner();
+ DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- String queues =
+ String groups =
"[{\"name\":\"smallQueue\", \"maxSize\":\"32M\"}, {\"name\":\"largeQueue\", \"type\":\"internal\", \"foo\":\"bar\", \"queue\":\"broken\"}]";
- final InitParameters queueParams = getInitParamQueues(defaultConf, queues);
- assertNotNull(queueParams);
- var e = assertThrows(JsonParseException.class, () -> QueuePlanner.init(queueParams),
- "Failed to throw error");
+ final InitParameters params = getInitParams(defaultConf, groups);
+ assertNotNull(params);
+ var e =
+ assertThrows(JsonParseException.class, () -> planner.init(params), "Failed to throw error");
assertTrue(e.getMessage().contains("[type, foo, queue]"),
"Error message didn't contain '[type, foo, queue]'");
- 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(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
- */
- @Test
- public void testErrorInternalTypeNoNumThreads() {
- DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- String executors = "[{'name':'small','type':'internal','maxSize':'32M'},"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':2},"
- + "{'name':'large','type':'internal','maxSize':'512M','numThreads':3}]";
-
- var e = assertThrows(NullPointerException.class,
- () -> planner.init(getInitParams(defaultConf, executors)), "Failed to throw error");
- assertTrue(e.getMessage().contains("numThreads"), "Error message didn't contain numThreads");
- }
-
- /**
- * Test external type executor with numThreads set throws error.
- */
- @Test
- public void testErrorExternalTypeNumThreads() {
- DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- String executors = "[{'name':'small','type':'internal','maxSize':'32M', 'numThreads':1},"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':2},"
- + "{'name':'large','type':'external','maxSize':'512M','numThreads':3}]";
-
- var e = assertThrows(IllegalArgumentException.class,
- () -> planner.init(getInitParams(defaultConf, executors)), "Failed to throw error");
- assertTrue(e.getMessage().contains("numThreads"), "Error message didn't contain numThreads");
- }
-
- /**
- * Tests external type executor missing group throws error
- */
- @Test
- public void testErrorExternalNoQueue() {
- DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- String executors = "[{'name':'small','type':'internal','maxSize':'32M', 'numThreads':1},"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':2},"
- + "{'name':'large','type':'external','maxSize':'512M'}]";
-
- var e = assertThrows(NullPointerException.class,
- () -> planner.init(getInitParams(defaultConf, executors)), "Failed to throw error");
- assertTrue(e.getMessage().contains("group"), "Error message didn't contain group");
}
/**
- * Tests queue with missing name throws error
+ * Tests group with missing name throws error
*/
@Test
- public void testErrorQueueNoName() {
+ public void testErrorGroupNoName() {
DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- String queues = "[{\"name\":\"smallQueue\", \"maxSize\":\"32M\"}, {\"maxSize\":\"120M\"}]";
+ String groups = "[{\"name\":\"smallQueue\", \"maxSize\":\"32M\"}, {\"maxSize\":\"120M\"}]";
- final InitParameters params = getInitParamQueues(defaultConf, queues);
+ final InitParameters params = getInitParams(defaultConf, groups);
assertNotNull(params);
var e = assertThrows(NullPointerException.class, () -> planner.init(params),
@@ -554,65 +488,55 @@ public class DefaultCompactionPlannerTest {
}
/**
- * Tests not having executors or queues throws errors
+ * Tests not having groups throws errors
*/
@Test
- public void testErrorNoExecutors() {
+ public void testErrorNoGroups() {
DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- 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(defaultConf, "");
- assertNotNull(params);
+ var groupParams = getInitParams(defaultConf, "");
+ assertNotNull(groupParams);
- var e2 = assertThrows(IllegalStateException.class, () -> planner.init(params),
+ var e = assertThrows(IllegalStateException.class, () -> planner.init(groupParams),
"Failed to throw error");
- assertEquals("No defined executors or queues for this planner", e2.getMessage(),
+ assertEquals("No defined compactor groups for this planner", e.getMessage(),
"Error message was not equal");
}
/**
- * Tests executors can only have one without a max size.
+ * Tests groups can only have one group without a max size set.
*/
@Test
public void testErrorOnlyOneMaxSize() {
DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- String executors = "[{'name':'small','type':'internal','maxSize':'32M', 'numThreads':1},"
- + "{'name':'medium','type':'internal','numThreads':2},"
- + "{'name':'large','type':'internal','numThreads':'3'}]";
+ String groups =
+ "[{\"name\":\"small\", \"maxSize\":\"32M\"}, {\"name\":\"medium\"}, {\"name\":\"large\"}]";
var e = assertThrows(IllegalArgumentException.class,
- () -> planner.init(getInitParams(defaultConf, executors)), "Failed to throw error");
- assertTrue(e.getMessage().contains("maxSize"), "Error message didn't contain maxSize");
+ () -> planner.init(getInitParams(defaultConf, groups)), "Failed to throw error");
+ assertTrue(e.getMessage().contains("Can only have one group w/o a maxSize"),
+ "Error message didn't contain maxSize");
}
/**
- * Tests executors can only have one without a max size.
+ * Tests groups cannot have the same max size set.
*/
@Test
public void testErrorDuplicateMaxSize() {
DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- String executors = "[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
- + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2},"
- + "{'name':'large','type': 'internal','maxSize':'128M','numThreads':3}]";
-
+ String groups =
+ "[{\"name\":\"small\", \"maxSize\":\"32M\"}, {\"name\":\"medium\", \"maxSize\":\"32M\"}, {\"name\":\"large\"}]";
var e = assertThrows(IllegalArgumentException.class,
- () -> planner.init(getInitParams(defaultConf, executors)), "Failed to throw error");
- assertTrue(e.getMessage().contains("maxSize"), "Error message didn't contain maxSize");
+ () -> planner.init(getInitParams(defaultConf, groups)), "Failed to throw error");
+ assertTrue(e.getMessage().contains("Duplicate maxSize set in groups"),
+ "Error message didn't contain maxSize");
}
// Test cases where a tablet has more than table.file.max files, but no files were found using the
// compaction ratio. The planner should try to find the highest ratio that will result in a
// compaction.
@Test
- public void testMaxTabletFiles() throws Exception {
- String executors = "[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
- + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2},"
- + "{'name':'large','type': 'internal','numThreads':3}]";
+ public void testMaxTabletFiles() {
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large'}]";
Map<String,String> overrides = new HashMap<>();
overrides.put(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.maxOpen", "10");
@@ -620,7 +544,7 @@ public class DefaultCompactionPlannerTest {
var conf = new ConfigurationImpl(SiteConfiguration.empty().withOverrides(overrides).build());
// For this case need to compact three files and the highest ratio that achieves that is 1.8
- var planner = createPlanner(conf, executors);
+ var planner = createPlanner(conf, groups);
var all = createCFs(1000, 1.1, 1.9, 1.8, 1.6, 1.3, 1.4, 1.3, 1.2, 1.1);
var params = createPlanningParams(all, all, Set.of(), 3, CompactionKind.SYSTEM, conf);
var plan = planner.makePlan(params);
@@ -686,10 +610,9 @@ public class DefaultCompactionPlannerTest {
}
@Test
- public void testMaxTabletFilesNoCompaction() throws Exception {
- 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}]";
+ public void testMaxTabletFilesNoCompaction() {
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large', 'maxSize':'512M'}]";
Map<String,String> overrides = new HashMap<>();
overrides.put(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.maxOpen", "10");
@@ -697,7 +620,7 @@ public class DefaultCompactionPlannerTest {
var conf = new ConfigurationImpl(SiteConfiguration.empty().withOverrides(overrides).build());
// ensure that when a compaction would be over the max size limit that it is not planned
- var planner = createPlanner(conf, executors);
+ var planner = createPlanner(conf, groups);
var all = createCFs(1_000_000_000, 2, 2, 2, 2, 2, 2, 2);
var params = createPlanningParams(all, all, Set.of(), 3, CompactionKind.SYSTEM, conf);
var plan = planner.makePlan(params);
@@ -707,7 +630,7 @@ public class DefaultCompactionPlannerTest {
// ensure when a compaction is running and we are over files max but below the compaction ratio
// that a compaction is not planned
all = createCFs(1_000, 2, 2, 2, 2, 2, 2, 2);
- var job = new CompactionJobImpl((short) 1, CompactionExecutorIdImpl.externalId("ee1"),
+ var job = new CompactionJobImpl((short) 1, CompactorGroupIdImpl.groupId("ee1"),
createCFs("F1", "1000"), CompactionKind.SYSTEM, Optional.of(false));
params = createPlanningParams(all, all, Set.of(job), 3, CompactionKind.SYSTEM, conf);
plan = planner.makePlan(params);
@@ -724,10 +647,9 @@ public class DefaultCompactionPlannerTest {
// Test to ensure that plugin falls back from TABLE_FILE_MAX to TSERV_SCAN_MAX_OPENFILES
@Test
- public void testMaxTableFilesFallback() throws Exception {
- String executors = "[{'name':'small','type': 'internal','maxSize':'32M','numThreads':1},"
- + "{'name':'medium','type': 'internal','maxSize':'128M','numThreads':2},"
- + "{'name':'large','type': 'internal','numThreads':3}]";
+ public void testMaxTableFilesFallback() {
+ String groups = "[{'name':'small','maxSize':'32M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large'}]";
Map<String,String> overrides = new HashMap<>();
overrides.put(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.maxOpen", "10");
@@ -735,7 +657,7 @@ public class DefaultCompactionPlannerTest {
overrides.put(Property.TSERV_SCAN_MAX_OPENFILES.getKey(), "5");
var conf = new ConfigurationImpl(SiteConfiguration.empty().withOverrides(overrides).build());
- var planner = createPlanner(conf, executors);
+ var planner = createPlanner(conf, groups);
var all = createCFs(1000, 1.9, 1.8, 1.7, 1.6, 1.5, 1.4, 1.3, 1.2, 1.1);
var params = createPlanningParams(all, all, Set.of(), 3, CompactionKind.SYSTEM, conf);
var plan = planner.makePlan(params);
@@ -746,8 +668,8 @@ public class DefaultCompactionPlannerTest {
private CompactionJob createJob(CompactionKind kind, Set<CompactableFile> all,
Set<CompactableFile> files) {
return new CompactionPlanImpl.BuilderImpl(kind, all, all)
- .addJob((short) all.size(), CompactionExecutorIdImpl.internalId(csid, "small"), files)
- .build().getJobs().iterator().next();
+ .addJob((short) all.size(), CompactorGroupIdImpl.groupId("small"), files).build().getJobs()
+ .iterator().next();
}
// Create a set of files whose sizes would require certain compaction ratios to compact
@@ -894,26 +816,10 @@ public class DefaultCompactionPlannerTest {
};
}
- private static CompactionPlanner.InitParameters getInitParamQueues(Configuration conf,
- String queues) {
-
- Map<String,String> options = new HashMap<>();
- options.put("queues", queues.replaceAll("'", "\""));
- options.put("maxOpen", "15");
-
- ServiceEnvironment senv = EasyMock.createMock(ServiceEnvironment.class);
- EasyMock.expect(senv.getConfiguration()).andReturn(conf).anyTimes();
- EasyMock.replay(senv);
-
- return new CompactionPlannerInitParams(csid, prefix, options, senv);
- }
-
- private static CompactionPlanner.InitParameters getInitParams(Configuration conf,
- String executors) {
-
+ private static CompactionPlanner.InitParameters getInitParams(Configuration conf, String groups) {
String maxOpen = conf.get(prefix + "cs1.planner.opts.maxOpen");
Map<String,String> options = new HashMap<>();
- options.put("executors", executors.replaceAll("'", "\""));
+ options.put("groups", groups.replaceAll("'", "\""));
if (maxOpen != null) {
options.put("maxOpen", maxOpen);
@@ -928,10 +834,9 @@ public class DefaultCompactionPlannerTest {
return new CompactionPlannerInitParams(csid, prefix, options, senv);
}
- private static DefaultCompactionPlanner createPlanner(Configuration conf, String executors) {
+ private static DefaultCompactionPlanner createPlanner(Configuration conf, String groups) {
DefaultCompactionPlanner planner = new DefaultCompactionPlanner();
- var initParams = getInitParams(conf, executors);
-
+ var initParams = getInitParams(conf, groups);
planner.init(initParams);
return planner;
}
diff --git a/core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionPrioritizerTest.java b/core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionPrioritizerTest.java
index 6d50f8ecb2..a112dbd6b1 100644
--- a/core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionPrioritizerTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionPrioritizerTest.java
@@ -44,7 +44,7 @@ public class CompactionPrioritizerTest {
// TODO pass numFiles
return new CompactionJobImpl(
CompactionJobPrioritizer.createPriority(kind, totalFiles, numFiles),
- CompactionExecutorIdImpl.externalId("test"), files, kind, Optional.of(false));
+ CompactorGroupIdImpl.groupId("test"), files, kind, Optional.of(false));
}
@Test
diff --git a/core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionServicesConfigTest.java b/core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionServicesConfigTest.java
index 57eb2e330a..981f9d1d6b 100644
--- a/core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionServicesConfigTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/util/compaction/CompactionServicesConfigTest.java
@@ -19,8 +19,6 @@
package org.apache.accumulo.core.util.compaction;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.Map;
@@ -31,75 +29,33 @@ import org.junit.jupiter.api.Test;
public class CompactionServicesConfigTest {
- @SuppressWarnings("deprecation")
- private final Property oldPrefix = Property.TSERV_COMPACTION_SERVICE_PREFIX;
- private final Property newPrefix = Property.COMPACTION_SERVICE_PREFIX;
+ private final Property prefix = Property.COMPACTION_SERVICE_PREFIX;
@Test
public void testCompactionProps() {
ConfigurationCopy conf = new ConfigurationCopy();
- conf.set(newPrefix.getKey() + "default.planner", DefaultCompactionPlanner.class.getName());
- conf.set(newPrefix.getKey() + "default.planner.opts.maxOpen", "10");
- conf.set(newPrefix.getKey() + "default.planner.opts.executors",
- "[{'name':'small','type':'internal','maxSize':'32M','numThreads':2},{'name':'medium','type':'internal','maxSize':'128M','numThreads':2},{'name':'large','type':'internal','numThreads':2}]");
+ conf.set(prefix.getKey() + "default.planner", DefaultCompactionPlanner.class.getName());
+ conf.set(prefix.getKey() + "default.planner.opts.maxOpen", "10");
+ conf.set(prefix.getKey() + "default.planner.opts.groups",
+ "[{'name':'small','maxSize':'32M'},{'name':'medium','maxSize':'128M'},{'name':'large'}]");
- conf.set(oldPrefix.getKey() + "default.planner.opts.ignoredProp", "1");
- conf.set(newPrefix.getKey() + "default.planner.opts.validProp", "1");
- conf.set(oldPrefix.getKey() + "default.planner.opts.validProp", "a");
+ conf.set(prefix.getKey() + "default.planner.opts.validProp", "1");
var compactionConfig = new CompactionServicesConfig(conf);
- assertEquals(Map.of("maxOpen", "10", "executors",
- "[{'name':'small','type':'internal','maxSize':'32M','numThreads':2},{'name':'medium','type':'internal','maxSize':'128M','numThreads':2},{'name':'large','type':'internal','numThreads':2}]",
+ assertEquals(Map.of("maxOpen", "10", "groups",
+ "[{'name':'small','maxSize':'32M'},{'name':'medium','maxSize':'128M'},{'name':'large'}]",
"validProp", "1"), compactionConfig.getOptions().get("default"));
}
- @Test
- public void testDuplicateCompactionPlannerDefs() {
- ConfigurationCopy conf = new ConfigurationCopy();
-
- String planner = DefaultCompactionPlanner.class.getName();
- String oldPlanner = "OldPlanner";
-
- conf.set(newPrefix.getKey() + "default.planner", planner);
- conf.set(oldPrefix.getKey() + "default.planner", oldPlanner);
-
- conf.set(oldPrefix.getKey() + "old.planner", oldPlanner);
-
- var compactionConfig = new CompactionServicesConfig(conf);
- assertEquals(Map.of("default", planner, "old", oldPlanner), compactionConfig.getPlanners());
- }
-
- @Test
- public void testCompactionPlannerOldDef() {
- ConfigurationCopy conf = new ConfigurationCopy();
-
- conf.set(oldPrefix.getKey() + "cs1.planner", DefaultCompactionPlanner.class.getName());
- conf.set(oldPrefix.getKey() + "cs1.planner.opts.maxOpen", "10");
- conf.set(oldPrefix.getKey() + "cs1.planner.opts.executors",
- "[{'name':'small','type':'internal','maxSize':'32M','numThreads':2},{'name':'medium','type':'internal','maxSize':'128M','numThreads':2},{'name':'large','type':'internal','numThreads':2}]");
- conf.set(oldPrefix.getKey() + "cs1.planner.opts.foo", "1");
-
- var compactionConfig = new CompactionServicesConfig(conf);
- assertTrue(compactionConfig.getOptions().get("cs1").containsKey("foo"));
- assertEquals("1", compactionConfig.getOptions().get("cs1").get("foo"));
- }
-
@Test
public void testCompactionRateLimits() {
ConfigurationCopy conf = new ConfigurationCopy();
CompactionServicesConfig compactionConfig;
- conf.set(oldPrefix.getKey() + "cs1.planner", DefaultCompactionPlanner.class.getName());
- conf.set(oldPrefix.getKey() + "cs1.rate.limit", "2M");
+ conf.set(prefix.getKey() + "cs1.planner", DefaultCompactionPlanner.class.getName());
+ conf.set(prefix.getKey() + "cs1.rate.limit", "2M");
compactionConfig = new CompactionServicesConfig(conf);
assertEquals(2097152, compactionConfig.getRateLimits().get("cs1"));
-
- // Test service collision
- conf.set(newPrefix.getKey() + "cs1.rate.limit", "4M");
- var e = assertThrows(IllegalArgumentException.class, () -> new CompactionServicesConfig(conf),
- "failed to throw error");
- assertEquals("Incomplete compaction service definition, missing planner class: cs1.rate.limit",
- e.getMessage(), "Error message was not equal");
}
}
diff --git a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
index f039e0d0b2..68c8cbd798 100644
--- a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
+++ b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloClusterImpl.java
@@ -704,16 +704,9 @@ public class MiniAccumuloClusterImpl implements AccumuloCluster {
var initParams = new CompactionPlannerInitParams(CompactionServiceId.of(serviceId),
csc.getPlannerPrefix(serviceId), csc.getOptions().get(serviceId), senv);
cp.init(initParams);
- initParams.getRequestedExternalExecutors().forEach(ceid -> {
- String id = ceid.canonical();
- if (id.startsWith("e.")) {
- groupNames.add(id.substring(2));
- } else {
- groupNames.add(id);
- }
- });
+ initParams.getRequestedGroups().forEach(gid -> groupNames.add(gid.canonical()));
} catch (Exception e) {
- log.error("For compaction service {}, failed to get compaction queues from planner {}.",
+ log.error("For compaction service {}, failed to get compactor groups from planner {}.",
serviceId, plannerClass, e);
}
}
diff --git a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloConfigImpl.java b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloConfigImpl.java
index 7dd9d90bb5..cb9d86837a 100644
--- a/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloConfigImpl.java
+++ b/minicluster/src/main/java/org/apache/accumulo/miniclusterImpl/MiniAccumuloConfigImpl.java
@@ -166,21 +166,21 @@ public class MiniAccumuloConfigImpl {
mergeProp(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_SIZE.getKey(),
Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_SIZE.getDefaultValue());
- mergeProp(Property.TSERV_COMPACTION_SERVICE_ROOT_PLANNER.getKey(),
- Property.TSERV_COMPACTION_SERVICE_ROOT_PLANNER.getDefaultValue());
- mergeProp(Property.TSERV_COMPACTION_SERVICE_ROOT_EXECUTORS.getKey(),
- Property.TSERV_COMPACTION_SERVICE_ROOT_EXECUTORS.getDefaultValue());
+ mergeProp(Property.COMPACTION_SERVICE_ROOT_PLANNER.getKey(),
+ Property.COMPACTION_SERVICE_ROOT_PLANNER.getDefaultValue());
+ mergeProp(Property.COMPACTION_SERVICE_ROOT_GROUPS.getKey(),
+ Property.COMPACTION_SERVICE_ROOT_GROUPS.getDefaultValue());
- mergeProp(Property.TSERV_COMPACTION_SERVICE_META_PLANNER.getKey(),
- Property.TSERV_COMPACTION_SERVICE_META_PLANNER.getDefaultValue());
- mergeProp(Property.TSERV_COMPACTION_SERVICE_META_EXECUTORS.getKey(),
- Property.TSERV_COMPACTION_SERVICE_META_EXECUTORS.getDefaultValue());
+ mergeProp(Property.COMPACTION_SERVICE_META_PLANNER.getKey(),
+ Property.COMPACTION_SERVICE_META_PLANNER.getDefaultValue());
+ mergeProp(Property.COMPACTION_SERVICE_META_GROUPS.getKey(),
+ Property.COMPACTION_SERVICE_META_GROUPS.getDefaultValue());
- mergeProp(Property.TSERV_COMPACTION_SERVICE_DEFAULT_PLANNER.getKey(),
- Property.TSERV_COMPACTION_SERVICE_DEFAULT_PLANNER.getDefaultValue());
+ mergeProp(Property.COMPACTION_SERVICE_DEFAULT_PLANNER.getKey(),
+ Property.COMPACTION_SERVICE_DEFAULT_PLANNER.getDefaultValue());
- mergeProp(Property.TSERV_COMPACTION_SERVICE_DEFAULT_EXECUTORS.getKey(),
- Property.TSERV_COMPACTION_SERVICE_DEFAULT_EXECUTORS.getDefaultValue());
+ mergeProp(Property.COMPACTION_SERVICE_DEFAULT_GROUPS.getKey(),
+ Property.COMPACTION_SERVICE_DEFAULT_GROUPS.getDefaultValue());
if (isUseCredentialProvider()) {
updateConfigForCredentialProvider();
diff --git a/server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionJobGenerator.java b/server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionJobGenerator.java
index cf8d5b7f3a..adc013fd0a 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionJobGenerator.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/compaction/CompactionJobGenerator.java
@@ -266,7 +266,7 @@ public class CompactionJobGenerator {
Collection<CompactableFile> files = ecMeta.getJobFiles().stream()
.map(f -> new CompactableFileImpl(f, allFiles2.get(f))).collect(Collectors.toList());
CompactionJob job = new CompactionJobImpl(ecMeta.getPriority(),
- ecMeta.getCompactionExecutorId(), files, ecMeta.getKind(), Optional.empty());
+ ecMeta.getCompactionGroupId(), files, ecMeta.getKind(), Optional.empty());
return job;
}).collect(Collectors.toUnmodifiableList());
}
diff --git a/server/base/src/main/java/org/apache/accumulo/server/conf/CheckCompactionConfig.java b/server/base/src/main/java/org/apache/accumulo/server/conf/CheckCompactionConfig.java
index c55fd75a59..46d65f9e1a 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/conf/CheckCompactionConfig.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/conf/CheckCompactionConfig.java
@@ -117,16 +117,9 @@ public class CheckCompactionConfig implements KeywordExecutable {
planner.init(initParams);
- initParams.getRequestedExecutors()
- .forEach((execId, numThreads) -> log.info(
- "Compaction service '{}' requested creation of thread pool '{}' with {} threads.",
- serviceId, execId, numThreads));
-
- initParams.getRequestedExternalExecutors()
- .forEach(execId -> log.info(
- "Compaction service '{}' requested with external execution group '{}'", serviceId,
- execId));
-
+ initParams.getRequestedGroups().forEach(
+ (groupId -> log.info("Compaction service '{}' requested with compactor group '{}'",
+ serviceId, groupId)));
}
log.info("Properties file has passed all checks.");
diff --git a/server/base/src/test/java/org/apache/accumulo/server/conf/CheckCompactionConfigTest.java b/server/base/src/test/java/org/apache/accumulo/server/conf/CheckCompactionConfigTest.java
index a8b871eb4d..052bb9c6f8 100644
--- a/server/base/src/test/java/org/apache/accumulo/server/conf/CheckCompactionConfigTest.java
+++ b/server/base/src/test/java/org/apache/accumulo/server/conf/CheckCompactionConfigTest.java
@@ -35,6 +35,8 @@ import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.gson.JsonParseException;
+
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "path not set by user input")
@@ -49,10 +51,9 @@ public class CheckCompactionConfigTest extends WithTestNames {
public void testValidInput1() throws Exception {
String inputString = ("compaction.service.cs1.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':4},\\\n"
- + "{'name':'large','type':'internal','numThreads':2}]").replaceAll("'", "\"");
+ + "compaction.service.cs1.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'},{'name':'medium','maxSize':'128M'},\\\n"
+ + "{'name':'large'}]").replaceAll("'", "\"");
String filePath = writeToFileAndReturnPath(inputString);
@@ -63,16 +64,13 @@ public class CheckCompactionConfigTest extends WithTestNames {
public void testValidInput2() throws Exception {
String inputString = ("compaction.service.cs1.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':4},\\\n"
- + "{'name':'large','type':'internal','numThreads':2}] \n"
- + "compaction.service.cs2.planner="
+ + "compaction.service.cs1.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'},{'name':'medium','maxSize':'128M'},\\\n"
+ + "{'name':'large'}] \ncompaction.service.cs2.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs2.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':7},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':5},\\\n"
- + "{'name':'large','type':'external','group':'DCQ1'}]").replaceAll("'", "\"");
+ + "compaction.service.cs2.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'},{'name':'medium','maxSize':'128M'},\\\n"
+ + "{'name':'large'}]").replaceAll("'", "\"");
String filePath = writeToFileAndReturnPath(inputString);
@@ -83,19 +81,15 @@ public class CheckCompactionConfigTest extends WithTestNames {
public void testValidInput3() throws Exception {
String inputString = ("compaction.service.cs1.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':4},\\\n"
- + "{'name':'large','type':'internal','numThreads':2}] \n"
- + "compaction.service.cs2.planner="
+ + "compaction.service.cs1.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'},{'name':'medium','maxSize':'128M'},\\\n"
+ + "{'name':'large'}] \ncompaction.service.cs2.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs2.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':7},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':5},\\\n"
- + "{'name':'large','type':'external','group':'DCQ1'}] \n"
- + "compaction.service.cs3.planner="
+ + "compaction.service.cs2.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'}, {'name':'medium','maxSize':'128M'},\\\n"
+ + "{'name':'large'}] \ncompaction.service.cs3.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs3.planner.opts.queues=\\\n"
+ + "compaction.service.cs3.planner.opts.groups=\\\n"
+ "[{'name':'small','maxSize':'16M'},{'name':'large'}]").replaceAll("'", "\"");
String filePath = writeToFileAndReturnPath(inputString);
@@ -103,46 +97,29 @@ public class CheckCompactionConfigTest extends WithTestNames {
}
@Test
- public void testThrowsExternalNumThreadsError() throws IOException {
+ public void testThrowsInvalidFieldsError() throws IOException {
String inputString = ("compaction.service.cs1.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'external','maxSize':'128M', 'group':'DCQ1', 'numThreads':4},\\\n"
- + "{'name':'large','type':'internal','numThreads':2}]").replaceAll("'", "\"");
- String expectedErrorMsg = "'numThreads' should not be specified for external compactions";
+ + "compaction.service.cs1.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'},{'name':'medium','maxSize':'128M'},\\\n"
+ + "{'name':'large','numThreads':2}]").replaceAll("'", "\"");
+ String expectedErrorMsg =
+ "Invalid fields: [numThreads] provided for class: org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner$GroupConfig";
String filePath = writeToFileAndReturnPath(inputString);
- var e = assertThrows(IllegalArgumentException.class,
+ var e = assertThrows(JsonParseException.class,
() -> CheckCompactionConfig.main(new String[] {filePath}));
assertEquals(expectedErrorMsg, e.getMessage());
}
- @Test
- public void testNegativeThreadCount() throws IOException {
- String inputString = ("compaction.service.cs1.planner="
- + "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':-4},\\\n"
- + "{'name':'large','type':'internal','numThreads':2}]").replaceAll("'", "\"");
- String expectedErrorMsg = "Positive number of threads required : -4";
-
- String filePath = writeToFileAndReturnPath(inputString);
-
- var e = assertThrows(IllegalArgumentException.class,
- () -> CheckCompactionConfig.main(new String[] {filePath}));
- assertEquals(e.getMessage(), expectedErrorMsg);
- }
-
@Test
public void testNoPlanner() throws Exception {
- String inputString = ("compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':4},\\\n"
- + "{'name':'large','type':'internal','numThreads':2}]").replaceAll("'", "\"");
- String expectedErrorMsg = "Incomplete compaction service definition, missing planner class";
+ String inputString = ("compaction.service.cs1.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'}, {'name':'medium','maxSize':'128M'},\\\n"
+ + "{'name':'large'}]").replaceAll("'", "\"");
+ String expectedErrorMsg =
+ "Incomplete compaction service definitions, missing planner class [cs1]";
String filePath = writeToFileAndReturnPath(inputString);
@@ -152,52 +129,16 @@ public class CheckCompactionConfigTest extends WithTestNames {
}
@Test
- public void testRepeatedCompactionExecutorID() throws Exception {
- String inputString = ("compaction.service.cs1.planner="
- + "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':4},\\\n"
- + "{'name':'small','type':'internal','numThreads':2}]").replaceAll("'", "\"");
- String expectedErrorMsg = "Duplicate Compaction Executor ID found";
-
- final String filePath = writeToFileAndReturnPath(inputString);
-
- var e = assertThrows(IllegalStateException.class,
- () -> CheckCompactionConfig.main(new String[] {filePath}));
- assertTrue(e.getMessage().startsWith(expectedErrorMsg));
- }
-
- @Test
- public void testRepeatedQueueName() throws Exception {
+ public void testRepeatedCompactionGroup() throws Exception {
String inputString = ("compaction.service.cs1.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'external','maxSize':'16M','group':'failedQueue'}] \n"
- + "compaction.service.cs1.planner.opts.queues=[{'name':'failedQueue'}]")
- .replaceAll("'", "\"");
-
- String expectedErrorMsg = "Duplicate external executor for group failedQueue";
+ + "compaction.service.cs1.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'},{'name':'medium','maxSize':'128M'},\\\n"
+ + "{'name':'small'}]").replaceAll("'", "\"");
+ String expectedErrorMsg = "Duplicate compactor group for group: small";
final String filePath = writeToFileAndReturnPath(inputString);
- var err = assertThrows(IllegalArgumentException.class,
- () -> CheckCompactionConfig.main(new String[] {filePath}));
- assertEquals(err.getMessage(), expectedErrorMsg);
- }
-
- @Test
- public void testInvalidTypeValue() throws Exception {
- String inputString = ("compaction.service.cs1.planner="
- + "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'128M','numThreads':4},\\\n"
- + "{'name':'large','type':'internl','numThreads':2}]").replaceAll("'", "\"");
- String expectedErrorMsg = "type must be 'internal' or 'external'";
-
- String filePath = writeToFileAndReturnPath(inputString);
-
var e = assertThrows(IllegalArgumentException.class,
() -> CheckCompactionConfig.main(new String[] {filePath}));
assertTrue(e.getMessage().startsWith(expectedErrorMsg));
@@ -207,10 +148,9 @@ public class CheckCompactionConfigTest extends WithTestNames {
public void testInvalidMaxSize() throws Exception {
String inputString = ("compaction.service.cs1.planner="
+ "org.apache.accumulo.core.spi.compaction.DefaultCompactionPlanner \n"
- + "compaction.service.cs1.planner.opts.executors=\\\n"
- + "[{'name':'small','type':'internal','maxSize':'16M','numThreads':8},\\\n"
- + "{'name':'medium','type':'internal','maxSize':'0M','numThreads':4},\\\n"
- + "{'name':'large','type':'internal','numThreads':2}]").replaceAll("'", "\"");
+ + "compaction.service.cs1.planner.opts.groups=\\\n"
+ + "[{'name':'small','maxSize':'16M'},{'name':'medium','maxSize':'0M'},\\\n"
+ + "{'name':'large'}]").replaceAll("'", "\"");
String expectedErrorMsg = "Invalid value for maxSize";
String filePath = writeToFileAndReturnPath(inputString);
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
index faea953da6..41196776c2 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/CompactionCoordinator.java
@@ -87,9 +87,9 @@ import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.schema.Ample;
import org.apache.accumulo.core.metadata.schema.Ample.Refreshes.RefreshEntry;
import org.apache.accumulo.core.metadata.schema.Ample.RejectionHandler;
+import org.apache.accumulo.core.metadata.schema.CompactionMetadata;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
-import org.apache.accumulo.core.metadata.schema.ExternalCompactionMetadata;
import org.apache.accumulo.core.metadata.schema.SelectedFiles;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
@@ -107,7 +107,7 @@ import org.apache.accumulo.core.tabletserver.thrift.TabletServerClientService;
import org.apache.accumulo.core.util.Retry;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.accumulo.core.util.cache.Caches.CacheName;
-import org.apache.accumulo.core.util.compaction.CompactionExecutorIdImpl;
+import org.apache.accumulo.core.util.compaction.CompactorGroupIdImpl;
import org.apache.accumulo.core.util.compaction.ExternalCompactionUtil;
import org.apache.accumulo.core.util.compaction.RunningCompaction;
import org.apache.accumulo.core.util.threads.ThreadPools;
@@ -409,8 +409,7 @@ public class CompactionCoordinator
TExternalCompactionJob result = null;
- CompactionJobQueues.MetaJob metaJob =
- jobQueues.poll(CompactionExecutorIdImpl.externalId(groupName));
+ CompactionJobQueues.MetaJob metaJob = jobQueues.poll(CompactorGroupIdImpl.groupId(groupName));
while (metaJob != null) {
@@ -418,7 +417,7 @@ public class CompactionCoordinator
// this method may reread the metadata, do not use the metadata in metaJob for anything after
// this method
- ExternalCompactionMetadata ecm = null;
+ CompactionMetadata ecm = null;
var kind = metaJob.getJob().getKind();
@@ -442,7 +441,7 @@ public class CompactionCoordinator
} else {
LOG.debug("Unable to reserve compaction job for {}, pulling another off the queue ",
metaJob.getTabletMetadata().getExtent());
- metaJob = jobQueues.poll(CompactionExecutorIdImpl.externalId(groupName));
+ metaJob = jobQueues.poll(CompactorGroupIdImpl.groupId(groupName));
}
}
@@ -527,7 +526,7 @@ public class CompactionCoordinator
}
}
- private ExternalCompactionMetadata createExternalCompactionMetadata(CompactionJob job,
+ private CompactionMetadata createExternalCompactionMetadata(CompactionJob job,
Set<StoredTabletFile> jobFiles, TabletMetadata tablet, String compactorAddress,
ExternalCompactionId externalCompactionId) {
boolean propDels;
@@ -556,12 +555,12 @@ public class CompactionCoordinator
ReferencedTabletFile newFile = TabletNameGenerator.getNextDataFilenameForMajc(propDels, ctx,
tablet, directoryCreator, externalCompactionId);
- return new ExternalCompactionMetadata(jobFiles, newFile, compactorAddress, job.getKind(),
- job.getPriority(), job.getExecutor(), propDels, fateTxId);
+ return new CompactionMetadata(jobFiles, newFile, compactorAddress, job.getKind(),
+ job.getPriority(), job.getGroup(), propDels, fateTxId);
}
- private ExternalCompactionMetadata reserveCompaction(CompactionJobQueues.MetaJob metaJob,
+ private CompactionMetadata reserveCompaction(CompactionJobQueues.MetaJob metaJob,
String compactorAddress, ExternalCompactionId externalCompactionId) {
Preconditions.checkArgument(metaJob.getJob().getKind() == CompactionKind.SYSTEM
@@ -616,9 +615,8 @@ public class CompactionCoordinator
return null;
}
- TExternalCompactionJob createThriftJob(String externalCompactionId,
- ExternalCompactionMetadata ecm, CompactionJobQueues.MetaJob metaJob,
- Optional<CompactionConfig> compactionConfig) {
+ TExternalCompactionJob createThriftJob(String externalCompactionId, CompactionMetadata ecm,
+ CompactionJobQueues.MetaJob metaJob, Optional<CompactionConfig> compactionConfig) {
Set<CompactableFile> selectedFiles;
if (metaJob.getJob().getKind() == CompactionKind.SYSTEM) {
@@ -809,7 +807,7 @@ public class CompactionCoordinator
return;
}
- ExternalCompactionMetadata ecm = tabletMeta.getExternalCompactions().get(ecid);
+ CompactionMetadata ecm = tabletMeta.getExternalCompactions().get(ecid);
// ELASTICITY_TODO this code does not handle race conditions or faults. Need to ensure refresh
// happens in the case of manager process death between commit and refresh.
@@ -852,7 +850,7 @@ public class CompactionCoordinator
}
private Optional<ReferencedTabletFile> renameOrDeleteFile(TCompactionStats stats,
- ExternalCompactionMetadata ecm, ReferencedTabletFile newDatafile) throws IOException {
+ CompactionMetadata ecm, ReferencedTabletFile newDatafile) throws IOException {
if (stats.getEntriesWritten() == 0) {
// the compaction produced no output so do not need to rename or add a file to the metadata
// table, only delete the input files.
@@ -906,7 +904,7 @@ public class CompactionCoordinator
return false;
}
- ExternalCompactionMetadata ecm = tabletMetadata.getExternalCompactions().get(ecid);
+ CompactionMetadata ecm = tabletMetadata.getExternalCompactions().get(ecid);
if (ecm == null) {
LOG.debug("Received completion notification for unknown compaction {} {}", ecid, extent);
@@ -961,7 +959,7 @@ public class CompactionCoordinator
.logInterval(3, MINUTES).createRetry();
while (canCommitCompaction(ecid, tablet)) {
- ExternalCompactionMetadata ecm = tablet.getExternalCompactions().get(ecid);
+ CompactionMetadata ecm = tablet.getExternalCompactions().get(ecid);
// the compacted files should not exists in the tablet already
var tablet2 = tablet;
@@ -1026,7 +1024,7 @@ public class CompactionCoordinator
private void updateTabletForCompaction(TCompactionStats stats, ExternalCompactionId ecid,
TabletMetadata tablet, Optional<ReferencedTabletFile> newDatafile, KeyExtent extent,
- ExternalCompactionMetadata ecm, Ample.ConditionalTabletMutator tabletMutator) {
+ CompactionMetadata ecm, Ample.ConditionalTabletMutator tabletMutator) {
// ELASTICITY_TODO improve logging adapt to use existing tablet files logging
if (ecm.getKind() == CompactionKind.USER) {
if (tablet.getSelectedFiles().getFiles().equals(ecm.getJobFiles())) {
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/QueueMetrics.java b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/QueueMetrics.java
index 24a9ccae5d..cb24ddd2d0 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/QueueMetrics.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/coordinator/QueueMetrics.java
@@ -27,7 +27,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.accumulo.core.metrics.MetricsProducer;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.manager.compaction.queue.CompactionJobPriorityQueue;
import org.apache.accumulo.manager.compaction.queue.CompactionJobQueues;
@@ -50,7 +50,7 @@ public class QueueMetrics implements MetricsProducer {
private final Gauge jobsRejected;
private final Gauge jobsLowestPriority;
- public QueueMeters(MeterRegistry meterRegistry, CompactionExecutorId queueId,
+ public QueueMeters(MeterRegistry meterRegistry, CompactorGroupId queueId,
CompactionJobPriorityQueue queue) {
length =
Gauge.builder(METRICS_COMPACTOR_JOB_PRIORITY_QUEUE_LENGTH, queue, q -> q.getMaxSize())
@@ -99,7 +99,7 @@ public class QueueMetrics implements MetricsProducer {
private static final long DEFAULT_MIN_REFRESH_DELAY = TimeUnit.SECONDS.toMillis(5);
private volatile MeterRegistry meterRegistry = null;
private final CompactionJobQueues compactionJobQueues;
- private final Map<CompactionExecutorId,QueueMeters> perQueueMetrics = new HashMap<>();
+ private final Map<CompactorGroupId,QueueMeters> perQueueMetrics = new HashMap<>();
private Gauge queueCountMeter = null;
public QueueMetrics(CompactionJobQueues compactionJobQueues) {
@@ -124,20 +124,20 @@ public class QueueMetrics implements MetricsProducer {
}
LOG.debug("update - cjq queues: {}", compactionJobQueues.getQueueIds());
- Set<CompactionExecutorId> definedQueues = compactionJobQueues.getQueueIds();
+ Set<CompactorGroupId> definedQueues = compactionJobQueues.getQueueIds();
LOG.debug("update - defined queues: {}", definedQueues);
- Set<CompactionExecutorId> queuesWithMetrics = perQueueMetrics.keySet();
+ Set<CompactorGroupId> queuesWithMetrics = perQueueMetrics.keySet();
LOG.debug("update - queues with metrics: {}", queuesWithMetrics);
- SetView<CompactionExecutorId> queuesWithoutMetrics =
+ SetView<CompactorGroupId> queuesWithoutMetrics =
Sets.difference(definedQueues, queuesWithMetrics);
queuesWithoutMetrics.forEach(q -> {
LOG.debug("update - creating meters for queue: {}", q);
perQueueMetrics.put(q, new QueueMeters(localRegistry, q, compactionJobQueues.getQueue(q)));
});
- SetView<CompactionExecutorId> metricsWithoutQueues =
+ SetView<CompactorGroupId> metricsWithoutQueues =
Sets.difference(queuesWithMetrics, definedQueues);
metricsWithoutQueues.forEach(q -> {
LOG.debug("update - removing meters for queue: {}", q);
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobPriorityQueue.java b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobPriorityQueue.java
index 74f168a985..12d82ff1f9 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobPriorityQueue.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobPriorityQueue.java
@@ -31,8 +31,8 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionJob;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
import org.apache.accumulo.core.util.compaction.CompactionJobPrioritizer;
import com.google.common.base.Preconditions;
@@ -48,7 +48,7 @@ import com.google.common.base.Preconditions;
*/
public class CompactionJobPriorityQueue {
// ELASTICITY_TODO unit test this class
- private final CompactionExecutorId executorId;
+ private final CompactorGroupId groupId;
private class CjpqKey implements Comparable<CjpqKey> {
private final CompactionJob job;
@@ -106,11 +106,11 @@ public class CompactionJobPriorityQueue {
private boolean closed = false;
- public CompactionJobPriorityQueue(CompactionExecutorId executorId, int maxSize) {
+ public CompactionJobPriorityQueue(CompactorGroupId groupId, int maxSize) {
this.jobQueue = new TreeMap<>();
this.maxSize = maxSize;
this.tabletJobs = new HashMap<>();
- this.executorId = executorId;
+ this.groupId = groupId;
this.rejectedJobs = new AtomicLong(0);
this.dequeuedJobs = new AtomicLong(0);
}
@@ -120,8 +120,7 @@ public class CompactionJobPriorityQueue {
return false;
}
- Preconditions
- .checkArgument(jobs.stream().allMatch(job -> job.getExecutor().equals(executorId)));
+ Preconditions.checkArgument(jobs.stream().allMatch(job -> job.getGroup().equals(groupId)));
removePreviousSubmissions(tabletMetadata.getExtent());
diff --git a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueues.java b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueues.java
index f6090effd2..2d6f88e448 100644
--- a/server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueues.java
+++ b/server/manager/src/main/java/org/apache/accumulo/manager/compaction/queue/CompactionJobQueues.java
@@ -24,8 +24,8 @@ import java.util.concurrent.ConcurrentHashMap.KeySetView;
import java.util.stream.Collectors;
import org.apache.accumulo.core.metadata.schema.TabletMetadata;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionJob;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,7 +39,7 @@ public class CompactionJobQueues {
// can be observed that scoped locks are acquired. Other concurrent map impls may run the compute
// lambdas concurrently for a given key, which may still be correct but is more difficult to
// analyze.
- private final ConcurrentHashMap<CompactionExecutorId,CompactionJobPriorityQueue> priorityQueues =
+ private final ConcurrentHashMap<CompactorGroupId,CompactionJobPriorityQueue> priorityQueues =
new ConcurrentHashMap<>();
private final int queueSize;
@@ -50,44 +50,44 @@ public class CompactionJobQueues {
public void add(TabletMetadata tabletMetadata, Collection<CompactionJob> jobs) {
if (jobs.size() == 1) {
- var executorId = jobs.iterator().next().getExecutor();
+ var executorId = jobs.iterator().next().getGroup();
add(tabletMetadata, executorId, jobs);
} else {
- jobs.stream().collect(Collectors.groupingBy(CompactionJob::getExecutor)).forEach(
- ((executorId, compactionJobs) -> add(tabletMetadata, executorId, compactionJobs)));
+ jobs.stream().collect(Collectors.groupingBy(CompactionJob::getGroup))
+ .forEach(((groupId, compactionJobs) -> add(tabletMetadata, groupId, compactionJobs)));
}
}
- public KeySetView<CompactionExecutorId,CompactionJobPriorityQueue> getQueueIds() {
+ public KeySetView<CompactorGroupId,CompactionJobPriorityQueue> getQueueIds() {
return priorityQueues.keySet();
}
- public CompactionJobPriorityQueue getQueue(CompactionExecutorId executorId) {
- return priorityQueues.get(executorId);
+ public CompactionJobPriorityQueue getQueue(CompactorGroupId groupId) {
+ return priorityQueues.get(groupId);
}
- public long getQueueMaxSize(CompactionExecutorId executorId) {
- var prioQ = priorityQueues.get(executorId);
+ public long getQueueMaxSize(CompactorGroupId groupId) {
+ var prioQ = priorityQueues.get(groupId);
return prioQ == null ? 0 : prioQ.getMaxSize();
}
- public long getQueuedJobs(CompactionExecutorId executorId) {
- var prioQ = priorityQueues.get(executorId);
+ public long getQueuedJobs(CompactorGroupId groupId) {
+ var prioQ = priorityQueues.get(groupId);
return prioQ == null ? 0 : prioQ.getQueuedJobs();
}
- public long getDequeuedJobs(CompactionExecutorId executorId) {
- var prioQ = priorityQueues.get(executorId);
+ public long getDequeuedJobs(CompactorGroupId groupId) {
+ var prioQ = priorityQueues.get(groupId);
return prioQ == null ? 0 : prioQ.getDequeuedJobs();
}
- public long getRejectedJobs(CompactionExecutorId executorId) {
- var prioQ = priorityQueues.get(executorId);
+ public long getRejectedJobs(CompactorGroupId groupId) {
+ var prioQ = priorityQueues.get(groupId);
return prioQ == null ? 0 : prioQ.getRejectedJobs();
}
- public long getLowestPriority(CompactionExecutorId executorId) {
- var prioQ = priorityQueues.get(executorId);
+ public long getLowestPriority(CompactorGroupId groupId) {
+ var prioQ = priorityQueues.get(groupId);
return prioQ == null ? 0 : prioQ.getLowestPriority();
}
@@ -123,15 +123,15 @@ public class CompactionJobQueues {
}
}
- public MetaJob poll(CompactionExecutorId executorId) {
- var prioQ = priorityQueues.get(executorId);
+ public MetaJob poll(CompactorGroupId groupId) {
+ var prioQ = priorityQueues.get(groupId);
if (prioQ == null) {
return null;
}
MetaJob mj = prioQ.poll();
if (mj == null) {
- priorityQueues.computeIfPresent(executorId, (eid, pq) -> {
+ priorityQueues.computeIfPresent(groupId, (eid, pq) -> {
if (pq.closeIfEmpty()) {
return null;
} else {
@@ -142,23 +142,23 @@ public class CompactionJobQueues {
return mj;
}
- private void add(TabletMetadata tabletMetadata, CompactionExecutorId executorId,
+ private void add(TabletMetadata tabletMetadata, CompactorGroupId groupId,
Collection<CompactionJob> jobs) {
// TODO log level
if (log.isDebugEnabled()) {
- log.debug("Adding jobs to queue {} {} {}", executorId, tabletMetadata.getExtent(),
+ log.debug("Adding jobs to queue {} {} {}", groupId, tabletMetadata.getExtent(),
jobs.stream().map(job -> "#files:" + job.getFiles().size() + ",prio:" + job.getPriority()
+ ",kind:" + job.getKind()).collect(Collectors.toList()));
}
- var pq = priorityQueues.computeIfAbsent(executorId,
- eid -> new CompactionJobPriorityQueue(eid, queueSize));
+ var pq = priorityQueues.computeIfAbsent(groupId,
+ gid -> new CompactionJobPriorityQueue(gid, queueSize));
while (!pq.add(tabletMetadata, jobs)) {
// This loop handles race condition where poll() closes empty priority queues. The queue could
// be closed after its obtained from the map and before add is called.
- pq = priorityQueues.computeIfAbsent(executorId,
- eid -> new CompactionJobPriorityQueue(eid, queueSize));
+ pq = priorityQueues.computeIfAbsent(groupId,
+ gid -> new CompactionJobPriorityQueue(gid, queueSize));
}
}
}
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 5ab11e6629..4bdd5a8eca 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
@@ -69,9 +69,11 @@ public class BadCompactionServiceConfigIT extends SharedMiniClusterBase {
@Override
public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
Map<String,String> siteCfg = new HashMap<>();
+ siteCfg.put(CSP + "default.planner", DefaultCompactionPlanner.class.getName());
+ siteCfg.put(CSP + "default.planner.opts.groups", "[{\"name\":\"default_group\"}]");
siteCfg.put(CSP + "cs1.planner", DefaultCompactionPlanner.class.getName());
// place invalid json in the planners config
- siteCfg.put(CSP + "cs1.planner.opts.executors", "{{'name]");
+ siteCfg.put(CSP + "cs1.planner.opts.groups", "{{'name]");
cfg.setSiteConfig(siteCfg);
}
}
@@ -126,8 +128,8 @@ public class BadCompactionServiceConfigIT extends SharedMiniClusterBase {
.collect(MoreCollectors.onlyElement()));
}
- var value = "[{'name':'all', 'type': 'external', 'group':'cs1q1'}]".replaceAll("'", "\"");
- client.instanceOperations().setProperty(CSP + "cs1.planner.opts.executors", value);
+ var value = "[{'name':'cs1q1'}]".replaceAll("'", "\"");
+ client.instanceOperations().setProperty(CSP + "cs1.planner.opts.groups", value);
// start the compactor, it was not started initially because of bad config
getCluster().getConfig().getClusterServerConfiguration()
@@ -152,7 +154,7 @@ public class BadCompactionServiceConfigIT extends SharedMiniClusterBase {
// misconfigure the service, test how going from good config to bad config works. The test
// started with an initial state of bad config.
- client.instanceOperations().setProperty(CSP + "cs1.planner.opts.executors", "]o.o[");
+ client.instanceOperations().setProperty(CSP + "cs1.planner.opts.groups", "]o.o[");
try (var writer = client.createBatchWriter(table)) {
writer.addMutation(new Mutation("0").at().family("f").qualifier("q").put("v"));
}
@@ -164,8 +166,8 @@ public class BadCompactionServiceConfigIT extends SharedMiniClusterBase {
fixerFuture = executorService.submit(() -> {
try {
Thread.sleep(2000);
- var value = "[{'name':'all', 'type': 'external', 'group':'cs1q1'}]".replaceAll("'", "\"");
- client.instanceOperations().setProperty(CSP + "cs1.planner.opts.executors", value);
+ var value = "[{'name':'cs1q1'}]".replaceAll("'", "\"");
+ client.instanceOperations().setProperty(CSP + "cs1.planner.opts.groups", value);
} catch (Exception e) {
throw new RuntimeException(e);
}
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 22e8438695..273d6c93b1 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
@@ -45,15 +45,14 @@ public class CompactionConfigChangeIT extends AccumuloClusterHarness {
@Override
public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
- cfg.getClusterServerConfiguration().addCompactorResourceGroup("e4", 1);
- cfg.getClusterServerConfiguration().addCompactorResourceGroup("e5", 1);
+ cfg.getClusterServerConfiguration().addCompactorResourceGroup("little", 1);
+ cfg.getClusterServerConfiguration().addCompactorResourceGroup("big", 1);
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.executors",
- ("[{'name':'small','type':'external','maxSize':'2M', 'group': 'e1'},"
- + "{'name':'medium','type':'external','maxSize':'128M', 'group':'e2'},"
- + "{'name':'large','type':'external','group':'e3'}]").replaceAll("'", "\""));
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.groups",
+ ("[{'name':'small','maxSize':'2M'}, {'name':'medium','maxSize':'128M'},"
+ + "{'name':'large'}]").replaceAll("'", "\""));
// use raw local file system so walogs sync and flush will work
hadoopCoreSite.set("fs.file.impl", RawLocalFileSystem.class.getName());
@@ -97,14 +96,13 @@ public class CompactionConfigChangeIT extends AccumuloClusterHarness {
// give some time for compactions to start running
Wait.waitFor(() -> countFiles(client, table, "F") < 95);
- // Change config deleting executors named small, medium, and large. There was bug where
- // deleting executors running compactions would leave the tablet in a bad state for future
+ // Change config deleting groups named small, medium, and large. There was bug where
+ // deleting groups running compactions would leave the tablet in a bad state for future
// compactions. Because the compactions are running slow, expect this config change to overlap
// with running compactions.
client.instanceOperations().setProperty(
- Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.executors",
- ("[{'name':'little','type':'external','maxSize':'128M','group':'e4'},"
- + "{'name':'big','type':'external','group':'e5'}]").replaceAll("'", "\""));
+ Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.groups",
+ ("[{'name':'little','maxSize':'128M'},{'name':'big'}]").replaceAll("'", "\""));
Wait.waitFor(() -> countFiles(client, table, "F") == 0, 60000);
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionExecutorIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/CompactionExecutorIT.java
index 19f6d0bb30..10d4dd1cd8 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionExecutorIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/CompactionExecutorIT.java
@@ -65,10 +65,10 @@ import org.apache.accumulo.core.iterators.user.RegExFilter;
import org.apache.accumulo.core.metadata.StoredTabletFile;
import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType;
import org.apache.accumulo.core.metadata.schema.TabletsMetadata;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionKind;
import org.apache.accumulo.core.spi.compaction.CompactionPlan;
import org.apache.accumulo.core.spi.compaction.CompactionPlanner;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
import org.apache.accumulo.harness.MiniClusterConfigurationCallback;
import org.apache.accumulo.harness.SharedMiniClusterBase;
import org.apache.accumulo.minicluster.ServerType;
@@ -93,27 +93,27 @@ public class CompactionExecutorIT extends SharedMiniClusterBase {
public static class TestPlanner implements CompactionPlanner {
- private static class ExecutorConfig {
- String group;
+ private static class GroupConfig {
+ String name;
}
private int filesPerCompaction;
- private List<CompactionExecutorId> executorIds;
+ private List<CompactorGroupId> groupIds;
private EnumSet<CompactionKind> kindsToProcess = EnumSet.noneOf(CompactionKind.class);
@Override
public void init(InitParameters params) {
- var executors = params.getOptions().get("executors");
+ var groups = params.getOptions().get("groups");
this.filesPerCompaction = Integer.parseInt(params.getOptions().get("filesPerCompaction"));
- this.executorIds = new ArrayList<>();
+ this.groupIds = new ArrayList<>();
for (String kind : params.getOptions().get("process").split(",")) {
kindsToProcess.add(CompactionKind.valueOf(kind.toUpperCase()));
}
- for (JsonElement element : GSON.get().fromJson(executors, JsonArray.class)) {
- ExecutorConfig executorConfig = GSON.get().fromJson(element, ExecutorConfig.class);
- var ceid = params.getExecutorManager().getExternalExecutor(executorConfig.group);
- executorIds.add(ceid);
+ for (JsonElement element : GSON.get().fromJson(groups, JsonArray.class)) {
+ GroupConfig groupConfig = GSON.get().fromJson(element, GroupConfig.class);
+ var cgid = params.getGroupManager().getGroup(groupConfig.name);
+ groupIds.add(cgid);
}
}
@@ -127,7 +127,7 @@ public class CompactionExecutorIT extends SharedMiniClusterBase {
if (Boolean.parseBoolean(params.getExecutionHints().getOrDefault("compact_all", "false"))) {
return params
.createPlanBuilder().addJob((short) 1,
- executorIds.get(RANDOM.get().nextInt(executorIds.size())), params.getCandidates())
+ groupIds.get(RANDOM.get().nextInt(groupIds.size())), params.getCandidates())
.build();
}
@@ -140,8 +140,7 @@ public class CompactionExecutorIT extends SharedMiniClusterBase {
params.getCandidates().stream().collect(Collectors.groupingBy(TestPlanner::getFirstChar))
.values().forEach(files -> {
for (int i = filesPerCompaction; i <= files.size(); i += filesPerCompaction) {
- planBuilder.addJob((short) 1,
- executorIds.get(RANDOM.get().nextInt(executorIds.size())),
+ planBuilder.addJob((short) 1, groupIds.get(RANDOM.get().nextInt(groupIds.size())),
files.subList(i - filesPerCompaction, i));
}
});
@@ -158,37 +157,29 @@ public class CompactionExecutorIT extends SharedMiniClusterBase {
public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration conf) {
var csp = Property.COMPACTION_SERVICE_PREFIX.getKey();
cfg.setProperty(csp + "cs1.planner", TestPlanner.class.getName());
- cfg.setProperty(csp + "cs1.planner.opts.executors",
- "[{'name':'one', 'type':'external', 'group':'e1'},"
- + "{'name':'two', 'type':'external', 'group':'e2'}, {'name':'three', 'type':'external','group':'e3'}]");
+ cfg.setProperty(csp + "cs1.planner.opts.groups",
+ "[{'name':'e1'},{'name':'e2'},{'name':'e3'}]");
cfg.setProperty(csp + "cs1.planner.opts.filesPerCompaction", "5");
cfg.setProperty(csp + "cs1.planner.opts.process", "SYSTEM");
cfg.setProperty(csp + "cs2.planner", TestPlanner.class.getName());
- cfg.setProperty(csp + "cs2.planner.opts.executors",
- "[{'name':'one', 'type':'external', 'group':'f1'},"
- + "{'name':'two', 'type':'external', 'group':'f2'}]");
+ cfg.setProperty(csp + "cs2.planner.opts.groups", "[{'name':'f1'},{'name':'f2'}]");
cfg.setProperty(csp + "cs2.planner.opts.filesPerCompaction", "7");
cfg.setProperty(csp + "cs2.planner.opts.process", "SYSTEM");
cfg.setProperty(csp + "cs3.planner", TestPlanner.class.getName());
- cfg.setProperty(csp + "cs3.planner.opts.executors",
- "[{'name':'one', 'type':'external', 'group':'g1'}]");
+ cfg.setProperty(csp + "cs3.planner.opts.groups", "[{'name':'g1'}]");
cfg.setProperty(csp + "cs3.planner.opts.filesPerCompaction", "3");
cfg.setProperty(csp + "cs3.planner.opts.process", "USER");
cfg.setProperty(csp + "cs4.planner", TestPlanner.class.getName());
- cfg.setProperty(csp + "cs4.planner.opts.executors",
- "[{'name':'one', 'type':'external', 'group':'h1'},"
- + "{'name':'two', 'type':'external', 'group':'h2'}]");
+ cfg.setProperty(csp + "cs4.planner.opts.groups", "[{'name':'h1'},{'name':'h2'}]");
cfg.setProperty(csp + "cs4.planner.opts.filesPerCompaction", "11");
cfg.setProperty(csp + "cs4.planner.opts.process", "USER");
// this is meant to be dynamically reconfigured
cfg.setProperty(csp + "recfg.planner", TestPlanner.class.getName());
- cfg.setProperty(csp + "recfg.planner.opts.executors",
- "[{'name':'one', 'type':'external', 'group':'i1'},"
- + "{'name':'two', 'type':'external', 'group':'i2'}]");
+ cfg.setProperty(csp + "recfg.planner.opts.groups", "[{'name':'i1'},{'name':'i2'}]");
cfg.setProperty(csp + "recfg.planner.opts.filesPerCompaction", "11");
cfg.setProperty(csp + "recfg.planner.opts.process", "SYSTEM");
@@ -262,8 +253,8 @@ public class CompactionExecutorIT extends SharedMiniClusterBase {
Property.COMPACTION_SERVICE_PREFIX.getKey() + "recfg.planner.opts.filesPerCompaction",
"5");
client.instanceOperations().setProperty(
- Property.COMPACTION_SERVICE_PREFIX.getKey() + "recfg.planner.opts.executors",
- "[{'name':'small', 'type':'external', 'group':'group1'}]");
+ Property.COMPACTION_SERVICE_PREFIX.getKey() + "recfg.planner.opts.groups",
+ "[{'name':'group1'}]");
getCluster().getConfig().getClusterServerConfiguration().addCompactorResourceGroup("group1",
1);
compactionGroups.add("group1");
@@ -288,9 +279,8 @@ public class CompactionExecutorIT extends SharedMiniClusterBase {
client.instanceOperations().setProperty(
Property.COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner.opts.process", "SYSTEM");
client.instanceOperations().setProperty(
- Property.COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner.opts.executors",
- "[{'name':'one', 'type':'external', 'group':'add1'},"
- + "{'name':'two', 'type':'external', 'group':'add2'}, {'name':'three', 'type':'external','group':'add3'}]");
+ Property.COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner.opts.groups",
+ "[{'name':'add1'},{'name':'add2'}, {'name':'add3'}]");
client.instanceOperations().setProperty(
Property.COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner",
TestPlanner.class.getName());
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java b/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
index cc5a423eab..f56dfd668f 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/CompactionPriorityQueueMetricsIT.java
@@ -155,9 +155,8 @@ public class CompactionPriorityQueueMetricsIT extends SharedMiniClusterBase {
// Create a new queue with zero compactors.
cfg.setProperty("tserver.compaction.major.service." + QUEUE1_SERVICE + ".planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(
- "tserver.compaction.major.service." + QUEUE1_SERVICE + ".planner.opts.executors",
- "[{'name':'all', 'type': 'external', 'group': '" + QUEUE1 + "'}]");
+ cfg.setProperty("tserver.compaction.major.service." + QUEUE1_SERVICE + ".planner.opts.groups",
+ "[{'name':'" + QUEUE1 + "'}]");
cfg.setProperty(Property.MANAGER_COMPACTION_SERVICE_PRIORITY_QUEUE_SIZE, "6");
cfg.getClusterServerConfiguration().addCompactorResourceGroup(QUEUE1, 0);
diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionTestUtils.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionTestUtils.java
index 6fa0fce8cd..3dbdc228b4 100644
--- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionTestUtils.java
+++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionTestUtils.java
@@ -197,36 +197,36 @@ public class ExternalCompactionTestUtils {
// configure the compaction services to use the queues
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.executors",
- "[{'name':'all', 'type': 'external', 'group': '" + GROUP1 + "'}]");
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs1.planner.opts.groups",
+ "[{'name':'" + GROUP1 + "'}]");
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs2.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs2.planner.opts.executors",
- "[{'name':'all', 'type': 'external','group': '" + GROUP2 + "'}]");
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs2.planner.opts.groups",
+ "[{'name':'" + GROUP2 + "'}]");
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs3.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs3.planner.opts.executors",
- "[{'name':'all', 'type': 'external','group': '" + GROUP3 + "'}]");
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs3.planner.opts.groups",
+ "[{'name':'" + GROUP3 + "'}]");
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs4.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs4.planner.opts.executors",
- "[{'name':'all', 'type': 'external','group': '" + GROUP4 + "'}]");
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs4.planner.opts.groups",
+ "[{'name':'" + GROUP4 + "'}]");
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs5.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs5.planner.opts.executors",
- "[{'name':'all', 'type': 'external','group': '" + GROUP5 + "'}]");
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs5.planner.opts.groups",
+ "[{'name':'" + GROUP5 + "'}]");
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs6.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs6.planner.opts.executors",
- "[{'name':'all', 'type': 'external','group': '" + GROUP6 + "'}]");
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs6.planner.opts.groups",
+ "[{'name':'" + GROUP6 + "'}]");
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs7.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs7.planner.opts.executors",
- "[{'name':'all', 'type': 'external','group': '" + GROUP7 + "'}]");
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs7.planner.opts.groups",
+ "[{'name':'" + GROUP7 + "'}]");
cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs8.planner",
DefaultCompactionPlanner.class.getName());
- cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs8.planner.opts.executors",
- "[{'name':'all', 'type': 'external','group': '" + GROUP8 + "'}]");
+ cfg.setProperty(Property.COMPACTION_SERVICE_PREFIX.getKey() + "cs8.planner.opts.groups",
+ "[{'name':'" + GROUP8 + "'}]");
cfg.setProperty(Property.COMPACTION_COORDINATOR_FINALIZER_COMPLETION_CHECK_INTERVAL, "5s");
cfg.setProperty(Property.COMPACTION_COORDINATOR_DEAD_COMPACTOR_CHECK_INTERVAL, "5s");
cfg.setProperty(Property.COMPACTION_COORDINATOR_TSERVER_COMPACTION_CHECK_INTERVAL, "3s");
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/IdleProcessMetricsIT.java b/test/src/main/java/org/apache/accumulo/test/functional/IdleProcessMetricsIT.java
index bc39619d56..0f214f218c 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/IdleProcessMetricsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/IdleProcessMetricsIT.java
@@ -44,12 +44,12 @@ public class IdleProcessMetricsIT extends SharedMiniClusterBase {
// Configure all compaction planners to use the default resource group so
// that only 1 compactor is started by MiniAccumuloCluster
- cfg.setProperty("tserver.compaction.major.service.root.planner.opts.executors",
- "[{'name':'all','type':'external','group':'default'}]".replaceAll("'", "\""));
- cfg.setProperty("tserver.compaction.major.service.meta.planner.opts.executors",
- "[{'name':'all','type':'external','group':'default'}]".replaceAll("'", "\""));
- cfg.setProperty("tserver.compaction.major.service.default.planner.opts.executors",
- "[{'name':'all','type':'external','group':'default'}]".replaceAll("'", "\""));
+ cfg.setProperty("compaction.service.root.planner.opts.groups",
+ "[{'name':'default'}]".replaceAll("'", "\""));
+ cfg.setProperty("compaction.service.meta.planner.opts.groups",
+ "[{'name':'default'}]".replaceAll("'", "\""));
+ cfg.setProperty("compaction.service.default.planner.opts.groups",
+ "[{'name':'default'}]".replaceAll("'", "\""));
// Disable the default scan servers and compactors, just start 1
// tablet server in the default group to host the root and metadata
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/MergeIT.java b/test/src/main/java/org/apache/accumulo/test/functional/MergeIT.java
index 856b778f19..1244d2d6cf 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/MergeIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/MergeIT.java
@@ -62,17 +62,17 @@ import org.apache.accumulo.core.dataImpl.KeyExtent;
import org.apache.accumulo.core.metadata.MetadataTable;
import org.apache.accumulo.core.metadata.ReferencedTabletFile;
import org.apache.accumulo.core.metadata.StoredTabletFile;
+import org.apache.accumulo.core.metadata.schema.CompactionMetadata;
import org.apache.accumulo.core.metadata.schema.DataFileValue;
import org.apache.accumulo.core.metadata.schema.ExternalCompactionId;
-import org.apache.accumulo.core.metadata.schema.ExternalCompactionMetadata;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.HostingColumnFamily;
import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily;
import org.apache.accumulo.core.security.Authorizations;
-import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionKind;
+import org.apache.accumulo.core.spi.compaction.CompactorGroupId;
import org.apache.accumulo.core.util.FastFormat;
import org.apache.accumulo.core.util.Merge;
-import org.apache.accumulo.core.util.compaction.CompactionExecutorIdImpl;
+import org.apache.accumulo.core.util.compaction.CompactorGroupIdImpl;
import org.apache.accumulo.harness.AccumuloClusterHarness;
import org.apache.accumulo.test.TestIngest;
import org.apache.accumulo.test.TestIngest.IngestParams;
@@ -694,11 +694,11 @@ public class MergeIT extends AccumuloClusterHarness {
ReferencedTabletFile tmpFile =
ReferencedTabletFile.of(new Path("file:///accumulo/tables/t-0/b-0/c1.rf"));
- CompactionExecutorId ceid = CompactionExecutorIdImpl.externalId("G1");
+ CompactorGroupId ceid = CompactorGroupIdImpl.groupId("G1");
Set<StoredTabletFile> jobFiles =
Set.of(StoredTabletFile.of(new Path("file:///accumulo/tables/t-0/b-0/b2.rf")));
- ExternalCompactionMetadata ecMeta = new ExternalCompactionMetadata(jobFiles, tmpFile,
- "localhost:4444", CompactionKind.SYSTEM, (short) 2, ceid, false, 44L);
+ CompactionMetadata ecMeta = new CompactionMetadata(jobFiles, tmpFile, "localhost:4444",
+ CompactionKind.SYSTEM, (short) 2, ceid, false, 44L);
tablet.putExternalCompaction(ecid, ecMeta);
tablet.mutate();
}
diff --git a/test/src/main/java/org/apache/accumulo/test/shell/ConfigSetIT.java b/test/src/main/java/org/apache/accumulo/test/shell/ConfigSetIT.java
index 71a5e92ecb..c62ce13133 100644
--- a/test/src/main/java/org/apache/accumulo/test/shell/ConfigSetIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/shell/ConfigSetIT.java
@@ -18,8 +18,8 @@
*/
package org.apache.accumulo.test.shell;
+import static org.apache.accumulo.core.conf.Property.COMPACTION_SERVICE_ROOT_GROUPS;
import static org.apache.accumulo.core.conf.Property.MONITOR_RESOURCES_EXTERNAL;
-import static org.apache.accumulo.core.conf.Property.TSERV_COMPACTION_SERVICE_ROOT_EXECUTORS;
import static org.apache.accumulo.harness.AccumuloITBase.MINI_CLUSTER_ONLY;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -53,17 +53,14 @@ public class ConfigSetIT extends SharedMiniClusterBase {
public void setInvalidJson() throws Exception {
log.debug("Starting setInvalidJson test ------------------");
- String validJson =
- "[{'name':'small','type':'internal','maxSize':'64M','numThreads':2},{'name':'huge','type':'internal','numThreads':2}]"
- .replaceAll("'", "\"");
+ String validJson = "[{'name':'small','maxSize':'64M'},{'name':'huge'}]".replaceAll("'", "\"");
// missing first value
String invalidJson = "notJson";
try (AccumuloClient client =
getCluster().createAccumuloClient("root", new PasswordToken(getRootPassword()))) {
- client.instanceOperations().setProperty(TSERV_COMPACTION_SERVICE_ROOT_EXECUTORS.getKey(),
- validJson);
+ client.instanceOperations().setProperty(COMPACTION_SERVICE_ROOT_GROUPS.getKey(), validJson);
assertThrows(AccumuloException.class, () -> client.instanceOperations()
.setProperty(MONITOR_RESOURCES_EXTERNAL.getKey(), invalidJson));