Allow empty tiered replicants map for load rules (#14432)

Changes:
- Add property `useDefaultTierForNull` for all load rules. This property determines the default
value of `tieredReplicants` if it is not specified. When true, the default is `_default_tier => 2 replicas`.
When false, the default is empty, i.e. no replicas on any tier.
- Fix validation to allow empty replicants map, so that the segment is used but not loaded anywhere.

docs/operations/rule-configuration.md

@@ -109,7 +109,16 @@ In the web console you can use the up and down arrows on the right side of the i
 
 Load rules define how Druid assigns segments to [historical process tiers](./mixed-workloads.md#historical-tiering), and how many replicas of a segment exist in each tier.
 
-If you have a single tier, Druid automatically names the tier `_default` and loads all segments onto it. If you define an additional tier, you must define a load rule to specify which segments to load on that tier. Until you define a load rule, your new tier remains empty.
+If you have a single tier, Druid automatically names the tier `_default`. If you define an additional tier, you must define a load rule to specify which segments to load on that tier. Until you define a load rule, your new tier remains empty.
+
+All load rules can have these properties:
+
+|Property|Description|Required|Default value|
+|---------|-----------|---------|-------------|
+| `tieredReplicants`| Map from tier names to the respective number of segment replicas to be loaded on those tiers. The number of replicas for each tier must be either 0 or a positive integer.| No | When `useDefaultTierForNull` is `true`, the default value is `{"_default_tier": 2}` i.e. 2 replicas to be loaded on the `_default_tier`.<br/><br/>When `useDefaultTierForNull` is `false`, the default value is `{}` i.e. no replicas to be loaded on any tier. |
+|`useDefaultTierForNull`|Determines the default value of `tieredReplicants` if it is not specified or set to `null`.| No | `true`|
+
+Specific types of load rules discussed below may have other properties too.
 
 ### Forever load rule
 
@@ -130,6 +139,7 @@ The following example places one replica of each segment on a custom tier named
 Set the following property:
 
 - `tieredReplicants`: a map of tier names to the number of segment replicas for that tier.
+- `useDefaultTierForNull`: This parameter determines the default value of `tieredReplicants` and only has an effect if the field is not present. The default value of `useDefaultTierForNull` is true.
 
 ### Period load rule
 
@@ -158,6 +168,7 @@ Set the following properties:
 
   You can use this property to load segments with future start and end dates, where "future" is relative to the time when the Coordinator evaluates data against the rule. Defaults to `true`.
 - `tieredReplicants`: a map of tier names to the number of segment replicas for that tier.
+- `useDefaultTierForNull`: This parameter determines the default value of `tieredReplicants` and only has an effect if the field is not present. The default value of `useDefaultTierForNull` is true.
 
 ### Interval load rule
 
@@ -180,6 +191,7 @@ Set the following properties:
 
 - `interval`: the load interval specified as an [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) range encoded as a string.
 - `tieredReplicants`: a map of tier names to the number of segment replicas for that tier.
+- `useDefaultTierForNull`: This parameter determines the default value of `tieredReplicants` and only has an effect if the field is not present. The default value of `useDefaultTierForNull` is true.
 
 ## Drop rules
 

server/src/main/java/org/apache/druid/metadata/SQLMetadataRuleManager.java

@@ -96,7 +96,8 @@ public class SQLMetadataRuleManager implements MetadataRuleManager
                       ImmutableMap.of(
                           DruidServer.DEFAULT_TIER,
                           DruidServer.DEFAULT_NUM_REPLICANTS
-                      )
+                      ),
+                      null
                   )
               );
               final String version = DateTimes.nowUtc().toString();

server/src/main/java/org/apache/druid/server/coordinator/rules/ForeverLoadRule.java

@@ -21,30 +21,24 @@ package org.apache.druid.server.coordinator.rules;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.collect.ImmutableMap;
-import org.apache.druid.client.DruidServer;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
+import javax.annotation.Nullable;
 import java.util.Map;
-import java.util.Objects;
 
 /**
  */
 public class ForeverLoadRule extends LoadRule
 {
-  private final Map<String, Integer> tieredReplicants;
-
   @JsonCreator
   public ForeverLoadRule(
-      @JsonProperty("tieredReplicants") Map<String, Integer> tieredReplicants
+      @JsonProperty("tieredReplicants") Map<String, Integer> tieredReplicants,
+      @JsonProperty("useDefaultTierForNull") @Nullable Boolean useDefaultTierForNull
   )
   {
-    this.tieredReplicants = tieredReplicants == null
+    super(tieredReplicants, useDefaultTierForNull);
-                            ? ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS)
-                            : tieredReplicants;
-    validateTieredReplicants(this.tieredReplicants);
   }
 
   @Override
@@ -54,20 +48,6 @@ public class ForeverLoadRule extends LoadRule
     return "loadForever";
   }
 
-  @Override
-  @JsonProperty
-  public Map<String, Integer> getTieredReplicants()
-  {
-    return tieredReplicants;
-  }
-
-  @Override
-  public int getNumReplicants(String tier)
-  {
-    Integer retVal = tieredReplicants.get(tier);
-    return (retVal == null) ? 0 : retVal;
-  }
-
   @Override
   public boolean appliesTo(DataSegment segment, DateTime referenceTimestamp)
   {
@@ -80,22 +60,4 @@ public class ForeverLoadRule extends LoadRule
     return true;
   }
 
-  @Override
-  public boolean equals(Object o)
-  {
-    if (this == o) {
-      return true;
-    }
-    if (o == null || getClass() != o.getClass()) {
-      return false;
-    }
-    ForeverLoadRule that = (ForeverLoadRule) o;
-    return Objects.equals(tieredReplicants, that.tieredReplicants);
-  }
-
-  @Override
-  public int hashCode()
-  {
-    return Objects.hash(tieredReplicants);
-  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/IntervalLoadRule.java

@@ -21,14 +21,14 @@ package org.apache.druid.server.coordinator.rules;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.collect.ImmutableMap;
-import org.apache.druid.client.DruidServer;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 
+import javax.annotation.Nullable;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  */
@@ -37,16 +37,15 @@ public class IntervalLoadRule extends LoadRule
   private static final Logger log = new Logger(IntervalLoadRule.class);
 
   private final Interval interval;
-  private final Map<String, Integer> tieredReplicants;
 
   @JsonCreator
   public IntervalLoadRule(
       @JsonProperty("interval") Interval interval,
-      @JsonProperty("tieredReplicants") Map<String, Integer> tieredReplicants
+      @JsonProperty("tieredReplicants") Map<String, Integer> tieredReplicants,
+      @JsonProperty("useDefaultTierForNull") @Nullable Boolean useDefaultTierForNull
   )
   {
-    this.tieredReplicants = tieredReplicants == null ? ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS) : tieredReplicants;
+    super(tieredReplicants, useDefaultTierForNull);
-    validateTieredReplicants(this.tieredReplicants);
     this.interval = interval;
   }
 
