diff --git a/docs/operations/rule-configuration.md b/docs/operations/rule-configuration.md
index 0d75cf54e89a..0bf803355aad 100644
--- a/docs/operations/rule-configuration.md
+++ b/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
 
diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataRuleManager.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataRuleManager.java
index b3d0fa9b27f1..b7cdb2f7ec91 100644
--- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataRuleManager.java
+++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataRuleManager.java
@@ -96,7 +96,8 @@ public Void withHandle(Handle handle) throws Exception
                       ImmutableMap.of(
                           DruidServer.DEFAULT_TIER,
                           DruidServer.DEFAULT_NUM_REPLICANTS
-                      )
+                      ),
+                      null
                   )
               );
               final String version = DateTimes.nowUtc().toString();
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/rules/ForeverLoadRule.java b/server/src/main/java/org/apache/druid/server/coordinator/rules/ForeverLoadRule.java
index dc1798f13e03..35f22fa555f8 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/rules/ForeverLoadRule.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/rules/ForeverLoadRule.java
@@ -21,30 +21,24 @@
 
 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
-                            ? ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS)
-                            : tieredReplicants;
-    validateTieredReplicants(this.tieredReplicants);
+    super(tieredReplicants, useDefaultTierForNull);
   }
 
   @Override
@@ -54,20 +48,6 @@ public String getType()
     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 boolean appliesTo(Interval interval, DateTime referenceTimestamp)
     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);
-  }
 }
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/rules/IntervalLoadRule.java b/server/src/main/java/org/apache/druid/server/coordinator/rules/IntervalLoadRule.java
index 2e944bf28544..209f5c24d1e4 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/rules/IntervalLoadRule.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/rules/IntervalLoadRule.java
@@ -21,14 +21,14 @@
 
 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;
-    validateTieredReplicants(this.tieredReplicants);
+    super(tieredReplicants, useDefaultTierForNull);
     this.interval = interval;
   }
 
@@ -57,20 +56,6 @@ public String getType()
     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 boolean equals(Object o)
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-
-    IntervalLoadRule that = (IntervalLoadRule) o;
-
-    if (interval != null ? !interval.equals(that.interval) : that.interval != null) {
-      return false;
-    }
-    if (tieredReplicants != null ? !tieredReplicants.equals(that.tieredReplicants) : that.tieredReplicants != null) {
+    if (!super.equals(o)) {
       return false;
     }
-
-    return true;
+    IntervalLoadRule that = (IntervalLoadRule) o;
+    return Objects.equals(interval, that.interval);
   }
 
   @Override
   public int hashCode()
   {
-    int result = interval != null ? interval.hashCode() : 0;
-    result = 31 * result + (tieredReplicants != null ? tieredReplicants.hashCode() : 0);
-    return result;
+    return Objects.hash(super.hashCode(), interval);
   }
 }
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/rules/LoadRule.java b/server/src/main/java/org/apache/druid/server/coordinator/rules/LoadRule.java
index 548c25cad546..5d7b724c8451 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/rules/LoadRule.java
+++ b/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) {
-      throw new IAE("A rule with empty tiered replicants is invalid");
+    if (useDefaultTierForNull) {
+      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);
+  }
 }
diff --git a/server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodLoadRule.java b/server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodLoadRule.java
index 0d6e2e099a74..1d2b4e187716 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodLoadRule.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/rules/PeriodLoadRule.java
@@ -21,15 +21,15 @@
 
 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;
-    validateTieredReplicants(this.tieredReplicants);
+    super(tieredReplicants, useDefaultTierForNull);
     this.period = period;
     this.includeFuture = includeFuture == null ? DEFAULT_INCLUDE_FUTURE : includeFuture;
   }
@@ -75,28 +74,36 @@ public boolean isIncludeFuture()
   }
 
   @Override
-  @JsonProperty
-  public Map<String, Integer> getTieredReplicants()
+  public boolean appliesTo(DataSegment segment, DateTime referenceTimestamp)
   {
-    return tieredReplicants;
+    return appliesTo(segment.getInterval(), referenceTimestamp);
   }
 
   @Override
