From a791327f84f75fc9be20a784f923f0fe7b7b9d0a Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Mon, 10 May 2021 23:02:04 -0700 Subject: [PATCH 1/9] add auto cleanup --- docs/configuration/index.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/configuration/index.md b/docs/configuration/index.md index e422a3fd4fdd..f1e14f122777 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -753,6 +753,8 @@ These Coordinator static configurations can be defined in the `coordinator/runti |`druid.coordinator.kill.audit.on`| Boolean value for whether to enable automatic deletion of audit logs. If set to true, Coordinator will periodically remove audit logs from the audit table entries in metadata storage.| No | False| |`druid.coordinator.kill.audit.period`| How often to do automatic deletion of audit logs 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.audit.on` is set to "True".| No| `P1D`| |`druid.coordinator.kill.audit.durationToRetain`| Duration of audit logs to be retained from created time in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if `druid.coordinator.kill.audit.on` is set to "True".| Yes if `druid.coordinator.kill.audit.on` is set to "True".| None| +|`druid.coordinator.kill.compaction.on`| Boolean value for whether to enable automatic deletion of compaction configurations. If set to true, Coordinator will periodically remove compaction configuration of inactive datasource (datasource with no used and unused segments) from the config table in metadata storage. | No | False| +|`druid.coordinator.kill.compaction.period`| How often to do automatic deletion of compaction configurations 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.compaction.on` is set to "True".| No| `P1D`| |`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| From 8154096d0ef2d7718c1ef15274560f50b30c5232 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 11 May 2021 01:34:22 -0700 Subject: [PATCH 2/9] add auto cleanup --- docs/operations/metrics.md | 1 + .../coordinator/DruidCoordinatorConfig.java | 4 + .../duty/KillCompactionConfig.java | 144 ++++++++++++++++++ .../TestDruidCoordinatorConfig.java | 9 ++ 4 files changed, 158 insertions(+) create mode 100644 server/src/main/java/org/apache/druid/server/coordinator/duty/KillCompactionConfig.java diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md index b68dcca58307..d5a7b88ce7a7 100644 --- a/docs/operations/metrics.md +++ b/docs/operations/metrics.md @@ -258,6 +258,7 @@ These metrics are for the Druid Coordinator and are reset each time the Coordina |`coordinator/global/time`|Approximate runtime of a full coordination cycle in milliseconds. The `dutyGroup` dimension indicates what type of coordination this run was. i.e. Historical Management vs Indexing|`dutyGroup`|Varies.| |`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/compaction/count`|Total number of compaction configurations that were automatically deleted from metadata store per each Coordinator kill compaction configuration duty run. Note that this metric is only emitted when `druid.coordinator.kill.compaction.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.| 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 7985b5c48329..f6d11a02b1e3 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 @@ -71,6 +71,10 @@ public abstract class DruidCoordinatorConfig @Default("PT-1s") public abstract Duration getCoordinatorAuditKillDurationToRetain(); + @Config("druid.coordinator.kill.compaction.period") + @Default("P1D") + public abstract Duration getCoordinatorCompactionKillPeriod(); + @Config("druid.coordinator.kill.rule.period") @Default("P1D") public abstract Duration getCoordinatorRuleKillPeriod(); 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 new file mode 100644 index 000000000000..fa9773d7347e --- /dev/null +++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillCompactionConfig.java @@ -0,0 +1,144 @@ +/* + * 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.collect.ImmutableList; +import com.google.inject.Inject; +import org.apache.druid.audit.AuditInfo; +import org.apache.druid.common.config.ConfigManager; +import org.apache.druid.common.config.JacksonConfigManager; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.metadata.SqlSegmentsMetadataManager; +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 java.util.Map; +import java.util.Set; +import java.util.concurrent.ThreadLocalRandom; +import java.util.function.Function; +import java.util.stream.Collectors; + +public class KillCompactionConfig implements CoordinatorDuty +{ + private static final Logger log = new Logger(KillCompactionConfig.class); + private static final long UPDATE_RETRY_DELAY = 1000; + private static final int UPDATE_NUM_RETRY = 5; + + private final long period; + private long lastKillTime = 0; + + private final JacksonConfigManager jacksonConfigManager; + private final SqlSegmentsMetadataManager sqlSegmentsMetadataManager; + + @Inject + public KillCompactionConfig( + DruidCoordinatorConfig config, + SqlSegmentsMetadataManager sqlSegmentsMetadataManager, + JacksonConfigManager jacksonConfigManager + ) + { + this.sqlSegmentsMetadataManager = sqlSegmentsMetadataManager; + this.jacksonConfigManager = jacksonConfigManager; + this.period = config.getCoordinatorCompactionKillPeriod().getMillis(); + Preconditions.checkArgument( + this.period >= config.getCoordinatorMetadataStoreManagementPeriod().getMillis(), + "Coordinator compaction configuration kill period must be >= druid.coordinator.period.metadataStoreManagementPeriod" + ); + log.debug( + "Compaction Configuration Kill Task scheduling enabled with period [%s]", + this.period + ); + } + + @Override + public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params) + { + long currentTimeMillis = System.currentTimeMillis(); + if ((lastKillTime + period) < currentTimeMillis) { + lastKillTime = currentTimeMillis; + // If current compaction config is empty then there is nothing to do + CoordinatorCompactionConfig current = CoordinatorCompactionConfig.current(jacksonConfigManager); + if (CoordinatorCompactionConfig.empty().equals(current)) { + log.info("Finished running KillCompactionConfig duty. Nothing to do as compaction config is already empty."); + return params; + } + int attemps = 0; + ConfigManager.SetResult setResult = null; + int compactionConfigRemoved = 0; + try { + while (attemps < UPDATE_NUM_RETRY) { + // Get all active datasources + // Note that we get all active datasources after getting compaction config to prevent race condition if new + // datasource and config are added. + Set activeDatasources = sqlSegmentsMetadataManager.retrieveAllDataSourceNames(); + + final Map updated = current + .getCompactionConfigs() + .stream() + .filter(dataSourceCompactionConfig -> activeDatasources.contains(dataSourceCompactionConfig.getDataSource())) + .collect(Collectors.toMap(DataSourceCompactionConfig::getDataSource, Function.identity())); + + // Calculate number of compaction configs to remove for logging + compactionConfigRemoved = current.getCompactionConfigs().size() - updated.size(); + + setResult = jacksonConfigManager.set( + CoordinatorCompactionConfig.CONFIG_KEY, + // Do database insert without swap if the current config is empty as this means the config may be null in the database + current, + CoordinatorCompactionConfig.from(current, ImmutableList.copyOf(updated.values())), + new AuditInfo("KillCompactionConfig", "CoordinatorDuty for automatic deletion of compaction config", "") + ); + + if (setResult.isOk() || !setResult.isRetryable()) { + break; + } + attemps++; + updateRetryDelay(); + // Refresh current view of Compaction Configuration before trying again + current = CoordinatorCompactionConfig.current(jacksonConfigManager); + } + } + catch (Exception e) { + log.error(e, "Failed to kill compaction configurations"); + return params; + } + + if (setResult.isOk()) { + log.info("Finished running KillCompactionConfig duty. Removed %,d compaction configs", compactionConfigRemoved); + } else { + log.error(setResult.getException(), "Failed to kill compaction configurations"); + } + } + return params; + } + + private void updateRetryDelay() + { + try { + Thread.sleep(ThreadLocalRandom.current().nextLong(UPDATE_RETRY_DELAY)); + } + catch (InterruptedException ie) { + throw new RuntimeException(ie); + } + } +} 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 2c1b7067cc4a..7b58675ef91b 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 @@ -35,6 +35,7 @@ public class TestDruidCoordinatorConfig extends DruidCoordinatorConfig 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 getLoadQueuePeonRepeatDelay; @@ -52,6 +53,7 @@ public TestDruidCoordinatorConfig( Duration coordinatorSupervisorKillDurationToRetain, Duration coordinatorAuditKillPeriod, Duration coordinatorAuditKillDurationToRetain, + Duration coordinatorCompactionKillPeriod, Duration coordinatorRuleKillPeriod, Duration coordinatorRuleKillDurationToRetain, int coordinatorKillMaxSegments, @@ -69,6 +71,7 @@ public TestDruidCoordinatorConfig( this.coordinatorSupervisorKillDurationToRetain = coordinatorSupervisorKillDurationToRetain; this.coordinatorAuditKillPeriod = coordinatorAuditKillPeriod; this.coordinatorAuditKillDurationToRetain = coordinatorAuditKillDurationToRetain; + this.coordinatorCompactionKillPeriod = coordinatorCompactionKillPeriod; this.coordinatorRuleKillPeriod = coordinatorRuleKillPeriod; this.coordinatorRuleKillDurationToRetain = coordinatorRuleKillDurationToRetain; this.coordinatorKillMaxSegments = coordinatorKillMaxSegments; @@ -135,6 +138,12 @@ public Duration getCoordinatorAuditKillDurationToRetain() return coordinatorAuditKillDurationToRetain; } + @Override + public Duration getCoordinatorCompactionKillPeriod() + { + return coordinatorCompactionKillPeriod; + } + @Override public Duration getCoordinatorRuleKillPeriod() { From dec954966e280f78623aba2d66590edb78638afb Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 11 May 2021 01:40:35 -0700 Subject: [PATCH 3/9] add auto cleanup --- .../server/coordinator/CuratorDruidCoordinatorTest.java | 1 + .../druid/server/coordinator/DruidCoordinatorTest.java | 1 + .../druid/server/coordinator/HttpLoadQueuePeonTest.java | 1 + .../apache/druid/server/coordinator/LoadQueuePeonTest.java | 3 +++ .../druid/server/coordinator/LoadQueuePeonTester.java | 1 + .../druid/server/coordinator/duty/KillAuditLogTest.java | 4 ++++ .../server/coordinator/duty/KillDatasourceMetadataTest.java | 5 +++++ .../apache/druid/server/coordinator/duty/KillRulesTest.java | 4 ++++ .../druid/server/coordinator/duty/KillSupervisorsTest.java | 4 ++++ .../server/coordinator/duty/KillUnusedSegmentsTest.java | 1 + .../src/main/java/org/apache/druid/cli/CliCoordinator.java | 6 ++++++ 11 files changed, 31 insertions(+) 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 adad57e7a8ea..a7214d3567cf 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 @@ -180,6 +180,7 @@ public void setUp() throws Exception 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 2d88d6ee4f13..2a96bf5b5a4a 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 @@ -152,6 +152,7 @@ public void setUp() throws Exception 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 4991f1373f2f..f83c155dcfa2 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 @@ -89,6 +89,7 @@ public class HttpLoadQueuePeonTest 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 ea4322a02f8a..c5317ba14ebc 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 @@ -104,6 +104,7 @@ public void testMultipleLoadDropSegments() throws Exception null, null, null, + null, 10, Duration.millis(0) ) @@ -308,6 +309,7 @@ public void testFailAssignForNonTimeoutFailures() throws Exception null, null, null, + null, 10, new Duration("PT1s") ) @@ -369,6 +371,7 @@ public void testFailAssignForLoadDropTimeout() throws Exception 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 326bc764113c..73c0b64d7fc7 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 @@ -53,6 +53,7 @@ public LoadQueuePeonTester() null, null, null, + null, 10, new Duration("PT1s") ) 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 4b883f38e783..5f3139d76cac 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 @@ -70,6 +70,7 @@ public void testRunSkipIfLastRunLessThanPeriod() null, null, null, + null, 10, null ); @@ -98,6 +99,7 @@ public void testRunNotSkipIfLastRunMoreThanPeriod() null, null, null, + null, 10, null ); @@ -126,6 +128,7 @@ public void testConstructorFailIfInvalidPeriod() null, null, null, + null, 10, null ); @@ -153,6 +156,7 @@ public void testConstructorFailIfInvalidRetainDuration() null, null, null, + null, 10, null ); 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 ff29136cc2e8..dee5461bb84e 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 @@ -77,6 +77,7 @@ public void testRunSkipIfLastRunLessThanPeriod() null, null, null, + null, new Duration(Long.MAX_VALUE), new Duration("PT1S"), 10, @@ -107,6 +108,7 @@ public void testRunNotSkipIfLastRunMoreThanPeriod() null, null, null, + null, new Duration("PT6S"), new Duration("PT1S"), 10, @@ -135,6 +137,7 @@ public void testConstructorFailIfInvalidPeriod() null, null, null, + null, new Duration("PT3S"), new Duration("PT1S"), 10, @@ -162,6 +165,7 @@ public void testConstructorFailIfInvalidRetainDuration() null, null, null, + null, new Duration("PT6S"), new Duration("PT-1S"), 10, @@ -191,6 +195,7 @@ public void testRunWithEmptyFilterExcludedDatasource() null, null, null, + null, new Duration("PT6S"), new Duration("PT1S"), 10, 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 8d63e9136554..b95e8d70bef4 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 @@ -73,6 +73,7 @@ public void testRunSkipIfLastRunLessThanPeriod() null, null, null, + null, new Duration(Long.MAX_VALUE), new Duration("PT1S"), null, @@ -101,6 +102,7 @@ public void testRunNotSkipIfLastRunMoreThanPeriod() null, null, null, + null, new Duration("PT6S"), new Duration("PT1S"), null, @@ -129,6 +131,7 @@ public void testConstructorFailIfInvalidPeriod() null, null, null, + null, new Duration("PT3S"), new Duration("PT1S"), null, @@ -156,6 +159,7 @@ public void testConstructorFailIfInvalidRetainDuration() null, null, null, + null, new Duration("PT6S"), new Duration("PT-1S"), 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 9ab3fb6b3084..bd88bb9b0295 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 @@ -70,6 +70,7 @@ public void testRunSkipIfLastRunLessThanPeriod() null, null, null, + null, 10, null ); @@ -98,6 +99,7 @@ public void testRunNotSkipIfLastRunMoreThanPeriod() null, null, null, + null, 10, null ); @@ -126,6 +128,7 @@ public void testConstructorFailIfInvalidPeriod() null, null, null, + null, 10, null ); @@ -153,6 +156,7 @@ public void testConstructorFailIfInvalidRetainDuration() 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 8a6eef3b62f4..371e64b57f28 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 @@ -117,6 +117,7 @@ private void testFindIntervalForKill(List segmentIntervals, Interval e 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 c864592b3ee9..e888e6bfb736 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.LoadQueueTaskMaster; import org.apache.druid.server.coordinator.duty.CoordinatorDuty; import org.apache.druid.server.coordinator.duty.KillAuditLog; +import org.apache.druid.server.coordinator.duty.KillCompactionConfig; import org.apache.druid.server.coordinator.duty.KillDatasourceMetadata; import org.apache.druid.server.coordinator.duty.KillRules; import org.apache.druid.server.coordinator.duty.KillSupervisors; @@ -274,6 +275,11 @@ public void configure(Binder binder) Predicates.equalTo("true"), KillDatasourceMetadata.class ); + conditionalMetadataStoreManagementDutyMultibind.addConditionBinding( + "druid.coordinator.kill.compaction.on", + Predicates.equalTo("true"), + KillCompactionConfig.class + ); bindNodeRoleAndAnnouncer( binder, From a17bd55b56ff98d242d5be80f20ebef024858920 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 11 May 2021 02:28:04 -0700 Subject: [PATCH 4/9] add tests --- .../duty/KillCompactionConfig.java | 20 +- .../duty/KillCompactionConfigTest.java | 367 ++++++++++++++++++ 2 files changed, 386 insertions(+), 1 deletion(-) create mode 100644 server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java 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 fa9773d7347e..106ff3f5bff6 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 @@ -26,6 +26,8 @@ import org.apache.druid.common.config.ConfigManager; import org.apache.druid.common.config.JacksonConfigManager; 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.SqlSegmentsMetadataManager; import org.apache.druid.server.coordinator.CoordinatorCompactionConfig; import org.apache.druid.server.coordinator.DataSourceCompactionConfig; @@ -44,6 +46,8 @@ public class KillCompactionConfig implements CoordinatorDuty private static final long UPDATE_RETRY_DELAY = 1000; private static final int UPDATE_NUM_RETRY = 5; + static final String COUNT_METRIC = "metadata/kill/compaction/count"; + private final long period; private long lastKillTime = 0; @@ -80,6 +84,7 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params) CoordinatorCompactionConfig current = CoordinatorCompactionConfig.current(jacksonConfigManager); if (CoordinatorCompactionConfig.empty().equals(current)) { log.info("Finished running KillCompactionConfig duty. Nothing to do as compaction config is already empty."); + emitMetric(params.getEmitter(), 0); return params; } int attemps = 0; @@ -100,7 +105,7 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params) // Calculate number of compaction configs to remove for logging compactionConfigRemoved = current.getCompactionConfigs().size() - updated.size(); - + setResult = jacksonConfigManager.set( CoordinatorCompactionConfig.CONFIG_KEY, // Do database insert without swap if the current config is empty as this means the config may be null in the database @@ -120,13 +125,16 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params) } catch (Exception e) { log.error(e, "Failed to kill compaction configurations"); + emitMetric(params.getEmitter(), 0); return params; } if (setResult.isOk()) { log.info("Finished running KillCompactionConfig duty. Removed %,d compaction configs", compactionConfigRemoved); + emitMetric(params.getEmitter(), compactionConfigRemoved); } else { log.error(setResult.getException(), "Failed to kill compaction configurations"); + emitMetric(params.getEmitter(), 0); } } return params; @@ -141,4 +149,14 @@ private void updateRetryDelay() throw new RuntimeException(ie); } } + + private void emitMetric(ServiceEmitter emitter, int compactionConfigRemoved) + { + emitter.emit( + new ServiceMetricEvent.Builder().build( + COUNT_METRIC, + compactionConfigRemoved + ) + ); + } } 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 new file mode 100644 index 000000000000..47f845b310f2 --- /dev/null +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java @@ -0,0 +1,367 @@ +/* + * 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.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import org.apache.druid.common.config.ConfigManager; +import org.apache.druid.common.config.JacksonConfigManager; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.java.util.emitter.service.ServiceEmitter; +import org.apache.druid.java.util.emitter.service.ServiceEventBuilder; +import org.apache.druid.metadata.SqlSegmentsMetadataManager; +import org.apache.druid.server.coordinator.CoordinatorCompactionConfig; +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.joda.time.Duration; +import org.joda.time.Period; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatchers; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.junit.MockitoJUnitRunner; +import org.mockito.stubbing.Answer; + +import java.util.concurrent.atomic.AtomicReference; + +@RunWith(MockitoJUnitRunner.class) +public class KillCompactionConfigTest +{ + @Mock + private DruidCoordinatorRuntimeParams mockDruidCoordinatorRuntimeParams; + + @Mock + private ServiceEmitter mockServiceEmitter; + + @Mock + private SqlSegmentsMetadataManager mockSqlSegmentsMetadataManager; + + @Mock + private JacksonConfigManager mockJacksonConfigManager; + + @Rule + public ExpectedException exception = ExpectedException.none(); + + private KillCompactionConfig killCompactionConfig; + + @Test + public void testRunSkipIfLastRunLessThanPeriod() + { + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + null, + null, + null, + null, + new Duration(Long.MAX_VALUE), + null, + null, + null, + null, + 10, + null + ); + killCompactionConfig = new KillCompactionConfig(druidCoordinatorConfig, mockSqlSegmentsMetadataManager, mockJacksonConfigManager); + killCompactionConfig.run(mockDruidCoordinatorRuntimeParams); + Mockito.verifyZeroInteractions(mockSqlSegmentsMetadataManager); + Mockito.verifyZeroInteractions(mockJacksonConfigManager); + Mockito.verifyZeroInteractions(mockServiceEmitter); + } + + @Test + public void testConstructorFailIfInvalidPeriod() + { + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + null, + null, + null, + null, + new Duration("PT3S"), + null, + null, + null, + null, + 10, + null + ); + exception.expect(IllegalArgumentException.class); + exception.expectMessage("Coordinator compaction configuration kill period must be >= druid.coordinator.period.metadataStoreManagementPeriod"); + killCompactionConfig = new KillCompactionConfig(druidCoordinatorConfig, mockSqlSegmentsMetadataManager, mockJacksonConfigManager); + } + + + @Test + public void testRunDoNothingIfCurrentConfigIsEmpty() + { + Mockito.when(mockDruidCoordinatorRuntimeParams.getEmitter()).thenReturn(mockServiceEmitter); + // Set current compaction config to an empty compaction config + Mockito.when(mockJacksonConfigManager.watch( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.eq(CoordinatorCompactionConfig.class), + ArgumentMatchers.eq(CoordinatorCompactionConfig.empty())) + ).thenReturn(new AtomicReference<>(CoordinatorCompactionConfig.empty())); + + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + null, + null, + null, + null, + new Duration("PT6S"), + null, + null, + null, + null, + 10, + null + ); + killCompactionConfig = new KillCompactionConfig(druidCoordinatorConfig, mockSqlSegmentsMetadataManager, mockJacksonConfigManager); + killCompactionConfig.run(mockDruidCoordinatorRuntimeParams); + Mockito.verifyZeroInteractions(mockSqlSegmentsMetadataManager); + final ArgumentCaptor emittedEventCaptor = ArgumentCaptor.forClass(ServiceEventBuilder.class); + Mockito.verify(mockServiceEmitter).emit(emittedEventCaptor.capture()); + Assert.assertEquals(KillCompactionConfig.COUNT_METRIC, emittedEventCaptor.getValue().build(ImmutableMap.of()).toMap().get("metric")); + Assert.assertEquals(0, emittedEventCaptor.getValue().build(ImmutableMap.of()).toMap().get("value")); + Mockito.verify(mockJacksonConfigManager).watch( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.eq(CoordinatorCompactionConfig.class), + ArgumentMatchers.eq(CoordinatorCompactionConfig.empty()) + ); + Mockito.verifyNoMoreInteractions(mockJacksonConfigManager); + } + + @Test + public void testRunRemoveInactiveDatasourceCompactionConfig() + { + String inactiveDatasourceName = "inactive_datasource"; + String activeDatasourceName = "active_datasource"; + DataSourceCompactionConfig inactiveDatasourceConfig = new DataSourceCompactionConfig( + inactiveDatasourceName, + null, + 500L, + null, + new Period(3600), + null, + new UserCompactionTaskGranularityConfig(Granularities.HOUR, null), + null, + ImmutableMap.of("key", "val") + ); + + DataSourceCompactionConfig activeDatasourceConfig = new DataSourceCompactionConfig( + activeDatasourceName, + null, + 500L, + null, + new Period(3600), + null, + new UserCompactionTaskGranularityConfig(Granularities.HOUR, null), + null, + ImmutableMap.of("key", "val") + ); + CoordinatorCompactionConfig originalCurrentConfig = CoordinatorCompactionConfig.from(ImmutableList.of(inactiveDatasourceConfig, activeDatasourceConfig)); + Mockito.when(mockDruidCoordinatorRuntimeParams.getEmitter()).thenReturn(mockServiceEmitter); + Mockito.when(mockJacksonConfigManager.watch( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.eq(CoordinatorCompactionConfig.class), + ArgumentMatchers.eq(CoordinatorCompactionConfig.empty())) + ).thenReturn(new AtomicReference<>(originalCurrentConfig)); + Mockito.when(mockSqlSegmentsMetadataManager.retrieveAllDataSourceNames()).thenReturn(ImmutableSet.of(activeDatasourceName)); + final ArgumentCaptor oldConfigCaptor = ArgumentCaptor.forClass(CoordinatorCompactionConfig.class); + final ArgumentCaptor newConfigCaptor = ArgumentCaptor.forClass(CoordinatorCompactionConfig.class); + Mockito.when(mockJacksonConfigManager.set( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + oldConfigCaptor.capture(), + newConfigCaptor.capture(), + ArgumentMatchers.any()) + ).thenReturn(ConfigManager.SetResult.ok()); + + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + null, + null, + null, + null, + new Duration("PT6S"), + null, + null, + null, + null, + 10, + null + ); + killCompactionConfig = new KillCompactionConfig(druidCoordinatorConfig, mockSqlSegmentsMetadataManager, mockJacksonConfigManager); + killCompactionConfig.run(mockDruidCoordinatorRuntimeParams); + + // Verify and Assert + Assert.assertNotNull(oldConfigCaptor.getValue()); + Assert.assertEquals(oldConfigCaptor.getValue(), originalCurrentConfig); + Assert.assertNotNull(newConfigCaptor.getValue()); + // The updated config should only contains one compaction config for the active datasource + Assert.assertEquals(1, newConfigCaptor.getValue().getCompactionConfigs().size()); + + Assert.assertEquals(activeDatasourceConfig, newConfigCaptor.getValue().getCompactionConfigs().get(0)); + final ArgumentCaptor emittedEventCaptor = ArgumentCaptor.forClass(ServiceEventBuilder.class); + Mockito.verify(mockServiceEmitter).emit(emittedEventCaptor.capture()); + Assert.assertEquals(KillCompactionConfig.COUNT_METRIC, emittedEventCaptor.getValue().build(ImmutableMap.of()).toMap().get("metric")); + // Should delete 1 config + Assert.assertEquals(1, emittedEventCaptor.getValue().build(ImmutableMap.of()).toMap().get("value")); + + Mockito.verify(mockJacksonConfigManager).watch( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.eq(CoordinatorCompactionConfig.class), + ArgumentMatchers.eq(CoordinatorCompactionConfig.empty()) + ); + Mockito.verify(mockJacksonConfigManager).set( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.any(CoordinatorCompactionConfig.class), + ArgumentMatchers.any(CoordinatorCompactionConfig.class), + ArgumentMatchers.any() + ); + Mockito.verifyNoMoreInteractions(mockJacksonConfigManager); + Mockito.verify(mockSqlSegmentsMetadataManager).retrieveAllDataSourceNames(); + Mockito.verifyNoMoreInteractions(mockSqlSegmentsMetadataManager); + } + + @Test + public void testRunRetryForRetryableException() + { + String inactiveDatasourceName = "inactive_datasource"; + DataSourceCompactionConfig inactiveDatasourceConfig = new DataSourceCompactionConfig( + inactiveDatasourceName, + null, + 500L, + null, + new Period(3600), + null, + new UserCompactionTaskGranularityConfig(Granularities.HOUR, null), + null, + ImmutableMap.of("key", "val") + ); + + CoordinatorCompactionConfig originalCurrentConfig = CoordinatorCompactionConfig.from(ImmutableList.of(inactiveDatasourceConfig)); + Mockito.when(mockDruidCoordinatorRuntimeParams.getEmitter()).thenReturn(mockServiceEmitter); + Mockito.when(mockJacksonConfigManager.watch( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.eq(CoordinatorCompactionConfig.class), + ArgumentMatchers.eq(CoordinatorCompactionConfig.empty())) + ).thenReturn(new AtomicReference<>(originalCurrentConfig)); + Mockito.when(mockSqlSegmentsMetadataManager.retrieveAllDataSourceNames()).thenReturn(ImmutableSet.of()); + Mockito.when(mockJacksonConfigManager.set( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.any(CoordinatorCompactionConfig.class), + ArgumentMatchers.any(CoordinatorCompactionConfig.class), + ArgumentMatchers.any()) + ).thenAnswer(new Answer() { + private int count = 0; + @Override + public Object answer(InvocationOnMock invocation) { + if (count++ < 3) { + // Return fail result with RetryableException the first three call to updated set + return ConfigManager.SetResult.fail(new Exception(), true); + } else { + // Return success ok on the fourth call to set updated config + return ConfigManager.SetResult.ok(); + } + } + }); + + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + null, + null, + null, + null, + new Duration("PT6S"), + null, + null, + null, + null, + 10, + null + ); + killCompactionConfig = new KillCompactionConfig(druidCoordinatorConfig, mockSqlSegmentsMetadataManager, mockJacksonConfigManager); + killCompactionConfig.run(mockDruidCoordinatorRuntimeParams); + + // Verify and Assert + final ArgumentCaptor emittedEventCaptor = ArgumentCaptor.forClass(ServiceEventBuilder.class); + Mockito.verify(mockServiceEmitter).emit(emittedEventCaptor.capture()); + Assert.assertEquals(KillCompactionConfig.COUNT_METRIC, emittedEventCaptor.getValue().build(ImmutableMap.of()).toMap().get("metric")); + // Should delete 1 config + Assert.assertEquals(1, emittedEventCaptor.getValue().build(ImmutableMap.of()).toMap().get("value")); + + // Should call watch (to refresh current compaction config) four times due to RetryableException when failed + Mockito.verify(mockJacksonConfigManager, Mockito.times(4)).watch( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.eq(CoordinatorCompactionConfig.class), + ArgumentMatchers.eq(CoordinatorCompactionConfig.empty()) + ); + // Should call set (to try set new updated compaction config) four times due to RetryableException when failed + Mockito.verify(mockJacksonConfigManager, Mockito.times(4)).set( + ArgumentMatchers.eq(CoordinatorCompactionConfig.CONFIG_KEY), + ArgumentMatchers.any(CoordinatorCompactionConfig.class), + ArgumentMatchers.any(CoordinatorCompactionConfig.class), + ArgumentMatchers.any() + ); + Mockito.verifyNoMoreInteractions(mockJacksonConfigManager); + // Should call retrieveAllDataSourceNames four times due to RetryableException when failed + Mockito.verify(mockSqlSegmentsMetadataManager, Mockito.times(4)).retrieveAllDataSourceNames(); + Mockito.verifyNoMoreInteractions(mockSqlSegmentsMetadataManager); + } +} From 6a497b6634d8303e1be24d2ce906743261bc9304 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 11 May 2021 02:29:08 -0700 Subject: [PATCH 5/9] add tests --- .../server/coordinator/duty/KillCompactionConfigTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 47f845b310f2..555c99814739 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 @@ -305,7 +305,8 @@ public void testRunRetryForRetryableException() ).thenAnswer(new Answer() { private int count = 0; @Override - public Object answer(InvocationOnMock invocation) { + public Object answer(InvocationOnMock invocation) + { if (count++ < 3) { // Return fail result with RetryableException the first three call to updated set return ConfigManager.SetResult.fail(new Exception(), true); From 0726707e2594ba82f53b78a7bcdeb845ccd98111 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 11 May 2021 12:35:13 -0700 Subject: [PATCH 6/9] use retryutils --- .../druid/java/util/RetryableException.java | 31 +++++ .../duty/KillCompactionConfig.java | 117 ++++++++---------- 2 files changed, 86 insertions(+), 62 deletions(-) create mode 100644 core/src/main/java/org/apache/druid/java/util/RetryableException.java diff --git a/core/src/main/java/org/apache/druid/java/util/RetryableException.java b/core/src/main/java/org/apache/druid/java/util/RetryableException.java new file mode 100644 index 000000000000..c5a44669813f --- /dev/null +++ b/core/src/main/java/org/apache/druid/java/util/RetryableException.java @@ -0,0 +1,31 @@ +/* + * 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.java.util; + +import org.apache.druid.java.util.common.StringUtils; + +public class RetryableException extends Exception +{ + public RetryableException(Throwable t) + { + super(t); + } + +} 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 106ff3f5bff6..9b6d02eff3a5 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 @@ -25,6 +25,8 @@ import org.apache.druid.audit.AuditInfo; import org.apache.druid.common.config.ConfigManager; import org.apache.druid.common.config.JacksonConfigManager; +import org.apache.druid.java.util.RetryableException; +import org.apache.druid.java.util.common.RetryUtils; 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; @@ -36,14 +38,12 @@ import java.util.Map; import java.util.Set; -import java.util.concurrent.ThreadLocalRandom; import java.util.function.Function; import java.util.stream.Collectors; public class KillCompactionConfig implements CoordinatorDuty { private static final Logger log = new Logger(KillCompactionConfig.class); - private static final long UPDATE_RETRY_DELAY = 1000; private static final int UPDATE_NUM_RETRY = 5; static final String COUNT_METRIC = "metadata/kill/compaction/count"; @@ -80,76 +80,69 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params) long currentTimeMillis = System.currentTimeMillis(); if ((lastKillTime + period) < currentTimeMillis) { lastKillTime = currentTimeMillis; - // If current compaction config is empty then there is nothing to do - CoordinatorCompactionConfig current = CoordinatorCompactionConfig.current(jacksonConfigManager); - if (CoordinatorCompactionConfig.empty().equals(current)) { - log.info("Finished running KillCompactionConfig duty. Nothing to do as compaction config is already empty."); - emitMetric(params.getEmitter(), 0); - return params; - } - int attemps = 0; - ConfigManager.SetResult setResult = null; - int compactionConfigRemoved = 0; try { - while (attemps < UPDATE_NUM_RETRY) { - // Get all active datasources - // Note that we get all active datasources after getting compaction config to prevent race condition if new - // datasource and config are added. - Set activeDatasources = sqlSegmentsMetadataManager.retrieveAllDataSourceNames(); - - final Map updated = current - .getCompactionConfigs() - .stream() - .filter(dataSourceCompactionConfig -> activeDatasources.contains(dataSourceCompactionConfig.getDataSource())) - .collect(Collectors.toMap(DataSourceCompactionConfig::getDataSource, Function.identity())); - - // Calculate number of compaction configs to remove for logging - compactionConfigRemoved = current.getCompactionConfigs().size() - updated.size(); - - setResult = jacksonConfigManager.set( - CoordinatorCompactionConfig.CONFIG_KEY, - // Do database insert without swap if the current config is empty as this means the config may be null in the database - current, - CoordinatorCompactionConfig.from(current, ImmutableList.copyOf(updated.values())), - new AuditInfo("KillCompactionConfig", "CoordinatorDuty for automatic deletion of compaction config", "") - ); - - if (setResult.isOk() || !setResult.isRetryable()) { - break; - } - attemps++; - updateRetryDelay(); - // Refresh current view of Compaction Configuration before trying again - current = CoordinatorCompactionConfig.current(jacksonConfigManager); - } + RetryUtils.retry( + () -> { + CoordinatorCompactionConfig current = CoordinatorCompactionConfig.current(jacksonConfigManager); + // If current compaction config is empty then there is nothing to do + if (CoordinatorCompactionConfig.empty().equals(current)) { + log.info( + "Finished running KillCompactionConfig duty. Nothing to do as compaction config is already empty."); + emitMetric(params.getEmitter(), 0); + return ConfigManager.SetResult.ok(); + } + + // Get all active datasources + // Note that we get all active datasources after getting compaction config to prevent race condition if new + // datasource and config are added. + Set activeDatasources = sqlSegmentsMetadataManager.retrieveAllDataSourceNames(); + final Map updated = current + .getCompactionConfigs() + .stream() + .filter(dataSourceCompactionConfig -> activeDatasources.contains(dataSourceCompactionConfig.getDataSource())) + .collect(Collectors.toMap(DataSourceCompactionConfig::getDataSource, Function.identity())); + + // Calculate number of compaction configs to remove for logging + int compactionConfigRemoved = current.getCompactionConfigs().size() - updated.size(); + + ConfigManager.SetResult result = jacksonConfigManager.set( + CoordinatorCompactionConfig.CONFIG_KEY, + // Do database insert without swap if the current config is empty as this means the config may be null in the database + current, + CoordinatorCompactionConfig.from(current, ImmutableList.copyOf(updated.values())), + new AuditInfo( + "KillCompactionConfig", + "CoordinatorDuty for automatic deletion of compaction config", + "" + ) + ); + if (result.isOk()) { + log.info( + "Finished running KillCompactionConfig duty. Removed %,d compaction configs", + compactionConfigRemoved + ); + emitMetric(params.getEmitter(), compactionConfigRemoved); + } else if (result.isRetryable()) { + log.debug("Retrying KillCompactionConfig duty"); + throw new RetryableException(result.getException()); + } else { + log.error(result.getException(), "Failed to kill compaction configurations"); + emitMetric(params.getEmitter(), 0); + } + return result; + }, + e -> e instanceof RetryableException, + UPDATE_NUM_RETRY + ); } catch (Exception e) { log.error(e, "Failed to kill compaction configurations"); emitMetric(params.getEmitter(), 0); - return params; - } - - if (setResult.isOk()) { - log.info("Finished running KillCompactionConfig duty. Removed %,d compaction configs", compactionConfigRemoved); - emitMetric(params.getEmitter(), compactionConfigRemoved); - } else { - log.error(setResult.getException(), "Failed to kill compaction configurations"); - emitMetric(params.getEmitter(), 0); } } return params; } - private void updateRetryDelay() - { - try { - Thread.sleep(ThreadLocalRandom.current().nextLong(UPDATE_RETRY_DELAY)); - } - catch (InterruptedException ie) { - throw new RuntimeException(ie); - } - } - private void emitMetric(ServiceEmitter emitter, int compactionConfigRemoved) { emitter.emit( From fbd755a6ec44dc195960d0f37362af5584fcfc05 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 11 May 2021 12:35:53 -0700 Subject: [PATCH 7/9] use retryutils --- .../java/org/apache/druid/java/util/RetryableException.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/core/src/main/java/org/apache/druid/java/util/RetryableException.java b/core/src/main/java/org/apache/druid/java/util/RetryableException.java index c5a44669813f..d9a3fdfef471 100644 --- a/core/src/main/java/org/apache/druid/java/util/RetryableException.java +++ b/core/src/main/java/org/apache/druid/java/util/RetryableException.java @@ -19,8 +19,6 @@ package org.apache.druid.java.util; -import org.apache.druid.java.util.common.StringUtils; - public class RetryableException extends Exception { public RetryableException(Throwable t) From 49ee419cb6e8803808a8b131ac1ff1987d1206d9 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 11 May 2021 12:36:49 -0700 Subject: [PATCH 8/9] use retryutils --- .../druid/server/coordinator/duty/KillCompactionConfig.java | 2 ++ 1 file changed, 2 insertions(+) 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 9b6d02eff3a5..e7f7e37e120a 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 @@ -123,9 +123,11 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params) ); emitMetric(params.getEmitter(), compactionConfigRemoved); } else if (result.isRetryable()) { + // Failed but is retryable log.debug("Retrying KillCompactionConfig duty"); throw new RetryableException(result.getException()); } else { + // Failed and not retryable log.error(result.getException(), "Failed to kill compaction configurations"); emitMetric(params.getEmitter(), 0); } From 22fb520ebb7130c230d93eb6e5a0135f0e118e2b Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 11 May 2021 16:29:52 -0700 Subject: [PATCH 9/9] address comments --- .../druid/java/util/RetryableException.java | 10 ++++++++++ .../java/util/common/RetryUtilsTest.java | 19 +++++++++++++++++++ .../duty/KillCompactionConfig.java | 5 +++++ 3 files changed, 34 insertions(+) diff --git a/core/src/main/java/org/apache/druid/java/util/RetryableException.java b/core/src/main/java/org/apache/druid/java/util/RetryableException.java index d9a3fdfef471..43ed5d5fce0e 100644 --- a/core/src/main/java/org/apache/druid/java/util/RetryableException.java +++ b/core/src/main/java/org/apache/druid/java/util/RetryableException.java @@ -19,6 +19,16 @@ package org.apache.druid.java.util; +import com.google.common.base.Predicate; +import org.apache.druid.java.util.common.RetryUtils; + +/** + * This Exception class can be use with {@link RetryUtils}. + * The method {@link RetryUtils#retry(RetryUtils.Task, Predicate, int)} retry condition (Predicate argument) + * requires an exception to be thrown and applying the predicate to the thrown exception. + * For cases where the task method does not throw an exception but still needs retrying, + * the method can throw this RetryableException so that the RetryUtils can then retry the task + */ public class RetryableException extends Exception { public RetryableException(Throwable t) diff --git a/core/src/test/java/org/apache/druid/java/util/common/RetryUtilsTest.java b/core/src/test/java/org/apache/druid/java/util/common/RetryUtilsTest.java index 7cfc7f59b76d..e4fc226f3ce8 100644 --- a/core/src/test/java/org/apache/druid/java/util/common/RetryUtilsTest.java +++ b/core/src/test/java/org/apache/druid/java/util/common/RetryUtilsTest.java @@ -22,6 +22,7 @@ import com.google.common.base.Predicate; import com.google.common.base.Predicates; import org.apache.commons.lang3.mutable.MutableInt; +import org.apache.druid.java.util.RetryableException; import org.apache.druid.java.util.common.concurrent.Execs; import org.hamcrest.CoreMatchers; import org.junit.Assert; @@ -190,4 +191,22 @@ public void testInterruptRetryLoop() throws ExecutionException, InterruptedExcep } } + @Test + public void testExceptionPredicateForRetryableException() throws Exception + { + final AtomicInteger count = new AtomicInteger(); + String result = RetryUtils.retry( + () -> { + if (count.incrementAndGet() >= 2) { + return "hey"; + } else { + throw new RetryableException(new RuntimeException("uhh")); + } + }, + e -> e instanceof RetryableException, + 3 + ); + Assert.assertEquals(result, "hey"); + Assert.assertEquals("count", 2, count.get()); + } } 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 e7f7e37e120a..cd54cf3d8d6c 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 @@ -41,6 +41,11 @@ import java.util.function.Function; import java.util.stream.Collectors; +/** + * CoordinatorDuty for automatic deletion of compaction configurations from the config table in metadata storage. + * Note that this will delete compaction configuration for inactive datasources + * (datasource with no used and unused segments) immediately. + */ public class KillCompactionConfig implements CoordinatorDuty { private static final Logger log = new Logger(KillCompactionConfig.class);