@@ -57,20 +56,6 @@ public class IntervalLoadRule extends LoadRule
     return "loadByInterval";
   }
 
-  @Override
-  @JsonProperty
-  public Map<String, Integer> getTieredReplicants()
-  {
-    return tieredReplicants;
-  }
-
-  @Override
-  public int getNumReplicants(String tier)
-  {
-    final Integer retVal = tieredReplicants.get(tier);
-    return retVal == null ? 0 : retVal;
-  }
-
   @JsonProperty
   public Interval getInterval()
   {
@@ -98,24 +83,16 @@ public class IntervalLoadRule extends LoadRule
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-
+    if (!super.equals(o)) {
+      return false;
+    }
     IntervalLoadRule that = (IntervalLoadRule) o;
-
+    return Objects.equals(interval, that.interval);
-    if (interval != null ? !interval.equals(that.interval) : that.interval != null) {
-      return false;
-    }
-    if (tieredReplicants != null ? !tieredReplicants.equals(that.tieredReplicants) : that.tieredReplicants != null) {
-      return false;
-    }
-
-    return true;
   }
 
   @Override
   public int hashCode()
   {
-    int result = interval != null ? interval.hashCode() : 0;
+    return Objects.hash(super.hashCode(), interval);
-    result = 31 * result + (tieredReplicants != null ? tieredReplicants.hashCode() : 0);
-    return result;
   }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/LoadRule.java

@@ -19,39 +19,106 @@
 
 package org.apache.druid.server.coordinator.rules;
 
-import org.apache.druid.java.util.common.IAE;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.error.InvalidInput;
 import org.apache.druid.timeline.DataSegment;
 
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * LoadRules indicate the number of replicants a segment should have in a given tier.
  */
 public abstract class LoadRule implements Rule
 {
+  private final Map<String, Integer> tieredReplicants;
+  /**
+   * Used to determing the default value if tieredReplicants is null in {@link #handleNullTieredReplicants}.
+   */
+  private final boolean useDefaultTierForNull;
+
+  protected LoadRule(Map<String, Integer> tieredReplicants, Boolean useDefaultTierForNull)
+  {
+    this.useDefaultTierForNull = Configs.valueOrDefault(useDefaultTierForNull, true);
+    this.tieredReplicants = handleNullTieredReplicants(tieredReplicants, this.useDefaultTierForNull);
+    validateTieredReplicants(this.tieredReplicants);
+  }
+
+  @JsonProperty
+  public Map<String, Integer> getTieredReplicants()
+  {
+    return tieredReplicants;
+  }
+
+  @JsonProperty
+  public boolean useDefaultTierForNull()
+  {
+    return useDefaultTierForNull;
+  }
+
   @Override
   public void run(DataSegment segment, SegmentActionHandler handler)
   {
     handler.replicateSegment(segment, getTieredReplicants());
   }
 
-  protected static void validateTieredReplicants(final Map<String, Integer> tieredReplicants)
+  /**
+   * Returns the given {@code tieredReplicants} map unchanged if it is non-null (including empty).
+   * Returns the following default values if the given map is null.
+   * <ul>
+   * <li>If {@code useDefaultTierForNull} is true, returns a singleton map from {@link DruidServer#DEFAULT_TIER} to {@link DruidServer#DEFAULT_NUM_REPLICANTS}.</li>
+   * <li>If {@code useDefaultTierForNull} is false, returns an empty map. This causes segments to have a replication factor of 0 and not get assigned to any historical.</li>
+   * </ul>
+   */
+  private static Map<String, Integer> handleNullTieredReplicants(final Map<String, Integer> tieredReplicants, boolean useDefaultTierForNull)
   {
-    if (tieredReplicants.size() == 0) {
+    if (useDefaultTierForNull) {
-      throw new IAE("A rule with empty tiered replicants is invalid");
+      return Configs.valueOrDefault(tieredReplicants, ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS));
+    } else {
+      return Configs.valueOrDefault(tieredReplicants, ImmutableMap.of());
     }
+  }
+
+  private static void validateTieredReplicants(final Map<String, Integer> tieredReplicants)
+  {
     for (Map.Entry<String, Integer> entry : tieredReplicants.entrySet()) {
       if (entry.getValue() == null) {
-        throw new IAE("Replicant value cannot be empty");
+        throw InvalidInput.exception("Invalid number of replicas for tier [%s]. Value must not be null.", entry.getKey());
       }
       if (entry.getValue() < 0) {
-        throw new IAE("Replicant value [%d] is less than 0, which is not allowed", entry.getValue());
+        throw InvalidInput.exception("Invalid number of replicas for tier [%s]. Value [%d] must be positive.", entry.getKey(), entry.getValue());
       }
     }
   }
 
-  public abstract Map<String, Integer> getTieredReplicants();
+  public int getNumReplicants(String tier)
+  {
+    Integer retVal = getTieredReplicants().get(tier);
+    return (retVal == null) ? 0 : retVal;
+  }
 
-  public abstract int getNumReplicants(String tier);
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    LoadRule loadRule = (LoadRule) o;
+    return useDefaultTierForNull == loadRule.useDefaultTierForNull && Objects.equals(
+        tieredReplicants,
+        loadRule.tieredReplicants
+    );
+  }
 
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(tieredReplicants, useDefaultTierForNull);
+  }
 }

server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodLoadRule.java

@@ -21,15 +21,15 @@ package org.apache.druid.server.coordinator.rules;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.collect.ImmutableMap;
-import org.apache.druid.client.DruidServer;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.timeline.DataSegment;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 import org.joda.time.Period;
 
+import javax.annotation.Nullable;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  */
@@ -40,17 +40,16 @@ public class PeriodLoadRule extends LoadRule
 
   private final Period period;
   private final boolean includeFuture;
-  private final Map<String, Integer> tieredReplicants;
 
   @JsonCreator
   public PeriodLoadRule(
       @JsonProperty("period") Period period,
       @JsonProperty("includeFuture") Boolean includeFuture,
-      @JsonProperty("tieredReplicants") Map<String, Integer> tieredReplicants
+      @JsonProperty("tieredReplicants") Map<String, Integer> tieredReplicants,
+      @JsonProperty("useDefaultTierForNull") @Nullable Boolean useDefaultTierForNull
   )
   {
-    this.tieredReplicants = tieredReplicants == null ? ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS) : tieredReplicants;
+    super(tieredReplicants, useDefaultTierForNull);
-    validateTieredReplicants(this.tieredReplicants);
     this.period = period;
     this.includeFuture = includeFuture == null ? DEFAULT_INCLUDE_FUTURE : includeFuture;
   }