-  public int getNumReplicants(String tier)
+  public boolean appliesTo(Interval interval, DateTime referenceTimestamp)
   {
-    final Integer retVal = tieredReplicants.get(tier);
-    return retVal == null ? 0 : retVal;
+    return Rules.eligibleForLoad(period, interval, referenceTimestamp, includeFuture);
   }
 
   @Override
-  public boolean appliesTo(DataSegment segment, DateTime referenceTimestamp)
+  public boolean equals(Object o)
   {
-    return appliesTo(segment.getInterval(), referenceTimestamp);
+    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 boolean appliesTo(Interval interval, DateTime referenceTimestamp)
+  public int hashCode()
   {
-    return Rules.eligibleForLoad(period, interval, referenceTimestamp, includeFuture);
+    return Objects.hash(super.hashCode(), period, includeFuture);
   }
 }
diff --git a/server/src/test/java/org/apache/druid/metadata/SQLMetadataRuleManagerTest.java b/server/src/test/java/org/apache/druid/metadata/SQLMetadataRuleManagerTest.java
index 61bc5d9080db..20ffdb81188a 100644
--- a/server/src/test/java/org/apache/druid/metadata/SQLMetadataRuleManagerTest.java
+++ b/server/src/test/java/org/apache/druid/metadata/SQLMetadataRuleManagerTest.java
@@ -108,7 +108,8 @@ public void testRuleInsert()
     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 void testAuditEntryCreated() throws Exception
     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");
@@ -200,9 +202,10 @@ public void testFetchAuditEntriesForAllDataSources() throws Exception
     List<Rule> rules = Collections.singletonList(
         new IntervalLoadRule(
             Intervals.of("2015-01-01/2015-02-01"), ImmutableMap.of(
-            DruidServer.DEFAULT_TIER,
-            DruidServer.DEFAULT_NUM_REPLICANTS
-        )
+                DruidServer.DEFAULT_TIER,
+                DruidServer.DEFAULT_NUM_REPLICANTS
+        ),
+            null
         )
     );
     final AuditInfo auditInfo = createAuditInfo("test_comment");
@@ -232,7 +235,8 @@ public void testRemoveRulesOlderThanWithNonExistenceDatasourceAndOlderThanTimest
     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 void testRemoveRulesOlderThanWithNonExistenceDatasourceAndNewerThanTimest
     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 void testRemoveRulesOlderThanWithActiveDatasourceShouldNotDelete() throws
     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"));
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsProfiler.java b/server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsProfiler.java
index b117b96520fa..4a45e16bf96b 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/BalanceSegmentsProfiler.java
+++ b/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
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 6c0ab813b371..15fc5f5ac919 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
@@ -207,7 +207,7 @@ public void testCoordinatorRun() throws Exception
     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 void testCoordinatorRun() throws Exception
   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 coldTier = new ForeverLoadRule(ImmutableMap.of(coldTierName, 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), 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);
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java
index abc218d89460..cd9c1e228da1 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/RunRulesTest.java
@@ -129,7 +129,8 @@ public void testOneTierTwoReplicantsWithStrictReplicantLimit()
         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 void testTwoTiersTwoReplicantsWithStrictReplicantLimit()
         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 void testRunThreeTiersOneReplicant()
         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 void testRunTwoTiersTwoReplicants()
         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 void testRunTwoTiersWithExistingSegments()
         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 void testRunTwoTiersTierDoesNotExist()
         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 void testRunRuleDoesNotExist()
             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 void testDropRemove()
         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 void testDropTooManyInSameTier()
         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 void testDropTooManyInDifferentTiers()
         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 void testDontDropInDifferentTiers()
         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 void testDropServerActuallyServesSegment()
             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 void testNoThrottleWhenSegmentNotLoadedInTier()
             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 void testReplicantThrottleAcrossTiers()
             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 void testDropReplicantThrottle()
             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 void testRulesRunOnNonOvershadowedSegmentsOnly()
     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 void testTwoNodesOneTierThreeReplicantsRandomStrategyNotEnoughNodes()
 
     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 void testOneNodesOneTierOneReplicantRandomStrategyEnoughSpace()
 
     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 void testOneNodesOneTierOneReplicantRandomStrategyNotEnoughSpace()
     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 void testOneNodesOneTierOneReplicantCostBalancerStrategyNotEnoughSpace()
     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);
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/UnloadUnusedSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/UnloadUnusedSegmentsTest.java
index 8faa4d361d01..83aa11fb303e 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/duty/UnloadUnusedSegmentsTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/UnloadUnusedSegmentsTest.java
@@ -316,7 +316,8 @@ private static void mockRuleManager(MetadataRuleManager metadataRuleManager)
                 ImmutableMap.of(
                     DruidServer.DEFAULT_TIER, 1,
                     "tier2", 1
-                )
+                ),
+                null
             )
         )).anyTimes();
 
