You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by ab...@apache.org on 2024/04/29 18:37:19 UTC
(druid) branch master updated: Remove usage of skife from DruidCoordinatorConfig (#15705)
This is an automated email from the ASF dual-hosted git repository.
abhishekrb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new aa463149715 Remove usage of skife from DruidCoordinatorConfig (#15705)
aa463149715 is described below
commit aa4631497158d657bafb3d1831f7ac4cc2c6ca9d
Author: Kashif Faraz <ka...@gmail.com>
AuthorDate: Tue Apr 30 00:07:13 2024 +0530
Remove usage of skife from DruidCoordinatorConfig (#15705)
* Remove usage of skife from DruidCoordinatorConfig
* Remove old config class
* Address static checks
* Fix tests
* Remove unnecessary mocks
* Fix config typos
* Fix config condition
* Fix test, spotbug check
* Move validation to DruidCoordinatorConfig
* Move DruidCoordinatorConfig to different package
* Fix validation of killunusedconfig
* Simplify and fix KillSupervisorsCustomDuty
* Address review comments
* Fix new tests
* Add KillUnusedSchemasConfig
* Remove KillUnusedSchemasConfig
* Minor renames
---
.../druid/server/coordinator/DruidCoordinator.java | 40 +-
.../server/coordinator/DruidCoordinatorConfig.java | 181 -------
.../coordinator/config/CoordinatorKillConfigs.java | 157 ++++++
.../config/CoordinatorPeriodConfig.java | 57 +++
.../coordinator/config/CoordinatorRunConfig.java | 54 ++
.../coordinator/config/DruidCoordinatorConfig.java | 157 ++++++
.../config/HttpLoadQueuePeonConfig.java | 71 +++
.../config/KillUnusedSegmentsConfig.java | 125 +++++
.../coordinator/config/MetadataCleanupConfig.java | 89 ++++
.../server/coordinator/duty/KillAuditLog.java | 14 +-
.../coordinator/duty/KillCompactionConfig.java | 15 +-
.../coordinator/duty/KillDatasourceMetadata.java | 14 +-
.../druid/server/coordinator/duty/KillRules.java | 14 +-
.../server/coordinator/duty/KillSupervisors.java | 14 +-
.../duty/KillSupervisorsCustomDuty.java | 36 +-
.../duty/KillUnreferencedSegmentSchemaDuty.java | 14 +-
.../coordinator/duty/KillUnusedSegments.java | 56 +-
.../coordinator/duty/MetadataCleanupDuty.java | 63 +--
.../coordinator/loading/CuratorLoadQueuePeon.java | 10 +-
.../coordinator/loading/HttpLoadQueuePeon.java | 20 +-
.../coordinator/loading/LoadQueueTaskMaster.java | 35 +-
.../druid/server/coordinator/stats/Stats.java | 2 +
.../coordinator/CuratorDruidCoordinatorTest.java | 11 +-
.../coordinator/DruidCoordinatorConfigTest.java | 552 ++++++++++++++++++--
.../server/coordinator/DruidCoordinatorTest.java | 35 +-
.../coordinator/TestDruidCoordinatorConfig.java | 565 ---------------------
.../coordinator/duty/CompactSegmentsTest.java | 2 +-
.../server/coordinator/duty/KillAuditLogTest.java | 66 +--
.../coordinator/duty/KillCompactionConfigTest.java | 68 +--
.../duty/KillDatasourceMetadataTest.java | 85 +---
.../server/coordinator/duty/KillRulesTest.java | 65 +--
.../duty/KillSupervisorsCustomDutyTest.java | 48 +-
.../coordinator/duty/KillSupervisorsTest.java | 66 +--
.../KillUnreferencedSegmentSchemaDutyTest.java | 42 +-
.../coordinator/duty/KillUnusedSegmentsTest.java | 97 +---
.../coordinator/loading/HttpLoadQueuePeonTest.java | 6 +-
.../coordinator/loading/LoadQueuePeonTest.java | 20 +-
.../simulate/CoordinatorSimulationBuilder.java | 104 ++--
services/pom.xml | 4 -
.../java/org/apache/druid/cli/CliCoordinator.java | 35 +-
40 files changed, 1505 insertions(+), 1604 deletions(-)
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
index 68668e76029..816cf44090b 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java
@@ -56,6 +56,9 @@ import org.apache.druid.segment.metadata.CoordinatorSegmentMetadataCache;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordinator.balancer.BalancerStrategyFactory;
import org.apache.druid.server.coordinator.compact.CompactionSegmentSearchPolicy;
+import org.apache.druid.server.coordinator.config.CoordinatorKillConfigs;
+import org.apache.druid.server.coordinator.config.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.KillUnusedSegmentsConfig;
import org.apache.druid.server.coordinator.duty.BalanceSegments;
import org.apache.druid.server.coordinator.duty.CollectSegmentAndServerStats;
import org.apache.druid.server.coordinator.duty.CompactSegments;
@@ -189,7 +192,6 @@ public class DruidCoordinator
ServiceAnnouncer serviceAnnouncer,
@Self DruidNode self,
CoordinatorCustomDutyGroups customDutyGroups,
- BalancerStrategyFactory balancerStrategyFactory,
LookupCoordinatorManager lookupCoordinatorManager,
@Coordinator DruidLeaderSelector coordLeaderSelector,
CompactionSegmentSearchPolicy compactionSegmentSearchPolicy,
@@ -209,7 +211,7 @@ public class DruidCoordinator
this.executorFactory = scheduledExecutorFactory;
- this.balancerStrategyFactory = balancerStrategyFactory;
+ this.balancerStrategyFactory = config.getBalancerStrategyFactory();
this.lookupCoordinatorManager = lookupCoordinatorManager;
this.coordLeaderSelector = coordLeaderSelector;
this.compactSegments = initializeCompactSegmentsDuty(compactionSegmentSearchPolicy);
@@ -562,10 +564,13 @@ public class DruidCoordinator
List<CoordinatorDuty> makeIndexingServiceDuties()
{
final List<CoordinatorDuty> duties = new ArrayList<>();
- if (config.isKillUnusedSegmentsEnabled()) {
- duties.add(new KillUnusedSegments(metadataManager.segments(), overlordClient, config));
+ final KillUnusedSegmentsConfig killUnusedConfig = config.getKillConfigs().unusedSegments(
+ config.getCoordinatorIndexingPeriod()
+ );
+ if (killUnusedConfig.isCleanupEnabled()) {
+ duties.add(new KillUnusedSegments(metadataManager.segments(), overlordClient, killUnusedConfig));
}
- if (config.isKillPendingSegmentsEnabled()) {
+ if (config.getKillConfigs().pendingSegments().isCleanupEnabled()) {
duties.add(new KillStalePendingSegments(overlordClient));
}
@@ -582,16 +587,23 @@ public class DruidCoordinator
private List<CoordinatorDuty> makeMetadataStoreManagementDuties()
{
- List<CoordinatorDuty> duties = new ArrayList<>();
-
- duties.add(new KillSupervisors(config, metadataManager.supervisors()));
- duties.add(new KillAuditLog(config, metadataManager.audit()));
- duties.add(new KillRules(config, metadataManager.rules()));
- duties.add(new KillDatasourceMetadata(config, metadataManager.indexer(), metadataManager.supervisors()));
- duties.add(new KillCompactionConfig(config, metadataManager.segments(), metadataManager.configs()));
-
+ final CoordinatorKillConfigs killConfigs = config.getKillConfigs();
+ final List<CoordinatorDuty> duties = new ArrayList<>();
+ duties.add(new KillSupervisors(killConfigs.supervisors(), metadataManager.supervisors()));
+ duties.add(new KillAuditLog(killConfigs.auditLogs(), metadataManager.audit()));
+ duties.add(new KillRules(killConfigs.rules(), metadataManager.rules()));
+ duties.add(
+ new KillDatasourceMetadata(
+ killConfigs.datasources(),
+ metadataManager.indexer(),
+ metadataManager.supervisors()
+ )
+ );
+ duties.add(
+ new KillCompactionConfig(killConfigs.compactionConfigs(), metadataManager.segments(), metadataManager.configs())
+ );
if (centralizedDatasourceSchemaConfig.isEnabled()) {
- duties.add(new KillUnreferencedSegmentSchemaDuty(config, metadataManager.schemas()));
+ duties.add(new KillUnreferencedSegmentSchemaDuty(killConfigs.segmentSchemas(), metadataManager.schemas()));
}
return duties;
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java
deleted file mode 100644
index 66123ab5ea6..00000000000
--- a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.druid.server.coordinator;
-
-import org.joda.time.Duration;
-import org.skife.config.Config;
-import org.skife.config.Default;
-
-public abstract class DruidCoordinatorConfig
-{
- @Config("druid.coordinator.startDelay")
- @Default("PT300s")
- public abstract Duration getCoordinatorStartDelay();
-
- @Config("druid.coordinator.period")
- @Default("PT60s")
- public abstract Duration getCoordinatorPeriod();
-
- @Config("druid.coordinator.period.indexingPeriod")
- @Default("PT1800s")
- public abstract Duration getCoordinatorIndexingPeriod();
-
- @Config("druid.coordinator.period.metadataStoreManagementPeriod")
- @Default("PT1H")
- public abstract Duration getCoordinatorMetadataStoreManagementPeriod();
-
- @Config("druid.coordinator.kill.on")
- @Default("false")
- public abstract boolean isKillUnusedSegmentsEnabled();
-
- /**
- * @return The period at which the coordinator cleans up unused segments. If the config isn't explicitly set,
- * it defaults to the coordinator's indexing period.
- */
- @Config("druid.coordinator.kill.period")
- public Duration getCoordinatorKillPeriod()
- {
- return getCoordinatorIndexingPeriod();
- }
-
- @Config("druid.coordinator.kill.durationToRetain")
- @Default("P90D")
- public abstract Duration getCoordinatorKillDurationToRetain();
-
- @Config("druid.coordinator.kill.ignoreDurationToRetain")
- @Default("false")
- public abstract boolean getCoordinatorKillIgnoreDurationToRetain();
-
- @Config("druid.coordinator.kill.bufferPeriod")
- @Default("P30D")
- public abstract Duration getCoordinatorKillBufferPeriod();
-
- @Config("druid.coordinator.kill.maxSegments")
- @Default("100")
- public abstract int getCoordinatorKillMaxSegments();
-
- @Config("druid.coordinator.kill.pendingSegments.on")
- @Default("true")
- public abstract boolean isKillPendingSegmentsEnabled();
-
- @Config("druid.coordinator.kill.supervisor.on")
- @Default("true")
- public abstract boolean isSupervisorKillEnabled();
-
- @Config("druid.coordinator.kill.supervisor.period")
- @Default("P1D")
- public abstract Duration getCoordinatorSupervisorKillPeriod();
-
- @Config("druid.coordinator.kill.supervisor.durationToRetain")
- @Default("P90D")
- public abstract Duration getCoordinatorSupervisorKillDurationToRetain();
-
- @Config("druid.coordinator.kill.audit.on")
- @Default("true")
- public abstract boolean isAuditKillEnabled();
-
- @Config("druid.coordinator.kill.audit.period")
- @Default("P1D")
- public abstract Duration getCoordinatorAuditKillPeriod();
-
- @Config("druid.coordinator.kill.audit.durationToRetain")
- @Default("P90D")
- public abstract Duration getCoordinatorAuditKillDurationToRetain();
-
- @Config("druid.coordinator.kill.compaction.on")
- @Default("false")
- public abstract boolean isCompactionKillEnabled();
-
- @Config("druid.coordinator.kill.compaction.period")
- @Default("P1D")
- public abstract Duration getCoordinatorCompactionKillPeriod();
-
- @Config("druid.coordinator.kill.rule.on")
- @Default("true")
- public abstract boolean isRuleKillEnabled();
-
- @Config("druid.coordinator.kill.rule.period")
- @Default("P1D")
- public abstract Duration getCoordinatorRuleKillPeriod();
-
- @Config("druid.coordinator.kill.rule.durationToRetain")
- @Default("P90D")
- public abstract Duration getCoordinatorRuleKillDurationToRetain();
-
- @Config("druid.coordinator.kill.datasource.on")
- @Default("true")
- public abstract boolean isDatasourceKillEnabled();
-
- @Config("druid.coordinator.kill.datasource.period")
- @Default("P1D")
- public abstract Duration getCoordinatorDatasourceKillPeriod();
-
- @Config("druid.coordinator.kill.datasource.durationToRetain")
- @Default("P90D")
- public abstract Duration getCoordinatorDatasourceKillDurationToRetain();
-
- @Config("druid.coordinator.load.timeout")
- public Duration getLoadTimeoutDelay()
- {
- return new Duration(15 * 60 * 1000);
- }
-
- @Config("druid.coordinator.loadqueuepeon.type")
- public String getLoadQueuePeonType()
- {
- return "http";
- }
-
- @Config("druid.coordinator.curator.loadqueuepeon.numCallbackThreads")
- public int getNumCuratorCallBackThreads()
- {
- return 2;
- }
-
- @Config("druid.coordinator.loadqueuepeon.http.repeatDelay")
- public Duration getHttpLoadQueuePeonRepeatDelay()
- {
- return Duration.millis(60000);
- }
-
- @Config("druid.coordinator.loadqueuepeon.http.hostTimeout")
- public Duration getHttpLoadQueuePeonHostTimeout()
- {
- return Duration.millis(300000);
- }
-
- @Config("druid.coordinator.loadqueuepeon.http.batchSize")
- public int getHttpLoadQueuePeonBatchSize()
- {
- return 1;
- }
-
- @Config("druid.coordinator.kill.segmentSchema.on")
- @Default("true")
- public abstract boolean isSegmentSchemaKillEnabled();
-
- @Config("druid.coordinator.kill.segmentSchema.period")
- @Default("PT1H")
- public abstract Duration getSegmentSchemaKillPeriod();
-
- @Config("druid.coordinator.kill.segmentSchema.durationToRetain")
- @Default("PT6H")
- public abstract Duration getSegmentSchemaKillDurationToRetain();
-}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorKillConfigs.java b/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorKillConfigs.java
new file mode 100644
index 00000000000..dfe08a00772
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorKillConfigs.java
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
+import org.joda.time.Duration;
+
+public class CoordinatorKillConfigs
+{
+ public static CoordinatorKillConfigs DEFAULT
+ = new CoordinatorKillConfigs(null, null, null, null, null, null, null, null, null, null, null, null, null);
+
+ @JsonProperty("supervisor")
+ private final MetadataCleanupConfig supervisors;
+
+ @JsonProperty("audit")
+ private final MetadataCleanupConfig auditLogs;
+
+ @JsonProperty("datasource")
+ private final MetadataCleanupConfig datasources;
+
+ @JsonProperty("rule")
+ private final MetadataCleanupConfig rules;
+
+ @JsonProperty("compaction")
+ private final MetadataCleanupConfig compactionConfigs;
+
+ @JsonProperty("pendingSegments")
+ private final MetadataCleanupConfig pendingSegments;
+
+ @JsonProperty("segmentSchema")
+ private final MetadataCleanupConfig segmentSchemas;
+
+ // Raw configs for killing unused segments
+ // These have been added as fields to make JsonConfigurator happy
+ @JsonProperty("on")
+ private final Boolean killUnusedEnabled;
+
+ @JsonProperty("period")
+ private final Duration killUnusedPeriod;
+
+ @JsonProperty("durationToRetain")
+ private final Duration killUnusedDurationToRetain;
+
+ @JsonProperty("ignoreDurationToRetain")
+ private final Boolean killUnusedIgnoreDurationToRetain;
+
+ @JsonProperty("bufferPeriod")
+ private final Duration killUnusedBufferPeriod;
+
+ @JsonProperty("maxSegments")
+ private final Integer killUnusedMaxSegments;
+
+ @JsonCreator
+ public CoordinatorKillConfigs(
+ @JsonProperty("pendingSegments") MetadataCleanupConfig pendingSegments,
+ @JsonProperty("supervisor") MetadataCleanupConfig supervisors,
+ @JsonProperty("audit") MetadataCleanupConfig auditLogs,
+ @JsonProperty("datasource") MetadataCleanupConfig datasources,
+ @JsonProperty("rule") MetadataCleanupConfig rules,
+ @JsonProperty("compaction") MetadataCleanupConfig compactionConfigs,
+ @JsonProperty("segmentSchema") MetadataCleanupConfig segmentSchemas,
+ // Configs for cleanup of unused segments
+ @JsonProperty("on") Boolean killUnusedEnabled,
+ @JsonProperty("period") Duration killUnusedPeriod,
+ @JsonProperty("durationToRetain") Duration killUnusedDurationToRetain,
+ @JsonProperty("ignoreDurationToRetain") Boolean killUnusedIgnoreDurationToRetain,
+ @JsonProperty("bufferPeriod") Duration killUnusedBufferPeriod,
+ @JsonProperty("maxSegments") Integer killUnusedMaxSegments
+ )
+ {
+ this.pendingSegments = Configs.valueOrDefault(pendingSegments, MetadataCleanupConfig.DEFAULT);
+ this.supervisors = Configs.valueOrDefault(supervisors, MetadataCleanupConfig.DEFAULT);
+ this.auditLogs = Configs.valueOrDefault(auditLogs, MetadataCleanupConfig.DEFAULT);
+ this.datasources = Configs.valueOrDefault(datasources, MetadataCleanupConfig.DEFAULT);
+ this.rules = Configs.valueOrDefault(rules, MetadataCleanupConfig.DEFAULT);
+ this.compactionConfigs = Configs.valueOrDefault(compactionConfigs, MetadataCleanupConfig.DEFAULT);
+ this.segmentSchemas = Configs.valueOrDefault(segmentSchemas, MetadataCleanupConfig.DEFAULT);
+
+ this.killUnusedEnabled = killUnusedEnabled;
+ this.killUnusedPeriod = killUnusedPeriod;
+ this.killUnusedDurationToRetain = killUnusedDurationToRetain;
+ this.killUnusedBufferPeriod = killUnusedBufferPeriod;
+ this.killUnusedIgnoreDurationToRetain = killUnusedIgnoreDurationToRetain;
+ this.killUnusedMaxSegments = killUnusedMaxSegments;
+ }
+
+ public MetadataCleanupConfig auditLogs()
+ {
+ return auditLogs;
+ }
+
+ public MetadataCleanupConfig datasources()
+ {
+ return datasources;
+ }
+
+ public MetadataCleanupConfig rules()
+ {
+ return rules;
+ }
+
+ public MetadataCleanupConfig compactionConfigs()
+ {
+ return compactionConfigs;
+ }
+
+ public MetadataCleanupConfig pendingSegments()
+ {
+ return pendingSegments;
+ }
+
+ public MetadataCleanupConfig supervisors()
+ {
+ return supervisors;
+ }
+
+ public MetadataCleanupConfig segmentSchemas()
+ {
+ return segmentSchemas;
+ }
+
+ /**
+ * Creates a KillUnusedSegmentsConfig. This config is initialized lazily as
+ * it uses the indexingPeriod as the default cleanup period.
+ */
+ public KillUnusedSegmentsConfig unusedSegments(Duration indexingPeriod)
+ {
+ return new KillUnusedSegmentsConfig(
+ killUnusedEnabled,
+ Configs.valueOrDefault(killUnusedPeriod, indexingPeriod),
+ killUnusedDurationToRetain,
+ killUnusedIgnoreDurationToRetain,
+ killUnusedBufferPeriod,
+ killUnusedMaxSegments
+ );
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorPeriodConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorPeriodConfig.java
new file mode 100644
index 00000000000..5deba3da3a0
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorPeriodConfig.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
+import org.joda.time.Duration;
+
+public class CoordinatorPeriodConfig
+{
+ @JsonProperty
+ private final Duration metadataStoreManagementPeriod;
+
+ @JsonProperty
+ private final Duration indexingPeriod;
+
+ @JsonCreator
+ public CoordinatorPeriodConfig(
+ @JsonProperty("metadataStoreManagementPeriod") Duration metadataStoreManagementPeriod,
+ @JsonProperty("indexingPeriod") Duration indexingPeriod
+ )
+ {
+ this.indexingPeriod = Configs.valueOrDefault(indexingPeriod, Duration.standardMinutes(30));
+ this.metadataStoreManagementPeriod = Configs.valueOrDefault(
+ metadataStoreManagementPeriod,
+ Duration.standardMinutes(60)
+ );
+ }
+
+ public Duration getMetadataStoreManagementPeriod()
+ {
+ return metadataStoreManagementPeriod;
+ }
+
+ public Duration getIndexingPeriod()
+ {
+ return indexingPeriod;
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorRunConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorRunConfig.java
new file mode 100644
index 00000000000..b35527e71ac
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/coordinator/config/CoordinatorRunConfig.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
+import org.joda.time.Duration;
+
+public class CoordinatorRunConfig
+{
+ @JsonProperty
+ private final Duration startDelay;
+
+ @JsonProperty
+ private final Duration period;
+
+ @JsonCreator
+ public CoordinatorRunConfig(
+ @JsonProperty("startDelay") Duration startDelay,
+ @JsonProperty("period") Duration period
+ )
+ {
+ this.startDelay = Configs.valueOrDefault(startDelay, Duration.standardMinutes(5));
+ this.period = Configs.valueOrDefault(period, Duration.standardMinutes(1));
+ }
+
+ public Duration getPeriod()
+ {
+ return period;
+ }
+
+ public Duration getStartDelay()
+ {
+ return startDelay;
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/config/DruidCoordinatorConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/config/DruidCoordinatorConfig.java
new file mode 100644
index 00000000000..6004c5b1ba4
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/coordinator/config/DruidCoordinatorConfig.java
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.config;
+
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.server.coordinator.balancer.BalancerStrategyFactory;
+import org.joda.time.Duration;
+
+/**
+ * Contains all static configs for the Coordinator.
+ */
+public class DruidCoordinatorConfig
+{
+ private final CoordinatorRunConfig runConfig;
+ private final CoordinatorPeriodConfig periodConfig;
+ private final CoordinatorKillConfigs killConfigs;
+ private final BalancerStrategyFactory balancerStrategyFactory;
+ private final HttpLoadQueuePeonConfig httpLoadQueuePeonConfig;
+
+ @Inject
+ public DruidCoordinatorConfig(
+ CoordinatorRunConfig runConfig,
+ CoordinatorPeriodConfig periodConfig,
+ CoordinatorKillConfigs killConfigs,
+ BalancerStrategyFactory balancerStrategyFactory,
+ HttpLoadQueuePeonConfig httpLoadQueuePeonConfig
+ )
+ {
+ this.killConfigs = killConfigs;
+ this.runConfig = runConfig;
+ this.periodConfig = periodConfig;
+ this.balancerStrategyFactory = balancerStrategyFactory;
+ this.httpLoadQueuePeonConfig = httpLoadQueuePeonConfig;
+
+ validateKillConfigs();
+ }
+
+ public Duration getCoordinatorStartDelay()
+ {
+ return runConfig.getStartDelay();
+ }
+
+ public Duration getCoordinatorPeriod()
+ {
+ return runConfig.getPeriod();
+ }
+
+ public Duration getCoordinatorIndexingPeriod()
+ {
+ return periodConfig.getIndexingPeriod();
+ }
+
+ public Duration getCoordinatorMetadataStoreManagementPeriod()
+ {
+ return periodConfig.getMetadataStoreManagementPeriod();
+ }
+
+ public CoordinatorKillConfigs getKillConfigs()
+ {
+ return killConfigs;
+ }
+
+ public BalancerStrategyFactory getBalancerStrategyFactory()
+ {
+ return balancerStrategyFactory;
+ }
+
+ public HttpLoadQueuePeonConfig getHttpLoadQueuePeonConfig()
+ {
+ return httpLoadQueuePeonConfig;
+ }
+
+ private void validateKillConfigs()
+ {
+ validateKillConfig(killConfigs.auditLogs(), "audit");
+ validateKillConfig(killConfigs.compactionConfigs(), "compaction");
+ validateKillConfig(killConfigs.datasources(), "datasource");
+ validateKillConfig(killConfigs.rules(), "rule");
+ validateKillConfig(killConfigs.supervisors(), "supervisor");
+ validateKillConfig(killConfigs.segmentSchemas(), "segmentSchema");
+
+ // Validate config for killing unused segments
+ final KillUnusedSegmentsConfig killUnusedConfig
+ = killConfigs.unusedSegments(getCoordinatorIndexingPeriod());
+ if (killUnusedConfig.getCleanupPeriod().getMillis() < getCoordinatorIndexingPeriod().getMillis()) {
+ throw newInvalidInputExceptionForOperator(
+ "'druid.coordinator.kill.period'[%s] must be greater than or equal to"
+ + " 'druid.coordinator.period.indexingPeriod'[%s]",
+ killUnusedConfig.getCleanupPeriod(), getCoordinatorIndexingPeriod()
+ );
+ }
+ if (killUnusedConfig.getMaxSegments() < 0) {
+ throw newInvalidInputExceptionForOperator(
+ "'druid.coordinator.kill.maxSegments'[%s] must be a positive integer.",
+ killUnusedConfig.getMaxSegments()
+ );
+ }
+ }
+
+ private void validateKillConfig(MetadataCleanupConfig config, String propertyPrefix)
+ {
+ if (!config.isCleanupEnabled()) {
+ // Do not perform validation if cleanup is disabled
+ return;
+ }
+
+ final Duration metadataManagementPeriod = getCoordinatorMetadataStoreManagementPeriod();
+ final Duration period = config.getCleanupPeriod();
+ if (period == null || period.getMillis() < metadataManagementPeriod.getMillis()) {
+ throw newInvalidInputExceptionForOperator(
+ "'druid.coordinator.kill.%s.period'[%s] must be greater than"
+ + " 'druid.coordinator.period.metadataStoreManagementPeriod'[%s]",
+ propertyPrefix, period, metadataManagementPeriod
+ );
+ }
+
+ final Duration retainDuration = config.getDurationToRetain();
+ if (retainDuration == null || retainDuration.getMillis() < 0) {
+ throw newInvalidInputExceptionForOperator(
+ "'druid.coordinator.kill.%s.durationToRetain'[%s] must be 0 milliseconds or higher",
+ propertyPrefix, retainDuration
+ );
+ }
+ if (retainDuration.getMillis() > System.currentTimeMillis()) {
+ throw newInvalidInputExceptionForOperator(
+ "'druid.coordinator.kill.%s.durationToRetain'[%s] cannot be greater than current time in milliseconds",
+ propertyPrefix, retainDuration
+ );
+ }
+ }
+
+ private static DruidException newInvalidInputExceptionForOperator(String msgFormat, Object... args)
+ {
+ return DruidException.forPersona(DruidException.Persona.OPERATOR)
+ .ofCategory(DruidException.Category.INVALID_INPUT)
+ .build(msgFormat, args);
+ }
+
+}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/config/HttpLoadQueuePeonConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/config/HttpLoadQueuePeonConfig.java
new file mode 100644
index 00000000000..f6f402037d1
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/coordinator/config/HttpLoadQueuePeonConfig.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
+import org.joda.time.Duration;
+
+public class HttpLoadQueuePeonConfig
+{
+ private static final Duration DEFAULT_LOAD_TIMEOUT = Duration.standardMinutes(15);
+
+ @JsonProperty
+ private final Duration hostTimeout;
+
+ @JsonProperty
+ private final Duration repeatDelay;
+
+ @JsonProperty
+ private final int batchSize;
+
+ @JsonCreator
+ public HttpLoadQueuePeonConfig(
+ @JsonProperty("hostTimeout") Duration hostTimeout,
+ @JsonProperty("repeatDelay") Duration repeatDelay,
+ @JsonProperty("batchSize") Integer batchSize
+ )
+ {
+ this.hostTimeout = Configs.valueOrDefault(hostTimeout, Duration.standardMinutes(5));
+ this.repeatDelay = Configs.valueOrDefault(repeatDelay, Duration.standardMinutes(1));
+ this.batchSize = Configs.valueOrDefault(batchSize, 1);
+ }
+
+ public int getBatchSize()
+ {
+ return batchSize;
+ }
+
+ public Duration getHostTimeout()
+ {
+ return hostTimeout;
+ }
+
+ public Duration getRepeatDelay()
+ {
+ return repeatDelay;
+ }
+
+ public Duration getLoadTimeout()
+ {
+ return DEFAULT_LOAD_TIMEOUT;
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/config/KillUnusedSegmentsConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/config/KillUnusedSegmentsConfig.java
new file mode 100644
index 00000000000..cbf8cc55bf4
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/coordinator/config/KillUnusedSegmentsConfig.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.config;
+
+import org.apache.druid.common.config.Configs;
+import org.joda.time.Duration;
+
+public class KillUnusedSegmentsConfig extends MetadataCleanupConfig
+{
+ private final int maxSegments;
+ private final boolean ignoreDurationToRetain;
+ private final Duration bufferPeriod;
+
+ public KillUnusedSegmentsConfig(
+ Boolean cleanupEnabled,
+ Duration cleanupPeriod,
+ Duration durationToRetain,
+ Boolean ignoreDurationToRetain,
+ Duration bufferPeriod,
+ Integer maxSegments
+ )
+ {
+ super(
+ Configs.valueOrDefault(cleanupEnabled, false),
+ cleanupPeriod,
+ durationToRetain
+ );
+ this.ignoreDurationToRetain = Configs.valueOrDefault(ignoreDurationToRetain, false);
+ this.bufferPeriod = Configs.valueOrDefault(bufferPeriod, Duration.standardDays(30));
+ this.maxSegments = Configs.valueOrDefault(maxSegments, 100);
+ }
+
+ public int getMaxSegments()
+ {
+ return maxSegments;
+ }
+
+ public Duration getBufferPeriod()
+ {
+ return bufferPeriod;
+ }
+
+ public boolean isIgnoreDurationToRetain()
+ {
+ return ignoreDurationToRetain;
+ }
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ public static class Builder
+ {
+ private Duration durationToRetain;
+ private Duration cleanupPeriod;
+ private Boolean ignoreDurationToRetain;
+ private Duration bufferPeriod;
+ private Integer maxSegments;
+
+ private Builder()
+ {
+
+ }
+
+ public KillUnusedSegmentsConfig build()
+ {
+ return new KillUnusedSegmentsConfig(
+ true,
+ cleanupPeriod,
+ durationToRetain,
+ ignoreDurationToRetain,
+ bufferPeriod,
+ maxSegments
+ );
+ }
+
+ public Builder withCleanupPeriod(Duration cleanupPeriod)
+ {
+ this.cleanupPeriod = cleanupPeriod;
+ return this;
+ }
+
+ public Builder withDurationToRetain(Duration durationToRetain)
+ {
+ this.durationToRetain = durationToRetain;
+ return this;
+ }
+
+ public Builder withIgnoreDurationToRetain(Boolean ignoreDurationToRetain)
+ {
+ this.ignoreDurationToRetain = ignoreDurationToRetain;
+ return this;
+ }
+
+ public Builder withMaxSegmentsToKill(Integer maxSegments)
+ {
+ this.maxSegments = maxSegments;
+ return this;
+ }
+
+ public Builder withBufferPeriod(Duration bufferPeriod)
+ {
+ this.bufferPeriod = bufferPeriod;
+ return this;
+ }
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/config/MetadataCleanupConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/config/MetadataCleanupConfig.java
new file mode 100644
index 00000000000..c1e1a956fc2
--- /dev/null
+++ b/server/src/main/java/org/apache/druid/server/coordinator/config/MetadataCleanupConfig.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.coordinator.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
+import org.joda.time.Duration;
+
+import java.util.Objects;
+
+public class MetadataCleanupConfig
+{
+ public static final MetadataCleanupConfig DEFAULT = new MetadataCleanupConfig(null, null, null);
+
+ @JsonProperty("on")
+ private final boolean cleanupEnabled;
+
+ @JsonProperty("period")
+ private final Duration cleanupPeriod;
+
+ @JsonProperty("durationToRetain")
+ private final Duration durationToRetain;
+
+ @JsonCreator
+ public MetadataCleanupConfig(
+ @JsonProperty("on") Boolean cleanupEnabled,
+ @JsonProperty("period") Duration cleanupPeriod,
+ @JsonProperty("durationToRetain") Duration durationToRetain
+ )
+ {
+ this.cleanupEnabled = Configs.valueOrDefault(cleanupEnabled, true);
+ this.cleanupPeriod = Configs.valueOrDefault(cleanupPeriod, Duration.standardDays(1));
+ this.durationToRetain = Configs.valueOrDefault(durationToRetain, Duration.standardDays(90));
+ }
+
+ public Duration getCleanupPeriod()
+ {
+ return cleanupPeriod;
+ }
+
+ public Duration getDurationToRetain()
+ {
+ return durationToRetain;
+ }
+
+ public boolean isCleanupEnabled()
+ {
+ return cleanupEnabled;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ MetadataCleanupConfig that = (MetadataCleanupConfig) o;
+ return cleanupEnabled == that.cleanupEnabled
+ && Objects.equals(cleanupPeriod, that.cleanupPeriod)
+ && Objects.equals(durationToRetain, that.durationToRetain);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(cleanupEnabled, cleanupPeriod, durationToRetain);
+ }
+}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillAuditLog.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillAuditLog.java
index e5248d48d7e..3d3e58edfa8 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillAuditLog.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillAuditLog.java
@@ -20,7 +20,7 @@
package org.apache.druid.server.coordinator.duty;
import org.apache.druid.audit.AuditManager;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.DateTime;
@@ -28,17 +28,9 @@ public class KillAuditLog extends MetadataCleanupDuty
{
private final AuditManager auditManager;
- public KillAuditLog(DruidCoordinatorConfig config, AuditManager auditManager)
+ public KillAuditLog(MetadataCleanupConfig config, AuditManager auditManager)
{
- super(
- "audit logs",
- "druid.coordinator.kill.audit",
- config.isAuditKillEnabled(),
- config.getCoordinatorAuditKillPeriod(),
- config.getCoordinatorAuditKillDurationToRetain(),
- Stats.Kill.AUDIT_LOGS,
- config
- );
+ super("audit logs", config, Stats.Kill.AUDIT_LOGS);
this.auditManager = auditManager;
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillCompactionConfig.java
index 304b54e7cf8..c231c31c543 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillCompactionConfig.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillCompactionConfig.java
@@ -29,10 +29,9 @@ import org.apache.druid.metadata.SegmentsMetadataManager;
import org.apache.druid.server.coordinator.CoordinatorCompactionConfig;
import org.apache.druid.server.coordinator.CoordinatorConfigManager;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.DateTime;
-import org.joda.time.Duration;
import java.util.Map;
import java.util.Set;
@@ -54,20 +53,12 @@ public class KillCompactionConfig extends MetadataCleanupDuty
private final CoordinatorConfigManager configManager;
public KillCompactionConfig(
- DruidCoordinatorConfig config,
+ MetadataCleanupConfig config,
SegmentsMetadataManager sqlSegmentsMetadataManager,
CoordinatorConfigManager configManager
)
{
- super(
- "compaction configs",
- "druid.coordinator.kill.compaction",
- config.isCompactionKillEnabled(),
- config.getCoordinatorCompactionKillPeriod(),
- Duration.millis(1), // Retain duration is ignored
- Stats.Kill.COMPACTION_CONFIGS,
- config
- );
+ super("compaction configs", config, Stats.Kill.COMPACTION_CONFIGS);
this.sqlSegmentsMetadataManager = sqlSegmentsMetadataManager;
this.configManager = configManager;
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadata.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadata.java
index 9f2d15e1d33..8e31e503d49 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadata.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadata.java
@@ -23,7 +23,7 @@ import com.google.common.base.Strings;
import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
import org.apache.druid.metadata.MetadataSupervisorManager;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.DateTime;
@@ -44,20 +44,12 @@ public class KillDatasourceMetadata extends MetadataCleanupDuty
private final MetadataSupervisorManager metadataSupervisorManager;
public KillDatasourceMetadata(
- DruidCoordinatorConfig config,
+ MetadataCleanupConfig config,
IndexerMetadataStorageCoordinator indexerMetadataStorageCoordinator,
MetadataSupervisorManager metadataSupervisorManager
)
{
- super(
- "datasources",
- "druid.coordinator.kill.datasource",
- config.isDatasourceKillEnabled(),
- config.getCoordinatorDatasourceKillPeriod(),
- config.getCoordinatorDatasourceKillDurationToRetain(),
- Stats.Kill.DATASOURCES,
- config
- );
+ super("datasources", config, Stats.Kill.DATASOURCES);
this.indexerMetadataStorageCoordinator = indexerMetadataStorageCoordinator;
this.metadataSupervisorManager = metadataSupervisorManager;
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillRules.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillRules.java
index 6b4693bdac9..a2c56a6e17b 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillRules.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillRules.java
@@ -20,7 +20,7 @@
package org.apache.druid.server.coordinator.duty;
import org.apache.druid.metadata.MetadataRuleManager;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.DateTime;
@@ -29,19 +29,11 @@ public class KillRules extends MetadataCleanupDuty
private final MetadataRuleManager metadataRuleManager;
public KillRules(
- DruidCoordinatorConfig config,
+ MetadataCleanupConfig config,
MetadataRuleManager metadataRuleManager
)
{
- super(
- "rules",
- "druid.coordinator.kill.rule",
- config.isRuleKillEnabled(),
- config.getCoordinatorRuleKillPeriod(),
- config.getCoordinatorRuleKillDurationToRetain(),
- Stats.Kill.RULES,
- config
- );
+ super("rules", config, Stats.Kill.RULES);
this.metadataRuleManager = metadataRuleManager;
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java
index 8a41e4b8f20..7f9065f7062 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java
@@ -20,7 +20,7 @@
package org.apache.druid.server.coordinator.duty;
import org.apache.druid.metadata.MetadataSupervisorManager;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.DateTime;
@@ -32,19 +32,11 @@ public class KillSupervisors extends MetadataCleanupDuty
private final MetadataSupervisorManager metadataSupervisorManager;
public KillSupervisors(
- DruidCoordinatorConfig config,
+ MetadataCleanupConfig config,
MetadataSupervisorManager metadataSupervisorManager
)
{
- super(
- "supervisors",
- "druid.coordinator.kill.supervisor",
- config.isSupervisorKillEnabled(),
- config.getCoordinatorSupervisorKillPeriod(),
- config.getCoordinatorSupervisorKillDurationToRetain(),
- Stats.Kill.SUPERVISOR_SPECS,
- config
- );
+ super("supervisors", config, Stats.Kill.SUPERVISOR_SPECS);
this.metadataSupervisorManager = metadataSupervisorManager;
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisorsCustomDuty.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisorsCustomDuty.java
index 9cdf8010718..9069a48bc21 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisorsCustomDuty.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisorsCustomDuty.java
@@ -25,9 +25,8 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.druid.guice.annotations.UnstableApi;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.metadata.MetadataSupervisorManager;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
-import org.apache.druid.server.coordinator.stats.Stats;
-import org.joda.time.DateTime;
+import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.joda.time.Duration;
/**
@@ -36,45 +35,34 @@ import org.joda.time.Duration;
* as {@link KillSupervisors} but uses a different configuration style as
* detailed in {@link CoordinatorCustomDuty}.
* <p>
- * This duty is only an example to demostrate the usage of coordinator custom
+ * This duty is only an example to demonstrate the usage of coordinator custom
* duties. All production clusters should continue using {@link KillSupervisors}.
- * <p>
- * In the future, we might migrate all metadata management coordinator duties to
- * {@link CoordinatorCustomDuty} but until then this class will remain undocumented.
*/
@UnstableApi
-public class KillSupervisorsCustomDuty extends MetadataCleanupDuty implements CoordinatorCustomDuty
+public class KillSupervisorsCustomDuty implements CoordinatorCustomDuty
{
private static final Logger log = new Logger(KillSupervisorsCustomDuty.class);
- private final MetadataSupervisorManager metadataSupervisorManager;
+ private final KillSupervisors delegate;
@JsonCreator
public KillSupervisorsCustomDuty(
@JsonProperty("durationToRetain") Duration retainDuration,
- @JacksonInject MetadataSupervisorManager metadataSupervisorManager,
- @JacksonInject DruidCoordinatorConfig coordinatorConfig
+ @JacksonInject MetadataSupervisorManager metadataSupervisorManager
)
{
- super(
- "supervisors",
- "KillSupervisorsCustomDuty",
- true,
- // Use the same period as metadata store management so that validation passes
- // Actual period of custom duties is configured by the user
- coordinatorConfig.getCoordinatorMetadataStoreManagementPeriod(),
- retainDuration,
- Stats.Kill.SUPERVISOR_SPECS,
- coordinatorConfig
+ this.delegate = new KillSupervisors(
+ // Pass period as zero here, actual period of custom duties is configured at the duty group level
+ new MetadataCleanupConfig(true, Duration.ZERO, retainDuration),
+ metadataSupervisorManager
);
- this.metadataSupervisorManager = metadataSupervisorManager;
log.warn("This is only an example implementation of a custom duty and"
+ " must not be used in production. Use KillSupervisors duty instead.");
}
@Override
- protected int cleanupEntriesCreatedBefore(DateTime minCreatedTime)
+ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
{
- return metadataSupervisorManager.removeTerminatedSupervisorsOlderThan(minCreatedTime.getMillis());
+ return delegate.run(params);
}
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnreferencedSegmentSchemaDuty.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnreferencedSegmentSchemaDuty.java
index 0c5ca6ad973..82ad4cedc77 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnreferencedSegmentSchemaDuty.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnreferencedSegmentSchemaDuty.java
@@ -21,7 +21,7 @@ package org.apache.druid.server.coordinator.duty;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.segment.metadata.SegmentSchemaManager;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.DateTime;
@@ -55,19 +55,11 @@ public class KillUnreferencedSegmentSchemaDuty extends MetadataCleanupDuty
private final SegmentSchemaManager segmentSchemaManager;
public KillUnreferencedSegmentSchemaDuty(
- DruidCoordinatorConfig config,
+ MetadataCleanupConfig config,
SegmentSchemaManager segmentSchemaManager
)
{
- super(
- "segmentSchema",
- "druid.coordinator.kill.segmentSchema",
- config.isSegmentSchemaKillEnabled(),
- config.getSegmentSchemaKillPeriod(),
- config.getSegmentSchemaKillDurationToRetain(),
- Stats.Kill.RULES,
- config
- );
+ super("segmentSchema", config, Stats.Kill.SEGMENT_SCHEMA);
this.segmentSchemaManager = segmentSchemaManager;
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnusedSegments.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnusedSegments.java
index 1d7876da387..575c320b9e0 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnusedSegments.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnusedSegments.java
@@ -21,17 +21,15 @@ package org.apache.druid.server.coordinator.duty;
import com.google.common.base.Predicate;
import org.apache.druid.common.guava.FutureUtils;
-import org.apache.druid.error.DruidException;
import org.apache.druid.indexer.TaskStatusPlus;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.JodaUtils;
-import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.metadata.SegmentsMetadataManager;
import org.apache.druid.rpc.indexing.OverlordClient;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+import org.apache.druid.server.coordinator.config.KillUnusedSegmentsConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Dimension;
import org.apache.druid.server.coordinator.stats.RowKey;
@@ -84,50 +82,31 @@ public class KillUnusedSegments implements CoordinatorDuty
*/
private final Map<String, DateTime> datasourceToLastKillIntervalEnd;
private DateTime lastKillTime;
- private final long bufferPeriod;
+ private final Duration bufferPeriod;
private final SegmentsMetadataManager segmentsMetadataManager;
private final OverlordClient overlordClient;
public KillUnusedSegments(
- final SegmentsMetadataManager segmentsMetadataManager,
- final OverlordClient overlordClient,
- final DruidCoordinatorConfig config
+ SegmentsMetadataManager segmentsMetadataManager,
+ OverlordClient overlordClient,
+ KillUnusedSegmentsConfig killConfig
)
{
- if (config.getCoordinatorKillPeriod().getMillis() < config.getCoordinatorIndexingPeriod().getMillis()) {
- throw DruidException.forPersona(DruidException.Persona.OPERATOR)
- .ofCategory(DruidException.Category.INVALID_INPUT)
- .build(
- StringUtils.format(
- "druid.coordinator.kill.period[%s] is invalid. It must be greater than or "
- + "equal to druid.coordinator.period.indexingPeriod[%s].",
- config.getCoordinatorKillPeriod(),
- config.getCoordinatorIndexingPeriod()
- )
- );
- }
- if (config.getCoordinatorKillMaxSegments() < 0) {
- throw DruidException.forPersona(DruidException.Persona.OPERATOR)
- .ofCategory(DruidException.Category.INVALID_INPUT)
- .build(StringUtils.format(
- "druid.coordinator.kill.maxSegments[%d] is invalid. It must be a positive integer.",
- config.getCoordinatorKillMaxSegments()
- )
- );
- }
- this.period = config.getCoordinatorKillPeriod();
- this.ignoreDurationToRetain = config.getCoordinatorKillIgnoreDurationToRetain();
- this.durationToRetain = config.getCoordinatorKillDurationToRetain();
+ this.period = killConfig.getCleanupPeriod();
+
+ this.maxSegmentsToKill = killConfig.getMaxSegments();
+ this.ignoreDurationToRetain = killConfig.isIgnoreDurationToRetain();
+ this.durationToRetain = killConfig.getDurationToRetain();
if (this.ignoreDurationToRetain) {
log.info(
"druid.coordinator.kill.durationToRetain[%s] will be ignored when discovering segments to kill "
+ "because druid.coordinator.kill.ignoreDurationToRetain is set to true.",
- this.durationToRetain
+ durationToRetain
);
}
- this.bufferPeriod = config.getCoordinatorKillBufferPeriod().getMillis();
- this.maxSegmentsToKill = config.getCoordinatorKillMaxSegments();
+ this.bufferPeriod = killConfig.getBufferPeriod();
+
datasourceToLastKillIntervalEnd = new ConcurrentHashMap<>();
log.info(
@@ -145,16 +124,15 @@ public class KillUnusedSegments implements CoordinatorDuty
@Override
public DruidCoordinatorRuntimeParams run(final DruidCoordinatorRuntimeParams params)
{
- if (!canDutyRun()) {
+ if (canDutyRun()) {
+ return runInternal(params);
+ } else {
log.debug(
"Skipping KillUnusedSegments until period[%s] has elapsed after lastKillTime[%s].",
- period,
- lastKillTime
+ period, lastKillTime
);
return params;
}
-
- return runInternal(params);
}
private DruidCoordinatorRuntimeParams runInternal(final DruidCoordinatorRuntimeParams params)
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/MetadataCleanupDuty.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/MetadataCleanupDuty.java
index 8514a7f7177..e24f9396689 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/duty/MetadataCleanupDuty.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/MetadataCleanupDuty.java
@@ -19,21 +19,19 @@
package org.apache.druid.server.coordinator.duty;
-import com.google.common.base.Preconditions;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.logger.Logger;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorStat;
import org.joda.time.DateTime;
-import org.joda.time.Duration;
import javax.annotation.Nullable;
/**
* Performs cleanup of stale metadata entries created before a configured retain duration.
* <p>
- * In every invocation of {@link #run}, the duty checks if the {@link #cleanupPeriod}
+ * In every invocation of {@link #run}, the duty checks if the {@code cleanupPeriod}
* has elapsed since the {@link #lastCleanupTime}. If it has, then the method
* {@link #cleanupEntriesCreatedBefore(DateTime)} is invoked. Otherwise, the duty
* completes immediately without making any changes.
@@ -42,40 +40,27 @@ public abstract class MetadataCleanupDuty implements CoordinatorDuty
{
private static final Logger log = new Logger(MetadataCleanupDuty.class);
- private final String propertyPrefix;
private final String entryType;
private final CoordinatorStat cleanupCountStat;
- private final boolean cleanupEnabled;
- private final Duration cleanupPeriod;
- private final Duration retainDuration;
+ private final MetadataCleanupConfig cleanupConfig;
private DateTime lastCleanupTime = DateTimes.utc(0);
protected MetadataCleanupDuty(
String entryType,
- String propertyPrefix,
- boolean cleanupEnabled,
- Duration cleanupPeriod,
- Duration retainDuration,
- CoordinatorStat cleanupCountStat,
- DruidCoordinatorConfig coordinatorConfig
+ MetadataCleanupConfig cleanupConfig,
+ CoordinatorStat cleanupCountStat
)
{
- this.propertyPrefix = propertyPrefix;
this.entryType = entryType;
- this.cleanupEnabled = cleanupEnabled;
- this.cleanupPeriod = cleanupPeriod;
- this.retainDuration = retainDuration;
+ this.cleanupConfig = cleanupConfig;
this.cleanupCountStat = cleanupCountStat;
- if (this.cleanupEnabled) {
- validatePeriod(cleanupPeriod, coordinatorConfig.getCoordinatorMetadataStoreManagementPeriod());
- validateRetainDuration(retainDuration);
-
+ if (cleanupConfig.isCleanupEnabled()) {
log.debug(
"Enabled cleanup of [%s] with period [%s] and durationToRetain [%s].",
- entryType, cleanupPeriod, retainDuration
+ entryType, cleanupConfig.getCleanupPeriod(), cleanupConfig.getDurationToRetain()
);
}
}
@@ -84,18 +69,18 @@ public abstract class MetadataCleanupDuty implements CoordinatorDuty
@Override
public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
{
- if (!cleanupEnabled) {
+ if (!cleanupConfig.isCleanupEnabled()) {
return params;
}
final DateTime now = getCurrentTime();
// Perform cleanup only if cleanup period has elapsed
- if (lastCleanupTime.plus(cleanupPeriod).isBefore(now)) {
+ if (lastCleanupTime.plus(cleanupConfig.getCleanupPeriod()).isBefore(now)) {
lastCleanupTime = now;
try {
- DateTime minCreatedTime = now.minus(retainDuration);
+ DateTime minCreatedTime = now.minus(cleanupConfig.getDurationToRetain());
int deletedEntries = cleanupEntriesCreatedBefore(minCreatedTime);
log.info("Removed [%,d] [%s] created before [%s].", deletedEntries, entryType, minCreatedTime);
@@ -112,38 +97,16 @@ public abstract class MetadataCleanupDuty implements CoordinatorDuty
/**
* Cleans up metadata entries created before the {@code minCreatedTime}.
* <p>
- * This method is not invoked if the {@link #cleanupPeriod} has not elapsed
+ * This method is not invoked if the {@code cleanupPeriod} has not elapsed
* since the {@link #lastCleanupTime}.
*
* @return Number of deleted metadata entries
*/
protected abstract int cleanupEntriesCreatedBefore(DateTime minCreatedTime);
- private void validatePeriod(Duration period, Duration metadataManagementPeriod)
- {
- Preconditions.checkArgument(
- period != null && period.getMillis() >= metadataManagementPeriod.getMillis(),
- "[%s.period] must be greater than [druid.coordinator.period.metadataStoreManagementPeriod]",
- propertyPrefix
- );
- }
-
- private void validateRetainDuration(Duration retainDuration)
- {
- Preconditions.checkArgument(
- retainDuration != null && retainDuration.getMillis() >= 0,
- "[%s.durationToRetain] must be 0 milliseconds or higher",
- propertyPrefix
- );
- Preconditions.checkArgument(
- retainDuration.getMillis() < System.currentTimeMillis(),
- "[%s.durationToRetain] cannot be greater than current time in milliseconds",
- propertyPrefix
- );
- }
-
protected DateTime getCurrentTime()
{
return DateTimes.nowUtc();
}
+
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeon.java b/server/src/main/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeon.java
index 7ea78a132fe..a508251ab58 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeon.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeon.java
@@ -28,13 +28,13 @@ import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.server.coordination.SegmentChangeRequestNoop;
import org.apache.druid.server.coordinator.DruidCoordinator;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.apache.druid.timeline.DataSegment;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
+import org.joda.time.Duration;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -72,7 +72,7 @@ public class CuratorLoadQueuePeon implements LoadQueuePeon
* with loading or dropping segments.
*/
private final ExecutorService callBackExecutor;
- private final DruidCoordinatorConfig config;
+ private final Duration loadTimeout;
private final AtomicLong queuedSize = new AtomicLong(0);
private final CoordinatorRunStats stats = new CoordinatorRunStats();
@@ -116,7 +116,7 @@ public class CuratorLoadQueuePeon implements LoadQueuePeon
ObjectMapper jsonMapper,
ScheduledExecutorService processingExecutor,
ExecutorService callbackExecutor,
- DruidCoordinatorConfig config
+ Duration loadTimeout
)
{
this.curator = curator;
@@ -124,7 +124,7 @@ public class CuratorLoadQueuePeon implements LoadQueuePeon
this.jsonMapper = jsonMapper;
this.callBackExecutor = callbackExecutor;
this.processingExecutor = processingExecutor;
- this.config = config;
+ this.loadTimeout = loadTimeout;
}
@JsonProperty
@@ -309,7 +309,7 @@ public class CuratorLoadQueuePeon implements LoadQueuePeon
failAssign(segmentHolder, false, e);
}
},
- config.getLoadTimeoutDelay().getMillis(),
+ loadTimeout.getMillis(),
TimeUnit.MILLISECONDS
);
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java b/server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java
index 92dd4714a45..4ec6f2427a3 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeon.java
@@ -37,7 +37,7 @@ import org.apache.druid.server.coordination.DataSegmentChangeHandler;
import org.apache.druid.server.coordination.DataSegmentChangeRequest;
import org.apache.druid.server.coordination.SegmentLoadDropHandler;
import org.apache.druid.server.coordinator.BytesAccumulatingResponseHandler;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.HttpLoadQueuePeonConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.CoordinatorStat;
import org.apache.druid.server.coordinator.stats.Dimension;
@@ -110,7 +110,7 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
private final Object lock = new Object();
- private final DruidCoordinatorConfig config;
+ private final HttpLoadQueuePeonConfig config;
private final ObjectMapper jsonMapper;
private final HttpClient httpClient;
@@ -126,7 +126,7 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
String baseUrl,
ObjectMapper jsonMapper,
HttpClient httpClient,
- DruidCoordinatorConfig config,
+ HttpLoadQueuePeonConfig config,
ScheduledExecutorService processingExecutor,
ExecutorService callBackExecutor
)
@@ -144,7 +144,7 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
new URL(baseUrl),
StringUtils.nonStrictFormat(
"druid-internal/v1/segments/changeRequests?timeout=%d",
- config.getHttpLoadQueuePeonHostTimeout().getMillis()
+ config.getHostTimeout().getMillis()
)
);
}
@@ -160,7 +160,7 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
return;
}
- final int batchSize = config.getHttpLoadQueuePeonBatchSize();
+ final int batchSize = config.getBatchSize();
final List<DataSegmentChangeRequest> newRequests = new ArrayList<>(batchSize);
@@ -194,7 +194,7 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
serverId,
segmentsToLoad.size(),
segmentsToDrop.size(),
- config.getHttpLoadQueuePeonBatchSize()
+ config.getBatchSize()
);
mainLoopInProgress.set(false);
return;
@@ -209,7 +209,7 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
.addHeader(HttpHeaders.Names.CONTENT_TYPE, MediaType.APPLICATION_JSON)
.setContent(requestBodyWriter.writeValueAsBytes(newRequests)),
responseHandler,
- new Duration(config.getHttpLoadQueuePeonHostTimeout().getMillis() + 5000)
+ new Duration(config.getHostTimeout().getMillis() + 5000)
);
Futures.addCallback(
@@ -345,7 +345,7 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
ScheduledExecutors.scheduleAtFixedRate(
processingExecutor,
- config.getHttpLoadQueuePeonRepeatDelay(),
+ config.getRepeatDelay(),
() -> {
if (!stopped) {
doSegmentManagement();
@@ -510,13 +510,13 @@ public class HttpLoadQueuePeon implements LoadQueuePeon
* A request is considered to have timed out if the time elapsed since it was
* first sent to the server is greater than the configured load timeout.
*
- * @see DruidCoordinatorConfig#getLoadTimeoutDelay()
+ * @see HttpLoadQueuePeonConfig#getLoadTimeout()
*/
private boolean hasRequestTimedOut(SegmentHolder holder, long currentTimeMillis)
{
return holder.isRequestSentToServer()
&& currentTimeMillis - holder.getFirstRequestMillis()
- > config.getLoadTimeoutDelay().getMillis();
+ > config.getLoadTimeout().getMillis();
}
private void onRequestFailed(SegmentHolder holder, String failureCause)
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/loading/LoadQueueTaskMaster.java b/server/src/main/java/org/apache/druid/server/coordinator/loading/LoadQueueTaskMaster.java
index 7a97a61fa31..647c9e0c413 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/loading/LoadQueueTaskMaster.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/loading/LoadQueueTaskMaster.java
@@ -22,14 +22,10 @@ package org.apache.druid.server.coordinator.loading;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import com.google.errorprone.annotations.concurrent.GuardedBy;
-import com.google.inject.Provider;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.utils.ZKPaths;
import org.apache.druid.client.ImmutableDruidServer;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.java.util.http.client.HttpClient;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
-import org.apache.druid.server.initialization.ZkPathsConfig;
+import org.apache.druid.server.coordinator.config.HttpLoadQueuePeonConfig;
import java.util.List;
import java.util.Map;
@@ -46,14 +42,11 @@ public class LoadQueueTaskMaster
{
private static final Logger log = new Logger(LoadQueueTaskMaster.class);
- private final Provider<CuratorFramework> curatorFrameworkProvider;
private final ObjectMapper jsonMapper;
private final ScheduledExecutorService peonExec;
private final ExecutorService callbackExec;
- private final DruidCoordinatorConfig config;
+ private final HttpLoadQueuePeonConfig config;
private final HttpClient httpClient;
- private final ZkPathsConfig zkPaths;
- private final boolean httpLoading;
@GuardedBy("this")
private final AtomicBoolean isLeader = new AtomicBoolean(false);
@@ -61,39 +54,23 @@ public class LoadQueueTaskMaster
private final ConcurrentHashMap<String, LoadQueuePeon> loadManagementPeons = new ConcurrentHashMap<>();
public LoadQueueTaskMaster(
- Provider<CuratorFramework> curatorFrameworkProvider,
ObjectMapper jsonMapper,
ScheduledExecutorService peonExec,
ExecutorService callbackExec,
- DruidCoordinatorConfig config,
- HttpClient httpClient,
- ZkPathsConfig zkPaths
+ HttpLoadQueuePeonConfig config,
+ HttpClient httpClient
)
{
- this.curatorFrameworkProvider = curatorFrameworkProvider;
this.jsonMapper = jsonMapper;
this.peonExec = peonExec;
this.callbackExec = callbackExec;
this.config = config;
this.httpClient = httpClient;
- this.zkPaths = zkPaths;
- this.httpLoading = "http".equalsIgnoreCase(config.getLoadQueuePeonType());
}
private LoadQueuePeon createPeon(ImmutableDruidServer server)
{
- if (httpLoading) {
- return new HttpLoadQueuePeon(server.getURL(), jsonMapper, httpClient, config, peonExec, callbackExec);
- } else {
- return new CuratorLoadQueuePeon(
- curatorFrameworkProvider.get(),
- ZKPaths.makePath(zkPaths.getLoadQueuePath(), server.getName()),
- jsonMapper,
- peonExec,
- callbackExec,
- config
- );
- }
+ return new HttpLoadQueuePeon(server.getURL(), jsonMapper, httpClient, config, peonExec, callbackExec);
}
public Map<String, LoadQueuePeon> getAllPeons()
@@ -161,6 +138,6 @@ public class LoadQueueTaskMaster
public boolean isHttpLoading()
{
- return httpLoading;
+ return true;
}
}
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java b/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
index bae0e407037..e7b901e3ed9 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
@@ -141,6 +141,8 @@ public class Stats
= CoordinatorStat.toDebugAndEmit("killedSupervisorSpecs", "metadata/kill/supervisor/count");
public static final CoordinatorStat RULES
= CoordinatorStat.toDebugAndEmit("killedRules", "metadata/kill/rule/count");
+ public static final CoordinatorStat SEGMENT_SCHEMA
+ = CoordinatorStat.toDebugAndEmit("killSchemas", "metadata/kill/segmentSchema/count");
public static final CoordinatorStat AUDIT_LOGS
= CoordinatorStat.toDebugAndEmit("killedAuditLogs", "metadata/kill/audit/count");
public static final CoordinatorStat DATASOURCES
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java b/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java
index 2908f7e009f..c312cb493b5 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java
@@ -131,13 +131,6 @@ public class CuratorDruidCoordinatorTest extends CuratorTestBase
curator.create().creatingParentsIfNeeded().forPath(DESTINATION_LOAD_PATH);
final ObjectMapper objectMapper = new DefaultObjectMapper();
- DruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withCoordinatorStartDelay(new Duration(COORDINATOR_START_DELAY))
- .withCoordinatorPeriod(new Duration(COORDINATOR_PERIOD))
- .withCoordinatorKillPeriod(new Duration(COORDINATOR_PERIOD))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
sourceLoadQueueChildrenCache = new PathChildrenCache(
curator,
SOURCE_LOAD_PATH,
@@ -158,7 +151,7 @@ public class CuratorDruidCoordinatorTest extends CuratorTestBase
objectMapper,
peonExec,
callbackExec,
- druidCoordinatorConfig
+ Duration.standardMinutes(15)
);
destinationLoadQueuePeon = new CuratorLoadQueuePeon(
curator,
@@ -166,7 +159,7 @@ public class CuratorDruidCoordinatorTest extends CuratorTestBase
objectMapper,
peonExec,
callbackExec,
- druidCoordinatorConfig
+ Duration.standardMinutes(15)
);
}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorConfigTest.java
index 6f1a0504587..cb064509c40 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorConfigTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorConfigTest.java
@@ -19,71 +19,531 @@
package org.apache.druid.server.coordinator;
-import org.apache.druid.java.util.common.config.Config;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Injector;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.DruidExceptionMatcher;
+import org.apache.druid.guice.GuiceInjectors;
+import org.apache.druid.guice.JsonConfigProvider;
+import org.apache.druid.guice.JsonConfigurator;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.server.coordinator.config.CoordinatorKillConfigs;
+import org.apache.druid.server.coordinator.config.CoordinatorPeriodConfig;
+import org.apache.druid.server.coordinator.config.CoordinatorRunConfig;
+import org.apache.druid.server.coordinator.config.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.HttpLoadQueuePeonConfig;
+import org.apache.druid.server.coordinator.config.KillUnusedSegmentsConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
+import org.hamcrest.MatcherAssert;
import org.joda.time.Duration;
+import org.joda.time.Period;
import org.junit.Assert;
import org.junit.Test;
-import org.skife.config.ConfigurationObjectFactory;
import java.util.Properties;
public class DruidCoordinatorConfigTest
{
+
+ @Test
+ public void testCoordinatorRunConfigDefaultValues()
+ {
+ final Properties props = new Properties();
+ final CoordinatorRunConfig config = deserializeFrom(props, "druid.coordinator", CoordinatorRunConfig.class);
+
+ Assert.assertEquals(Duration.standardMinutes(1), config.getPeriod());
+ Assert.assertEquals(Duration.standardMinutes(5), config.getStartDelay());
+ }
+
+ @Test
+ public void testCoordinatorRunConfigOverrideValues()
+ {
+ final Properties props = new Properties();
+ props.setProperty("druid.coordinator.startDelay", "PT10M");
+ props.setProperty("druid.coordinator.period", "PT30S");
+ final CoordinatorRunConfig config = deserializeFrom(props, "druid.coordinator", CoordinatorRunConfig.class);
+
+ Assert.assertEquals(Duration.standardSeconds(30), config.getPeriod());
+ Assert.assertEquals(Duration.standardMinutes(10), config.getStartDelay());
+ }
+
+ @Test
+ public void testCoordinatorPeriodConfigDefaultValues()
+ {
+ final Properties props = new Properties();
+ final CoordinatorPeriodConfig config
+ = deserializeFrom(props, "druid.coordinator.period", CoordinatorPeriodConfig.class);
+
+ Assert.assertEquals(Duration.standardMinutes(30), config.getIndexingPeriod());
+ Assert.assertEquals(Duration.standardMinutes(60), config.getMetadataStoreManagementPeriod());
+ }
+
+ @Test
+ public void testCoordinatorPeriodConfigOverrideValues()
+ {
+ final Properties props = new Properties();
+ props.setProperty("druid.coordinator.period.indexingPeriod", "PT1M");
+ props.setProperty("druid.coordinator.period.metadataStoreManagementPeriod", "PT3M");
+ final CoordinatorPeriodConfig config
+ = deserializeFrom(props, "druid.coordinator.period", CoordinatorPeriodConfig.class);
+
+ Assert.assertEquals(Duration.standardMinutes(1), config.getIndexingPeriod());
+ Assert.assertEquals(Duration.standardMinutes(3), config.getMetadataStoreManagementPeriod());
+ }
+
@Test
- public void testCoordinatorConfigWithDefaults()
+ public void testLoadQueuePeonConfigDefaultValues()
{
- final ConfigurationObjectFactory factory = Config.createFactory(new Properties());
- final DruidCoordinatorConfig config = factory.build(DruidCoordinatorConfig.class);
+ final Properties props = new Properties();
+ final HttpLoadQueuePeonConfig config
+ = deserializeFrom(props, "druid.coordinator.loadqueuepeon.http", HttpLoadQueuePeonConfig.class);
- Assert.assertEquals(new Duration("PT300s"), config.getCoordinatorStartDelay());
- Assert.assertEquals(new Duration("PT60s"), config.getCoordinatorPeriod());
- Assert.assertEquals(new Duration("PT1800s"), config.getCoordinatorIndexingPeriod());
- Assert.assertEquals(new Duration("PT1800s"), config.getCoordinatorKillPeriod());
- Assert.assertEquals(new Duration("PT7776000s"), config.getCoordinatorKillDurationToRetain());
- Assert.assertEquals(100, config.getCoordinatorKillMaxSegments());
- Assert.assertEquals(new Duration(15 * 60 * 1000), config.getLoadTimeoutDelay());
- Assert.assertFalse(config.getCoordinatorKillIgnoreDurationToRetain());
- Assert.assertEquals("http", config.getLoadQueuePeonType());
+ Assert.assertEquals(Duration.standardMinutes(1), config.getRepeatDelay());
+ Assert.assertEquals(Duration.standardMinutes(5), config.getHostTimeout());
+ Assert.assertEquals(Duration.standardMinutes(15), config.getLoadTimeout());
+ Assert.assertEquals(1, config.getBatchSize());
}
@Test
- public void testCoordinatorConfigWithOverrides()
+ public void testLoadQueuePeonConfigOverrideValues()
{
final Properties props = new Properties();
- props.setProperty("druid.coordinator.startDelay", "PT1s");
- props.setProperty("druid.coordinator.period", "PT1s");
- props.setProperty("druid.coordinator.period.indexingPeriod", "PT1s");
- props.setProperty("druid.coordinator.kill.on", "true");
- props.setProperty("druid.coordinator.kill.period", "PT10s");
- props.setProperty("druid.coordinator.kill.durationToRetain", "PT1s");
- props.setProperty("druid.coordinator.kill.maxSegments", "10000");
- props.setProperty("druid.coordinator.kill.pendingSegments.on", "true");
- props.setProperty("druid.coordinator.load.timeout", "PT1s");
- props.setProperty("druid.coordinator.loadqueuepeon.repeatDelay", "PT0.100s");
- props.setProperty("druid.coordinator.kill.ignoreDurationToRetain", "true");
-
- final ConfigurationObjectFactory factory = Config.createFactory(props);
- final DruidCoordinatorConfig config = factory.build(DruidCoordinatorConfig.class);
-
- Assert.assertEquals(new Duration("PT1s"), config.getCoordinatorStartDelay());
- Assert.assertEquals(new Duration("PT1s"), config.getCoordinatorPeriod());
- Assert.assertEquals(new Duration("PT1s"), config.getCoordinatorIndexingPeriod());
- Assert.assertEquals(new Duration("PT10s"), config.getCoordinatorKillPeriod());
- Assert.assertEquals(new Duration("PT1s"), config.getCoordinatorKillDurationToRetain());
- Assert.assertEquals(10000, config.getCoordinatorKillMaxSegments());
- Assert.assertEquals(new Duration("PT1s"), config.getLoadTimeoutDelay());
- Assert.assertTrue(config.getCoordinatorKillIgnoreDurationToRetain());
- }
-
- @Test
- public void testCoordinatorConfigWithNegativeDurationToRetain()
+ props.setProperty("druid.coordinator.loadqueuepeon.http.repeatDelay", "PT20M");
+ props.setProperty("druid.coordinator.loadqueuepeon.http.hostTimeout", "PT10M");
+ props.setProperty("druid.coordinator.loadqueuepeon.http.batchSize", "100");
+
+ final HttpLoadQueuePeonConfig config
+ = deserializeFrom(props, "druid.coordinator.loadqueuepeon.http", HttpLoadQueuePeonConfig.class);
+
+ Assert.assertEquals(Duration.standardMinutes(20), config.getRepeatDelay());
+ Assert.assertEquals(Duration.standardMinutes(10), config.getHostTimeout());
+ Assert.assertEquals(Duration.standardMinutes(15), config.getLoadTimeout());
+ Assert.assertEquals(100, config.getBatchSize());
+ }
+
+ @Test
+ public void testMetadataCleanupConfigDefaultValues()
+ {
+ final MetadataCleanupConfig config = new MetadataCleanupConfig(null, null, null);
+
+ Assert.assertTrue(config.isCleanupEnabled());
+ Assert.assertEquals(Duration.standardDays(1), config.getCleanupPeriod());
+ Assert.assertEquals(Duration.standardDays(90), config.getDurationToRetain());
+ }
+
+ @Test
+ public void testMetadataCleanupConfigOverrideValues()
+ {
+ final MetadataCleanupConfig config = new MetadataCleanupConfig(
+ false,
+ Period.parse("PT5H").toStandardDuration(),
+ Period.parse("P1D").toStandardDuration()
+ );
+
+ Assert.assertFalse(config.isCleanupEnabled());
+ Assert.assertEquals(Duration.standardHours(5), config.getCleanupPeriod());
+ Assert.assertEquals(Duration.standardHours(24), config.getDurationToRetain());
+ }
+
+ @Test
+ public void testKillUnusedSegmentsConfigDefaultValues()
+ {
+ CoordinatorKillConfigs killConfigs = createKillConfig().build();
+ final KillUnusedSegmentsConfig config = killConfigs.unusedSegments(null);
+
+ Assert.assertFalse(config.isCleanupEnabled());
+ Assert.assertFalse(config.isIgnoreDurationToRetain());
+ Assert.assertEquals(Duration.standardDays(1), config.getCleanupPeriod());
+ Assert.assertEquals(Duration.standardDays(30), config.getBufferPeriod());
+ Assert.assertEquals(Duration.standardDays(90), config.getDurationToRetain());
+ Assert.assertEquals(100, config.getMaxSegments());
+ }
+
+ @Test
+ public void testKillUnusedSegmentsConfigOverrideValues()
+ {
+ KillUnusedSegmentsConfig inputConfig = new KillUnusedSegmentsConfig(
+ true,
+ Period.parse("PT30M").toStandardDuration(),
+ Period.parse("PT12H").toStandardDuration(),
+ true,
+ Period.parse("PT60M").toStandardDuration(),
+ 500
+ );
+ CoordinatorKillConfigs killConfigs = createKillConfig().unusedSegments(inputConfig).build();
+
+ final KillUnusedSegmentsConfig config = killConfigs.unusedSegments(null);
+ Assert.assertTrue(config.isCleanupEnabled());
+ Assert.assertTrue(config.isIgnoreDurationToRetain());
+ Assert.assertEquals(Duration.standardMinutes(30), config.getCleanupPeriod());
+ Assert.assertEquals(Duration.standardMinutes(60), config.getBufferPeriod());
+ Assert.assertEquals(Duration.standardHours(12), config.getDurationToRetain());
+ Assert.assertEquals(500, config.getMaxSegments());
+ }
+
+ @Test
+ public void testKillUnusedSegmentsPeriodLessThanIndexingPeriod()
+ {
+ KillUnusedSegmentsConfig killUnusedConfig
+ = KillUnusedSegmentsConfig.builder()
+ .withCleanupPeriod(Duration.standardSeconds(5))
+ .build();
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().unusedSegments(killUnusedConfig).build(),
+ new CoordinatorPeriodConfig(null, Duration.standardSeconds(10)),
+ "'druid.coordinator.kill.period'[PT5S] must be greater than or equal to"
+ + " 'druid.coordinator.period.indexingPeriod'[PT10S]"
+ );
+ }
+
+ @Test
+ public void testKillUnusedSegmentsMaxSegmentsNegative()
+ {
+ KillUnusedSegmentsConfig killUnusedConfig
+ = KillUnusedSegmentsConfig.builder()
+ .withMaxSegmentsToKill(-5)
+ .build();
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().unusedSegments(killUnusedConfig).build(),
+ new CoordinatorPeriodConfig(null, Duration.standardSeconds(10)),
+ "'druid.coordinator.kill.maxSegments'[-5] must be a positive integer."
+ );
+ }
+
+ @Test
+ public void testCoordinatorKillConfigDefaultValues()
+ {
+ final CoordinatorKillConfigs killConfigs
+ = deserializeFrom(new Properties(), "druid.coordinator.kill", CoordinatorKillConfigs.class);
+
+ Assert.assertEquals(MetadataCleanupConfig.DEFAULT, killConfigs.auditLogs());
+ Assert.assertEquals(MetadataCleanupConfig.DEFAULT, killConfigs.supervisors());
+ Assert.assertEquals(MetadataCleanupConfig.DEFAULT, killConfigs.compactionConfigs());
+ Assert.assertEquals(MetadataCleanupConfig.DEFAULT, killConfigs.datasources());
+ Assert.assertEquals(MetadataCleanupConfig.DEFAULT, killConfigs.rules());
+ Assert.assertEquals(MetadataCleanupConfig.DEFAULT, killConfigs.pendingSegments());
+ Assert.assertEquals(MetadataCleanupConfig.DEFAULT, killConfigs.segmentSchemas());
+ }
+
+ @Test
+ public void testCoordinatorKillConfigOverrideValues()
{
final Properties props = new Properties();
- props.setProperty("druid.coordinator.kill.durationToRetain", "PT-1s");
+ props.setProperty("druid.coordinator.kill.audit.on", "false");
+ props.setProperty("druid.coordinator.kill.audit.period", "PT10H");
+ props.setProperty("druid.coordinator.kill.audit.durationToRetain", "PT20H");
+
+ props.setProperty("druid.coordinator.kill.compaction.on", "false");
+ props.setProperty("druid.coordinator.kill.compaction.period", "PT20H");
+ props.setProperty("druid.coordinator.kill.compaction.durationToRetain", "PT30H");
+
+ props.setProperty("druid.coordinator.kill.datasource.on", "false");
+ props.setProperty("druid.coordinator.kill.datasource.period", "PT5H");
+ props.setProperty("druid.coordinator.kill.datasource.durationToRetain", "PT10H");
+
+ props.setProperty("druid.coordinator.kill.rule.on", "false");
+ props.setProperty("druid.coordinator.kill.rule.period", "PT11H");
+ props.setProperty("druid.coordinator.kill.rule.durationToRetain", "PT12H");
+
+ props.setProperty("druid.coordinator.kill.supervisor.on", "false");
+ props.setProperty("druid.coordinator.kill.supervisor.period", "PT1H");
+ props.setProperty("druid.coordinator.kill.supervisor.durationToRetain", "PT2H");
+
+ props.setProperty("druid.coordinator.kill.pendingSegments.on", "false");
+
+ props.setProperty("druid.coordinator.kill.segmentSchema.on", "false");
+ props.setProperty("druid.coordinator.kill.segmentSchema.period", "PT2H");
+ props.setProperty("druid.coordinator.kill.segmentSchema.durationToRetain", "PT8H");
+
+ final CoordinatorKillConfigs killConfigs
+ = deserializeFrom(props, "druid.coordinator.kill", CoordinatorKillConfigs.class);
+
+ Assert.assertEquals(
+ new MetadataCleanupConfig(false, Duration.standardHours(10), Duration.standardHours(20)),
+ killConfigs.auditLogs()
+ );
+ Assert.assertEquals(
+ new MetadataCleanupConfig(false, Duration.standardHours(20), Duration.standardHours(30)),
+ killConfigs.compactionConfigs()
+ );
+ Assert.assertEquals(
+ new MetadataCleanupConfig(false, Duration.standardHours(5), Duration.standardHours(10)),
+ killConfigs.datasources()
+ );
+ Assert.assertEquals(
+ new MetadataCleanupConfig(false, Duration.standardHours(11), Duration.standardHours(12)),
+ killConfigs.rules()
+ );
+ Assert.assertEquals(
+ new MetadataCleanupConfig(false, Duration.standardHours(1), Duration.standardHours(2)),
+ killConfigs.supervisors()
+ );
+ Assert.assertEquals(
+ new MetadataCleanupConfig(false, Duration.standardHours(2), Duration.standardHours(8)),
+ killConfigs.segmentSchemas()
+ );
+ Assert.assertFalse(killConfigs.pendingSegments().isCleanupEnabled());
+ }
+
+ @Test
+ public void testCoordinatorConfigFailsWhenCleanupPeriodIsInvalid()
+ {
+ // Validation fails when cleanup period is less than metadata store management period
+ final MetadataCleanupConfig cleanupConfig
+ = new MetadataCleanupConfig(true, Duration.standardMinutes(30), null);
+ final CoordinatorPeriodConfig periodConfig = new CoordinatorPeriodConfig(Duration.standardHours(1), null);
+
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().audit(cleanupConfig).build(),
+ periodConfig,
+ "'druid.coordinator.kill.audit.period'[PT1800S] must be greater than"
+ + " 'druid.coordinator.period.metadataStoreManagementPeriod'[PT3600S]"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().compaction(cleanupConfig).build(),
+ periodConfig,
+ "'druid.coordinator.kill.compaction.period'[PT1800S] must be greater than"
+ + " 'druid.coordinator.period.metadataStoreManagementPeriod'[PT3600S]"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().datasource(cleanupConfig).build(),
+ periodConfig,
+ "'druid.coordinator.kill.datasource.period'[PT1800S] must be greater than"
+ + " 'druid.coordinator.period.metadataStoreManagementPeriod'[PT3600S]"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().rules(cleanupConfig).build(),
+ periodConfig,
+ "'druid.coordinator.kill.rule.period'[PT1800S] must be greater than"
+ + " 'druid.coordinator.period.metadataStoreManagementPeriod'[PT3600S]"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().supervisors(cleanupConfig).build(),
+ periodConfig,
+ "'druid.coordinator.kill.supervisor.period'[PT1800S] must be greater than"
+ + " 'druid.coordinator.period.metadataStoreManagementPeriod'[PT3600S]"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().segmentSchema(cleanupConfig).build(),
+ periodConfig,
+ "'druid.coordinator.kill.segmentSchema.period'[PT1800S] must be greater than"
+ + " 'druid.coordinator.period.metadataStoreManagementPeriod'[PT3600S]"
+ );
+ }
+
+ @Test
+ public void testCoordinatorConfigFailsWhenRetainDurationIsNegative()
+ {
+ final MetadataCleanupConfig cleanupConfig
+ = new MetadataCleanupConfig(true, null, Duration.standardSeconds(1).negated());
+
+ final CoordinatorPeriodConfig defaultPeriodConfig = new CoordinatorPeriodConfig(null, null);
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().audit(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.audit.durationToRetain'[PT-1S] must be 0 milliseconds or higher"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().compaction(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.compaction.durationToRetain'[PT-1S] must be 0 milliseconds or higher"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().datasource(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.datasource.durationToRetain'[PT-1S] must be 0 milliseconds or higher"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().rules(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.rule.durationToRetain'[PT-1S] must be 0 milliseconds or higher"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().supervisors(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.supervisor.durationToRetain'[PT-1S] must be 0 milliseconds or higher"
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().segmentSchema(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.segmentSchema.durationToRetain'[PT-1S] must be 0 milliseconds or higher"
+ );
+ }
+
+ @Test
+ public void testCoordinatorConfigFailsWhenRetainDurationIsHigherThanCurrentTime()
+ {
+ final Duration futureRetainDuration = Duration.millis(System.currentTimeMillis()).plus(10_000);
+ final MetadataCleanupConfig cleanupConfig = new MetadataCleanupConfig(true, null, futureRetainDuration);
+ final CoordinatorPeriodConfig defaultPeriodConfig = new CoordinatorPeriodConfig(null, null);
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().audit(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.audit.durationToRetain'[%s] cannot be greater than current time in milliseconds",
+ futureRetainDuration
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().compaction(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.compaction.durationToRetain'[%s] cannot be greater than current time in milliseconds",
+ futureRetainDuration
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().datasource(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.datasource.durationToRetain'[%s] cannot be greater than current time in milliseconds",
+ futureRetainDuration
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().rules(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.rule.durationToRetain'[%s] cannot be greater than current time in milliseconds",
+ futureRetainDuration
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().supervisors(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.supervisor.durationToRetain'[%s] cannot be greater than current time in milliseconds",
+ futureRetainDuration
+ );
+ verifyCoordinatorConfigFailsWith(
+ createKillConfig().segmentSchema(cleanupConfig).build(),
+ defaultPeriodConfig,
+ "'druid.coordinator.kill.segmentSchema.durationToRetain'[%s] cannot be"
+ + " greater than current time in milliseconds",
+ futureRetainDuration
+ );
+ }
+
+ @Test
+ public void testSomeBasicStuff()
+ {
+ Duration millis = Duration.millis(1);
+ Duration days = Duration.standardDays(1);
+ System.out.println(millis.getMillis() + ", " + days);
+ }
+
+ private KillConfigBuilder createKillConfig()
+ {
+ return new KillConfigBuilder();
+ }
+
+ private void verifyCoordinatorConfigFailsWith(
+ CoordinatorKillConfigs killConfig,
+ CoordinatorPeriodConfig periodConfig,
+ String expectedMessageFormat,
+ Object... args
+ )
+ {
+ DruidException exception = Assert.assertThrows(
+ DruidException.class,
+ () -> new DruidCoordinatorConfig(
+ new CoordinatorRunConfig(null, null),
+ periodConfig,
+ killConfig,
+ null,
+ null
+ )
+ );
+
+ final String expectedMessage = StringUtils.format(expectedMessageFormat, args);
+ MatcherAssert.assertThat(
+ exception,
+ new DruidExceptionMatcher(
+ DruidException.Persona.OPERATOR,
+ DruidException.Category.INVALID_INPUT,
+ "general"
+ ).expectMessageIs(expectedMessage)
+ );
+ }
+
+ private Injector createInjector(String propertyPrefix, Class<?> clazz)
+ {
+ return GuiceInjectors.makeStartupInjectorWithModules(
+ ImmutableList.of(
+ binder -> JsonConfigProvider.bind(binder, propertyPrefix, clazz)
+ )
+ );
+ }
+
+ private <T> T deserializeFrom(Properties props, String propertyPrefix, Class<T> type)
+ {
+ final Injector injector = createInjector(propertyPrefix, type);
+ final JsonConfigProvider<T> provider = JsonConfigProvider.of(propertyPrefix, type);
+ provider.inject(props, injector.getInstance(JsonConfigurator.class));
+ return provider.get();
+ }
+
+ private static class KillConfigBuilder
+ {
+ MetadataCleanupConfig audit;
+ MetadataCleanupConfig compaction;
+ MetadataCleanupConfig datasource;
+ MetadataCleanupConfig rules;
+ MetadataCleanupConfig supervisors;
+ MetadataCleanupConfig pendingSegments;
+ MetadataCleanupConfig segmentSchema;
+ KillUnusedSegmentsConfig unusedSegments;
+
+ KillConfigBuilder audit(MetadataCleanupConfig config)
+ {
+ this.audit = config;
+ return this;
+ }
+
+ KillConfigBuilder compaction(MetadataCleanupConfig config)
+ {
+ this.compaction = config;
+ return this;
+ }
+
+ KillConfigBuilder datasource(MetadataCleanupConfig config)
+ {
+ this.datasource = config;
+ return this;
+ }
+
+ KillConfigBuilder rules(MetadataCleanupConfig config)
+ {
+ this.rules = config;
+ return this;
+ }
+
+ KillConfigBuilder supervisors(MetadataCleanupConfig config)
+ {
+ this.supervisors = config;
+ return this;
+ }
+
+ KillConfigBuilder segmentSchema(MetadataCleanupConfig config)
+ {
+ this.segmentSchema = config;
+ return this;
+ }
- final ConfigurationObjectFactory factory = Config.createFactory(props);
- final DruidCoordinatorConfig config = factory.build(DruidCoordinatorConfig.class);
+ KillConfigBuilder unusedSegments(KillUnusedSegmentsConfig config)
+ {
+ this.unusedSegments = config;
+ return this;
+ }
- Assert.assertEquals(new Duration("PT-1s"), config.getCoordinatorKillDurationToRetain());
+ CoordinatorKillConfigs build()
+ {
+ return new CoordinatorKillConfigs(
+ pendingSegments,
+ supervisors,
+ audit,
+ datasource,
+ rules,
+ compaction,
+ segmentSchema,
+ unusedSegments == null ? null : unusedSegments.isCleanupEnabled(),
+ unusedSegments == null ? null : unusedSegments.getCleanupPeriod(),
+ unusedSegments == null ? null : unusedSegments.getDurationToRetain(),
+ unusedSegments == null ? null : unusedSegments.isIgnoreDurationToRetain(),
+ unusedSegments == null ? null : unusedSegments.getBufferPeriod(),
+ unusedSegments == null ? null : unusedSegments.getMaxSegments()
+ );
+ }
}
}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
index 2a785764450..a5ed1616a6d 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
@@ -53,6 +53,10 @@ import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordination.ServerType;
import org.apache.druid.server.coordinator.balancer.CostBalancerStrategyFactory;
import org.apache.druid.server.coordinator.compact.NewestSegmentFirstPolicy;
+import org.apache.druid.server.coordinator.config.CoordinatorKillConfigs;
+import org.apache.druid.server.coordinator.config.CoordinatorPeriodConfig;
+import org.apache.druid.server.coordinator.config.CoordinatorRunConfig;
+import org.apache.druid.server.coordinator.config.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.duty.CompactSegments;
import org.apache.druid.server.coordinator.duty.CoordinatorCustomDuty;
import org.apache.druid.server.coordinator.duty.CoordinatorCustomDutyGroup;
@@ -88,6 +92,7 @@ import java.util.concurrent.atomic.AtomicReference;
public class DruidCoordinatorTest extends CuratorTestBase
{
private static final String LOADPATH = "/druid/loadqueue/localhost:1234";
+ private static final Duration LOAD_TIMEOUT = Duration.standardMinutes(15);
private static final long COORDINATOR_START_DELAY = 1;
private static final long COORDINATOR_PERIOD = 100;
@@ -142,10 +147,13 @@ public class DruidCoordinatorTest extends CuratorTestBase
curator.create().creatingParentsIfNeeded().forPath(LOADPATH);
objectMapper = new DefaultObjectMapper();
newestSegmentFirstPolicy = new NewestSegmentFirstPolicy(objectMapper);
- druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withCoordinatorStartDelay(new Duration(COORDINATOR_START_DELAY))
- .withCoordinatorPeriod(new Duration(COORDINATOR_PERIOD))
- .build();
+ druidCoordinatorConfig = new DruidCoordinatorConfig(
+ new CoordinatorRunConfig(new Duration(COORDINATOR_START_DELAY), new Duration(COORDINATOR_PERIOD)),
+ new CoordinatorPeriodConfig(null, null),
+ CoordinatorKillConfigs.DEFAULT,
+ new CostBalancerStrategyFactory(),
+ null
+ );
pathChildrenCache = new PathChildrenCache(
curator,
LOADPATH,
@@ -159,7 +167,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
objectMapper,
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_scheduled-%d"),
Execs.singleThreaded("coordinator_test_load_queue_peon-%d"),
- druidCoordinatorConfig
+ LOAD_TIMEOUT
);
loadQueuePeon.start();
druidNode = new DruidNode("hey", "what", false, 1234, null, true, false);
@@ -178,7 +186,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
new LatchableServiceAnnouncer(leaderAnnouncerLatch, leaderUnannouncerLatch),
druidNode,
new CoordinatorCustomDutyGroups(ImmutableSet.of()),
- new CostBalancerStrategyFactory(),
EasyMock.createNiceMock(LookupCoordinatorManager.class),
new TestDruidLeaderSelector(),
null,
@@ -354,7 +361,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
objectMapper,
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_cold_scheduled-%d"),
Execs.singleThreaded("coordinator_test_load_queue_peon_cold-%d"),
- druidCoordinatorConfig
+ LOAD_TIMEOUT
);
final PathChildrenCache pathChildrenCacheCold = new PathChildrenCache(
curator,
@@ -453,7 +460,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
objectMapper,
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_cold_scheduled-%d"),
Execs.singleThreaded("coordinator_test_load_queue_peon_cold-%d"),
- druidCoordinatorConfig
+ LOAD_TIMEOUT
);
final LoadQueuePeon loadQueuePeonBroker1 = new CuratorLoadQueuePeon(
@@ -462,7 +469,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
objectMapper,
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_broker1_scheduled-%d"),
Execs.singleThreaded("coordinator_test_load_queue_peon_broker1-%d"),
- druidCoordinatorConfig
+ LOAD_TIMEOUT
);
final LoadQueuePeon loadQueuePeonBroker2 = new CuratorLoadQueuePeon(
@@ -471,7 +478,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
objectMapper,
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_broker2_scheduled-%d"),
Execs.singleThreaded("coordinator_test_load_queue_peon_broker2-%d"),
- druidCoordinatorConfig
+ LOAD_TIMEOUT
);
final LoadQueuePeon loadQueuePeonPoenServer = new CuratorLoadQueuePeon(
@@ -480,7 +487,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
objectMapper,
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_peon_scheduled-%d"),
Execs.singleThreaded("coordinator_test_load_queue_peon_peon-%d"),
- druidCoordinatorConfig
+ LOAD_TIMEOUT
);
final PathChildrenCache pathChildrenCacheCold = new PathChildrenCache(
curator,
@@ -607,7 +614,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
new LatchableServiceAnnouncer(leaderAnnouncerLatch, leaderUnannouncerLatch),
druidNode,
emptyCustomDutyGroups,
- new CostBalancerStrategyFactory(),
EasyMock.createNiceMock(LookupCoordinatorManager.class),
new TestDruidLeaderSelector(),
null,
@@ -633,7 +639,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
CoordinatorCustomDutyGroup group = new CoordinatorCustomDutyGroup(
"group1",
Duration.standardSeconds(1),
- ImmutableList.of(new KillSupervisorsCustomDuty(new Duration("PT1S"), null, druidCoordinatorConfig))
+ ImmutableList.of(new KillSupervisorsCustomDuty(new Duration("PT1S"), null))
);
CoordinatorCustomDutyGroups customDutyGroups = new CoordinatorCustomDutyGroups(ImmutableSet.of(group));
coordinator = new DruidCoordinator(
@@ -648,7 +654,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
new LatchableServiceAnnouncer(leaderAnnouncerLatch, leaderUnannouncerLatch),
druidNode,
customDutyGroups,
- new CostBalancerStrategyFactory(),
EasyMock.createNiceMock(LookupCoordinatorManager.class),
new TestDruidLeaderSelector(),
null,
@@ -689,7 +694,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
new LatchableServiceAnnouncer(leaderAnnouncerLatch, leaderUnannouncerLatch),
druidNode,
customDutyGroups,
- new CostBalancerStrategyFactory(),
EasyMock.createNiceMock(LookupCoordinatorManager.class),
new TestDruidLeaderSelector(),
null,
@@ -795,7 +799,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
new LatchableServiceAnnouncer(leaderAnnouncerLatch, leaderUnannouncerLatch),
druidNode,
groups,
- new CostBalancerStrategyFactory(),
EasyMock.createNiceMock(LookupCoordinatorManager.class),
new TestDruidLeaderSelector(),
null,
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java b/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java
deleted file mode 100644
index 75bed00c3c3..00000000000
--- a/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java
+++ /dev/null
@@ -1,565 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.druid.server.coordinator;
-
-import org.joda.time.Duration;
-import org.joda.time.Period;
-
-public class TestDruidCoordinatorConfig extends DruidCoordinatorConfig
-{
- private final Duration coordinatorStartDelay;
- private final Duration coordinatorPeriod;
- private final Duration coordinatorIndexingPeriod;
- private final Duration metadataStoreManagementPeriod;
- private final Duration loadTimeoutDelay;
- private final Duration coordinatorKillPeriod;
- private final Duration coordinatorKillDurationToRetain;
- private final Duration coordinatorSupervisorKillPeriod;
- private final Duration coordinatorSupervisorKillDurationToRetain;
- private final Duration coordinatorAuditKillPeriod;
- private final Duration coordinatorAuditKillDurationToRetain;
- private final Duration coordinatorCompactionKillPeriod;
- private final Duration coordinatorRuleKillPeriod;
- private final Duration coordinatorRuleKillDurationToRetain;
- private final Duration coordinatorDatasourceKillPeriod;
- private final Duration coordinatorDatasourceKillDurationToRetain;
- private final int coordinatorKillMaxSegments;
- private final boolean coordinatorKillIgnoreDurationToRetain;
- private final String loadQueuePeonType;
- private final Duration httpLoadQueuePeonRepeatDelay;
- private final int curatorLoadQueuePeonNumCallbackThreads;
- private final Duration httpLoadQueuePeonHostTimeout;
- private final int httpLoadQueuePeonBatchSize;
- private final Duration coordinatorKillBufferPeriod;
- private final Duration segmentSchemaKillPeriod;
- private final Duration segmentSchemaKillDurationToRetain;
-
- public TestDruidCoordinatorConfig(
- Duration coordinatorStartDelay,
- Duration coordinatorPeriod,
- Duration coordinatorIndexingPeriod,
- Duration metadataStoreManagementPeriod,
- Duration loadTimeoutDelay,
- Duration coordinatorKillPeriod,
- Duration coordinatorKillDurationToRetain,
- Duration coordinatorSupervisorKillPeriod,
- Duration coordinatorSupervisorKillDurationToRetain,
- Duration coordinatorAuditKillPeriod,
- Duration coordinatorAuditKillDurationToRetain,
- Duration coordinatorCompactionKillPeriod,
- Duration coordinatorRuleKillPeriod,
- Duration coordinatorRuleKillDurationToRetain,
- Duration coordinatorDatasourceKillPeriod,
- Duration coordinatorDatasourceKillDurationToRetain,
- int coordinatorKillMaxSegments,
- boolean coordinatorKillIgnoreDurationToRetain,
- String loadQueuePeonType,
- Duration httpLoadQueuePeonRepeatDelay,
- Duration httpLoadQueuePeonHostTimeout,
- int httpLoadQueuePeonBatchSize,
- int curatorLoadQueuePeonNumCallbackThreads,
- Duration coordinatorKillBufferPeriod,
- Duration segmentSchemaKillPeriod,
- Duration segmentSchemaKillDurationToRetain
- )
- {
- this.coordinatorStartDelay = coordinatorStartDelay;
- this.coordinatorPeriod = coordinatorPeriod;
- this.coordinatorIndexingPeriod = coordinatorIndexingPeriod;
- this.metadataStoreManagementPeriod = metadataStoreManagementPeriod;
- this.loadTimeoutDelay = loadTimeoutDelay;
- this.coordinatorKillPeriod = coordinatorKillPeriod;
- this.coordinatorKillDurationToRetain = coordinatorKillDurationToRetain;
- this.coordinatorSupervisorKillPeriod = coordinatorSupervisorKillPeriod;
- this.coordinatorSupervisorKillDurationToRetain = coordinatorSupervisorKillDurationToRetain;
- this.coordinatorAuditKillPeriod = coordinatorAuditKillPeriod;
- this.coordinatorAuditKillDurationToRetain = coordinatorAuditKillDurationToRetain;
- this.coordinatorCompactionKillPeriod = coordinatorCompactionKillPeriod;
- this.coordinatorRuleKillPeriod = coordinatorRuleKillPeriod;
- this.coordinatorRuleKillDurationToRetain = coordinatorRuleKillDurationToRetain;
- this.coordinatorDatasourceKillPeriod = coordinatorDatasourceKillPeriod;
- this.coordinatorDatasourceKillDurationToRetain = coordinatorDatasourceKillDurationToRetain;
- this.coordinatorKillMaxSegments = coordinatorKillMaxSegments;
- this.coordinatorKillIgnoreDurationToRetain = coordinatorKillIgnoreDurationToRetain;
- this.loadQueuePeonType = loadQueuePeonType;
- this.httpLoadQueuePeonRepeatDelay = httpLoadQueuePeonRepeatDelay;
- this.httpLoadQueuePeonHostTimeout = httpLoadQueuePeonHostTimeout;
- this.httpLoadQueuePeonBatchSize = httpLoadQueuePeonBatchSize;
- this.curatorLoadQueuePeonNumCallbackThreads = curatorLoadQueuePeonNumCallbackThreads;
- this.coordinatorKillBufferPeriod = coordinatorKillBufferPeriod;
- this.segmentSchemaKillPeriod = segmentSchemaKillPeriod;
- this.segmentSchemaKillDurationToRetain = segmentSchemaKillDurationToRetain;
- }
-
- @Override
- public Duration getCoordinatorStartDelay()
- {
- return coordinatorStartDelay;
- }
-
- @Override
- public Duration getCoordinatorPeriod()
- {
- return coordinatorPeriod;
- }
-
- @Override
- public Duration getCoordinatorIndexingPeriod()
- {
- return coordinatorIndexingPeriod;
- }
-
- @Override
- public Duration getCoordinatorMetadataStoreManagementPeriod()
- {
- return metadataStoreManagementPeriod;
- }
-
- @Override
- public boolean isKillUnusedSegmentsEnabled()
- {
- return true;
- }
-
- @Override
- public boolean isKillPendingSegmentsEnabled()
- {
- return true;
- }
-
- @Override
- public Duration getCoordinatorKillPeriod()
- {
- return coordinatorKillPeriod;
- }
-
- @Override
- public Duration getCoordinatorKillDurationToRetain()
- {
- return coordinatorKillDurationToRetain;
- }
-
- @Override
- public Duration getCoordinatorSupervisorKillPeriod()
- {
- return coordinatorSupervisorKillPeriod;
- }
-
- @Override
- public boolean isSupervisorKillEnabled()
- {
- return true;
- }
-
- @Override
- public Duration getCoordinatorSupervisorKillDurationToRetain()
- {
- return coordinatorSupervisorKillDurationToRetain;
- }
-
- @Override
- public boolean isAuditKillEnabled()
- {
- return true;
- }
-
- @Override
- public Duration getCoordinatorAuditKillPeriod()
- {
- return coordinatorAuditKillPeriod;
- }
-
- @Override
- public Duration getCoordinatorAuditKillDurationToRetain()
- {
- return coordinatorAuditKillDurationToRetain;
- }
-
- @Override
- public boolean isCompactionKillEnabled()
- {
- return true;
- }
-
- @Override
- public Duration getCoordinatorCompactionKillPeriod()
- {
- return coordinatorCompactionKillPeriod;
- }
-
- @Override
- public boolean isRuleKillEnabled()
- {
- return true;
- }
-
- @Override
- public Duration getCoordinatorRuleKillPeriod()
- {
- return coordinatorRuleKillPeriod;
- }
-
- @Override
- public Duration getCoordinatorRuleKillDurationToRetain()
- {
- return coordinatorRuleKillDurationToRetain;
- }
-
- @Override
- public boolean isDatasourceKillEnabled()
- {
- return true;
- }
-
- @Override
- public Duration getCoordinatorDatasourceKillPeriod()
- {
- return coordinatorDatasourceKillPeriod;
- }
-
- @Override
- public Duration getCoordinatorDatasourceKillDurationToRetain()
- {
- return coordinatorDatasourceKillDurationToRetain;
- }
-
- @Override
- public int getCoordinatorKillMaxSegments()
- {
- return coordinatorKillMaxSegments;
- }
-
- @Override
- public Duration getLoadTimeoutDelay()
- {
- return loadTimeoutDelay == null ? super.getLoadTimeoutDelay() : loadTimeoutDelay;
- }
-
- @Override
- public boolean getCoordinatorKillIgnoreDurationToRetain()
- {
- return coordinatorKillIgnoreDurationToRetain;
- }
-
- @Override
- public String getLoadQueuePeonType()
- {
- return loadQueuePeonType;
- }
-
- @Override
- public Duration getHttpLoadQueuePeonRepeatDelay()
- {
- return httpLoadQueuePeonRepeatDelay;
- }
-
- @Override
- public int getNumCuratorCallBackThreads()
- {
- return curatorLoadQueuePeonNumCallbackThreads;
- }
-
- @Override
- public Duration getHttpLoadQueuePeonHostTimeout()
- {
- return httpLoadQueuePeonHostTimeout;
- }
-
- @Override
- public int getHttpLoadQueuePeonBatchSize()
- {
- return httpLoadQueuePeonBatchSize;
- }
-
- @Override
- public boolean isSegmentSchemaKillEnabled()
- {
- return true;
- }
-
- @Override
- public Duration getSegmentSchemaKillPeriod()
- {
- return segmentSchemaKillPeriod;
- }
-
- @Override
- public Duration getSegmentSchemaKillDurationToRetain()
- {
- return segmentSchemaKillDurationToRetain;
- }
-
- @Override
- public Duration getCoordinatorKillBufferPeriod()
- {
- return coordinatorKillBufferPeriod;
- }
-
- public static class Builder
- {
- private static final Duration DEFAULT_COORDINATOR_START_DELAY = Period.parse("PT300s").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_PERIOD = Period.parse("PT60s").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_INDEXING_PERIOD = Period.parse("PT1800s").toStandardDuration();
- private static final Duration DEFAULT_METADATA_STORE_MANAGEMENT_PERIOD = Period.parse("PT1H").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_KILL_DURATION_TO_RETAIN = Period.parse("P90D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_KILL_BUFFER_PERIOD = Period.parse("P30D").toStandardDuration();
- private static final boolean DEFAULT_COORDINATOR_KILL_IGNORE_DURATION_TO_RETAIN = false;
- private static final int DEFAULT_COORDINATOR_KILL_MAX_SEGMENTS = 100;
- private static final Duration DEFAULT_COORDINATOR_SUPERVISOR_KILL_PERIOD = Period.parse("P1D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_SUPERVISOR_KILL_DURATION_TO_RETAIN = Period.parse("P90D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_COMPACTION_KILL_PERIOD = Period.parse("P1D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_RULE_KILL_PERIOD = Period.parse("P1D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_RULE_KILL_DURATION_TO_RETAIN = Period.parse("P90D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_DATASOURCE_KILL_PERIOD = Period.parse("P1D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_DATASOURCE_KILL_DURATION_TO_RETAIN = Period.parse("P90D").toStandardDuration();
- private static final Duration DEFAULT_LOAD_TIMEOUT_DELAY = new Duration(15 * 60 * 1000);
- private static final String DEFAULT_LOAD_QUEUE_PEON_TYPE = "curator";
- private static final int DEFAULT_CURATOR_LOAD_QUEUE_PEON_NUM_CALLBACK_THREADS = 2;
- private static final Duration DEFAULT_HTTP_LOAD_QUEUE_PEON_REPEAT_DELAY = Duration.millis(60000);
- private static final Duration DEFAULT_HTTP_LOAD_QUEUE_PEON_HOST_TIMEOUT = Duration.millis(300000);
- private static final int DEFAULT_HTTP_LOAD_QUEUE_PEON_BATCH_SIZE = 1;
- private static final Duration DEFAULT_COORDINATOR_AUDIT_KILL_PERIOD = Period.parse("P1D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_AUTIT_KILL_DURATION_TO_RETAIN = Period.parse("P90D").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_SEGMENT_SCHEMA_KILL_PERIOD = Period.parse("PT1H").toStandardDuration();
- private static final Duration DEFAULT_COORDINATOR_SEGMENT_SCHEMA_KILL_DURATION_TO_RETAIN = Period.parse("PT6H").toStandardDuration();
-
- private Duration coordinatorStartDelay;
- private Duration coordinatorPeriod;
- private Duration coordinatorIndexingPeriod;
- private Duration metadataStoreManagementPeriod;
- private Duration coordinatorKillPeriod;
- private Duration coordinatorKillDurationToRetain;
- private Boolean coordinatorKillIgnoreDurationToRetain;
- private Integer coordinatorKillMaxSegments;
- private Duration coordinatorSupervisorKillPeriod;
- private Duration coordinatorSupervisorKillDurationToRetain;
- private Duration coordinatorCompactionKillPeriod;
- private Duration coordinatorRuleKillPeriod;
- private Duration coordinatorRuleKillDurationToRetain;
- private Duration coordinatorDatasourceKillPeriod;
- private Duration coordinatorDatasourceKillDurationToRetain;
- private Duration loadTimeoutDelay;
- private String loadQueuePeonType;
- private Duration httpLoadQueuePeonRepeatDelay;
- private Integer curatorLoadQueuePeonNumCallbackThreads;
- private Duration httpLoadQueuePeonHostTimeout;
- private Integer httpLoadQueuePeonBatchSize;
- private Duration coordinatorAuditKillPeriod;
- private Duration coordinatorAuditKillDurationToRetain;
- private Duration coordinatorKillBufferPeriod;
- private Duration segmentSchemaKillPeriod;
- private Duration segmentSchemaKillDurationToRetain;
-
- public Builder()
- {
- }
-
- public Builder withCoordinatorStartDelay(Duration coordinatorStartDelay)
- {
- this.coordinatorStartDelay = coordinatorStartDelay;
- return this;
- }
-
- public Builder withCoordinatorPeriod(Duration coordinatorPeriod)
- {
- this.coordinatorPeriod = coordinatorPeriod;
- return this;
- }
-
- public Builder withCoordinatorIndexingPeriod(Duration coordinatorIndexingPeriod)
- {
- this.coordinatorIndexingPeriod = coordinatorIndexingPeriod;
- return this;
- }
-
- public Builder withMetadataStoreManagementPeriod(Duration metadataStoreManagementPeriod)
- {
- this.metadataStoreManagementPeriod = metadataStoreManagementPeriod;
- return this;
- }
-
- public Builder withCoordinatorKillPeriod(Duration coordinatorKillPeriod)
- {
- this.coordinatorKillPeriod = coordinatorKillPeriod;
- return this;
- }
-
- public Builder withCoordinatorKillDurationToRetain(Duration coordinatorKillDurationToRetain)
- {
- this.coordinatorKillDurationToRetain = coordinatorKillDurationToRetain;
- return this;
- }
-
- public Builder withCoordinatorKillIgnoreDurationToRetain(boolean coordinatorKillIgnoreDurationToRetain)
- {
- this.coordinatorKillIgnoreDurationToRetain = coordinatorKillIgnoreDurationToRetain;
- return this;
- }
-
- public Builder withCoordinatorKillMaxSegments(int coordinatorKillMaxSegments)
- {
- this.coordinatorKillMaxSegments = coordinatorKillMaxSegments;
- return this;
- }
-
- public Builder withCoordinatorSupervisorKillPeriod(Duration coordinatorSupervisorKillPeriod)
- {
- this.coordinatorSupervisorKillPeriod = coordinatorSupervisorKillPeriod;
- return this;
- }
-
- public Builder withCoordinatorSupervisorKillDurationToRetain(Duration coordinatorSupervisorKillDurationToRetain)
- {
- this.coordinatorSupervisorKillDurationToRetain = coordinatorSupervisorKillDurationToRetain;
- return this;
- }
-
- public Builder withCoordinatorCompactionKillPeriod(Duration coordinatorCompactionKillPeriod)
- {
- this.coordinatorCompactionKillPeriod = coordinatorCompactionKillPeriod;
- return this;
- }
-
- public Builder withCoordinatorRuleKillPeriod(Duration coordinatorRuleKillPeriod)
- {
- this.coordinatorRuleKillPeriod = coordinatorRuleKillPeriod;
- return this;
- }
-
- public Builder withCoordinatorRuleKillDurationToRetain(Duration coordinatorRuleKillDurationToRetain)
- {
- this.coordinatorRuleKillDurationToRetain = coordinatorRuleKillDurationToRetain;
- return this;
- }
-
- public Builder withCoordinatorDatasourceKillPeriod(Duration coordinatorDatasourceKillPeriod)
- {
- this.coordinatorDatasourceKillPeriod = coordinatorDatasourceKillPeriod;
- return this;
- }
-
- public Builder withCoordinatorDatasourceKillDurationToRetain(Duration coordinatorDatasourceKillDurationToRetain)
- {
- this.coordinatorDatasourceKillDurationToRetain = coordinatorDatasourceKillDurationToRetain;
- return this;
- }
-
- public Builder withLoadTimeoutDelay(Duration loadTimeoutDelay)
- {
- this.loadTimeoutDelay = loadTimeoutDelay;
- return this;
- }
-
- public Builder withLoadQueuePeonType(String loadQueuePeonType)
- {
- this.loadQueuePeonType = loadQueuePeonType;
- return this;
- }
-
- public Builder withHttpLoadQueuePeonRepeatDelay(Duration httpLoadQueuePeonRepeatDelay)
- {
- this.httpLoadQueuePeonRepeatDelay = httpLoadQueuePeonRepeatDelay;
- return this;
- }
-
- public Builder withCuratorLoadQueuePeonNumCallbackThreads(int curatorLoadQueuePeonNumCallbackThreads)
- {
- this.curatorLoadQueuePeonNumCallbackThreads = curatorLoadQueuePeonNumCallbackThreads;
- return this;
- }
-
- public Builder withHttpLoadQueuePeonHostTimeout(Duration httpLoadQueuePeonHostTimeout)
- {
- this.httpLoadQueuePeonHostTimeout = httpLoadQueuePeonHostTimeout;
- return this;
- }
-
- public Builder withHttpLoadQueuePeonBatchSize(int httpLoadQueuePeonBatchSize)
- {
- this.httpLoadQueuePeonBatchSize = httpLoadQueuePeonBatchSize;
- return this;
- }
-
- public Builder withCoordianatorAuditKillPeriod(Duration coordinatorAuditKillPeriod)
- {
- this.coordinatorAuditKillPeriod = coordinatorAuditKillPeriod;
- return this;
- }
-
- public Builder withCoordinatorAuditKillDurationToRetain(Duration coordinatorAuditKillDurationToRetain)
- {
- this.coordinatorAuditKillDurationToRetain = coordinatorAuditKillDurationToRetain;
- return this;
- }
-
- public Builder withCoordinatorKillBufferPeriod(Duration coordinatorKillBufferPeriod)
- {
- this.coordinatorKillBufferPeriod = coordinatorKillBufferPeriod;
- return this;
- }
-
- public Builder withSegmentSchemaKillPeriod(Duration segmentSchemaKillPeriod)
- {
- this.segmentSchemaKillPeriod = segmentSchemaKillPeriod;
- return this;
- }
-
- public Builder withSegmentSchemaKillDurationToRetain(Duration segmentSchemaKillDurationToRetain)
- {
- this.segmentSchemaKillDurationToRetain = segmentSchemaKillDurationToRetain;
- return this;
- }
-
- public TestDruidCoordinatorConfig build()
- {
- return new TestDruidCoordinatorConfig(
- coordinatorStartDelay == null ? DEFAULT_COORDINATOR_START_DELAY : coordinatorStartDelay,
- coordinatorPeriod == null ? DEFAULT_COORDINATOR_PERIOD : coordinatorPeriod,
- coordinatorIndexingPeriod == null ? DEFAULT_COORDINATOR_INDEXING_PERIOD : coordinatorIndexingPeriod,
- metadataStoreManagementPeriod == null ? DEFAULT_METADATA_STORE_MANAGEMENT_PERIOD : metadataStoreManagementPeriod,
- loadTimeoutDelay == null ? DEFAULT_LOAD_TIMEOUT_DELAY : loadTimeoutDelay,
- coordinatorKillPeriod == null ? (coordinatorIndexingPeriod == null ? DEFAULT_COORDINATOR_INDEXING_PERIOD : coordinatorIndexingPeriod)
- : coordinatorKillPeriod,
- coordinatorKillDurationToRetain == null ? DEFAULT_COORDINATOR_KILL_DURATION_TO_RETAIN
- : coordinatorKillDurationToRetain,
- coordinatorSupervisorKillPeriod == null ? DEFAULT_COORDINATOR_SUPERVISOR_KILL_PERIOD : coordinatorSupervisorKillPeriod,
- coordinatorSupervisorKillDurationToRetain == null ? DEFAULT_COORDINATOR_SUPERVISOR_KILL_DURATION_TO_RETAIN : coordinatorSupervisorKillDurationToRetain,
- coordinatorAuditKillPeriod == null ? DEFAULT_COORDINATOR_AUDIT_KILL_PERIOD : coordinatorAuditKillPeriod,
- coordinatorAuditKillDurationToRetain == null ? DEFAULT_COORDINATOR_AUTIT_KILL_DURATION_TO_RETAIN : coordinatorAuditKillDurationToRetain,
- coordinatorCompactionKillPeriod == null ? DEFAULT_COORDINATOR_COMPACTION_KILL_PERIOD : coordinatorCompactionKillPeriod,
- coordinatorRuleKillPeriod == null ? DEFAULT_COORDINATOR_RULE_KILL_PERIOD : coordinatorRuleKillPeriod,
- coordinatorRuleKillDurationToRetain == null ? DEFAULT_COORDINATOR_RULE_KILL_DURATION_TO_RETAIN : coordinatorRuleKillDurationToRetain,
- coordinatorDatasourceKillPeriod == null ? DEFAULT_COORDINATOR_DATASOURCE_KILL_PERIOD : coordinatorDatasourceKillPeriod,
- coordinatorDatasourceKillDurationToRetain == null ? DEFAULT_COORDINATOR_DATASOURCE_KILL_DURATION_TO_RETAIN : coordinatorDatasourceKillDurationToRetain,
- coordinatorKillMaxSegments == null ? DEFAULT_COORDINATOR_KILL_MAX_SEGMENTS : coordinatorKillMaxSegments,
- coordinatorKillIgnoreDurationToRetain == null ? DEFAULT_COORDINATOR_KILL_IGNORE_DURATION_TO_RETAIN : coordinatorKillIgnoreDurationToRetain,
- loadQueuePeonType == null ? DEFAULT_LOAD_QUEUE_PEON_TYPE : loadQueuePeonType,
- httpLoadQueuePeonRepeatDelay == null ? DEFAULT_HTTP_LOAD_QUEUE_PEON_REPEAT_DELAY : httpLoadQueuePeonRepeatDelay,
- httpLoadQueuePeonHostTimeout == null ? DEFAULT_HTTP_LOAD_QUEUE_PEON_HOST_TIMEOUT : httpLoadQueuePeonHostTimeout,
- httpLoadQueuePeonBatchSize == null ? DEFAULT_HTTP_LOAD_QUEUE_PEON_BATCH_SIZE : httpLoadQueuePeonBatchSize,
- curatorLoadQueuePeonNumCallbackThreads == null ? DEFAULT_CURATOR_LOAD_QUEUE_PEON_NUM_CALLBACK_THREADS
- : curatorLoadQueuePeonNumCallbackThreads,
- coordinatorKillBufferPeriod == null ? DEFAULT_COORDINATOR_KILL_BUFFER_PERIOD : coordinatorKillBufferPeriod,
- segmentSchemaKillPeriod == null ? DEFAULT_COORDINATOR_SEGMENT_SCHEMA_KILL_PERIOD : segmentSchemaKillPeriod,
- segmentSchemaKillDurationToRetain == null ? DEFAULT_COORDINATOR_SEGMENT_SCHEMA_KILL_DURATION_TO_RETAIN : segmentSchemaKillDurationToRetain
- );
- }
- }
-}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java
index 43b1c50c969..173e4537cca 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java
@@ -69,7 +69,6 @@ import org.apache.druid.segment.transform.TransformSpec;
import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
import org.apache.druid.server.coordinator.CoordinatorCompactionConfig;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
import org.apache.druid.server.coordinator.UserCompactionTaskDimensionsConfig;
import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig;
@@ -78,6 +77,7 @@ import org.apache.druid.server.coordinator.UserCompactionTaskQueryTuningConfig;
import org.apache.druid.server.coordinator.UserCompactionTaskTransformConfig;
import org.apache.druid.server.coordinator.compact.CompactionSegmentSearchPolicy;
import org.apache.druid.server.coordinator.compact.NewestSegmentFirstPolicy;
+import org.apache.druid.server.coordinator.config.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.apache.druid.timeline.CompactionState;
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java
index f6475fbe89a..69ea92ac69a 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java
@@ -21,7 +21,7 @@ package org.apache.druid.server.coordinator.duty;
import org.apache.druid.audit.AuditManager;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.Duration;
@@ -56,14 +56,9 @@ public class KillAuditLogTest
@Test
public void testRunSkipIfLastRunLessThanPeriod()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5s"))
- .withCoordianatorAuditKillPeriod(new Duration(Long.MAX_VALUE))
- .withCoordinatorAuditKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- killAuditLog = new KillAuditLog(druidCoordinatorConfig, mockAuditManager);
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration(Long.MAX_VALUE), new Duration("PT1S"));
+ killAuditLog = new KillAuditLog(config, mockAuditManager);
killAuditLog.run(mockDruidCoordinatorRuntimeParams);
Mockito.verifyNoInteractions(mockAuditManager);
}
@@ -71,58 +66,11 @@ public class KillAuditLogTest
@Test
public void testRunNotSkipIfLastRunMoreThanPeriod()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5s"))
- .withCoordianatorAuditKillPeriod(new Duration("PT6S"))
- .withCoordinatorAuditKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- killAuditLog = new KillAuditLog(druidCoordinatorConfig, mockAuditManager);
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration("PT6S"), new Duration("PT1S"));
+ killAuditLog = new KillAuditLog(config, mockAuditManager);
killAuditLog.run(mockDruidCoordinatorRuntimeParams);
Mockito.verify(mockAuditManager).removeAuditLogsOlderThan(ArgumentMatchers.anyLong());
Assert.assertTrue(runStats.hasStat(Stats.Kill.AUDIT_LOGS));
}
-
- @Test
- public void testConstructorFailIfInvalidPeriod()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5s"))
- .withCoordianatorAuditKillPeriod(new Duration("PT3S"))
- .withCoordinatorAuditKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
-
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killAuditLog = new KillAuditLog(druidCoordinatorConfig, mockAuditManager)
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.audit.period] must be greater than"
- + " [druid.coordinator.period.metadataStoreManagementPeriod]",
- exception.getMessage()
- );
- }
-
- @Test
- public void testConstructorFailIfInvalidRetainDuration()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordianatorAuditKillPeriod(new Duration("PT6S"))
- .withCoordinatorAuditKillDurationToRetain(new Duration("PT-1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killAuditLog = new KillAuditLog(druidCoordinatorConfig, mockAuditManager)
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.audit.durationToRetain] must be 0 milliseconds or higher",
- exception.getMessage()
- );
- }
}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java
index d331204525a..ec4547c940c 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java
@@ -32,8 +32,8 @@ import org.apache.druid.server.coordinator.CoordinatorCompactionConfig;
import org.apache.druid.server.coordinator.CoordinatorConfigManager;
import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.Duration;
@@ -86,14 +86,10 @@ public class KillCompactionConfigTest
@Test
public void testRunSkipIfLastRunLessThanPeriod()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorCompactionKillPeriod(new Duration(Long.MAX_VALUE))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration(Long.MAX_VALUE), null);
killCompactionConfig = new KillCompactionConfig(
- druidCoordinatorConfig,
+ config,
mockSqlSegmentsMetadataManager,
coordinatorConfigManager
);
@@ -103,32 +99,6 @@ public class KillCompactionConfigTest
Assert.assertEquals(0, runStats.rowCount());
}
- @Test
- public void testConstructorFailIfInvalidPeriod()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorCompactionKillPeriod(new Duration("PT3S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
-
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killCompactionConfig = new KillCompactionConfig(
- druidCoordinatorConfig,
- mockSqlSegmentsMetadataManager,
- coordinatorConfigManager
- )
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.compaction.period] must be greater than"
- + " [druid.coordinator.period.metadataStoreManagementPeriod]",
- exception.getMessage()
- );
- }
-
-
@Test
public void testRunDoNothingIfCurrentConfigIsEmpty()
{
@@ -145,14 +115,10 @@ public class KillCompactionConfigTest
ArgumentMatchers.eq(CoordinatorCompactionConfig.empty()))
).thenReturn(CoordinatorCompactionConfig.empty());
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorCompactionKillPeriod(new Duration("PT6S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration("PT6S"), null);
killCompactionConfig = new KillCompactionConfig(
- druidCoordinatorConfig,
+ config,
mockSqlSegmentsMetadataManager,
coordinatorConfigManager
);
@@ -232,14 +198,10 @@ public class KillCompactionConfigTest
ArgumentMatchers.any())
).thenReturn(ConfigManager.SetResult.ok());
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorCompactionKillPeriod(new Duration("PT6S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration("PT6S"), null);
killCompactionConfig = new KillCompactionConfig(
- druidCoordinatorConfig,
+ config,
mockSqlSegmentsMetadataManager,
coordinatorConfigManager
);
@@ -324,14 +286,10 @@ public class KillCompactionConfigTest
ConfigManager.SetResult.ok()
);
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorCompactionKillPeriod(new Duration("PT6S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration("PT6S"), null);
killCompactionConfig = new KillCompactionConfig(
- druidCoordinatorConfig,
+ config,
mockSqlSegmentsMetadataManager,
coordinatorConfigManager
);
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadataTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadataTest.java
index 222d5a715fb..5e848cde073 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadataTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadataTest.java
@@ -23,7 +23,7 @@ import com.google.common.collect.ImmutableSet;
import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
import org.apache.druid.metadata.MetadataSupervisorManager;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.Duration;
@@ -61,14 +61,9 @@ public class KillDatasourceMetadataTest
@Test
public void testRunSkipIfLastRunLessThanPeriod()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorDatasourceKillPeriod(new Duration(Long.MAX_VALUE))
- .withCoordinatorDatasourceKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- killDatasourceMetadata = new KillDatasourceMetadata(druidCoordinatorConfig, mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration(Long.MAX_VALUE), new Duration("PT1S"));
+ killDatasourceMetadata = new KillDatasourceMetadata(config, mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
killDatasourceMetadata.run(mockDruidCoordinatorRuntimeParams);
Mockito.verifyNoInteractions(mockIndexerMetadataStorageCoordinator);
Mockito.verifyNoInteractions(mockMetadataSupervisorManager);
@@ -79,15 +74,10 @@ public class KillDatasourceMetadataTest
{
Mockito.when(mockDruidCoordinatorRuntimeParams.getCoordinatorStats()).thenReturn(runStats);
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorDatasourceKillPeriod(new Duration("PT6S"))
- .withCoordinatorDatasourceKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration("PT6S"), new Duration("PT1S"));
killDatasourceMetadata = new KillDatasourceMetadata(
- druidCoordinatorConfig,
+ config,
mockIndexerMetadataStorageCoordinator,
mockMetadataSupervisorManager
);
@@ -96,67 +86,12 @@ public class KillDatasourceMetadataTest
Assert.assertTrue(runStats.hasStat(Stats.Kill.DATASOURCES));
}
- @Test
- public void testConstructorFailIfInvalidPeriod()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorDatasourceKillPeriod(new Duration("PT3S"))
- .withCoordinatorDatasourceKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
-
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killDatasourceMetadata = new KillDatasourceMetadata(
- druidCoordinatorConfig,
- mockIndexerMetadataStorageCoordinator,
- mockMetadataSupervisorManager
- )
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.datasource.period] must be greater than"
- + " [druid.coordinator.period.metadataStoreManagementPeriod]",
- exception.getMessage()
- );
- }
-
- @Test
- public void testConstructorFailIfInvalidRetainDuration()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorDatasourceKillPeriod(new Duration("PT6S"))
- .withCoordinatorDatasourceKillDurationToRetain(new Duration("PT-1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killDatasourceMetadata = new KillDatasourceMetadata(
- druidCoordinatorConfig,
- mockIndexerMetadataStorageCoordinator,
- mockMetadataSupervisorManager
- )
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.datasource.durationToRetain] must be 0 milliseconds or higher",
- exception.getMessage()
- );
- }
-
@Test
public void testRunWithEmptyFilterExcludedDatasource()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorDatasourceKillPeriod(new Duration("PT6S"))
- .withCoordinatorDatasourceKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- killDatasourceMetadata = new KillDatasourceMetadata(druidCoordinatorConfig, mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration("PT6S"), new Duration("PT1S"));
+ killDatasourceMetadata = new KillDatasourceMetadata(config, mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
killDatasourceMetadata.run(mockDruidCoordinatorRuntimeParams);
Mockito.verify(mockIndexerMetadataStorageCoordinator).removeDataSourceMetadataOlderThan(ArgumentMatchers.anyLong(), ArgumentMatchers.eq(ImmutableSet.of()));
Assert.assertTrue(runStats.hasStat(Stats.Kill.DATASOURCES));
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillRulesTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillRulesTest.java
index f1663dffda5..fde2c2a28b9 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillRulesTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillRulesTest.java
@@ -21,7 +21,7 @@ package org.apache.druid.server.coordinator.duty;
import org.apache.druid.metadata.MetadataRuleManager;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.Duration;
@@ -56,14 +56,9 @@ public class KillRulesTest
@Test
public void testRunSkipIfLastRunLessThanPeriod()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorRuleKillPeriod(new Duration(Long.MAX_VALUE))
- .withCoordinatorRuleKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- killRules = new KillRules(druidCoordinatorConfig, mockRuleManager);
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration(Long.MAX_VALUE), new Duration("PT1S"));
+ killRules = new KillRules(config, mockRuleManager);
killRules.run(mockDruidCoordinatorRuntimeParams);
Mockito.verifyNoInteractions(mockRuleManager);
}
@@ -71,57 +66,11 @@ public class KillRulesTest
@Test
public void testRunNotSkipIfLastRunMoreThanPeriod()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorRuleKillPeriod(new Duration("PT6S"))
- .withCoordinatorRuleKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- killRules = new KillRules(druidCoordinatorConfig, mockRuleManager);
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration("PT6S"), new Duration("PT1S"));
+ killRules = new KillRules(config, mockRuleManager);
killRules.run(mockDruidCoordinatorRuntimeParams);
Mockito.verify(mockRuleManager).removeRulesForEmptyDatasourcesOlderThan(ArgumentMatchers.anyLong());
Assert.assertTrue(runStats.hasStat(Stats.Kill.RULES));
}
-
- @Test
- public void testConstructorFailIfInvalidPeriod()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorRuleKillPeriod(new Duration("PT3S"))
- .withCoordinatorRuleKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killRules = new KillRules(druidCoordinatorConfig, mockRuleManager)
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.rule.period] must be greater than"
- + " [druid.coordinator.period.metadataStoreManagementPeriod]",
- exception.getMessage()
- );
- }
-
- @Test
- public void testConstructorFailIfInvalidRetainDuration()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorRuleKillPeriod(new Duration("PT6S"))
- .withCoordinatorRuleKillDurationToRetain(new Duration("PT-1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killRules = new KillRules(druidCoordinatorConfig, mockRuleManager)
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.rule.durationToRetain] must be 0 milliseconds or higher",
- exception.getMessage()
- );
- }
}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsCustomDutyTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsCustomDutyTest.java
index e93e04f1b38..6dd296d3b51 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsCustomDutyTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsCustomDutyTest.java
@@ -20,13 +20,11 @@
package org.apache.druid.server.coordinator.duty;
import org.apache.druid.metadata.MetadataSupervisorManager;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.Duration;
import org.junit.Assert;
-import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentMatchers;
@@ -43,55 +41,14 @@ public class KillSupervisorsCustomDutyTest
@Mock
private DruidCoordinatorRuntimeParams mockDruidCoordinatorRuntimeParams;
- @Mock
- private DruidCoordinatorConfig coordinatorConfig;
-
private KillSupervisorsCustomDuty killSupervisors;
- @Before
- public void setup()
- {
- Mockito.when(coordinatorConfig.getCoordinatorMetadataStoreManagementPeriod())
- .thenReturn(new Duration(3600 * 1000));
- }
-
- @Test
- public void testConstructorFailIfRetainDurationNull()
- {
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killSupervisors = new KillSupervisorsCustomDuty(null, mockMetadataSupervisorManager, coordinatorConfig)
- );
- Assert.assertEquals(
- "[KillSupervisorsCustomDuty.durationToRetain] must be 0 milliseconds or higher",
- exception.getMessage()
- );
- }
-
- @Test
- public void testConstructorFailIfRetainDurationInvalid()
- {
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killSupervisors = new KillSupervisorsCustomDuty(
- new Duration("PT-1S"),
- mockMetadataSupervisorManager,
- coordinatorConfig
- )
- );
- Assert.assertEquals(
- "[KillSupervisorsCustomDuty.durationToRetain] must be 0 milliseconds or higher",
- exception.getMessage()
- );
- }
-
@Test
public void testConstructorSuccess()
{
killSupervisors = new KillSupervisorsCustomDuty(
new Duration("PT1S"),
- mockMetadataSupervisorManager,
- coordinatorConfig
+ mockMetadataSupervisorManager
);
Assert.assertNotNull(killSupervisors);
}
@@ -103,8 +60,7 @@ public class KillSupervisorsCustomDutyTest
Mockito.when(mockDruidCoordinatorRuntimeParams.getCoordinatorStats()).thenReturn(runStats);
killSupervisors = new KillSupervisorsCustomDuty(
new Duration("PT1S"),
- mockMetadataSupervisorManager,
- coordinatorConfig
+ mockMetadataSupervisorManager
);
killSupervisors.run(mockDruidCoordinatorRuntimeParams);
Mockito.verify(mockMetadataSupervisorManager).removeTerminatedSupervisorsOlderThan(ArgumentMatchers.anyLong());
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java
index bbe92a2cd10..7362c335d08 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java
@@ -21,7 +21,7 @@ package org.apache.druid.server.coordinator.duty;
import org.apache.druid.metadata.MetadataSupervisorManager;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Stats;
import org.joda.time.Duration;
@@ -56,14 +56,9 @@ public class KillSupervisorsTest
@Test
public void testRunSkipIfLastRunLessThanPeriod()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorSupervisorKillPeriod(new Duration(Long.MAX_VALUE))
- .withCoordinatorSupervisorKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- killSupervisors = new KillSupervisors(druidCoordinatorConfig, mockMetadataSupervisorManager);
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration(Long.MAX_VALUE), new Duration("PT1S"));
+ killSupervisors = new KillSupervisors(config, mockMetadataSupervisorManager);
killSupervisors.run(mockDruidCoordinatorRuntimeParams);
Mockito.verifyNoInteractions(mockMetadataSupervisorManager);
}
@@ -71,58 +66,11 @@ public class KillSupervisorsTest
@Test
public void testRunNotSkipIfLastRunMoreThanPeriod()
{
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorSupervisorKillPeriod(new Duration("PT6S"))
- .withCoordinatorSupervisorKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- killSupervisors = new KillSupervisors(druidCoordinatorConfig, mockMetadataSupervisorManager);
+ final MetadataCleanupConfig config
+ = new MetadataCleanupConfig(true, new Duration("PT6S"), new Duration("PT1S"));
+ killSupervisors = new KillSupervisors(config, mockMetadataSupervisorManager);
killSupervisors.run(mockDruidCoordinatorRuntimeParams);
Mockito.verify(mockMetadataSupervisorManager).removeTerminatedSupervisorsOlderThan(ArgumentMatchers.anyLong());
Assert.assertTrue(runStats.hasStat(Stats.Kill.SUPERVISOR_SPECS));
}
-
- @Test
- public void testConstructorFailIfInvalidPeriod()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorSupervisorKillPeriod(new Duration("PT3S"))
- .withCoordinatorSupervisorKillDurationToRetain(new Duration("PT1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killSupervisors = new KillSupervisors(druidCoordinatorConfig, mockMetadataSupervisorManager)
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.supervisor.period] must be greater than"
- + " [druid.coordinator.period.metadataStoreManagementPeriod]",
- exception.getMessage()
- );
- }
-
- @Test
- public void testConstructorFailIfInvalidRetainDuration()
- {
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(new Duration("PT5S"))
- .withCoordinatorSupervisorKillPeriod(new Duration("PT6S"))
- .withCoordinatorSupervisorKillDurationToRetain(new Duration("PT-1S"))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
-
- final IllegalArgumentException exception = Assert.assertThrows(
- IllegalArgumentException.class,
- () -> killSupervisors = new KillSupervisors(druidCoordinatorConfig, mockMetadataSupervisorManager)
- );
- Assert.assertEquals(
- "[druid.coordinator.kill.supervisor.durationToRetain] must be 0 milliseconds or higher",
- exception.getMessage()
- );
- }
}
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnreferencedSegmentSchemaDutyTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnreferencedSegmentSchemaDutyTest.java
index 25ad519eea6..a7701c4e104 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnreferencedSegmentSchemaDutyTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnreferencedSegmentSchemaDutyTest.java
@@ -36,9 +36,8 @@ import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
import org.apache.druid.segment.metadata.FingerprintGenerator;
import org.apache.druid.segment.metadata.SegmentSchemaManager;
import org.apache.druid.segment.metadata.SegmentSchemaTestUtils;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.MetadataCleanupConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
@@ -105,14 +104,13 @@ public class KillUnreferencedSegmentSchemaDutyTest
dateTimes.add(now.plusMinutes(61));
dateTimes.add(now.plusMinutes(6 * 60 + 1));
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(Period.parse("PT1H").toStandardDuration())
- .withSegmentSchemaKillPeriod(Period.parse("PT1H").toStandardDuration())
- .withSegmentSchemaKillDurationToRetain(Period.parse("PT6H").toStandardDuration())
- .build();
-
+ final MetadataCleanupConfig cleanupConfig = new MetadataCleanupConfig(
+ true,
+ Period.parse("PT1H").toStandardDuration(),
+ Period.parse("PT6H").toStandardDuration()
+ );
KillUnreferencedSegmentSchemaDuty duty =
- new TestKillUnreferencedSegmentSchemasDuty(druidCoordinatorConfig, segmentSchemaManager, dateTimes);
+ new TestKillUnreferencedSegmentSchemasDuty(cleanupConfig, segmentSchemaManager, dateTimes);
Set<DataSegment> segments = new HashSet<>();
List<SegmentSchemaManager.SegmentSchemaMetadataPlus> schemaMetadataPluses = new ArrayList<>();
@@ -211,14 +209,14 @@ public class KillUnreferencedSegmentSchemaDutyTest
dateTimes.add(now);
dateTimes.add(now.plusMinutes(61));
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(Period.parse("PT1H").toStandardDuration())
- .withSegmentSchemaKillPeriod(Period.parse("PT1H").toStandardDuration())
- .withSegmentSchemaKillDurationToRetain(Period.parse("PT6H").toStandardDuration())
- .build();
+ MetadataCleanupConfig cleanupConfig = new MetadataCleanupConfig(
+ true,
+ Period.parse("PT1H").toStandardDuration(),
+ Period.parse("PT6H").toStandardDuration()
+ );
KillUnreferencedSegmentSchemaDuty duty =
- new TestKillUnreferencedSegmentSchemasDuty(druidCoordinatorConfig, segmentSchemaManager, dateTimes);
+ new TestKillUnreferencedSegmentSchemasDuty(cleanupConfig, segmentSchemaManager, dateTimes);
RowSignature rowSignature = RowSignature.builder().add("c1", ColumnType.FLOAT).build();
@@ -283,14 +281,14 @@ public class KillUnreferencedSegmentSchemaDutyTest
dateTimes.add(now.plusMinutes(61));
dateTimes.add(now.plusMinutes(6 * 60 + 1));
- TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withMetadataStoreManagementPeriod(Period.parse("PT1H").toStandardDuration())
- .withSegmentSchemaKillPeriod(Period.parse("PT1H").toStandardDuration())
- .withSegmentSchemaKillDurationToRetain(Period.parse("PT6H").toStandardDuration())
- .build();
+ MetadataCleanupConfig cleanupConfig = new MetadataCleanupConfig(
+ true,
+ Period.parse("PT1H").toStandardDuration(),
+ Period.parse("PT6H").toStandardDuration()
+ );
KillUnreferencedSegmentSchemaDuty duty =
- new TestKillUnreferencedSegmentSchemasDuty(druidCoordinatorConfig, segmentSchemaManager, dateTimes);
+ new TestKillUnreferencedSegmentSchemasDuty(cleanupConfig, segmentSchemaManager, dateTimes);
// create 2 versions of same schema
// unreferenced one should get deleted
@@ -369,7 +367,7 @@ public class KillUnreferencedSegmentSchemaDutyTest
private int index = -1;
public TestKillUnreferencedSegmentSchemasDuty(
- DruidCoordinatorConfig config,
+ MetadataCleanupConfig config,
SegmentSchemaManager segmentSchemaManager,
List<DateTime> dateTimes
)
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java
index ca6ddc55029..2455610ceac 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java
@@ -28,8 +28,6 @@ import org.apache.druid.client.indexing.IndexingTotalWorkerCapacityInfo;
import org.apache.druid.client.indexing.IndexingWorker;
import org.apache.druid.client.indexing.IndexingWorkerInfo;
import org.apache.druid.client.indexing.NoopOverlordClient;
-import org.apache.druid.error.DruidException;
-import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.indexer.RunnerTaskState;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskState;
@@ -46,14 +44,13 @@ import org.apache.druid.segment.TestHelper;
import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.KillUnusedSegmentsConfig;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Dimension;
import org.apache.druid.server.coordinator.stats.RowKey;
import org.apache.druid.server.coordinator.stats.Stats;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.partition.NoneShardSpec;
-import org.hamcrest.MatcherAssert;
import org.joda.time.DateTime;
import org.joda.time.Duration;
import org.joda.time.Interval;
@@ -93,7 +90,7 @@ public class KillUnusedSegmentsTest
private CoordinatorDynamicConfig.Builder dynamicConfigBuilder;
private TestOverlordClient overlordClient;
- private TestDruidCoordinatorConfig.Builder configBuilder;
+ private KillUnusedSegmentsConfig.Builder configBuilder;
private DruidCoordinatorRuntimeParams.Builder paramsBuilder;
private KillUnusedSegments killDuty;
@@ -127,11 +124,11 @@ public class KillUnusedSegmentsTest
connector.createSegmentTable();
overlordClient = new TestOverlordClient();
- configBuilder = new TestDruidCoordinatorConfig.Builder()
- .withCoordinatorIndexingPeriod(Duration.standardSeconds(0))
- .withCoordinatorKillDurationToRetain(Duration.standardHours(36))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillBufferPeriod(Duration.standardSeconds(1));
+ configBuilder = KillUnusedSegmentsConfig.builder()
+ .withCleanupPeriod(Duration.standardSeconds(0))
+ .withDurationToRetain(Duration.standardHours(36))
+ .withMaxSegmentsToKill(10)
+ .withBufferPeriod(Duration.standardSeconds(1));
dynamicConfigBuilder = CoordinatorDynamicConfig.builder()
.withKillTaskSlotRatio(1.0);
paramsBuilder = DruidCoordinatorRuntimeParams.newBuilder(DateTimes.nowUtc());
@@ -140,7 +137,7 @@ public class KillUnusedSegmentsTest
@Test
public void testKillWithDefaultCoordinatorConfig()
{
- configBuilder = new TestDruidCoordinatorConfig.Builder();
+ configBuilder = KillUnusedSegmentsConfig.builder();
dynamicConfigBuilder = CoordinatorDynamicConfig.builder();
final DateTime sixtyDaysAgo = NOW.minusDays(60);
@@ -182,8 +179,7 @@ public class KillUnusedSegmentsTest
@Test
public void testKillWithMultipleDatasources()
{
- configBuilder.withCoordinatorKillIgnoreDurationToRetain(true);
- configBuilder.withCoordinatorKillMaxSegments(2);
+ configBuilder.withIgnoreDurationToRetain(true).withMaxSegmentsToKill(2);
createAndAddUnusedSegment(DS1, YEAR_OLD, VERSION, NOW.minusDays(1));
createAndAddUnusedSegment(DS1, MONTH_OLD, VERSION, NOW.minusDays(1));
@@ -240,8 +236,8 @@ public class KillUnusedSegmentsTest
@Test
public void testKillWithDifferentLastUpdatedTimesInWideInterval()
{
- configBuilder.withCoordinatorKillIgnoreDurationToRetain(true);
- configBuilder.withCoordinatorKillBufferPeriod(Duration.standardDays(3));
+ configBuilder.withIgnoreDurationToRetain(true)
+ .withBufferPeriod(Duration.standardDays(3));
createAndAddUnusedSegment(DS1, YEAR_OLD, VERSION, NOW.minusDays(10));
createAndAddUnusedSegment(DS1, MONTH_OLD, VERSION, NOW.minusDays(10));
@@ -269,8 +265,8 @@ public class KillUnusedSegmentsTest
@Test
public void testAddOlderSegmentsAfterInitialRun()
{
- configBuilder.withCoordinatorKillIgnoreDurationToRetain(true);
- configBuilder.withCoordinatorKillMaxSegments(2);
+ configBuilder.withIgnoreDurationToRetain(true)
+ .withMaxSegmentsToKill(2);
createAndAddUnusedSegment(DS1, DAY_OLD, VERSION, NOW.minusDays(1));
createAndAddUnusedSegment(DS1, NEXT_DAY, VERSION, NOW.minusDays(1));
@@ -347,7 +343,7 @@ public class KillUnusedSegmentsTest
@Test
public void testNegativeDurationToRetain()
{
- configBuilder.withCoordinatorKillDurationToRetain(Duration.standardHours(36).negated());
+ configBuilder.withDurationToRetain(Duration.standardHours(36).negated());
createAndAddUnusedSegment(DS1, YEAR_OLD, VERSION, NOW.minusDays(10));
createAndAddUnusedSegment(DS1, MONTH_OLD, VERSION, NOW.minusDays(10));
@@ -371,7 +367,7 @@ public class KillUnusedSegmentsTest
@Test
public void testIgnoreDurationToRetain()
{
- configBuilder.withCoordinatorKillIgnoreDurationToRetain(true);
+ configBuilder.withIgnoreDurationToRetain(true);
createAndAddUnusedSegment(DS1, YEAR_OLD, VERSION, NOW.minusDays(10));
createAndAddUnusedSegment(DS1, MONTH_OLD, VERSION, NOW.minusDays(10));
@@ -395,7 +391,7 @@ public class KillUnusedSegmentsTest
@Test
public void testLowerMaxSegmentsToKill()
{
- configBuilder.withCoordinatorKillMaxSegments(1);
+ configBuilder.withMaxSegmentsToKill(1);
createAndAddUnusedSegment(DS1, YEAR_OLD, VERSION, NOW.minusDays(10));
createAndAddUnusedSegment(DS1, MONTH_OLD, VERSION, NOW.minusDays(10));
@@ -418,7 +414,7 @@ public class KillUnusedSegmentsTest
@Test
public void testLargeKillPeriod()
{
- configBuilder.withCoordinatorKillPeriod(Duration.standardHours(1));
+ configBuilder.withCleanupPeriod(Duration.standardHours(1));
createAndAddUnusedSegment(DS1, YEAR_OLD, VERSION, NOW.minusDays(10));
createAndAddUnusedSegment(DS1, MONTH_OLD, VERSION, NOW.minusDays(10));
@@ -612,7 +608,7 @@ public class KillUnusedSegmentsTest
@Test
public void testKillFirstHalfEternitySegment()
{
- configBuilder.withCoordinatorKillIgnoreDurationToRetain(true);
+ configBuilder.withIgnoreDurationToRetain(true);
final Interval firstHalfEternity = new Interval(DateTimes.MIN, DateTimes.of("2024"));
createAndAddUnusedSegment(DS1, firstHalfEternity, VERSION, NOW.minusDays(60));
@@ -631,7 +627,7 @@ public class KillUnusedSegmentsTest
@Test
public void testKillEternitySegment()
{
- configBuilder.withCoordinatorKillIgnoreDurationToRetain(true);
+ configBuilder.withIgnoreDurationToRetain(true);
createAndAddUnusedSegment(DS1, Intervals.ETERNITY, VERSION, NOW.minusDays(60));
@@ -649,7 +645,7 @@ public class KillUnusedSegmentsTest
@Test
public void testKillSecondHalfEternitySegment()
{
- configBuilder.withCoordinatorKillIgnoreDurationToRetain(true);
+ configBuilder.withIgnoreDurationToRetain(true);
final Interval secondHalfEternity = new Interval(DateTimes.of("1970"), DateTimes.MAX);
createAndAddUnusedSegment(DS1, secondHalfEternity, VERSION, NOW.minusDays(60));
@@ -688,7 +684,7 @@ public class KillUnusedSegmentsTest
@Test
public void testKillMultipleSegmentsInSameInterval()
{
- configBuilder.withCoordinatorKillIgnoreDurationToRetain(true);
+ configBuilder.withIgnoreDurationToRetain(true);
createAndAddUnusedSegment(DS1, YEAR_OLD, "v1", NOW.minusDays(10));
createAndAddUnusedSegment(DS1, YEAR_OLD, "v2", NOW.minusDays(10));
@@ -705,56 +701,6 @@ public class KillUnusedSegmentsTest
validateLastKillStateAndReset(DS1, YEAR_OLD);
}
- @Test
- public void testKillPeriodShorterThanIndexingPeriod()
- {
- MatcherAssert.assertThat(
- Assert.assertThrows(
- DruidException.class,
- () -> new KillUnusedSegments(
- sqlSegmentsMetadataManager,
- overlordClient,
- new TestDruidCoordinatorConfig.Builder()
- .withCoordinatorIndexingPeriod(Duration.standardSeconds(10))
- .withCoordinatorKillPeriod(Duration.standardSeconds(5))
- .build()
- )
- ),
- new DruidExceptionMatcher(
- DruidException.Persona.OPERATOR,
- DruidException.Category.INVALID_INPUT,
- "general"
- ).expectMessageIs(
- "druid.coordinator.kill.period[PT5S] is invalid. It must be greater than or equal to"
- + " druid.coordinator.period.indexingPeriod[PT10S]."
- )
- );
- }
-
- @Test
- public void testNegativeMaxKillSegments()
- {
- MatcherAssert.assertThat(
- Assert.assertThrows(
- DruidException.class,
- () -> new KillUnusedSegments(
- sqlSegmentsMetadataManager,
- overlordClient,
- new TestDruidCoordinatorConfig.Builder()
- .withCoordinatorKillMaxSegments(-5)
- .build()
- )
- ),
- new DruidExceptionMatcher(
- DruidException.Persona.OPERATOR,
- DruidException.Category.INVALID_INPUT,
- "general"
- ).expectMessageIs(
- "druid.coordinator.kill.maxSegments[-5] is invalid. It must be a positive integer."
- )
- );
- }
-
private void validateLastKillStateAndReset(final String dataSource, @Nullable final Interval expectedKillInterval)
{
final Interval observedLastKillInterval = overlordClient.getLastKillInterval(dataSource);
@@ -811,7 +757,6 @@ public class KillUnusedSegmentsTest
);
}
-
private void initDuty()
{
killDuty = new KillUnusedSegments(sqlSegmentsMetadataManager, overlordClient, configBuilder.build());
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeonTest.java b/server/src/test/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeonTest.java
index c3d3fb0fbe8..de1801d7ad7 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeonTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/loading/HttpLoadQueuePeonTest.java
@@ -33,7 +33,7 @@ import org.apache.druid.server.coordination.DataSegmentChangeHandler;
import org.apache.druid.server.coordination.DataSegmentChangeRequest;
import org.apache.druid.server.coordination.SegmentLoadDropHandler;
import org.apache.druid.server.coordinator.CreateDataSegments;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.HttpLoadQueuePeonConfig;
import org.apache.druid.server.coordinator.simulate.BlockingExecutorService;
import org.apache.druid.server.coordinator.simulate.WrappingScheduledExecutorService;
import org.apache.druid.timeline.DataSegment;
@@ -89,9 +89,7 @@ public class HttpLoadQueuePeonTest
"http://dummy:4000",
ServerTestHelper.MAPPER,
httpClient,
- new TestDruidCoordinatorConfig.Builder()
- .withHttpLoadQueuePeonBatchSize(10)
- .build(),
+ new HttpLoadQueuePeonConfig(null, null, 10),
new WrappingScheduledExecutorService("HttpLoadQueuePeonTest-%s", processingExecutor, true),
callbackExecutor
);
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/loading/LoadQueuePeonTest.java b/server/src/test/java/org/apache/druid/server/coordinator/loading/LoadQueuePeonTest.java
index aba1c9b6be6..3e1c816716c 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/loading/LoadQueuePeonTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/loading/LoadQueuePeonTest.java
@@ -38,7 +38,6 @@ import org.apache.druid.server.coordination.DataSegmentChangeHandler;
import org.apache.druid.server.coordination.DataSegmentChangeRequest;
import org.apache.druid.server.coordination.SegmentChangeRequestDrop;
import org.apache.druid.server.coordination.SegmentChangeRequestLoad;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.NoneShardSpec;
@@ -89,10 +88,7 @@ public class LoadQueuePeonTest extends CuratorTestBase
jsonMapper,
Execs.scheduledSingleThreaded("test_load_queue_peon_scheduled-%d"),
Execs.singleThreaded("test_load_queue_peon-%d"),
- new TestDruidCoordinatorConfig.Builder()
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build()
+ Duration.standardMinutes(15)
);
loadQueuePeon.start();
@@ -279,11 +275,7 @@ public class LoadQueuePeonTest extends CuratorTestBase
Execs.scheduledSingleThreaded("test_load_queue_peon_scheduled-%d"),
Execs.singleThreaded("test_load_queue_peon-%d"),
// set time-out to 1 ms so that LoadQueuePeon will fail the assignment quickly
- new TestDruidCoordinatorConfig.Builder()
- .withLoadTimeoutDelay(new Duration(1))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build()
+ new Duration(1)
);
loadQueuePeon.start();
@@ -321,12 +313,8 @@ public class LoadQueuePeonTest extends CuratorTestBase
Execs.singleThreaded("test_load_queue_peon-%d"),
// The timeout here was set to 1ms, when this test was acting flakey. A cursory glance makes me wonder if
// there's a race where the timeout actually happens before other code can run. 1ms timeout seems aggressive.
- // 100ms is a great price to pay if it removes the flakeyness
- new TestDruidCoordinatorConfig.Builder()
- .withLoadTimeoutDelay(new Duration(100))
- .withCoordinatorKillMaxSegments(10)
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build()
+ // 100ms is a great price to pay if it removes the flakeyness,
+ new Duration(100)
);
loadQueuePeon.start();
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
index 023cd5977d9..7b6eb280a39 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
@@ -41,9 +41,7 @@ import org.apache.druid.server.coordinator.CoordinatorCompactionConfig;
import org.apache.druid.server.coordinator.CoordinatorConfigManager;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
import org.apache.druid.server.coordinator.DruidCoordinator;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.MetadataManager;
-import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
import org.apache.druid.server.coordinator.balancer.BalancerStrategyFactory;
import org.apache.druid.server.coordinator.balancer.CachingCostBalancerStrategyConfig;
import org.apache.druid.server.coordinator.balancer.CachingCostBalancerStrategyFactory;
@@ -52,6 +50,11 @@ import org.apache.druid.server.coordinator.balancer.DiskNormalizedCostBalancerSt
import org.apache.druid.server.coordinator.balancer.RandomBalancerStrategyFactory;
import org.apache.druid.server.coordinator.compact.CompactionSegmentSearchPolicy;
import org.apache.druid.server.coordinator.compact.NewestSegmentFirstPolicy;
+import org.apache.druid.server.coordinator.config.CoordinatorKillConfigs;
+import org.apache.druid.server.coordinator.config.CoordinatorPeriodConfig;
+import org.apache.druid.server.coordinator.config.CoordinatorRunConfig;
+import org.apache.druid.server.coordinator.config.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.HttpLoadQueuePeonConfig;
import org.apache.druid.server.coordinator.duty.CoordinatorCustomDutyGroups;
import org.apache.druid.server.coordinator.loading.LoadQueueTaskMaster;
import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
@@ -182,7 +185,8 @@ public class CoordinatorSimulationBuilder
serverInventoryView,
dynamicConfig,
loadImmediately,
- autoSyncInventory
+ autoSyncInventory,
+ balancerStrategy
);
if (segments != null) {
@@ -206,7 +210,6 @@ public class CoordinatorSimulationBuilder
new ServiceAnnouncer.Noop(),
null,
new CoordinatorCustomDutyGroups(Collections.emptySet()),
- createBalancerStrategy(env),
env.lookupCoordinatorManager,
env.leaderSelector,
COMPACTION_SEGMENT_SEARCH_POLICY,
@@ -217,40 +220,6 @@ public class CoordinatorSimulationBuilder
return new SimulationImpl(coordinator, env);
}
- private BalancerStrategyFactory createBalancerStrategy(Environment env)
- {
- if (balancerStrategy == null) {
- return new CostBalancerStrategyFactory();
- }
-
- switch (balancerStrategy) {
- case "cost":
- return new CostBalancerStrategyFactory();
- case "cachingCost":
- return buildCachingCostBalancerStrategy(env);
- case "diskNormalized":
- return new DiskNormalizedCostBalancerStrategyFactory();
- case "random":
- return new RandomBalancerStrategyFactory();
- default:
- throw new IAE("Unknown balancer stratgy: " + balancerStrategy);
- }
- }
-
- private BalancerStrategyFactory buildCachingCostBalancerStrategy(Environment env)
- {
- try {
- return new CachingCostBalancerStrategyFactory(
- env.coordinatorInventoryView,
- env.lifecycle,
- new CachingCostBalancerStrategyConfig()
- );
- }
- catch (Exception e) {
- throw new ISE(e, "Error building balancer strategy");
- }
- }
-
/**
* Implementation of {@link CoordinatorSimulation}.
*/
@@ -455,21 +424,13 @@ public class CoordinatorSimulationBuilder
TestServerInventoryView clusterInventory,
CoordinatorDynamicConfig dynamicConfig,
boolean loadImmediately,
- boolean autoSyncInventory
+ boolean autoSyncInventory,
+ String balancerStrategy
)
{
this.inventory = clusterInventory;
this.loadImmediately = loadImmediately;
this.autoSyncInventory = autoSyncInventory;
-
- this.coordinatorConfig = new TestDruidCoordinatorConfig.Builder()
- .withCoordinatorStartDelay(new Duration(1L))
- .withCoordinatorPeriod(Duration.standardMinutes(1))
- .withCoordinatorKillPeriod(Duration.millis(100))
- .withLoadQueuePeonType("http")
- .withCoordinatorKillIgnoreDurationToRetain(false)
- .build();
-
this.executorFactory = new ExecutorFactory(loadImmediately);
this.coordinatorInventoryView = autoSyncInventory
? clusterInventory
@@ -480,14 +441,19 @@ public class CoordinatorSimulationBuilder
executorFactory.create(1, ExecutorFactory.HISTORICAL_LOADER)
);
+ this.coordinatorConfig = new DruidCoordinatorConfig(
+ new CoordinatorRunConfig(new Duration(1L), Duration.standardMinutes(1)),
+ new CoordinatorPeriodConfig(null, null),
+ CoordinatorKillConfigs.DEFAULT,
+ createBalancerStrategy(balancerStrategy),
+ new HttpLoadQueuePeonConfig(null, null, null)
+ );
this.loadQueueTaskMaster = new LoadQueueTaskMaster(
- null,
OBJECT_MAPPER,
executorFactory.create(1, ExecutorFactory.LOAD_QUEUE_EXECUTOR),
executorFactory.create(1, ExecutorFactory.LOAD_CALLBACK_EXECUTOR),
- coordinatorConfig,
- httpClient,
- null
+ coordinatorConfig.getHttpLoadQueuePeonConfig(),
+ httpClient
);
this.loadQueueManager =
new SegmentLoadQueueManager(coordinatorInventoryView, loadQueueTaskMaster);
@@ -554,6 +520,40 @@ public class CoordinatorSimulationBuilder
return jacksonConfigManager;
}
+
+ private BalancerStrategyFactory createBalancerStrategy(String strategyName)
+ {
+ if (strategyName == null) {
+ return new CostBalancerStrategyFactory();
+ }
+
+ switch (strategyName) {
+ case "cost":
+ return new CostBalancerStrategyFactory();
+ case "cachingCost":
+ return buildCachingCostBalancerStrategy();
+ case "diskNormalized":
+ return new DiskNormalizedCostBalancerStrategyFactory();
+ case "random":
+ return new RandomBalancerStrategyFactory();
+ default:
+ throw new IAE("Unknown balancer stratgy: " + strategyName);
+ }
+ }
+
+ private BalancerStrategyFactory buildCachingCostBalancerStrategy()
+ {
+ try {
+ return new CachingCostBalancerStrategyFactory(
+ this.coordinatorInventoryView,
+ this.lifecycle,
+ new CachingCostBalancerStrategyConfig()
+ );
+ }
+ catch (Exception e) {
+ throw new ISE(e, "Error building balancer strategy");
+ }
+ }
}
/**
diff --git a/services/pom.xml b/services/pom.xml
index 45193df19ae..514de6de7e4 100644
--- a/services/pom.xml
+++ b/services/pom.xml
@@ -95,10 +95,6 @@
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
</dependency>
- <dependency>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-framework</artifactId>
- </dependency>
<dependency>
<groupId>org.apache.calcite.avatica</groupId>
<artifactId>avatica-core</artifactId>
diff --git a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
index f464e2a7868..7adaac5b88d 100644
--- a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
+++ b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
@@ -35,7 +35,6 @@ import com.google.inject.TypeLiteral;
import com.google.inject.multibindings.MapBinder;
import com.google.inject.name.Names;
import com.google.inject.util.Providers;
-import org.apache.curator.framework.CuratorFramework;
import org.apache.druid.client.CoordinatorSegmentWatcherConfig;
import org.apache.druid.client.CoordinatorServerView;
import org.apache.druid.client.DirectDruidClientFactory;
@@ -44,7 +43,6 @@ import org.apache.druid.client.InternalQueryConfig;
import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.discovery.DruidLeaderSelector;
import org.apache.druid.discovery.NodeRole;
-import org.apache.druid.guice.ConfigProvider;
import org.apache.druid.guice.DruidBinders;
import org.apache.druid.guice.Jerseys;
import org.apache.druid.guice.JsonConfigProvider;
@@ -101,12 +99,16 @@ import org.apache.druid.server.QueryScheduler;
import org.apache.druid.server.QuerySchedulerProvider;
import org.apache.druid.server.coordinator.CoordinatorConfigManager;
import org.apache.druid.server.coordinator.DruidCoordinator;
-import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
import org.apache.druid.server.coordinator.MetadataManager;
import org.apache.druid.server.coordinator.balancer.BalancerStrategyFactory;
import org.apache.druid.server.coordinator.balancer.CachingCostBalancerStrategyConfig;
import org.apache.druid.server.coordinator.compact.CompactionSegmentSearchPolicy;
import org.apache.druid.server.coordinator.compact.NewestSegmentFirstPolicy;
+import org.apache.druid.server.coordinator.config.CoordinatorKillConfigs;
+import org.apache.druid.server.coordinator.config.CoordinatorPeriodConfig;
+import org.apache.druid.server.coordinator.config.CoordinatorRunConfig;
+import org.apache.druid.server.coordinator.config.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.config.HttpLoadQueuePeonConfig;
import org.apache.druid.server.coordinator.duty.CoordinatorCustomDuty;
import org.apache.druid.server.coordinator.duty.CoordinatorCustomDutyGroup;
import org.apache.druid.server.coordinator.duty.CoordinatorCustomDutyGroups;
@@ -127,7 +129,6 @@ import org.apache.druid.server.http.RulesResource;
import org.apache.druid.server.http.SelfDiscoveryResource;
import org.apache.druid.server.http.ServersResource;
import org.apache.druid.server.http.TiersResource;
-import org.apache.druid.server.initialization.ZkPathsConfig;
import org.apache.druid.server.initialization.jetty.JettyServerInitializer;
import org.apache.druid.server.lookup.cache.LookupCoordinatorManager;
import org.apache.druid.server.lookup.cache.LookupCoordinatorManagerConfig;
@@ -211,13 +212,15 @@ public class CliCoordinator extends ServerRunnable
binder.bindConstant().annotatedWith(Names.named("servicePort")).to(8081);
binder.bindConstant().annotatedWith(Names.named("tlsServicePort")).to(8281);
- ConfigProvider.bind(binder, DruidCoordinatorConfig.class);
-
binder.bind(MetadataStorage.class).toProvider(MetadataStorageProvider.class);
JsonConfigProvider.bind(binder, SegmentsMetadataManagerConfig.CONFIG_PREFIX, SegmentsMetadataManagerConfig.class);
JsonConfigProvider.bind(binder, "druid.manager.rules", MetadataRuleManagerConfig.class);
JsonConfigProvider.bind(binder, "druid.manager.lookups", LookupCoordinatorManagerConfig.class);
+ JsonConfigProvider.bind(binder, "druid.coordinator", CoordinatorRunConfig.class);
+ JsonConfigProvider.bind(binder, "druid.coordinator.kill", CoordinatorKillConfigs.class);
+ JsonConfigProvider.bind(binder, "druid.coordinator.period", CoordinatorPeriodConfig.class);
+ JsonConfigProvider.bind(binder, "druid.coordinator.loadqueuepeon.http", HttpLoadQueuePeonConfig.class);
JsonConfigProvider.bind(binder, "druid.coordinator.balancer", BalancerStrategyFactory.class);
JsonConfigProvider.bind(binder, "druid.coordinator.segment", CoordinatorSegmentWatcherConfig.class);
JsonConfigProvider.bind(
@@ -226,6 +229,7 @@ public class CliCoordinator extends ServerRunnable
CachingCostBalancerStrategyConfig.class
);
JsonConfigProvider.bind(binder, "druid.coordinator.segmentMetadataCache", SegmentMetadataCacheConfig.class);
+ binder.bind(DruidCoordinatorConfig.class);
binder.bind(RedirectFilter.class).in(LazySingleton.class);
if (beOverlord) {
@@ -319,34 +323,21 @@ public class CliCoordinator extends ServerRunnable
@Provides
@LazySingleton
public LoadQueueTaskMaster getLoadQueueTaskMaster(
- Provider<CuratorFramework> curatorFrameworkProvider,
ObjectMapper jsonMapper,
ScheduledExecutorFactory factory,
DruidCoordinatorConfig config,
@EscalatedGlobal HttpClient httpClient,
- ZkPathsConfig zkPaths,
Lifecycle lifecycle
)
{
- boolean useHttpLoadQueuePeon = "http".equalsIgnoreCase(config.getLoadQueuePeonType());
- ExecutorService callBackExec;
- if (useHttpLoadQueuePeon) {
- callBackExec = Execs.singleThreaded("LoadQueuePeon-callbackexec--%d");
- } else {
- callBackExec = Execs.multiThreaded(
- config.getNumCuratorCallBackThreads(),
- "LoadQueuePeon-callbackexec--%d"
- );
- }
+ final ExecutorService callBackExec = Execs.singleThreaded("LoadQueuePeon-callbackexec--%d");
ExecutorServices.manageLifecycle(lifecycle, callBackExec);
return new LoadQueueTaskMaster(
- curatorFrameworkProvider,
jsonMapper,
factory.create(1, "Master-PeonExec--%d"),
callBackExec,
- config,
- httpClient,
- zkPaths
+ config.getHttpLoadQueuePeonConfig(),
+ httpClient
);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org