@@ -74,20 +73,6 @@ public class PeriodLoadRule extends LoadRule
     return includeFuture;
   }
 
-  @Override
-  @JsonProperty
-  public Map<String, Integer> getTieredReplicants()
-  {
-    return tieredReplicants;
-  }
-
-  @Override
-  public int getNumReplicants(String tier)
-  {
-    final Integer retVal = tieredReplicants.get(tier);
-    return retVal == null ? 0 : retVal;
-  }
-
   @Override
   public boolean appliesTo(DataSegment segment, DateTime referenceTimestamp)
   {
@@ -99,4 +84,26 @@ public class PeriodLoadRule extends LoadRule
   {
     return Rules.eligibleForLoad(period, interval, referenceTimestamp, includeFuture);
   }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    if (!super.equals(o)) {
+      return false;
+    }
+    PeriodLoadRule that = (PeriodLoadRule) o;
+    return includeFuture == that.includeFuture && Objects.equals(period, that.period);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(super.hashCode(), period, includeFuture);
+  }
 }

server/src/test/java/org/apache/druid/metadata/SQLMetadataRuleManagerTest.java

@@ -108,7 +108,8 @@ public class SQLMetadataRuleManagerTest
     List<Rule> rules = Collections.singletonList(
         new IntervalLoadRule(
             Intervals.of("2015-01-01/2015-02-01"),
-            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS)
+            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS),
+            null
         )
     );
     ruleManager.overrideRule(DATASOURCE, rules, createAuditInfo("override rule"));
@@ -171,7 +172,8 @@ public class SQLMetadataRuleManagerTest
     List<Rule> rules = Collections.singletonList(
         new IntervalLoadRule(
             Intervals.of("2015-01-01/2015-02-01"),
-            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS)
+            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS),
+            null
         )
     );
     final AuditInfo auditInfo = createAuditInfo("create audit entry");
@@ -202,7 +204,8 @@ public class SQLMetadataRuleManagerTest
             Intervals.of("2015-01-01/2015-02-01"), ImmutableMap.of(
                 DruidServer.DEFAULT_TIER,
                 DruidServer.DEFAULT_NUM_REPLICANTS
-        )
+        ),
+            null
         )
     );
     final AuditInfo auditInfo = createAuditInfo("test_comment");
@@ -232,7 +235,8 @@ public class SQLMetadataRuleManagerTest
     List<Rule> rules = ImmutableList.of(
         new IntervalLoadRule(
             Intervals.of("2015-01-01/2015-02-01"),
-            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS)
+            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS),
+            null
         )
     );
     ruleManager.overrideRule(DATASOURCE, rules, createAuditInfo("test"));
@@ -258,7 +262,8 @@ public class SQLMetadataRuleManagerTest
     List<Rule> rules = ImmutableList.of(
         new IntervalLoadRule(
             Intervals.of("2015-01-01/2015-02-01"),
-            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS)
+            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS),
+            null
         )
     );
     ruleManager.overrideRule(DATASOURCE, rules, createAuditInfo("update rules"));
@@ -286,7 +291,8 @@ public class SQLMetadataRuleManagerTest
     List<Rule> rules = ImmutableList.of(
         new IntervalLoadRule(
             Intervals.of("2015-01-01/2015-02-01"),
-            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS)
+            ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS),
+            null
         )
     );
     ruleManager.overrideRule(DATASOURCE, rules, createAuditInfo("update rules"));

server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsProfiler.java

@@ -59,7 +59,7 @@ public class BalanceSegmentsProfiler
   List<DataSegment> segments = new ArrayList<>();
   ServiceEmitter emitter;
   MetadataRuleManager manager;
-  PeriodLoadRule loadRule = new PeriodLoadRule(new Period("P5000Y"), null, ImmutableMap.of("normal", 3));
+  PeriodLoadRule loadRule = new PeriodLoadRule(new Period("P5000Y"), null, ImmutableMap.of("normal", 3), null);
   List<Rule> rules = ImmutableList.of(loadRule);
 
   @Before

server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java

@@ -207,7 +207,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
     String tier = "hot";
 
     // Setup MetadataRuleManager
-    Rule foreverLoadRule = new ForeverLoadRule(ImmutableMap.of(tier, 2));
+    Rule foreverLoadRule = new ForeverLoadRule(ImmutableMap.of(tier, 2), null);
     EasyMock.expect(metadataRuleManager.getRulesWithDefault(EasyMock.anyString()))
             .andReturn(ImmutableList.of(foreverLoadRule)).atLeastOnce();
 
@@ -325,8 +325,8 @@ public class DruidCoordinatorTest extends CuratorTestBase
   public void testCoordinatorTieredRun() throws Exception
   {
     final String dataSource = "dataSource", hotTierName = "hot", coldTierName = "cold";
-    final Rule hotTier = new IntervalLoadRule(Intervals.of("2018-01-01/P1M"), ImmutableMap.of(hotTierName, 1));
+    final Rule hotTier = new IntervalLoadRule(Intervals.of("2018-01-01/P1M"), ImmutableMap.of(hotTierName, 1), null);
-    final Rule coldTier = new ForeverLoadRule(ImmutableMap.of(coldTierName, 1));
+    final Rule coldTier = new ForeverLoadRule(ImmutableMap.of(coldTierName, 1), null);
     final String loadPathCold = "/druid/loadqueue/cold:1234";
     final DruidServer hotServer = new DruidServer("hot", "hot", null, 5L, ServerType.HISTORICAL, hotTierName, 0);
     final DruidServer coldServer = new DruidServer("cold", "cold", null, 5L, ServerType.HISTORICAL, coldTierName, 0);

server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java

@@ -129,7 +129,8 @@ public class RunRulesTest
         Collections.singletonList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01/2012-01-02"),
-                ImmutableMap.of("normal", 2)
+                ImmutableMap.of("normal", 2),
+                null
             )
         )).atLeastOnce();
     EasyMock.replay(databaseRuleManager);
@@ -185,7 +186,8 @@ public class RunRulesTest
         Collections.singletonList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"),
-                ImmutableMap.of("hot", 2, "normal", 2)
+                ImmutableMap.of("hot", 2, "normal", 2),
+                null
             )
         )).atLeastOnce();
     EasyMock.replay(databaseRuleManager);