@@ -326,7 +327,8 @@ private static void mockRuleManager(MetadataRuleManager metadataRuleManager)
                 ImmutableMap.of(
                     DruidServer.DEFAULT_TIER, 1,
                     "tier2", 1
-                )
+                ),
+                null
             )
         )).anyTimes();
 
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/rules/ForeverLoadRuleTest.java b/server/src/test/java/org/apache/druid/server/coordinator/rules/ForeverLoadRuleTest.java
index c9be7f63ddbe..c1497afe899b 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/rules/ForeverLoadRuleTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/rules/ForeverLoadRuleTest.java
@@ -22,8 +22,10 @@
 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 @@
 
 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(
-        null
-    );
+    ForeverLoadRule rule = new ForeverLoadRule(null, null);
 
     ObjectMapper jsonMapper = new DefaultObjectMapper();
     Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
@@ -47,45 +49,79 @@ public void testSerdeNullTieredReplicants() throws Exception
   }
 
   @Test
-  public void testMappingNullTieredReplicants() throws Exception
+  public void testCreatingNegativeTieredReplicants()
   {
-    String inputJson = "{\n"
-                      + " \"type\": \"loadForever\"\n"
-                      + "}";
-    String expectedJson = "    {\n"
-                          + "      \"tieredReplicants\": {\n"
-                          + "        \"" + DruidServer.DEFAULT_TIER + "\": " + DruidServer.DEFAULT_NUM_REPLICANTS + "\n"
-                          + "      },\n"
-                          + "      \"type\": \"loadForever\"\n"
-                          + "    }";
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
-    ForeverLoadRule inputForeverLoadRule = jsonMapper.readValue(inputJson, ForeverLoadRule.class);
-    ForeverLoadRule expectedForeverLoadRule = jsonMapper.readValue(expectedJson, ForeverLoadRule.class);
-    Assert.assertEquals(expectedForeverLoadRule.getTieredReplicants(), inputForeverLoadRule.getTieredReplicants());
+    MatcherAssert.assertThat(
+        Assert.assertThrows(DruidException.class, () ->
+            new ForeverLoadRule(
+                ImmutableMap.of(DruidServer.DEFAULT_TIER, -1),
+                null
+            )
+        ),
+        DruidExceptionMatcher.invalidInput().expectMessageContains(
+            "Invalid number of replicas for tier [_default_tier]. Value [-1] must be positive."
+        )
+    );
   }
 
-  @Test(expected = IAE.class)
+  @Test
   public void testEmptyTieredReplicants() throws Exception
   {
-    ForeverLoadRule rule = new ForeverLoadRule(
-        ImmutableMap.of()
-    );
+    ForeverLoadRule rule = new ForeverLoadRule(ImmutableMap.of(), false);
 
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
-    Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
+    LoadRule reread = (LoadRule) OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
+    Assert.assertEquals(ImmutableMap.of(), reread.getTieredReplicants());
   }
 
-  @Test(expected = IAE.class)
-  public void testEmptyReplicantValue() throws Exception
+  @Test
+  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
+
+    MatcherAssert.assertThat(
+        Assert.assertThrows(DruidException.class, () ->
+            new ForeverLoadRule(
+                tieredReplicants,
+                true
+            )
+        ),
+        DruidExceptionMatcher.invalidInput().expectMessageContains(
+            "Invalid number of replicas for tier [tier]. Value must not be null."
+        )
     );
+  }
 
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
-    Rule reread = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
+  @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());
   }
 }
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/rules/IntervalLoadRuleTest.java b/server/src/test/java/org/apache/druid/server/coordinator/rules/IntervalLoadRuleTest.java
index 52e892dd8cff..5eae342d2eda 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/rules/IntervalLoadRuleTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/rules/IntervalLoadRuleTest.java
@@ -22,25 +22,34 @@
 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 = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
