kfaraz commented on code in PR #20063:
URL: https://github.com/apache/druid/pull/20063#discussion_r3812675906
##########
server/src/main/java/org/apache/druid/metadata/MetadataRuleManager.java:
##########
@@ -41,6 +42,14 @@ public interface MetadataRuleManager
List<Rule> getRulesWithDefault(String dataSource);
+ /**
+ * Immutable snapshot of the rules of all datasources. A caller that needs
the rules
+ * of more than one datasource, or of the same datasource more than once,
must use a
+ * snapshot so that a concurrent rule update cannot change the answer part
way
+ * through.
Review Comment:
Slightly reworded to avoid confusion since the current comment somewhat
gives the idea that subsequent calls to this method would return the same
results.
```suggestion
* Current snapshot of the rules of all datasources. Using a single
snapshot while performing an
* operation (such as a Coordinator duty run) allows the steps within the
operation to remain
* consistent with each other, even if the rules are updated concurrently.
```
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/UnloadUnusedSegments.java:
##########
@@ -83,13 +80,14 @@ private int dropUnusedSegments(
ServerHolder serverHolder,
DruidCoordinatorRuntimeParams params,
CoordinatorRunStats stats,
- Map<String, Boolean> broadcastStatusByDatasource
+ Map<String, Boolean> broadcastStatusByDatasource,
+ RetentionRulesSnapshot rulesSnapshot
Review Comment:
Nit: we can skip the extra arg.
##########
server/src/main/java/org/apache/druid/metadata/MetadataRuleManager.java:
##########
@@ -41,6 +42,14 @@ public interface MetadataRuleManager
List<Rule> getRulesWithDefault(String dataSource);
Review Comment:
I feel like we should get rid of these methods now since the new method is
superior to all of these. We should just expose these methods on the snapshot
class now.
##########
server/src/main/java/org/apache/druid/server/coordinator/rules/RetentionRulesSnapshot.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.rules;
+
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Immutable snapshot of the retention rules of every datasource, taken once
at the
+ * start of a coordinator run and carried in
+ * {@link org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams}.
+ * <p>
+ * Duties must read rules from this snapshot rather than from
+ * {@link org.apache.druid.metadata.MetadataRuleManager} directly. That
manager swaps
+ * its entire rule map as soon as a rule update is submitted, so a duty
reading it
+ * per segment can apply the old rules to some segments and the new rules to
others
+ * within a single run. The new rules are then evaluated against values the
run has
+ * already snapshotted from {@link
org.apache.druid.server.coordinator.CoordinatorDynamicConfig},
+ * notably {@code historicalTierAliases}: a rule naming a virtual tier is
meaningful
Review Comment:
```suggestion
* e.g. {@code historicalTierAliases}: a rule naming a virtual tier is
meaningful
```
##########
server/src/main/java/org/apache/druid/server/coordinator/rules/RetentionRulesSnapshot.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.rules;
+
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Immutable snapshot of the retention rules of every datasource, taken once
at the
+ * start of a coordinator run and carried in
+ * {@link org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams}.
+ * <p>
+ * Duties must read rules from this snapshot rather than from
+ * {@link org.apache.druid.metadata.MetadataRuleManager} directly. That
manager swaps
+ * its entire rule map as soon as a rule update is submitted, so a duty
reading it
+ * per segment can apply the old rules to some segments and the new rules to
others
+ * within a single run. The new rules are then evaluated against values the
run has
+ * already snapshotted from {@link
org.apache.druid.server.coordinator.CoordinatorDynamicConfig},
Review Comment:
simplified a bit
```suggestion
* Duties within a run must use a single snapshot to remain consistent with
each other.
```
##########
server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java:
##########
@@ -1286,6 +1209,94 @@ public void
testSegmentWithZeroRequiredReplicasHasZeroReplicationFactor()
Assert.assertEquals(0, replicaCounts.requiredAndLoadable());
}
+ @Test
+ public void testRuleChangeDuringRunIsNotObservedInThatRun()
+ {
+ mockPeon.loadSegment(EasyMock.anyObject(), EasyMock.anyObject(),
EasyMock.anyObject());
+ EasyMock.expectLastCall().atLeastOnce();
+ mockEmptyPeon();
+
+ final TestMetadataRuleManager ruleManager = new TestMetadataRuleManager();
+ ruleManager.overrideRule(
+ DATASOURCE,
+ Collections.singletonList(new
ForeverLoadRule(ImmutableMap.of("normal", 1), null)),
+ AUDIT_INFO
+ );
+
+ final DruidCluster druidCluster = DruidCluster
+ .builder()
+ .addTier("normal", createServerHolder("server1", "normal", mockPeon))
+ .build();
+
+ final DruidCoordinatorRuntimeParams params =
createCoordinatorRuntimeParams(druidCluster)
+ .withRetentionRulesSnapshot(ruleManager.getRulesSnapshot())
+ .withBalancerStrategy(new CostBalancerStrategy(balancerExecutor))
+ .withDynamicConfigs(
+
CoordinatorDynamicConfig.builder().withSmartSegmentLoading(false).build()
+ )
+ .withSegmentAssignerUsing(loadQueueManager)
+ .build();
+
+ // The rules change after the snapshot was taken, but before the duty runs
+ ruleManager.overrideRule(DATASOURCE, Collections.singletonList(new
ForeverDropRule()), AUDIT_INFO);
Review Comment:
I feel like this test is a bit of an overkill.
Since we are explicitly passing a snapshot into the runtime params, it seems
unnecessary to verify that performing operations on the rule manager directly
would not affect the duties (as the duties have no way to access the rule
manager itself).
The tests in `SQLMetadataRuleManagerTest` should suffice.
##########
server/src/main/java/org/apache/druid/server/coordinator/rules/RetentionRulesSnapshot.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.rules;
+
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Immutable snapshot of the retention rules of every datasource, taken once
at the
+ * start of a coordinator run and carried in
+ * {@link org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams}.
+ * <p>
+ * Duties must read rules from this snapshot rather than from
+ * {@link org.apache.druid.metadata.MetadataRuleManager} directly. That
manager swaps
+ * its entire rule map as soon as a rule update is submitted, so a duty
reading it
+ * per segment can apply the old rules to some segments and the new rules to
others
+ * within a single run. The new rules are then evaluated against values the
run has
+ * already snapshotted from {@link
org.apache.druid.server.coordinator.CoordinatorDynamicConfig},
+ * notably {@code historicalTierAliases}: a rule naming a virtual tier is
meaningful
+ * only together with the alias config that resolves it, and a rule resolving
to no
+ * tier at all makes every existing replica look unwanted.
+ */
+public class RetentionRulesSnapshot
+{
+ private static final RetentionRulesSnapshot EMPTY = new
RetentionRulesSnapshot(Map.of(), List.of());
+
+ /**
+ * Rules of each datasource, already concatenated with {@link
#clusterDefaultRules}.
+ */
+ private final Map<String, List<Rule>> datasourceToRulesWithDefault;
+ private final List<Rule> clusterDefaultRules;
+
+ public static RetentionRulesSnapshot empty()
+ {
+ return EMPTY;
+ }
+
+ /**
+ * @param datasourceToRules Rules configured for each datasource, not
including the
+ * cluster defaults.
+ * @param clusterDefaultRules Rules configured for the default datasource.
These apply
+ * to every datasource after its own rules.
+ */
+ public RetentionRulesSnapshot(Map<String, List<Rule>> datasourceToRules,
List<Rule> clusterDefaultRules)
+ {
+ this.clusterDefaultRules = List.copyOf(clusterDefaultRules);
+
+ final Map<String, List<Rule>> rulesWithDefault =
Maps.newHashMapWithExpectedSize(datasourceToRules.size());
+ datasourceToRules.forEach((datasource, rules) -> {
+ final List<Rule> combined = new ArrayList<>(rules.size() +
this.clusterDefaultRules.size());
+ combined.addAll(rules);
+ combined.addAll(this.clusterDefaultRules);
+ rulesWithDefault.put(datasource, Collections.unmodifiableList(combined));
+ });
+ this.datasourceToRulesWithDefault = Map.copyOf(rulesWithDefault);
+ }
+
+ /**
+ * Rules of the given datasource followed by the cluster default rules, or
just the
+ * cluster defaults if the datasource has no rules of its own.
Review Comment:
```suggestion
* All retention rules applicable to segments of this datasource.
* The returned list contains the override rules specified for the
datasource followed by the cluster default rules.
```
##########
server/src/test/java/org/apache/druid/server/coordinator/duty/UnloadUnusedSegmentsTest.java:
##########
@@ -274,6 +280,36 @@ public void test_unloadUnusedSegmentsFromAllServers()
Assert.assertEquals(1L, stats.getSegmentStat(Stats.Segments.UNNEEDED,
"tier2", broadcastDatasource));
}
+ @Test
+ public void test_broadcastStatusComesFromRulesSnapshot()
Review Comment:
```suggestion
public void test_broadcastSegmentsAreDropped_ifSnapshotHasNoBroadcastRule()
```
##########
server/src/main/java/org/apache/druid/server/coordinator/rules/RetentionRulesSnapshot.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.rules;
+
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Immutable snapshot of the retention rules of every datasource, taken once
at the
+ * start of a coordinator run and carried in
+ * {@link org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams}.
+ * <p>
+ * Duties must read rules from this snapshot rather than from
+ * {@link org.apache.druid.metadata.MetadataRuleManager} directly. That
manager swaps
+ * its entire rule map as soon as a rule update is submitted, so a duty
reading it
+ * per segment can apply the old rules to some segments and the new rules to
others
+ * within a single run. The new rules are then evaluated against values the
run has
+ * already snapshotted from {@link
org.apache.druid.server.coordinator.CoordinatorDynamicConfig},
+ * notably {@code historicalTierAliases}: a rule naming a virtual tier is
meaningful
+ * only together with the alias config that resolves it, and a rule resolving
to no
+ * tier at all makes every existing replica look unwanted.
+ */
+public class RetentionRulesSnapshot
+{
+ private static final RetentionRulesSnapshot EMPTY = new
RetentionRulesSnapshot(Map.of(), List.of());
+
+ /**
+ * Rules of each datasource, already concatenated with {@link
#clusterDefaultRules}.
+ */
+ private final Map<String, List<Rule>> datasourceToRulesWithDefault;
+ private final List<Rule> clusterDefaultRules;
+
+ public static RetentionRulesSnapshot empty()
+ {
+ return EMPTY;
+ }
+
+ /**
+ * @param datasourceToRules Rules configured for each datasource, not
including the
+ * cluster defaults.
+ * @param clusterDefaultRules Rules configured for the default datasource.
These apply
+ * to every datasource after its own rules.
+ */
+ public RetentionRulesSnapshot(Map<String, List<Rule>> datasourceToRules,
List<Rule> clusterDefaultRules)
+ {
+ this.clusterDefaultRules = List.copyOf(clusterDefaultRules);
+
+ final Map<String, List<Rule>> rulesWithDefault =
Maps.newHashMapWithExpectedSize(datasourceToRules.size());
+ datasourceToRules.forEach((datasource, rules) -> {
+ final List<Rule> combined = new ArrayList<>(rules.size() +
this.clusterDefaultRules.size());
+ combined.addAll(rules);
+ combined.addAll(this.clusterDefaultRules);
+ rulesWithDefault.put(datasource, Collections.unmodifiableList(combined));
+ });
+ this.datasourceToRulesWithDefault = Map.copyOf(rulesWithDefault);
+ }
+
+ /**
+ * Rules of the given datasource followed by the cluster default rules, or
just the
+ * cluster defaults if the datasource has no rules of its own.
+ */
+ public List<Rule> getRulesWithDefault(String datasource)
Review Comment:
Should we rename this method to reduce ambiguity?
Maybe something like
```suggestion
public List<Rule> getAllRulesApplicableOn(String datasource)
```
##########
server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java:
##########
@@ -1286,6 +1209,94 @@ public void
testSegmentWithZeroRequiredReplicasHasZeroReplicationFactor()
Assert.assertEquals(0, replicaCounts.requiredAndLoadable());
}
+ @Test
+ public void testRuleChangeDuringRunIsNotObservedInThatRun()
+ {
+ mockPeon.loadSegment(EasyMock.anyObject(), EasyMock.anyObject(),
EasyMock.anyObject());
+ EasyMock.expectLastCall().atLeastOnce();
+ mockEmptyPeon();
+
+ final TestMetadataRuleManager ruleManager = new TestMetadataRuleManager();
+ ruleManager.overrideRule(
+ DATASOURCE,
+ Collections.singletonList(new
ForeverLoadRule(ImmutableMap.of("normal", 1), null)),
+ AUDIT_INFO
+ );
+
+ final DruidCluster druidCluster = DruidCluster
+ .builder()
+ .addTier("normal", createServerHolder("server1", "normal", mockPeon))
+ .build();
+
+ final DruidCoordinatorRuntimeParams params =
createCoordinatorRuntimeParams(druidCluster)
+ .withRetentionRulesSnapshot(ruleManager.getRulesSnapshot())
+ .withBalancerStrategy(new CostBalancerStrategy(balancerExecutor))
+ .withDynamicConfigs(
+
CoordinatorDynamicConfig.builder().withSmartSegmentLoading(false).build()
+ )
+ .withSegmentAssignerUsing(loadQueueManager)
+ .build();
+
+ // The rules change after the snapshot was taken, but before the duty runs
+ ruleManager.overrideRule(DATASOURCE, Collections.singletonList(new
ForeverDropRule()), AUDIT_INFO);
+
+ final CoordinatorRunStats stats = runDutyAndGetStats(params);
+
+ // The snapshotted load rule is applied, not the drop rule that replaced it
+ Assert.assertEquals(24L, stats.getSegmentStat(Stats.Segments.ASSIGNED,
"normal", DATASOURCE));
+ Assert.assertEquals(0L, stats.get(Stats.Segments.DELETED,
DATASOURCE_STAT_KEY));
+ }
+
+ @Test
+ public void testRuleChangeToUnresolvedAliasTierDoesNotDropReplicas()
Review Comment:
Is there also a test that verifies that unresolved alias tier does drop
replicas?
I feel like that would add more value than verifying that the snapshot does
not change.
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/RunRules.java:
##########
@@ -159,11 +156,14 @@ private void
alertForInvalidRules(StrategicSegmentAssigner segmentAssigner)
);
}
- private Set<String> getBroadcastDatasources(DruidCoordinatorRuntimeParams
params)
+ private Set<String> getBroadcastDatasources(
+ DruidCoordinatorRuntimeParams params,
+ RetentionRulesSnapshot rulesSnapshot
Review Comment:
Nit: We don't need this as a separate arg. It can be extracted from the
params.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