@@ -248,15 +250,18 @@ public class RunRulesTest
         Lists.newArrayList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T06:00:00.000Z"),
-                ImmutableMap.of("hot", 1)
+                ImmutableMap.of("hot", 1),
+                null
             ),
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T12:00:00.000Z"),
-                ImmutableMap.of("normal", 1)
+                ImmutableMap.of("normal", 1),
+                null
             ),
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"),
-                ImmutableMap.of("cold", 1)
+                ImmutableMap.of("cold", 1),
+                null
             )
         )).atLeastOnce();
     EasyMock.replay(databaseRuleManager);
@@ -353,11 +358,13 @@ public class RunRulesTest
         Lists.newArrayList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T06:00:00.000Z"),
-                ImmutableMap.of("hot", 2)
+                ImmutableMap.of("hot", 2),
+                null
             ),
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"),
-                ImmutableMap.of("cold", 1)
+                ImmutableMap.of("cold", 1),
+                null
             )
         )
     ).atLeastOnce();
@@ -399,11 +406,13 @@ public class RunRulesTest
         Lists.newArrayList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T12:00:00.000Z"),
-                ImmutableMap.of("hot", 1)
+                ImmutableMap.of("hot", 1),
+                null
             ),
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"),
-                ImmutableMap.of("normal", 1)
+                ImmutableMap.of("normal", 1),
+                null
             )
         )
     ).atLeastOnce();
@@ -445,11 +454,13 @@ public class RunRulesTest
         Lists.newArrayList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T12:00:00.000Z"),
-                ImmutableMap.of("hot", 1)
+                ImmutableMap.of("hot", 1),
+                null
             ),
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"),
-                ImmutableMap.of("normal", 1)
+                ImmutableMap.of("normal", 1),
+                null
             )
         )
     ).atLeastOnce();
@@ -480,7 +491,8 @@ public class RunRulesTest
             Collections.singletonList(
                 new IntervalLoadRule(
                     Intervals.of("2012-01-02T00:00:00.000Z/2012-01-03T00:00:00.000Z"),
-                    ImmutableMap.of("normal", 1)
+                    ImmutableMap.of("normal", 1),
+                    null
                 )
             )
         )
@@ -527,7 +539,8 @@ public class RunRulesTest
         Lists.newArrayList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T12:00:00.000Z"),
-                ImmutableMap.of("normal", 1)
+                ImmutableMap.of("normal", 1),
+                null
             ),
             new IntervalDropRule(Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"))
         )
@@ -564,7 +577,8 @@ public class RunRulesTest
         Lists.newArrayList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T12:00:00.000Z"),
-                ImmutableMap.of("normal", 1)
+                ImmutableMap.of("normal", 1),
+                null
             ),
             new IntervalDropRule(Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"))
         )
@@ -620,7 +634,8 @@ public class RunRulesTest
         Lists.newArrayList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T12:00:00.000Z"),
-                ImmutableMap.of("hot", 1)
+                ImmutableMap.of("hot", 1),
+                null
             ),
             new IntervalDropRule(Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"))
         )
@@ -663,7 +678,8 @@ public class RunRulesTest
         Lists.newArrayList(
             new IntervalLoadRule(
                 Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T12:00:00.000Z"),
-                ImmutableMap.of("hot", 1)
+                ImmutableMap.of("hot", 1),
+                null
             ),
             new IntervalDropRule(Intervals.of("2012-01-01T00:00:00.000Z/2012-01-02T00:00:00.000Z"))
         )
@@ -703,7 +719,8 @@ public class RunRulesTest
             Collections.singletonList(
                 new IntervalLoadRule(
                     Intervals.of("2012-01-01T00:00:00.000Z/2012-01-01T01:00:00.000Z"),
-                    ImmutableMap.of("normal", 0)
+                    ImmutableMap.of("normal", 0),
+                    null
                 )
             )
         )
@@ -768,7 +785,8 @@ public class RunRulesTest
             Collections.singletonList(
                 new IntervalLoadRule(
                     Intervals.of("2012-01-01T00:00:00.000Z/2013-01-01T00:00:00.000Z"),
-                    ImmutableMap.of("hot", 2)
+                    ImmutableMap.of("hot", 2),
+                    null
                 )
             )
         )
@@ -847,7 +865,8 @@ public class RunRulesTest
             Collections.singletonList(
                 new IntervalLoadRule(
                     Intervals.of("2012-01-01/2013-01-01"),
-                    ImmutableMap.of("hot", 1, DruidServer.DEFAULT_TIER, 1)
+                    ImmutableMap.of("hot", 1, DruidServer.DEFAULT_TIER, 1),
+                    null
                 )
             )
         )
@@ -899,7 +918,8 @@ public class RunRulesTest
             Collections.singletonList(
                 new IntervalLoadRule(
                     Intervals.of("2012-01-01/2013-01-02"),
-                    ImmutableMap.of("normal", 1)
+                    ImmutableMap.of("normal", 1),
+                    null
                 )
             )
         )