+    Rule reread = OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
 
     Assert.assertEquals(rule, reread);
   }
@@ -49,36 +58,85 @@ public void testSerde() throws Exception
   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 = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
+    Rule reread = OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
 
     Assert.assertEquals(rule, reread);
-    Assert.assertEquals(
-        ImmutableMap.of(DruidServer.DEFAULT_TIER, DruidServer.DEFAULT_NUM_REPLICANTS),
-        rule.getTieredReplicants()
+    Assert.assertEquals(ImmutableMap.of(), 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 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 testMappingNullTieredReplicants() throws Exception
+  public void testShouldCreateDefaultTier() throws Exception
   {
-    String inputJson = "    {\n"
+    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"
-                          + "      \"interval\": \"0000-01-01T00:00:00.000-05:50:36/3000-01-01T00:00:00.000-06:00\",\n"
-                          + "      \"tieredReplicants\": {\n"
-                          + "        \"" + DruidServer.DEFAULT_TIER + "\": " + DruidServer.DEFAULT_NUM_REPLICANTS + "\n"
-                          + "      },\n"
-                          + "      \"type\": \"loadByInterval\"\n"
-                          + "    }";
-    ObjectMapper jsonMapper = new DefaultObjectMapper();
-    IntervalLoadRule inputIntervalLoadRule = jsonMapper.readValue(inputJson, IntervalLoadRule.class);
-    IntervalLoadRule expectedIntervalLoadRule = jsonMapper.readValue(expectedJson, IntervalLoadRule.class);
-    Assert.assertEquals(expectedIntervalLoadRule, inputIntervalLoadRule);
+                       + "   }";
+    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"
+                       + "      \"type\": \"loadByInterval\",\n"
+                       + "      \"useDefaultTierForNull\": \"true\"\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 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());
   }
 }
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/rules/LoadRuleTest.java b/server/src/test/java/org/apache/druid/server/coordinator/rules/LoadRuleTest.java
index 85aeeb77db15..668013f7ff34 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/rules/LoadRuleTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/rules/LoadRuleTest.java
@@ -23,6 +23,7 @@
 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 @@ private DataSegment createDataSegment(String dataSource)
 
   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 @@ private static class Tier
     static final String T1 = "tier1";
     static final String T2 = "tier2";
   }
+
+  @Test
+  public void testEquals()
+  {
+    EqualsVerifier.forClass(LoadRule.class)
+                  .withNonnullFields("tieredReplicants")
+                  .usingGetClass()
+                  .verify();
+  }
 }
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/rules/PeriodLoadRuleTest.java b/server/src/test/java/org/apache/druid/server/coordinator/rules/PeriodLoadRuleTest.java
index 84b72614000f..7b1dd2085f02 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/rules/PeriodLoadRuleTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/rules/PeriodLoadRuleTest.java
@@ -21,22 +21,31 @@
 
 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 void testAppliesToAll()
     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 void testAppliesToPeriod()
     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));
@@ -80,7 +91,7 @@ public void testAppliesToPeriod()
     Assert.assertFalse(
         rule.appliesTo(
             BUILDER.interval(new Interval(now.plusDays(1), now.plusDays(2)))
-                       .build(),
+                   .build(),
             now
         )
     );
@@ -91,24 +102,25 @@ public void testAppliesToPartialOverlap()
   {
     DateTime now = DateTimes.of("2012-12-31T01:00:00");
     PeriodLoadRule rule = new PeriodLoadRule(
-            new Period("P1M"),
-            false,
-            ImmutableMap.of("", 0)
+        new Period("P1M"),
+        false,
+        ImmutableMap.of("", 0),
+        null
     );
 
     Assert.assertTrue(
-            rule.appliesTo(
-                    BUILDER.interval(new Interval(now.minusWeeks(1), now.plusWeeks(1))).build(),
-                    now
-            )
+        rule.appliesTo(
+            BUILDER.interval(new Interval(now.minusWeeks(1), now.plusWeeks(1))).build(),
+            now
+        )
     );
     Assert.assertTrue(
-            rule.appliesTo(
-                    BUILDER.interval(
-                            new Interval(now.minusMonths(1).minusWeeks(1), now.minusMonths(1).plusWeeks(1))
-                    ).build(),
-                    now
-            )
+        rule.appliesTo(
+            BUILDER.interval(
+                new Interval(now.minusMonths(1).minusWeeks(1), now.minusMonths(1).plusWeeks(1))
+            ).build(),
+            now
+        )
     );
   }
 
