This is an automated email from the ASF dual-hosted git repository.
maytasm 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 4326e69 Add feature to automatically remove datasource metadata based
on retention period (#11227)
4326e69 is described below
commit 4326e699bd3e146e1899f014e8765c52241f3cd3
Author: Maytas Monsereenusorn <mayt...@apache.org>
AuthorDate: Tue May 11 01:22:33 2021 -0700
Add feature to automatically remove datasource metadata based on retention
period (#11227)
* add auto clean up datasource metadata
* add test
* fix checkstyle
* add comments
* fix error
* address comments
* Address comments
* fix test
* fix test
* fix typo
* add comment
* fix test
* fix test
---
docs/configuration/index.md | 3 +
docs/operations/metrics.md | 1 +
.../TestIndexerMetadataStorageCoordinator.java | 8 ++
.../IndexerMetadataStorageCoordinator.java | 10 ++
.../IndexerSQLMetadataStorageCoordinator.java | 37 +++++++
.../druid/metadata/MetadataSupervisorManager.java | 19 ++++
.../metadata/SQLMetadataSupervisorManager.java | 41 +++++--
.../server/coordinator/DruidCoordinatorConfig.java | 8 ++
.../server/coordinator/duty/KillAuditLog.java | 32 +++---
.../coordinator/duty/KillDatasourceMetadata.java | 121 +++++++++++++++++++++
.../druid/server/coordinator/duty/KillRules.java | 32 +++---
.../server/coordinator/duty/KillSupervisors.java | 32 +++---
.../IndexerSQLMetadataStorageCoordinatorTest.java | 78 +++++++++++++
.../metadata/SQLMetadataSupervisorManagerTest.java | 40 +++++++
.../coordinator/CuratorDruidCoordinatorTest.java | 2 +
.../server/coordinator/DruidCoordinatorTest.java | 2 +
.../server/coordinator/HttpLoadQueuePeonTest.java | 2 +
.../server/coordinator/LoadQueuePeonTest.java | 6 +
.../server/coordinator/LoadQueuePeonTester.java | 2 +
.../coordinator/TestDruidCoordinatorConfig.java | 18 +++
.../server/coordinator/duty/KillAuditLogTest.java | 8 ++
...rsTest.java => KillDatasourceMetadataTest.java} | 85 ++++++++++++---
.../server/coordinator/duty/KillRulesTest.java | 8 ++
.../coordinator/duty/KillSupervisorsTest.java | 8 ++
.../coordinator/duty/KillUnusedSegmentsTest.java | 2 +
.../java/org/apache/druid/cli/CliCoordinator.java | 8 +-
26 files changed, 547 insertions(+), 66 deletions(-)
diff --git a/docs/configuration/index.md b/docs/configuration/index.md
index 986a448..fdb9365 100644
--- a/docs/configuration/index.md
+++ b/docs/configuration/index.md
@@ -756,6 +756,9 @@ These Coordinator static configurations can be defined in
the `coordinator/runti
|`druid.coordinator.kill.rule.on`| Boolean value for whether to enable
automatic deletion of rules. If set to true, Coordinator will periodically
remove rules of inactive datasource (datasource with no used and unused
segments) from the rule table in metadata storage.| No | False|
|`druid.coordinator.kill.rule.period`| How often to do automatic deletion of
rules in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format.
Value must be equal to or greater than
`druid.coordinator.period.metadataStoreManagementPeriod`. Only applies if
`druid.coordinator.kill.rule.on` is set to "True".| No| `P1D`|
|`druid.coordinator.kill.rule.durationToRetain`| Duration of rules to be
retained from created time in [ISO
8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if
`druid.coordinator.kill.rule.on` is set to "True".| Yes if
`druid.coordinator.kill.rule.on` is set to "True".| None|
+|`druid.coordinator.kill.datasource.on`| Boolean value for whether to enable
automatic deletion of datasource metadata (Note: datasource metadata only
exists for datasource created from supervisor). If set to true, Coordinator
will periodically remove datasource metadata of terminated supervisor from the
datasource table in metadata storage. | No | False|
+|`druid.coordinator.kill.datasource.period`| How often to do automatic
deletion of datasource metadata in [ISO
8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Value must be
equal to or greater than
`druid.coordinator.period.metadataStoreManagementPeriod`. Only applies if
`druid.coordinator.kill.datasource.on` is set to "True".| No| `P1D`|
+|`druid.coordinator.kill.datasource.durationToRetain`| Duration of datasource
metadata to be retained from created time in [ISO
8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if
`druid.coordinator.kill.datasource.on` is set to "True".| Yes if
`druid.coordinator.kill.datasource.on` is set to "True".| None|
##### Segment Management
|Property|Possible Values|Description|Default|
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index b68dcca..a4aa9ab 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -259,6 +259,7 @@ These metrics are for the Druid Coordinator and are reset
each time the Coordina
|`metadata/kill/supervisor/count`|Total number of terminated supervisors that
were automatically deleted from metadata store per each Coordinator kill
supervisor duty run. This metric can help adjust
`druid.coordinator.kill.supervisor.durationToRetain` configuration based on
whether more or less terminated supervisors need to be deleted per cycle. Note
that this metric is only emitted when `druid.coordinator.kill.supervisor.on` is
set to true.| |Varies.|
|`metadata/kill/audit/count`|Total number of audit logs that were
automatically deleted from metadata store per each Coordinator kill audit duty
run. This metric can help adjust
`druid.coordinator.kill.audit.durationToRetain` configuration based on whether
more or less audit logs need to be deleted per cycle. Note that this metric is
only emitted when `druid.coordinator.kill.audit.on` is set to true.| |Varies.|
|`metadata/kill/rule/count`|Total number of rules that were automatically
deleted from metadata store per each Coordinator kill rule duty run. This
metric can help adjust `druid.coordinator.kill.rule.durationToRetain`
configuration based on whether more or less rules need to be deleted per cycle.
Note that this metric is only emitted when `druid.coordinator.kill.rule.on` is
set to true.| |Varies.|
+|`metadata/kill/datasource/count`|Total number of datasource metadata that
were automatically deleted from metadata store per each Coordinator kill
datasource duty run (Note: datasource metadata only exists for datasource
created from supervisor). This metric can help adjust
`druid.coordinator.kill.datasource.durationToRetain` configuration based on
whether more or less datasource metadata need to be deleted per cycle. Note
that this metric is only emitted when `druid.coordinator.kill.da [...]
If `emitBalancingStats` is set to `true` in the Coordinator [dynamic
configuration](../configuration/index.md#dynamic-configuration), then [log
entries](../configuration/logging.md) for class
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
b/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
index decaff8..fb274f3 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
@@ -34,6 +34,7 @@ import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.partition.PartialShardSpec;
import org.joda.time.Interval;
+import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -140,6 +141,13 @@ public class TestIndexerMetadataStorageCoordinator
implements IndexerMetadataSto
}
@Override
+ public int removeDataSourceMetadataOlderThan(long timestamp, @Nullable
Set<String> excludeDatasources)
+ {
+ throw new UnsupportedOperationException("Not implemented, no test uses
this currently.");
+ }
+
+
+ @Override
public SegmentIdWithShardSpec allocatePendingSegment(
String dataSource,
String sequenceName,
diff --git
a/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
b/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
index 3a1d257..513f5b9 100644
---
a/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
+++
b/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
@@ -26,6 +26,7 @@ import org.apache.druid.timeline.partition.PartialShardSpec;
import org.joda.time.Interval;
import javax.annotation.Nullable;
+import javax.validation.constraints.NotNull;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
@@ -268,6 +269,15 @@ public interface IndexerMetadataStorageCoordinator
boolean insertDataSourceMetadata(String dataSource, DataSourceMetadata
dataSourceMetadata);
/**
+ * Remove datasource metadata created before the given timestamp and not in
given excludeDatasources set.
+ *
+ * @param timestamp timestamp in milliseconds
+ * @param excludeDatasources set of datasource names to exclude from removal
+ * @return number of datasource metadata removed
+ */
+ int removeDataSourceMetadataOlderThan(long timestamp, @NotNull Set<String>
excludeDatasources);
+
+ /**
* Similar to {@link #announceHistoricalSegments(Set)}, but meant for
streaming ingestion tasks for handling
* the case where the task ingested no records and created no segments, but
still needs to update the metadata
* with the progress that the task made.
diff --git
a/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
b/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
index 05fcb56..46d6c0e 100644
---
a/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
+++
b/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
@@ -57,6 +57,7 @@ import org.apache.druid.timeline.partition.PartialShardSpec;
import org.apache.druid.timeline.partition.PartitionChunk;
import org.apache.druid.timeline.partition.PartitionIds;
import org.apache.druid.timeline.partition.SingleDimensionShardSpec;
+import org.joda.time.DateTime;
import org.joda.time.Interval;
import org.joda.time.chrono.ISOChronology;
import org.skife.jdbi.v2.Batch;
@@ -73,6 +74,7 @@ import org.skife.jdbi.v2.tweak.HandleCallback;
import org.skife.jdbi.v2.util.ByteArrayMapper;
import javax.annotation.Nullable;
+import javax.validation.constraints.NotNull;
import java.io.IOException;
import java.sql.ResultSet;
import java.util.ArrayList;
@@ -1423,4 +1425,39 @@ public class IndexerSQLMetadataStorageCoordinator
implements IndexerMetadataStor
.execute()
);
}
+
+ @Override
+ public int removeDataSourceMetadataOlderThan(long timestamp, @NotNull
Set<String> excludeDatasources)
+ {
+ DateTime dateTime = DateTimes.utc(timestamp);
+ List<String> datasourcesToDelete = connector.getDBI().withHandle(
+ handle -> handle
+ .createQuery(
+ StringUtils.format(
+ "SELECT dataSource FROM %1$s WHERE created_date < '%2$s'",
+ dbTables.getDataSourceTable(),
+ dateTime.toString()
+ )
+ )
+ .mapTo(String.class)
+ .list()
+ );
+ datasourcesToDelete.removeAll(excludeDatasources);
+ return connector.getDBI().withHandle(
+ handle -> {
+ final PreparedBatch batch = handle.prepareBatch(
+ StringUtils.format(
+ "DELETE FROM %1$s WHERE dataSource = :dataSource AND
created_date < '%2$s'",
+ dbTables.getDataSourceTable(),
+ dateTime.toString()
+ )
+ );
+ for (String datasource : datasourcesToDelete) {
+ batch.bind("dataSource", datasource).add();
+ }
+ int[] result = batch.execute();
+ return IntStream.of(result).sum();
+ }
+ );
+ }
}
diff --git
a/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java
b/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java
index 591dee2..9eb254f 100644
---
a/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java
+++
b/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java
@@ -33,9 +33,28 @@ public interface MetadataSupervisorManager
Map<String, List<VersionedSupervisorSpec>> getAll();
+ /**
+ * Return latest supervisors (both active and terminated)
+ *
+ * @return latest terminated supervisors
+ */
Map<String, SupervisorSpec> getLatest();
/**
+ * Only return the latest active supervisors
+ *
+ * @return latest active supervisors
+ */
+ Map<String, SupervisorSpec> getLatestActiveOnly();
+
+ /**
+ * Only return the latest terminated supervisors
+ *
+ * @return latest terminated supervisors
+ */
+ Map<String, SupervisorSpec> getLatestTerminatedOnly();
+
+ /**
* Remove terminated supervisors created before the given timestamp.
*
* @param timestamp timestamp in milliseconds
diff --git
a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java
b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java
index e995a14..0480c45 100644
---
a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java
+++
b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java
@@ -254,10 +254,40 @@ public class SQLMetadataSupervisorManager implements
MetadataSupervisorManager
}
@Override
+ public Map<String, SupervisorSpec> getLatestActiveOnly()
+ {
+ Map<String, SupervisorSpec> supervisors = getLatest();
+ Map<String, SupervisorSpec> activeSupervisors = new HashMap<>();
+ for (Map.Entry<String, SupervisorSpec> entry : supervisors.entrySet()) {
+ // Terminated supervisor will have it's latest supervisorSpec as
NoopSupervisorSpec
+ // (NoopSupervisorSpec is used as a tombstone marker)
+ if (!(entry.getValue() instanceof NoopSupervisorSpec)) {
+ activeSupervisors.put(entry.getKey(), entry.getValue());
+ }
+ }
+ return ImmutableMap.copyOf(activeSupervisors);
+ }
+
+ @Override
+ public Map<String, SupervisorSpec> getLatestTerminatedOnly()
+ {
+ Map<String, SupervisorSpec> supervisors = getLatest();
+ Map<String, SupervisorSpec> activeSupervisors = new HashMap<>();
+ for (Map.Entry<String, SupervisorSpec> entry : supervisors.entrySet()) {
+ // Terminated supervisor will have it's latest supervisorSpec as
NoopSupervisorSpec
+ // (NoopSupervisorSpec is used as a tombstone marker)
+ if (entry.getValue() instanceof NoopSupervisorSpec) {
+ activeSupervisors.put(entry.getKey(), entry.getValue());
+ }
+ }
+ return ImmutableMap.copyOf(activeSupervisors);
+ }
+
+ @Override
public int removeTerminatedSupervisorsOlderThan(long timestamp)
{
DateTime dateTime = DateTimes.utc(timestamp);
- Map<String, SupervisorSpec> supervisors = getLatest();
+ Map<String, SupervisorSpec> terminatedSupervisors =
getLatestTerminatedOnly();
return dbi.withHandle(
handle -> {
final PreparedBatch batch = handle.prepareBatch(
@@ -267,13 +297,8 @@ public class SQLMetadataSupervisorManager implements
MetadataSupervisorManager
dateTime.toString()
)
);
- for (Map.Entry<String, SupervisorSpec> supervisor :
supervisors.entrySet()) {
- final SupervisorSpec spec = supervisor.getValue();
- // Terminated supervisor will have it's latest supervisorSpec as
NoopSupervisorSpec
- // (NoopSupervisorSpec is used as a tombstone marker)
- if (spec instanceof NoopSupervisorSpec) {
- batch.bind("spec_id", supervisor.getKey()).add();
- }
+ for (Map.Entry<String, SupervisorSpec> supervisor :
terminatedSupervisors.entrySet()) {
+ batch.bind("spec_id", supervisor.getKey()).add();
}
int[] result = batch.execute();
return IntStream.of(result).sum();
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
index 7985b5c..14da3c8 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java
@@ -79,6 +79,14 @@ public abstract class DruidCoordinatorConfig
@Default("PT-1s")
public abstract Duration getCoordinatorRuleKillDurationToRetain();
+ @Config("druid.coordinator.kill.datasource.period")
+ @Default("P1D")
+ public abstract Duration getCoordinatorDatasourceKillPeriod();
+
+ @Config("druid.coordinator.kill.datasource.durationToRetain")
+ @Default("PT-1s")
+ public abstract Duration getCoordinatorDatasourceKillDurationToRetain();
+
@Config("druid.coordinator.load.timeout")
public Duration getLoadTimeoutDelay()
{
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 651cf6b..ac735c6 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
@@ -51,6 +51,7 @@ public class KillAuditLog implements CoordinatorDuty
);
this.retainDuration =
config.getCoordinatorAuditKillDurationToRetain().getMillis();
Preconditions.checkArgument(this.retainDuration >= 0, "coordinator audit
kill retainDuration must be >= 0");
+ Preconditions.checkArgument(this.retainDuration <
System.currentTimeMillis(), "Coordinator audit kill retainDuration cannot be
greater than current time in ms");
log.debug(
"Audit Kill Task scheduling enabled with period [%s], retainDuration
[%s]",
this.period,
@@ -62,19 +63,24 @@ public class KillAuditLog implements CoordinatorDuty
@Override
public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams
params)
{
- if ((lastKillTime + period) < System.currentTimeMillis()) {
- lastKillTime = System.currentTimeMillis();
-
- long timestamp = System.currentTimeMillis() - retainDuration;
- int auditRemoved = auditManager.removeAuditLogsOlderThan(timestamp);
- ServiceEmitter emitter = params.getEmitter();
- emitter.emit(
- new ServiceMetricEvent.Builder().build(
- "metadata/kill/audit/count",
- auditRemoved
- )
- );
- log.info("Finished running KillAuditLog duty. Removed %,d audit logs",
auditRemoved);
+ long currentTimeMillis = System.currentTimeMillis();
+ if ((lastKillTime + period) < currentTimeMillis) {
+ lastKillTime = currentTimeMillis;
+ long timestamp = currentTimeMillis - retainDuration;
+ try {
+ int auditRemoved = auditManager.removeAuditLogsOlderThan(timestamp);
+ ServiceEmitter emitter = params.getEmitter();
+ emitter.emit(
+ new ServiceMetricEvent.Builder().build(
+ "metadata/kill/audit/count",
+ auditRemoved
+ )
+ );
+ log.info("Finished running KillAuditLog duty. Removed %,d audit logs",
auditRemoved);
+ }
+ catch (Exception e) {
+ log.error(e, "Failed to kill audit log");
+ }
}
return params;
}
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
new file mode 100644
index 0000000..9c9535b
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadata.java
@@ -0,0 +1,121 @@
+/*
+ * 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.duty;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.inject.Inject;
+import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataSupervisorManager;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * CoordinatorDuty for automatic deletion of datasource metadata from the
datasource table in metadata storage.
+ * (Note: datasource metadata only exists for datasource created from
supervisor).
+ * Note that this class relies on the supervisorSpec.getDataSources names to
match with the
+ * 'datasource' column of the datasource metadata table.
+ */
+public class KillDatasourceMetadata implements CoordinatorDuty
+{
+ private static final Logger log = new Logger(KillDatasourceMetadata.class);
+
+ private final long period;
+ private final long retainDuration;
+ private long lastKillTime = 0;
+
+ private final IndexerMetadataStorageCoordinator
indexerMetadataStorageCoordinator;
+ private final MetadataSupervisorManager metadataSupervisorManager;
+
+ @Inject
+ public KillDatasourceMetadata(
+ DruidCoordinatorConfig config,
+ IndexerMetadataStorageCoordinator indexerMetadataStorageCoordinator,
+ MetadataSupervisorManager metadataSupervisorManager
+ )
+ {
+ this.indexerMetadataStorageCoordinator = indexerMetadataStorageCoordinator;
+ this.metadataSupervisorManager = metadataSupervisorManager;
+ this.period = config.getCoordinatorDatasourceKillPeriod().getMillis();
+ Preconditions.checkArgument(
+ this.period >=
config.getCoordinatorMetadataStoreManagementPeriod().getMillis(),
+ "Coordinator datasource metadata kill period must be >=
druid.coordinator.period.metadataStoreManagementPeriod"
+ );
+ this.retainDuration =
config.getCoordinatorDatasourceKillDurationToRetain().getMillis();
+ Preconditions.checkArgument(this.retainDuration >= 0, "Coordinator
datasource metadata kill retainDuration must be >= 0");
+ Preconditions.checkArgument(this.retainDuration <
System.currentTimeMillis(), "Coordinator datasource metadata kill
retainDuration cannot be greater than current time in ms");
+ log.debug(
+ "Datasource Metadata Kill Task scheduling enabled with period [%s],
retainDuration [%s]",
+ this.period,
+ this.retainDuration
+ );
+ }
+
+ @Override
+ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams
params)
+ {
+ long currentTimeMillis = System.currentTimeMillis();
+ if ((lastKillTime + period) < currentTimeMillis) {
+ lastKillTime = currentTimeMillis;
+ long timestamp = currentTimeMillis - retainDuration;
+ try {
+ // Datasource metadata only exists for datasource with supervisor
+ // To determine if datasource metadata is still active, we check if
the supervisor for that particular datasource
+ // is still active or not
+ Map<String, SupervisorSpec> allActiveSupervisor =
metadataSupervisorManager.getLatestActiveOnly();
+ Set<String> allDatasourceWithActiveSupervisor =
allActiveSupervisor.values()
+
.stream()
+
.map(supervisorSpec -> supervisorSpec.getDataSources())
+
.flatMap(Collection::stream)
+
.filter(datasource -> !Strings.isNullOrEmpty(datasource))
+
.collect(Collectors.toSet());
+ // We exclude removing datasource metadata with active supervisor
+ int datasourceMetadataRemovedCount =
indexerMetadataStorageCoordinator.removeDataSourceMetadataOlderThan(
+ timestamp,
+ allDatasourceWithActiveSupervisor
+ );
+ ServiceEmitter emitter = params.getEmitter();
+ emitter.emit(
+ new ServiceMetricEvent.Builder().build(
+ "metadata/kill/datasource/count",
+ datasourceMetadataRemovedCount
+ )
+ );
+ log.info(
+ "Finished running KillDatasourceMetadata duty. Removed %,d
datasource metadata",
+ datasourceMetadataRemovedCount
+ );
+ }
+ catch (Exception e) {
+ log.error(e, "Failed to kill datasource metadata");
+ }
+ }
+ return params;
+ }
+}
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 eb4f018..50b1740 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
@@ -47,6 +47,7 @@ public class KillRules implements CoordinatorDuty
);
this.retainDuration =
config.getCoordinatorRuleKillDurationToRetain().getMillis();
Preconditions.checkArgument(this.retainDuration >= 0, "coordinator rule
kill retainDuration must be >= 0");
+ Preconditions.checkArgument(this.retainDuration <
System.currentTimeMillis(), "Coordinator rule kill retainDuration cannot be
greater than current time in ms");
log.debug(
"Rule Kill Task scheduling enabled with period [%s], retainDuration
[%s]",
this.period,
@@ -57,19 +58,24 @@ public class KillRules implements CoordinatorDuty
@Override
public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams
params)
{
- if ((lastKillTime + period) < System.currentTimeMillis()) {
- lastKillTime = System.currentTimeMillis();
-
- long timestamp = System.currentTimeMillis() - retainDuration;
- int ruleRemoved =
params.getDatabaseRuleManager().removeRulesForEmptyDatasourcesOlderThan(timestamp);
- ServiceEmitter emitter = params.getEmitter();
- emitter.emit(
- new ServiceMetricEvent.Builder().build(
- "metadata/kill/rule/count",
- ruleRemoved
- )
- );
- log.info("Finished running KillRules duty. Removed %,d rule",
ruleRemoved);
+ long currentTimeMillis = System.currentTimeMillis();
+ if ((lastKillTime + period) < currentTimeMillis) {
+ lastKillTime = currentTimeMillis;
+ long timestamp = currentTimeMillis - retainDuration;
+ try {
+ int ruleRemoved =
params.getDatabaseRuleManager().removeRulesForEmptyDatasourcesOlderThan(timestamp);
+ ServiceEmitter emitter = params.getEmitter();
+ emitter.emit(
+ new ServiceMetricEvent.Builder().build(
+ "metadata/kill/rule/count",
+ ruleRemoved
+ )
+ );
+ log.info("Finished running KillRules duty. Removed %,d rule",
ruleRemoved);
+ }
+ catch (Exception e) {
+ log.error(e, "Failed to kill rules metadata");
+ }
}
return params;
}
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 a1c2a3f..8a0b484 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
@@ -55,6 +55,7 @@ public class KillSupervisors implements CoordinatorDuty
);
this.retainDuration =
config.getCoordinatorSupervisorKillDurationToRetain().getMillis();
Preconditions.checkArgument(this.retainDuration >= 0, "Coordinator
supervisor kill retainDuration must be >= 0");
+ Preconditions.checkArgument(this.retainDuration <
System.currentTimeMillis(), "Coordinator supervisor kill retainDuration cannot
be greater than current time in ms");
log.debug(
"Supervisor Kill Task scheduling enabled with period [%s],
retainDuration [%s]",
this.period,
@@ -65,19 +66,24 @@ public class KillSupervisors implements CoordinatorDuty
@Override
public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams
params)
{
- if ((lastKillTime + period) < System.currentTimeMillis()) {
- lastKillTime = System.currentTimeMillis();
-
- long timestamp = System.currentTimeMillis() - retainDuration;
- int supervisorRemoved =
metadataSupervisorManager.removeTerminatedSupervisorsOlderThan(timestamp);
- ServiceEmitter emitter = params.getEmitter();
- emitter.emit(
- new ServiceMetricEvent.Builder().build(
- "metadata/kill/supervisor/count",
- supervisorRemoved
- )
- );
- log.info("Finished running KillSupervisors duty. Removed %,d
supervisor", supervisorRemoved);
+ long currentTimeMillis = System.currentTimeMillis();
+ if ((lastKillTime + period) < currentTimeMillis) {
+ lastKillTime = currentTimeMillis;
+ long timestamp = currentTimeMillis - retainDuration;
+ try {
+ int supervisorRemoved =
metadataSupervisorManager.removeTerminatedSupervisorsOlderThan(timestamp);
+ ServiceEmitter emitter = params.getEmitter();
+ emitter.emit(
+ new ServiceMetricEvent.Builder().build(
+ "metadata/kill/supervisor/count",
+ supervisorRemoved
+ )
+ );
+ log.info("Finished running KillSupervisors duty. Removed %,d
supervisor", supervisorRemoved);
+ }
+ catch (Exception e) {
+ log.error(e, "Failed to kill terminated supervisor metadata");
+ }
}
return params;
}
diff --git
a/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
b/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
index 924ad92..7acd90f 100644
---
a/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
+++
b/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
@@ -1471,4 +1471,82 @@ public class IndexerSQLMetadataStorageCoordinatorTest
Assert.assertEquals(IndexerSQLMetadataStorageCoordinator.DataStoreMetadataUpdateResult.TRY_AGAIN,
result);
}
}
+
+ @Test
+ public void
testRemoveDataSourceMetadataOlderThanDatasourceActiveShouldNotBeDeleted()
throws Exception
+ {
+ coordinator.announceHistoricalSegments(
+ ImmutableSet.of(defaultSegment),
+ ImmutableSet.of(),
+ new ObjectMetadata(null),
+ new ObjectMetadata(ImmutableMap.of("foo", "bar"))
+ );
+
+ Assert.assertEquals(
+ new ObjectMetadata(ImmutableMap.of("foo", "bar")),
+ coordinator.retrieveDataSourceMetadata("fooDataSource")
+ );
+
+ // Try delete. Datasource should not be deleted as it is in excluded set
+ int deletedCount =
coordinator.removeDataSourceMetadataOlderThan(System.currentTimeMillis(),
ImmutableSet.of("fooDataSource"));
+
+ // Datasource should not be deleted
+ Assert.assertEquals(
+ new ObjectMetadata(ImmutableMap.of("foo", "bar")),
+ coordinator.retrieveDataSourceMetadata("fooDataSource")
+ );
+ Assert.assertEquals(0, deletedCount);
+ }
+
+ @Test
+ public void
testRemoveDataSourceMetadataOlderThanDatasourceNotActiveAndOlderThanTimeShouldBeDeleted()
throws Exception
+ {
+ coordinator.announceHistoricalSegments(
+ ImmutableSet.of(defaultSegment),
+ ImmutableSet.of(),
+ new ObjectMetadata(null),
+ new ObjectMetadata(ImmutableMap.of("foo", "bar"))
+ );
+
+ Assert.assertEquals(
+ new ObjectMetadata(ImmutableMap.of("foo", "bar")),
+ coordinator.retrieveDataSourceMetadata("fooDataSource")
+ );
+
+ // Try delete. Datasource should be deleted as it is not in excluded set
and created time older than given time
+ int deletedCount =
coordinator.removeDataSourceMetadataOlderThan(System.currentTimeMillis(),
ImmutableSet.of());
+
+ // Datasource should be deleted
+ Assert.assertNull(
+ coordinator.retrieveDataSourceMetadata("fooDataSource")
+ );
+ Assert.assertEquals(1, deletedCount);
+ }
+
+ @Test
+ public void
testRemoveDataSourceMetadataOlderThanDatasourceNotActiveButNotOlderThanTimeShouldNotBeDeleted()
throws Exception
+ {
+ coordinator.announceHistoricalSegments(
+ ImmutableSet.of(defaultSegment),
+ ImmutableSet.of(),
+ new ObjectMetadata(null),
+ new ObjectMetadata(ImmutableMap.of("foo", "bar"))
+ );
+
+ Assert.assertEquals(
+ new ObjectMetadata(ImmutableMap.of("foo", "bar")),
+ coordinator.retrieveDataSourceMetadata("fooDataSource")
+ );
+
+ // Do delete. Datasource metadata should not be deleted. Datasource is not
active but it was created just now so it's
+ // created timestamp will be later than the timestamp 2012-01-01T00:00:00Z
+ int deletedCount =
coordinator.removeDataSourceMetadataOlderThan(DateTimes.of("2012-01-01T00:00:00Z").getMillis(),
ImmutableSet.of());
+
+ // Datasource should not be deleted
+ Assert.assertEquals(
+ new ObjectMetadata(ImmutableMap.of("foo", "bar")),
+ coordinator.retrieveDataSourceMetadata("fooDataSource")
+ );
+ Assert.assertEquals(0, deletedCount);
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java
b/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java
index 5e4f483..9547387 100644
---
a/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java
+++
b/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java
@@ -243,6 +243,46 @@ public class SQLMetadataSupervisorManagerTest
Assert.assertNull(specs.get(0).getSpec());
}
+ @Test
+ public void testGetLatestActiveOnly()
+ {
+ final String supervisor1 = "test-supervisor-1";
+ final String datasource1 = "datasource-1";
+ final String supervisor2 = "test-supervisor-2";
+ final Map<String, String> data1rev1 = ImmutableMap.of("key1-1",
"value1-1-1", "key1-2", "value1-2-1");
+ Assert.assertTrue(supervisorManager.getAll().isEmpty());
+ supervisorManager.insert(supervisor1, new TestSupervisorSpec(supervisor1,
data1rev1));
+ // supervisor1 is terminated
+ supervisorManager.insert(supervisor1, new NoopSupervisorSpec(supervisor1,
ImmutableList.of(datasource1)));
+ // supervisor2 is still active
+ supervisorManager.insert(supervisor2, new TestSupervisorSpec(supervisor2,
data1rev1));
+ // get latest active should only return supervisor2
+ Map<String, SupervisorSpec> actual =
supervisorManager.getLatestActiveOnly();
+ Assert.assertEquals(1, actual.size());
+ Assert.assertTrue(actual.containsKey(supervisor2));
+ }
+
+
+ @Test
+ public void testGetLatestTerminatedOnly()
+ {
+ final String supervisor1 = "test-supervisor-1";
+ final String datasource1 = "datasource-1";
+ final String supervisor2 = "test-supervisor-2";
+ final Map<String, String> data1rev1 = ImmutableMap.of("key1-1",
"value1-1-1", "key1-2", "value1-2-1");
+ Assert.assertTrue(supervisorManager.getAll().isEmpty());
+ supervisorManager.insert(supervisor1, new TestSupervisorSpec(supervisor1,
data1rev1));
+ // supervisor1 is terminated
+ supervisorManager.insert(supervisor1, new NoopSupervisorSpec(supervisor1,
ImmutableList.of(datasource1)));
+ // supervisor2 is still active
+ supervisorManager.insert(supervisor2, new TestSupervisorSpec(supervisor2,
data1rev1));
+ // get latest terminated should only return supervisor1
+ Map<String, SupervisorSpec> actual =
supervisorManager.getLatestTerminatedOnly();
+ Assert.assertEquals(1, actual.size());
+ Assert.assertTrue(actual.containsKey(supervisor1));
+ }
+
+
private static class BadSupervisorSpec implements SupervisorSpec
{
private final String id;
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 b8e3103..adad57e 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
@@ -178,6 +178,8 @@ public class CuratorDruidCoordinatorTest extends
CuratorTestBase
null,
null,
null,
+ null,
+ null,
10,
new Duration("PT0s")
);
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 97c220d..2d88d6e 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
@@ -150,6 +150,8 @@ public class DruidCoordinatorTest extends CuratorTestBase
null,
null,
null,
+ null,
+ null,
10,
new Duration("PT0s")
);
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java
index 2a85500..4991f13 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java
@@ -87,6 +87,8 @@ public class HttpLoadQueuePeonTest
null,
null,
null,
+ null,
+ null,
10,
Duration.ZERO
)
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java
index 8b2f486..ea4322a 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java
@@ -102,6 +102,8 @@ public class LoadQueuePeonTest extends CuratorTestBase
null,
null,
null,
+ null,
+ null,
10,
Duration.millis(0)
)
@@ -304,6 +306,8 @@ public class LoadQueuePeonTest extends CuratorTestBase
null,
null,
null,
+ null,
+ null,
10,
new Duration("PT1s")
)
@@ -363,6 +367,8 @@ public class LoadQueuePeonTest extends CuratorTestBase
null,
null,
null,
+ null,
+ null,
10,
new Duration("PT1s")
)
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java
b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java
index 57ac65b..326bc76 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java
@@ -51,6 +51,8 @@ public class LoadQueuePeonTester extends CuratorLoadQueuePeon
null,
null,
null,
+ null,
+ null,
10,
new Duration("PT1s")
)
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
index 2c1b706..e2c0570 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java
@@ -37,6 +37,8 @@ public class TestDruidCoordinatorConfig extends
DruidCoordinatorConfig
private final Duration coordinatorAuditKillDurationToRetain;
private final Duration coordinatorRuleKillPeriod;
private final Duration coordinatorRuleKillDurationToRetain;
+ private final Duration coordinatorDatasourceKillPeriod;
+ private final Duration coordinatorDatasourceKillDurationToRetain;
private final Duration getLoadQueuePeonRepeatDelay;
private final int coordinatorKillMaxSegments;
@@ -54,6 +56,8 @@ public class TestDruidCoordinatorConfig extends
DruidCoordinatorConfig
Duration coordinatorAuditKillDurationToRetain,
Duration coordinatorRuleKillPeriod,
Duration coordinatorRuleKillDurationToRetain,
+ Duration coordinatorDatasourceKillPeriod,
+ Duration coordinatorDatasourceKillDurationToRetain,
int coordinatorKillMaxSegments,
Duration getLoadQueuePeonRepeatDelay
)
@@ -71,6 +75,8 @@ public class TestDruidCoordinatorConfig extends
DruidCoordinatorConfig
this.coordinatorAuditKillDurationToRetain =
coordinatorAuditKillDurationToRetain;
this.coordinatorRuleKillPeriod = coordinatorRuleKillPeriod;
this.coordinatorRuleKillDurationToRetain =
coordinatorRuleKillDurationToRetain;
+ this.coordinatorDatasourceKillPeriod = coordinatorDatasourceKillPeriod;
+ this.coordinatorDatasourceKillDurationToRetain =
coordinatorDatasourceKillDurationToRetain;
this.coordinatorKillMaxSegments = coordinatorKillMaxSegments;
this.getLoadQueuePeonRepeatDelay = getLoadQueuePeonRepeatDelay;
}
@@ -148,6 +154,18 @@ public class TestDruidCoordinatorConfig extends
DruidCoordinatorConfig
}
@Override
+ public Duration getCoordinatorDatasourceKillPeriod()
+ {
+ return coordinatorDatasourceKillPeriod;
+ }
+
+ @Override
+ public Duration getCoordinatorDatasourceKillDurationToRetain()
+ {
+ return coordinatorDatasourceKillDurationToRetain;
+ }
+
+ @Override
public int getCoordinatorKillMaxSegments()
{
return coordinatorKillMaxSegments;
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 aeb0bd9..4b883f3 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
@@ -68,6 +68,8 @@ public class KillAuditLogTest
new Duration("PT1S"),
null,
null,
+ null,
+ null,
10,
null
);
@@ -94,6 +96,8 @@ public class KillAuditLogTest
new Duration("PT1S"),
null,
null,
+ null,
+ null,
10,
null
);
@@ -120,6 +124,8 @@ public class KillAuditLogTest
new Duration("PT1S"),
null,
null,
+ null,
+ null,
10,
null
);
@@ -145,6 +151,8 @@ public class KillAuditLogTest
new Duration("PT-1S"),
null,
null,
+ null,
+ null,
10,
null
);
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/KillDatasourceMetadataTest.java
similarity index 58%
copy from
server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java
copy to
server/src/test/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadataTest.java
index 9aeb992..ff29136 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/KillDatasourceMetadataTest.java
@@ -19,9 +19,12 @@
package org.apache.druid.server.coordinator.duty;
+import com.google.common.collect.ImmutableSet;
+import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.java.util.emitter.service.ServiceEventBuilder;
import org.apache.druid.metadata.MetadataSupervisorManager;
+import org.apache.druid.metadata.TestSupervisorSpec;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig;
import org.joda.time.Duration;
@@ -35,21 +38,27 @@ import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
-public class KillSupervisorsTest
+public class KillDatasourceMetadataTest
{
@Mock
+ private IndexerMetadataStorageCoordinator
mockIndexerMetadataStorageCoordinator;
+
+ @Mock
private MetadataSupervisorManager mockMetadataSupervisorManager;
@Mock
private DruidCoordinatorRuntimeParams mockDruidCoordinatorRuntimeParams;
@Mock
+ private TestSupervisorSpec mockKinesisSupervisorSpec;
+
+ @Mock
private ServiceEmitter mockServiceEmitter;
@Rule
public ExpectedException exception = ExpectedException.none();
- private KillSupervisors killSupervisors;
+ private KillDatasourceMetadata killDatasourceMetadata;
@Test
public void testRunSkipIfLastRunLessThanPeriod()
@@ -62,17 +71,20 @@ public class KillSupervisorsTest
null,
null,
null,
- new Duration(Long.MAX_VALUE),
- new Duration("PT1S"),
null,
null,
null,
null,
+ null,
+ null,
+ new Duration(Long.MAX_VALUE),
+ new Duration("PT1S"),
10,
null
);
- killSupervisors = new KillSupervisors(druidCoordinatorConfig,
mockMetadataSupervisorManager);
- killSupervisors.run(mockDruidCoordinatorRuntimeParams);
+ killDatasourceMetadata = new
KillDatasourceMetadata(druidCoordinatorConfig,
mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
+ killDatasourceMetadata.run(mockDruidCoordinatorRuntimeParams);
+ Mockito.verifyZeroInteractions(mockIndexerMetadataStorageCoordinator);
Mockito.verifyZeroInteractions(mockMetadataSupervisorManager);
}
@@ -80,6 +92,7 @@ public class KillSupervisorsTest
public void testRunNotSkipIfLastRunMoreThanPeriod()
{
Mockito.when(mockDruidCoordinatorRuntimeParams.getEmitter()).thenReturn(mockServiceEmitter);
+
TestDruidCoordinatorConfig druidCoordinatorConfig = new
TestDruidCoordinatorConfig(
null,
null,
@@ -88,18 +101,20 @@ public class KillSupervisorsTest
null,
null,
null,
- new Duration("PT6S"),
- new Duration("PT1S"),
null,
null,
null,
null,
+ null,
+ null,
+ new Duration("PT6S"),
+ new Duration("PT1S"),
10,
null
);
- killSupervisors = new KillSupervisors(druidCoordinatorConfig,
mockMetadataSupervisorManager);
- killSupervisors.run(mockDruidCoordinatorRuntimeParams);
-
Mockito.verify(mockMetadataSupervisorManager).removeTerminatedSupervisorsOlderThan(ArgumentMatchers.anyLong());
+ killDatasourceMetadata = new
KillDatasourceMetadata(druidCoordinatorConfig,
mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
+ killDatasourceMetadata.run(mockDruidCoordinatorRuntimeParams);
+
Mockito.verify(mockIndexerMetadataStorageCoordinator).removeDataSourceMetadataOlderThan(ArgumentMatchers.anyLong(),
ArgumentMatchers.anySet());
Mockito.verify(mockServiceEmitter).emit(ArgumentMatchers.any(ServiceEventBuilder.class));
}
@@ -114,18 +129,20 @@ public class KillSupervisorsTest
null,
null,
null,
- new Duration("PT3S"),
- new Duration("PT1S"),
null,
null,
null,
null,
+ null,
+ null,
+ new Duration("PT3S"),
+ new Duration("PT1S"),
10,
null
);
exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Coordinator supervisor kill period must be >=
druid.coordinator.period.metadataStoreManagementPeriod");
- killSupervisors = new KillSupervisors(druidCoordinatorConfig,
mockMetadataSupervisorManager);
+ exception.expectMessage("Coordinator datasource metadata kill period must
be >= druid.coordinator.period.metadataStoreManagementPeriod");
+ killDatasourceMetadata = new
KillDatasourceMetadata(druidCoordinatorConfig,
mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
}
@Test
@@ -139,17 +156,49 @@ public class KillSupervisorsTest
null,
null,
null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
new Duration("PT6S"),
new Duration("PT-1S"),
+ 10,
+ null
+ );
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("Coordinator datasource metadata kill
retainDuration must be >= 0");
+ killDatasourceMetadata = new
KillDatasourceMetadata(druidCoordinatorConfig,
mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
+ }
+
+ @Test
+ public void testRunWithEmptyFilterExcludedDatasource()
+ {
+
Mockito.when(mockDruidCoordinatorRuntimeParams.getEmitter()).thenReturn(mockServiceEmitter);
+
+ TestDruidCoordinatorConfig druidCoordinatorConfig = new
TestDruidCoordinatorConfig(
+ null,
+ null,
+ null,
+ new Duration("PT5S"),
+ null,
+ null,
+ null,
null,
null,
null,
null,
+ null,
+ null,
+ new Duration("PT6S"),
+ new Duration("PT1S"),
10,
null
);
- exception.expect(IllegalArgumentException.class);
- exception.expectMessage("Coordinator supervisor kill retainDuration must
be >= 0");
- killSupervisors = new KillSupervisors(druidCoordinatorConfig,
mockMetadataSupervisorManager);
+ killDatasourceMetadata = new
KillDatasourceMetadata(druidCoordinatorConfig,
mockIndexerMetadataStorageCoordinator, mockMetadataSupervisorManager);
+ killDatasourceMetadata.run(mockDruidCoordinatorRuntimeParams);
+
Mockito.verify(mockIndexerMetadataStorageCoordinator).removeDataSourceMetadataOlderThan(ArgumentMatchers.anyLong(),
ArgumentMatchers.eq(ImmutableSet.of()));
+
Mockito.verify(mockServiceEmitter).emit(ArgumentMatchers.any(ServiceEventBuilder.class));
}
}
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 d3268f2..8d63e91 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
@@ -75,6 +75,8 @@ public class KillRulesTest
null,
new Duration(Long.MAX_VALUE),
new Duration("PT1S"),
+ null,
+ null,
10,
null
);
@@ -101,6 +103,8 @@ public class KillRulesTest
null,
new Duration("PT6S"),
new Duration("PT1S"),
+ null,
+ null,
10,
null
);
@@ -127,6 +131,8 @@ public class KillRulesTest
null,
new Duration("PT3S"),
new Duration("PT1S"),
+ null,
+ null,
10,
null
);
@@ -152,6 +158,8 @@ public class KillRulesTest
null,
new Duration("PT6S"),
new Duration("PT-1S"),
+ null,
+ null,
10,
null
);
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 9aeb992..9ab3fb6 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
@@ -68,6 +68,8 @@ public class KillSupervisorsTest
null,
null,
null,
+ null,
+ null,
10,
null
);
@@ -94,6 +96,8 @@ public class KillSupervisorsTest
null,
null,
null,
+ null,
+ null,
10,
null
);
@@ -120,6 +124,8 @@ public class KillSupervisorsTest
null,
null,
null,
+ null,
+ null,
10,
null
);
@@ -145,6 +151,8 @@ public class KillSupervisorsTest
null,
null,
null,
+ null,
+ null,
10,
null
);
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 34d020f..8a6eef3 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
@@ -115,6 +115,8 @@ public class KillUnusedSegmentsTest
null,
null,
null,
+ null,
+ null,
1000,
Duration.ZERO
)
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 584023c..c864592 100644
--- a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
+++ b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java
@@ -73,6 +73,7 @@ import
org.apache.druid.server.coordinator.KillStalePendingSegments;
import org.apache.druid.server.coordinator.LoadQueueTaskMaster;
import org.apache.druid.server.coordinator.duty.CoordinatorDuty;
import org.apache.druid.server.coordinator.duty.KillAuditLog;
+import org.apache.druid.server.coordinator.duty.KillDatasourceMetadata;
import org.apache.druid.server.coordinator.duty.KillRules;
import org.apache.druid.server.coordinator.duty.KillSupervisors;
import org.apache.druid.server.coordinator.duty.KillUnusedSegments;
@@ -254,7 +255,7 @@ public class CliCoordinator extends ServerRunnable
CoordinatorMetadataStoreManagementDuty.class
);
conditionalMetadataStoreManagementDutyMultibind.addConditionBinding(
- "druid.coordinator.kill.rule.on",
+ "druid.coordinator.kill.supervisor.on",
Predicates.equalTo("true"),
KillSupervisors.class
);
@@ -268,6 +269,11 @@ public class CliCoordinator extends ServerRunnable
Predicates.equalTo("true"),
KillRules.class
);
+
conditionalMetadataStoreManagementDutyMultibind.addConditionBinding(
+ "druid.coordinator.kill.datasource.on",
+ Predicates.equalTo("true"),
+ KillDatasourceMetadata.class
+ );
bindNodeRoleAndAnnouncer(
binder,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@druid.apache.org
For additional commands, e-mail: commits-h...@druid.apache.org