@@ -982,7 +1002,7 @@ public class RunRulesTest
     mockEmptyPeon();
 
     EasyMock.expect(databaseRuleManager.getRulesWithDefault(EasyMock.anyObject())).andReturn(
-        Collections.singletonList(new ForeverLoadRule(ImmutableMap.of(DruidServer.DEFAULT_TIER, 1)))).atLeastOnce();
+        Collections.singletonList(new ForeverLoadRule(ImmutableMap.of(DruidServer.DEFAULT_TIER, 1), null))).atLeastOnce();
     EasyMock.replay(databaseRuleManager);
 
     DruidCluster druidCluster = DruidCluster.builder().add(
@@ -1019,7 +1039,7 @@ public class RunRulesTest
 
     EasyMock.expect(databaseRuleManager.getRulesWithDefault(EasyMock.anyObject())).andReturn(
         Collections.singletonList(
-            new ForeverLoadRule(ImmutableMap.of(DruidServer.DEFAULT_TIER, 3))
+            new ForeverLoadRule(ImmutableMap.of(DruidServer.DEFAULT_TIER, 3), null)
         )).atLeastOnce();
     EasyMock.replay(databaseRuleManager);
 
@@ -1081,7 +1101,7 @@ public class RunRulesTest
 
     EasyMock.expect(databaseRuleManager.getRulesWithDefault(EasyMock.anyObject())).andReturn(
         Collections.singletonList(
-            new ForeverLoadRule(ImmutableMap.of(DruidServer.DEFAULT_TIER, 1))
+            new ForeverLoadRule(ImmutableMap.of(DruidServer.DEFAULT_TIER, 1), null)
         )).atLeastOnce();
     EasyMock.replay(databaseRuleManager);
 
@@ -1133,7 +1153,8 @@ public class RunRulesTest
     EasyMock.expect(databaseRuleManager.getRulesWithDefault(EasyMock.anyObject())).andReturn(
         Collections.singletonList(
             new ForeverLoadRule(
-                ImmutableMap.of(DruidServer.DEFAULT_TIER, numReplicants)
+                ImmutableMap.of(DruidServer.DEFAULT_TIER, numReplicants),
+                null
             )
         )).atLeastOnce();
     EasyMock.replay(databaseRuleManager);
@@ -1194,7 +1215,8 @@ public class RunRulesTest
     EasyMock.expect(databaseRuleManager.getRulesWithDefault(EasyMock.anyObject())).andReturn(
         Collections.singletonList(
             new ForeverLoadRule(
-                ImmutableMap.of(DruidServer.DEFAULT_TIER, numReplicants)
+                ImmutableMap.of(DruidServer.DEFAULT_TIER, numReplicants),
+                null
             )
         )).atLeastOnce();
     EasyMock.replay(databaseRuleManager);

server/src/test/java/org/apache/druid/server/coordinator/duty/UnloadUnusedSegmentsTest.java

@@ -316,7 +316,8 @@ public class UnloadUnusedSegmentsTest
                 ImmutableMap.of(
                     DruidServer.DEFAULT_TIER, 1,
                     "tier2", 1
-                )
+                ),
+                null
             )
         )).anyTimes();
 
@@ -326,7 +327,8 @@ public class UnloadUnusedSegmentsTest
                 ImmutableMap.of(
                     DruidServer.DEFAULT_TIER, 1,
                     "tier2", 1
-                )
+                ),
+                null
             )
         )).anyTimes();
 

server/src/test/java/org/apache/druid/server/coordinator/rules/ForeverLoadRuleTest.java

@@ -22,8 +22,10 @@ package org.apache.druid.server.coordinator.rules;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.client.DruidServer;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.DruidExceptionMatcher;
 import org.apache.druid.jackson.DefaultObjectMapper;
-import org.apache.druid.java.util.common.IAE;
+import org.hamcrest.MatcherAssert;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -32,12 +34,12 @@ import java.util.Map;
 
 public class ForeverLoadRuleTest
 {
+  private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
+
   @Test
-  public void testSerdeNullTieredReplicants() throws Exception
+  public void testSerde() throws Exception
   {
-    ForeverLoadRule rule = new ForeverLoadRule(
+    ForeverLoadRule rule = new ForeverLoadRule(null, null);
-        null
-    );
 
     ObjectMapper jsonMapper = new DefaultObjectMapper();
     Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
@@ -47,45 +49,79 @@ public class ForeverLoadRuleTest
   }
 
   @Test
-  public void testMappingNullTieredReplicants() throws Exception
+  public void testCreatingNegativeTieredReplicants()
   {
-    String inputJson = "{\n"
+    MatcherAssert.assertThat(
-                      + " \"type\": \"loadForever\"\n"
+        Assert.assertThrows(DruidException.class, () ->
-                      + "}";
+            new ForeverLoadRule(
-    String expectedJson = "    {\n"
+                ImmutableMap.of(DruidServer.DEFAULT_TIER, -1),
-                          + "      \"tieredReplicants\": {\n"
+                null
-                          + "        \"" + DruidServer.DEFAULT_TIER + "\": " + DruidServer.DEFAULT_NUM_REPLICANTS + "\n"
+            )
-                          + "      },\n"
+        ),
-                          + "      \"type\": \"loadForever\"\n"
+        DruidExceptionMatcher.invalidInput().expectMessageContains(
-                          + "    }";
+            "Invalid number of replicas for tier [_default_tier]. Value [-1] must be positive."
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
+        )
-    ForeverLoadRule inputForeverLoadRule = jsonMapper.readValue(inputJson, ForeverLoadRule.class);
+    );
-    ForeverLoadRule expectedForeverLoadRule = jsonMapper.readValue(expectedJson, ForeverLoadRule.class);
-    Assert.assertEquals(expectedForeverLoadRule.getTieredReplicants(), inputForeverLoadRule.getTieredReplicants());
   }
 
-  @Test(expected = IAE.class)
+  @Test
   public void testEmptyTieredReplicants() throws Exception
   {
-    ForeverLoadRule rule = new ForeverLoadRule(
+    ForeverLoadRule rule = new ForeverLoadRule(ImmutableMap.of(), false);
-        ImmutableMap.of()
-    );
 
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
+    LoadRule reread = (LoadRule) OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
-    Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
+    Assert.assertEquals(ImmutableMap.of(), reread.getTieredReplicants());
   }
 
-  @Test(expected = IAE.class)
+  @Test
-  public void testEmptyReplicantValue() throws Exception
+  public void testNullReplicantValue()
   {
     // Immutable map does not allow null values
     Map<String, Integer> tieredReplicants = new HashMap<>();
     tieredReplicants.put("tier", null);
-    ForeverLoadRule rule = new ForeverLoadRule(
-        tieredReplicants
-    );
 
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
+    MatcherAssert.assertThat(
-    Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
+        Assert.assertThrows(DruidException.class, () ->
+            new ForeverLoadRule(
+                tieredReplicants,
+                true
+            )
+        ),
+        DruidExceptionMatcher.invalidInput().expectMessageContains(
+            "Invalid number of replicas for tier [tier]. Value must not be null."
+        )
+    );
+  }
+
+  @Test
+  public void testShouldCreateDefaultTier() throws Exception
+  {
+    String inputJson = "    {\n"
+                       + "     \"type\": \"loadForever\"\n"
+                       + "  }";
+    ForeverLoadRule inputForeverLoadRule = OBJECT_MAPPER.readValue(inputJson, ForeverLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS), inputForeverLoadRule.getTieredReplicants());
+  }
+
+  @Test
+  public void testUseDefaultTierAsTrueShouldCreateDefaultTier() throws Exception
+  {
+    String inputJson = "    {\n"
+                       + "     \"type\": \"loadForever\"\n,"
+                       + "     \"useDefaultTierForNull\": \"true\"\n"
+                       + "  }";
+    ForeverLoadRule inputForeverLoadRule = OBJECT_MAPPER.readValue(inputJson, ForeverLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS), inputForeverLoadRule.getTieredReplicants());
+  }
+
+  @Test
+  public void testUseDefaultTierAsFalseShouldCreateEmptyMap() throws Exception
+  {
+    String inputJson = "    {\n"
+                       + "     \"type\": \"loadForever\"\n,"
+                       + "     \"useDefaultTierForNull\": \"false\"\n"
+                       + "  }";
+    ForeverLoadRule inputForeverLoadRule = OBJECT_MAPPER.readValue(inputJson, ForeverLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(), inputForeverLoadRule.getTieredReplicants());
   }
 }