@@ -119,12 +131,14 @@ public void testIncludeFuture()
     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 void testIncludeFuture()
   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 = jsonMapper.readValue(jsonMapper.writeValueAsString(rule), Rule.class);
+    Rule reread = OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(rule), Rule.class);
 
     Assert.assertEquals(rule.getPeriod(), ((PeriodLoadRule) reread).getPeriod());
     Assert.assertEquals(rule.isIncludeFuture(), ((PeriodLoadRule) reread).isIncludeFuture());
@@ -174,16 +187,100 @@ public void testMappingNull() throws Exception
     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 = jsonMapper.readValue(inputJson, PeriodLoadRule.class);
-    PeriodLoadRule expectedPeriodLoadRule = jsonMapper.readValue(expectedJson, PeriodLoadRule.class);
+    PeriodLoadRule inputPeriodLoadRule = OBJECT_MAPPER.readValue(inputJson, PeriodLoadRule.class);
+    PeriodLoadRule expectedPeriodLoadRule = OBJECT_MAPPER.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();
+  }
 }
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java
index 99d4d9d0a837..e64ab0bcc46c 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java
@@ -276,7 +276,7 @@ Load andOn(String tier, int numReplicas)
 
     Rule forever()
     {
-      return new ForeverLoadRule(tieredReplicants);
+      return new ForeverLoadRule(tieredReplicants, null);
     }
   }
 
diff --git a/server/src/test/java/org/apache/druid/server/coordinator/simulate/TestMetadataRuleManager.java b/server/src/test/java/org/apache/druid/server/coordinator/simulate/TestMetadataRuleManager.java
index 9ca037b0cfbf..9bb877112cf6 100644
--- a/server/src/test/java/org/apache/druid/server/coordinator/simulate/TestMetadataRuleManager.java
+++ b/server/src/test/java/org/apache/druid/server/coordinator/simulate/TestMetadataRuleManager.java
@@ -40,7 +40,7 @@ public TestMetadataRuleManager()
   {
     rules.put(
         DEFAULT_DATASOURCE,
-        Collections.singletonList(new ForeverLoadRule(null))
+        Collections.singletonList(new ForeverLoadRule(null, null))
     );
   }
 
diff --git a/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java b/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
index 69200382f357..4e59e913e23b 100644
--- a/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
+++ b/server/src/test/java/org/apache/druid/server/http/DataSourcesResourceTest.java
@@ -631,7 +631,7 @@ public void testMarkAsUnusedAllSegmentsInDataSource()
   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);
diff --git a/services/src/test/java/org/apache/druid/server/router/CoordinatorRuleManagerTest.java b/services/src/test/java/org/apache/druid/server/router/CoordinatorRuleManagerTest.java
index 89741c0c706b..553b6ef7b278 100644
--- a/services/src/test/java/org/apache/druid/server/router/CoordinatorRuleManagerTest.java
+++ b/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 void testGetRulesWithKnownDatasourceReturningAllRulesWithDefaultRule()
     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 @@ private DruidLeaderClient mockClient()
   {
     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())
diff --git a/services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java b/services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java
index 670b61702e2f..8dbf03292360 100644
--- a/services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java
+++ b/services/src/test/java/org/apache/druid/server/router/TieredBrokerHostSelectorTest.java
@@ -418,11 +418,12 @@ public boolean isStarted()
     public List<Rule> getRulesWithDefault(String dataSource)
     {
       return Arrays.asList(
-          new IntervalLoadRule(Intervals.of("2013/2014"), ImmutableMap.of("hot", 1)),
-          new IntervalLoadRule(Intervals.of("2012/2013"), ImmutableMap.of("medium", 1)),
+          new IntervalLoadRule(Intervals.of("2013/2014"), ImmutableMap.of("hot", 1), null),
+          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
           )
       );
     }