Allow empty tiered replicants map for load rules

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();

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);
-  }
 }

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);
   }
 }

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

@@ -19,39 +19,103 @@
 
 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;
+  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);
+  }
 }

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);
   }
 }