server/src/test/java/org/apache/druid/server/coordinator/rules/IntervalLoadRuleTest.java

@@ -22,25 +22,34 @@ package org.apache.druid.server.coordinator.rules;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.client.DruidServer;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.DruidExceptionMatcher;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.Intervals;
+import org.hamcrest.MatcherAssert;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
+ * Unit tests for {@link IntervalLoadRule}
  */
 public class IntervalLoadRuleTest
 {
+  private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
+
   @Test
   public void testSerde() throws Exception
   {
     IntervalLoadRule rule = new IntervalLoadRule(
         Intervals.of("0/3000"),
-        ImmutableMap.of(DruidServer.DEFAULT_TIER, 2)
+        ImmutableMap.of(DruidServer.DEFAULT_TIER, 2),
+        null
     );
 
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
+    Rule reread = OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
-    Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
 
     Assert.assertEquals(rule, reread);
   }
@@ -49,36 +58,85 @@ public class IntervalLoadRuleTest
   public void testSerdeNullTieredReplicants() throws Exception
   {
     IntervalLoadRule rule = new IntervalLoadRule(
-        Intervals.of("0/3000"), null
+        Intervals.of("0/3000"), null, false
     );
 
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
+    Rule reread = OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
-    Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
 
     Assert.assertEquals(rule, reread);
-    Assert.assertEquals(
+    Assert.assertEquals(ImmutableMap.of(), rule.getTieredReplicants());
-        ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS),
+  }
-        rule.getTieredReplicants()
+
+  @Test
+  public void testCreatingNegativeTieredReplicants()
+  {
+    MatcherAssert.assertThat(
+        Assert.assertThrows(DruidException.class, () ->
+            new IntervalLoadRule(
+                Intervals.of("0/3000"),
+                ImmutableMap.of(DruidServer.DEFAULT_TIER, -1),
+                null
+            )
+        ),
+        DruidExceptionMatcher.invalidInput().expectMessageContains(
+            "Invalid number of replicas for tier [_default_tier]. Value [-1] must be positive."
+        )
     );
   }
 
   @Test
-  public void testMappingNullTieredReplicants() throws Exception
+  public void testNullReplicantValue()
+  {
+    // Immutable map does not allow null values
+    Map<String, Integer> tieredReplicants = new HashMap<>();
+    tieredReplicants.put("tier", null);
+
+    MatcherAssert.assertThat(
+        Assert.assertThrows(DruidException.class, () ->
+            new IntervalLoadRule(
+                Intervals.of("0/3000"),
+                tieredReplicants,
+                null
+            )
+        ),
+        DruidExceptionMatcher.invalidInput().expectMessageContains(
+            "Invalid number of replicas for tier [tier]. Value must not be null."
+        )
+    );
+  }
+
+  @Test
+  public void testShouldCreateDefaultTier() throws Exception
   {
     String inputJson = "     {\n"
                        + "      \"interval\": \"0000-01-01T00:00:00.000-05:50:36/3000-01-01T00:00:00.000-06:00\",\n"
                        + "      \"type\": \"loadByInterval\"\n"
                        + "   }";
-    String expectedJson = "{\n"
+    IntervalLoadRule inputIntervalLoadRule = OBJECT_MAPPER.readValue(inputJson, IntervalLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS), inputIntervalLoadRule.getTieredReplicants());
+  }
+
+  @Test
+  public void testUseDefaultTierAsTrueShouldCreateDefaultTier() throws Exception
+  {
+    String inputJson = "     {\n"
                        + "      \"interval\": \"0000-01-01T00:00:00.000-05:50:36/3000-01-01T00:00:00.000-06:00\",\n"
-                          + "      \"tieredReplicants\": {\n"
+                       + "      \"type\": \"loadByInterval\",\n"
-                          + "        \"" + DruidServer.DEFAULT_TIER + "\": " + DruidServer.DEFAULT_NUM_REPLICANTS + "\n"
+                       + "      \"useDefaultTierForNull\": \"true\"\n"
-                          + "      },\n"
-                          + "      \"type\": \"loadByInterval\"\n"
                        + "   }";
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
+    IntervalLoadRule inputIntervalLoadRule = OBJECT_MAPPER.readValue(inputJson, IntervalLoadRule.class);
-    IntervalLoadRule inputIntervalLoadRule = jsonMapper.readValue(inputJson, IntervalLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS), inputIntervalLoadRule.getTieredReplicants());
-    IntervalLoadRule expectedIntervalLoadRule = jsonMapper.readValue(expectedJson, IntervalLoadRule.class);
+  }
-    Assert.assertEquals(expectedIntervalLoadRule, inputIntervalLoadRule);
+
+  @Test
+  public void testUseDefaultTierAsFalseShouldCreateEmptyMap() throws Exception
+  {
+    String inputJson = "     {\n"
+                       + "      \"interval\": \"0000-01-01T00:00:00.000-05:50:36/3000-01-01T00:00:00.000-06:00\",\n"
+                       + "      \"type\": \"loadByInterval\",\n"
+                       + "      \"useDefaultTierForNull\": \"false\"\n"
+                       + "   }";
+    IntervalLoadRule inputIntervalLoadRule = OBJECT_MAPPER.readValue(inputJson, IntervalLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(), inputIntervalLoadRule.getTieredReplicants());
   }
 }

server/src/test/java/org/apache/druid/server/coordinator/rules/LoadRuleTest.java

@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
+import nl.jqno.equalsverifier.EqualsVerifier;
 import org.apache.druid.client.DruidServer;
 import org.apache.druid.client.ImmutableDruidServer;
 import org.apache.druid.java.util.common.DateTimes;
@@ -705,7 +706,7 @@ public class LoadRuleTest
 
   private static LoadRule loadForever(final Map<String, Integer> tieredReplicants)
   {
-    return new ForeverLoadRule(tieredReplicants);
+    return new ForeverLoadRule(tieredReplicants, null);
   }
 
   private static LoadQueuePeon createEmptyPeon()
@@ -764,4 +765,13 @@ public class LoadRuleTest
     static final String T1 = "tier1";
     static final String T2 = "tier2";
   }
+
+  @Test
+  public void testEquals()
+  {
+    EqualsVerifier.forClass(LoadRule.class)
+                  .withNonnullFields("tieredReplicants")
+                  .usingGetClass()
+                  .verify();
+  }
 }

server/src/test/java/org/apache/druid/server/coordinator/rules/PeriodLoadRuleTest.java

@@ -21,22 +21,31 @@ package org.apache.druid.server.coordinator.rules;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
+import nl.jqno.equalsverifier.EqualsVerifier;
 import org.apache.druid.client.DruidServer;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.DruidExceptionMatcher;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.partition.NoneShardSpec;
+import org.hamcrest.MatcherAssert;
 import org.joda.time.DateTime;
 import org.joda.time.Interval;
 import org.joda.time.Period;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  */
 public class PeriodLoadRuleTest
 {
+  private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
+
   private static final DataSegment.Builder BUILDER = DataSegment
       .builder()
       .dataSource("test")
@@ -51,7 +60,8 @@ public class PeriodLoadRuleTest
     PeriodLoadRule rule = new PeriodLoadRule(
         new Period("P5000Y"),
         false,
-        ImmutableMap.of("", 0)
+        ImmutableMap.of("", 0),
+        null
     );
 
     Assert.assertTrue(rule.appliesTo(BUILDER.interval(Intervals.of("2012-01-01/2012-12-31")).build(), now));
@@ -66,7 +76,8 @@ public class PeriodLoadRuleTest
     PeriodLoadRule rule = new PeriodLoadRule(
         new Period("P1M"),
         false,
-        ImmutableMap.of("", 0)
+        ImmutableMap.of("", 0),
+        null
     );
 
     Assert.assertTrue(rule.appliesTo(BUILDER.interval(new Interval(now.minusWeeks(1), now)).build(), now));
@@ -93,7 +104,8 @@ public class PeriodLoadRuleTest
     PeriodLoadRule rule = new PeriodLoadRule(
         new Period("P1M"),
         false,
-            ImmutableMap.of("", 0)
+        ImmutableMap.of("", 0),
+        null
     );
 
     Assert.assertTrue(
@@ -119,12 +131,14 @@ public class PeriodLoadRuleTest
     PeriodLoadRule includeFutureRule = new PeriodLoadRule(
         new Period("P2D"),
         true,
-        ImmutableMap.of("", 0)
+        ImmutableMap.of("", 0),
+        null
     );
     PeriodLoadRule notIncludeFutureRule = new PeriodLoadRule(
         new Period("P2D"),
         false,
-        ImmutableMap.of("", 0)
+        ImmutableMap.of("", 0),
+        null
     );
 
     Assert.assertTrue(
@@ -148,11 +162,10 @@ public class PeriodLoadRuleTest
   public void testSerdeNull() throws Exception
   {
     PeriodLoadRule rule = new PeriodLoadRule(
-        new Period("P1D"), null, null
+        new Period("P1D"), null, null, null
     );
 
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
+    Rule reread = OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
-    Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
 
     Assert.assertEquals(rule.getPeriod(), ((PeriodLoadRule) reread).getPeriod());
     Assert.assertEquals(rule.isIncludeFuture(), ((PeriodLoadRule) reread).isIncludeFuture());
@@ -174,16 +187,100 @@ public class PeriodLoadRuleTest
     String expectedJson = "{\n"
                           + "      \"period\": \"P1D\",\n"
                           + "      \"includeFuture\": " + PeriodLoadRule.DEFAULT_INCLUDE_FUTURE + ",\n"
-                          + "      \"tieredReplicants\": {\n"
-                          + "        \"" + DruidServer.DEFAULT_TIER + "\": " + DruidServer.DEFAULT_NUM_REPLICANTS + "\n"
-                          + "      },\n"
                           + "      \"type\": \"loadByPeriod\"\n"
                           + "    }";
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
+    PeriodLoadRule inputPeriodLoadRule = OBJECT_MAPPER.readValue(inputJson, PeriodLoadRule.class);
-    PeriodLoadRule inputPeriodLoadRule = jsonMapper.readValue(inputJson, PeriodLoadRule.class);
+    PeriodLoadRule expectedPeriodLoadRule = OBJECT_MAPPER.readValue(expectedJson, PeriodLoadRule.class);
-    PeriodLoadRule expectedPeriodLoadRule = jsonMapper.readValue(expectedJson, PeriodLoadRule.class);
     Assert.assertEquals(expectedPeriodLoadRule.getTieredReplicants(), inputPeriodLoadRule.getTieredReplicants());
     Assert.assertEquals(expectedPeriodLoadRule.getPeriod(), inputPeriodLoadRule.getPeriod());
     Assert.assertEquals(expectedPeriodLoadRule.isIncludeFuture(), inputPeriodLoadRule.isIncludeFuture());
   }
+
+  @Test
+  public void testCreatingNegativeTieredReplicants()
+  {
+    MatcherAssert.assertThat(
+        Assert.assertThrows(DruidException.class, () ->
+            new PeriodLoadRule(
+                Period.days(1),
+                true,
+                ImmutableMap.of(DruidServer.DEFAULT_TIER, -1),
+                true
+            )
+        ),
+        DruidExceptionMatcher.invalidInput().expectMessageContains(
+            "Invalid number of replicas for tier [_default_tier]. Value [-1] must be positive."
+        )
+    );
+  }
+
+  @Test
+  public void testNullReplicantValue()
+  {
+    // Immutable map does not allow null values
+    Map<String, Integer> tieredReplicants = new HashMap<>();
+    tieredReplicants.put("tier", null);
+
+    MatcherAssert.assertThat(
+        Assert.assertThrows(DruidException.class, () ->
+            new PeriodLoadRule(
+                Period.days(1),
+                true,
+                tieredReplicants,
+                true
+            )
+        ),
+        DruidExceptionMatcher.invalidInput().expectMessageContains(
+            "Invalid number of replicas for tier [tier]. Value must not be null."
+        )
+    );
+  }
+
+
+  @Test
+  public void testShouldCreateDefaultTier() throws Exception
+  {
+    String inputJson = "      {\n"
+                       + "      \"period\": \"P1D\",\n"
+                       + "      \"includeFuture\": " + PeriodLoadRule.DEFAULT_INCLUDE_FUTURE + ",\n"
+                       + "      \"type\": \"loadByPeriod\"\n"
+                       + "    }";
+    PeriodLoadRule inputPeriodLoadRule = OBJECT_MAPPER.readValue(inputJson, PeriodLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS), inputPeriodLoadRule.getTieredReplicants());
+  }
+
+  @Test
+  public void testUseDefaultTierAsTrueShouldCreateDefaultTier() throws Exception
+  {
+    String inputJson = "      {\n"
+                       + "      \"period\": \"P1D\",\n"
+                       + "      \"includeFuture\": " + PeriodLoadRule.DEFAULT_INCLUDE_FUTURE + ",\n"
+                       + "      \"useDefaultTierForNull\": \"true\",\n"
+                       + "      \"type\": \"loadByPeriod\"\n"
+                       + "    }";
+    PeriodLoadRule inputPeriodLoadRule = OBJECT_MAPPER.readValue(inputJson, PeriodLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS), inputPeriodLoadRule.getTieredReplicants());
+  }
+
+  @Test
+  public void testUseDefaultTierAsFalseShouldCreateEmptyMap() throws Exception
+  {
+    String inputJson = "    {\n"
+                       + "     \"period\": \"P1D\",\n"
+                       + "     \"includeFuture\": " + PeriodLoadRule.DEFAULT_INCLUDE_FUTURE + ",\n"
+                       + "     \"useDefaultTierForNull\": \"false\",\n"
+                       + "     \"type\": \"loadByPeriod\"\n"
+                       + "  }";
+    PeriodLoadRule inputPeriodLoadRule = OBJECT_MAPPER.readValue(inputJson, PeriodLoadRule.class);
+    Assert.assertEquals(ImmutableMap.of(), inputPeriodLoadRule.getTieredReplicants());
+  }
+
+  @Test
+  public void testEquals()
+  {
+    EqualsVerifier.forClass(PeriodLoadRule.class)
+                  .withNonnullFields("tieredReplicants")
+                  .usingGetClass()
+                  .verify();
+  }
 }

server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java

@@ -276,7 +276,7 @@ public abstract class CoordinatorSimulationBaseTest implements
 
     Rule forever()
     {
-      return new ForeverLoadRule(tieredReplicants);
+      return new ForeverLoadRule(tieredReplicants, null);
     }
   }
 

server/src/test/java/org/apache/druid/server/coordinator/simulate/TestMetadataRuleManager.java

@@ -40,7 +40,7 @@ public class TestMetadataRuleManager implements MetadataRuleManager
   {
     rules.put(
         DEFAULT_DATASOURCE,
-        Collections.singletonList(new ForeverLoadRule(null))
+        Collections.singletonList(new ForeverLoadRule(null, null))
     );
   }
 

server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java

@@ -631,7 +631,7 @@ public class DataSourcesResourceTest
   public void testIsHandOffComplete()
   {
     MetadataRuleManager databaseRuleManager = EasyMock.createMock(MetadataRuleManager.class);
-    Rule loadRule = new IntervalLoadRule(Intervals.of("2013-01-02T00:00:00Z/2013-01-03T00:00:00Z"), null);
+    Rule loadRule = new IntervalLoadRule(Intervals.of("2013-01-02T00:00:00Z/2013-01-03T00:00:00Z"), null, null);
     Rule dropRule = new IntervalDropRule(Intervals.of("2013-01-01T00:00:00Z/2013-01-02T00:00:00Z"));
     DataSourcesResource dataSourcesResource =
         new DataSourcesResource(inventoryView, null, databaseRuleManager, null, null, null);

services/src/test/java/org/apache/druid/server/router/CoordinatorRuleManagerTest.java

@@ -49,7 +49,7 @@ public class CoordinatorRuleManagerTest
   private static final String DATASOURCE1 = "datasource1";
   private static final String DATASOURCE2 = "datasource2";
   private static final List<Rule> DEFAULT_RULES = ImmutableList.of(
-      new ForeverLoadRule(ImmutableMap.of("__default", 2))
+      new ForeverLoadRule(ImmutableMap.of("__default", 2), null)
   );
 
   @org.junit.Rule
@@ -109,7 +109,7 @@ public class CoordinatorRuleManagerTest
     manager.poll();
     final List<Rule> rules = manager.getRulesWithDefault(DATASOURCE2);
     final List<Rule> expectedRules = new ArrayList<>();
-    expectedRules.add(new ForeverLoadRule(null));
+    expectedRules.add(new ForeverLoadRule(null, null));
     expectedRules.add(new IntervalDropRule(Intervals.of("2020-01-01/2020-01-02")));
     expectedRules.addAll(DEFAULT_RULES);
     Assert.assertEquals(expectedRules, rules);
@@ -119,16 +119,16 @@ public class CoordinatorRuleManagerTest
   {
     final Map<String, List<Rule>> rules = ImmutableMap.of(
         DATASOURCE1,
-        ImmutableList.of(new ForeverLoadRule(null)),
+        ImmutableList.of(new ForeverLoadRule(null, null)),
         DATASOURCE2,
-        ImmutableList.of(new ForeverLoadRule(null), new IntervalDropRule(Intervals.of("2020-01-01/2020-01-02"))),
+        ImmutableList.of(new ForeverLoadRule(null, null), new IntervalDropRule(Intervals.of("2020-01-01/2020-01-02"))),
         "datasource3",
         ImmutableList.of(
-            new PeriodLoadRule(new Period("P1M"), true, null),
+            new PeriodLoadRule(new Period("P1M"), true, null, null),
             new ForeverDropRule()
         ),
         TieredBrokerConfig.DEFAULT_RULE_NAME,
-        ImmutableList.of(new ForeverLoadRule(ImmutableMap.of("__default", 2)))
+        ImmutableList.of(new ForeverLoadRule(ImmutableMap.of("__default", 2), null))
     );
     final StringFullResponseHolder holder = EasyMock.niceMock(StringFullResponseHolder.class);
     EasyMock.expect(holder.getStatus())

services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java

@@ -418,11 +418,12 @@ public class TieredBrokerHostSelectorTest
     public List<Rule> getRulesWithDefault(String dataSource)
     {
       return Arrays.asList(
-          new IntervalLoadRule(Intervals.of("2013/2014"), ImmutableMap.of("hot", 1)),
+          new IntervalLoadRule(Intervals.of("2013/2014"), ImmutableMap.of("hot", 1), null),
-          new IntervalLoadRule(Intervals.of("2012/2013"), ImmutableMap.of("medium", 1)),
+          new IntervalLoadRule(Intervals.of("2012/2013"), ImmutableMap.of("medium", 1), null),
           new IntervalLoadRule(
               Intervals.of("2011/2012"),
-              ImmutableMap.of(DruidServer.DEFAULT_TIER, 1)
+              ImmutableMap.of(DruidServer.DEFAULT_TIER, 1),
+              null
           )
       );
     }