Priority on loading for primary replica

server/src/main/java/io/druid/server/coordinator/DruidCluster.java

@@ -20,19 +20,20 @@
 package io.druid.server.coordinator;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.MinMaxPriorityQueue;
-import com.google.common.collect.Ordering;
 import io.druid.client.ImmutableDruidServer;
 import io.druid.java.util.common.IAE;
 
 import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.NavigableSet;
 import java.util.Set;
+import java.util.TreeSet;
 
 /**
  * Contains a representation of the current state of the cluster by tier.
@@ -41,7 +42,7 @@
 public class DruidCluster
 {
   private final Set<ServerHolder> realtimes;
-  private final Map<String, MinMaxPriorityQueue<ServerHolder>> historicals;
+  private final Map<String, NavigableSet<ServerHolder>> historicals;
 
   public DruidCluster()
   {
@@ -52,7 +53,7 @@ public DruidCluster()
   @VisibleForTesting
   public DruidCluster(
       @Nullable Set<ServerHolder> realtimes,
-      Map<String, MinMaxPriorityQueue<ServerHolder>> historicals
+      Map<String, NavigableSet<ServerHolder>> historicals
   )
   {
     this.realtimes = realtimes == null ? new HashSet<>() : new HashSet<>(realtimes);
@@ -86,9 +87,9 @@ private void addRealtime(ServerHolder serverHolder)
   private void addHistorical(ServerHolder serverHolder)
   {
     final ImmutableDruidServer server = serverHolder.getServer();
-    final MinMaxPriorityQueue<ServerHolder> tierServers = historicals.computeIfAbsent(
+    final NavigableSet<ServerHolder> tierServers = historicals.computeIfAbsent(
         server.getTier(),
-        k -> MinMaxPriorityQueue.orderedBy(Ordering.natural().reverse()).create()
+        k -> new TreeSet<>(Collections.reverseOrder())
     );
     tierServers.add(serverHolder);
   }
@@ -98,7 +99,7 @@ public Set<ServerHolder> getRealtimes()
     return realtimes;
   }
 
-  public Map<String, MinMaxPriorityQueue<ServerHolder>> getHistoricals()
+  public Map<String, NavigableSet<ServerHolder>> getHistoricals()
   {
     return historicals;
   }
@@ -108,7 +109,7 @@ public Iterable<String> getTierNames()
     return historicals.keySet();
   }
 
-  public MinMaxPriorityQueue<ServerHolder> getHistoricalsByTier(String tier)
+  public NavigableSet<ServerHolder> getHistoricalsByTier(String tier)
   {
     return historicals.get(tier);
   }
@@ -124,7 +125,7 @@ public Collection<ServerHolder> getAllServers()
     return allServers;
   }
 
-  public Iterable<MinMaxPriorityQueue<ServerHolder>> getSortedHistoricalsByTier()
+  public Iterable<NavigableSet<ServerHolder>> getSortedHistoricalsByTier()
   {
     return historicals.values();
   }
@@ -146,7 +147,7 @@ public boolean hasRealtimes()
 
   public boolean hasTier(String tier)
   {
-    MinMaxPriorityQueue<ServerHolder> servers = historicals.get(tier);
-    return (servers == null) || servers.isEmpty();
+    NavigableSet<ServerHolder> servers = historicals.get(tier);
+    return (servers != null) && !servers.isEmpty();
   }
 }

server/src/main/java/io/druid/server/coordinator/SegmentReplicantLookup.java

@@ -21,12 +21,12 @@
 
 import com.google.common.collect.HashBasedTable;
 import com.google.common.collect.Maps;
-import com.google.common.collect.MinMaxPriorityQueue;
 import com.google.common.collect.Table;
 import io.druid.client.ImmutableDruidServer;
 import io.druid.timeline.DataSegment;
 
 import java.util.Map;
+import java.util.SortedSet;
 
 /**
  * A lookup for the number of replicants of a given segment for a certain tier.
@@ -38,7 +38,7 @@ public static SegmentReplicantLookup make(DruidCluster cluster)
     final Table<String, String, Integer> segmentsInCluster = HashBasedTable.create();
     final Table<String, String, Integer> loadingSegments = HashBasedTable.create();
 
-    for (MinMaxPriorityQueue<ServerHolder> serversByType : cluster.getSortedHistoricalsByTier()) {
+    for (SortedSet<ServerHolder> serversByType : cluster.getSortedHistoricalsByTier()) {
       for (ServerHolder serverHolder : serversByType) {
         ImmutableDruidServer server = serverHolder.getServer();
 

server/src/main/java/io/druid/server/coordinator/ServerHolder.java

@@ -19,6 +19,7 @@
 
 package io.druid.server.coordinator;
 
+import com.google.common.primitives.Longs;
 import io.druid.client.ImmutableDruidServer;
 import io.druid.java.util.common.logger.Logger;
 import io.druid.timeline.DataSegment;
@@ -52,32 +53,32 @@ public LoadQueuePeon getPeon()
     return peon;
   }
 
-  public Long getMaxSize()
+  public long getMaxSize()
   {
     return server.getMaxSize();
   }
 
-  public Long getCurrServerSize()
+  public long getCurrServerSize()
   {
     return server.getCurrSize();
   }
 
-  public Long getLoadQueueSize()
+  public long getLoadQueueSize()
   {
     return peon.getLoadQueueSize();
   }
 
-  public Long getSizeUsed()
+  public long getSizeUsed()
   {
     return getCurrServerSize() + getLoadQueueSize();
   }
 
-  public Double getPercentUsed()
+  public double getPercentUsed()
   {
-    return (100 * getSizeUsed().doubleValue()) / getMaxSize();
+    return (100.0 * getSizeUsed()) / getMaxSize();
   }
 
-  public Long getAvailableSize()
+  public long getAvailableSize()
   {
     long maxSize = getMaxSize();
     long sizeUsed = getSizeUsed();
@@ -114,7 +115,22 @@ public int getNumberOfSegmentsInQueue()
   @Override
   public int compareTo(ServerHolder serverHolder)
   {
-    return getAvailableSize().compareTo(serverHolder.getAvailableSize());
+    int result = Longs.compare(getAvailableSize(), serverHolder.getAvailableSize());
+    if (result != 0) {
+      return result;
+    }
+
+    result = server.getHost().compareTo(serverHolder.server.getHost());
+    if (result != 0) {
+      return result;
+    }
+
+    result = server.getTier().compareTo(serverHolder.server.getTier());
+    if (result != 0) {
+      return result;
+    }
+
+    return server.getType().compareTo(serverHolder.server.getType());
   }
 
   @Override

server/src/main/java/io/druid/server/coordinator/helper/DruidCoordinatorBalancer.java

@@ -20,7 +20,6 @@
 package io.druid.server.coordinator.helper;
 
 import com.google.common.collect.Lists;
-import com.google.common.collect.MinMaxPriorityQueue;
 import com.metamx.emitter.EmittingLogger;
 import io.druid.client.ImmutableDruidServer;
 import io.druid.java.util.common.StringUtils;
@@ -38,6 +37,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.NavigableSet;
+import java.util.SortedSet;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
@@ -78,7 +79,7 @@ protected void reduceLifetimes(String tier)
   public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
   {
     final CoordinatorStats stats = new CoordinatorStats();
-    params.getDruidCluster().getHistoricals().forEach((String tier, MinMaxPriorityQueue<ServerHolder> servers) -> {
+    params.getDruidCluster().getHistoricals().forEach((String tier, NavigableSet<ServerHolder> servers) -> {
       balanceTier(params, tier, servers, stats);
     });
     return params.buildFromExisting().withCoordinatorStats(stats).build();
@@ -87,7 +88,7 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
   private void balanceTier(
       DruidCoordinatorRuntimeParams params,
       String tier,
-      MinMaxPriorityQueue<ServerHolder> servers,
+      SortedSet<ServerHolder> servers,
       CoordinatorStats stats
   )
   {

server/src/main/java/io/druid/server/coordinator/helper/DruidCoordinatorCleanupOvershadowed.java

@@ -20,8 +20,6 @@
 package io.druid.server.coordinator.helper;
 
 import com.google.common.collect.Maps;
-import com.google.common.collect.MinMaxPriorityQueue;
-
 import io.druid.client.ImmutableDruidDataSource;
 import io.druid.client.ImmutableDruidServer;
 import io.druid.java.util.common.guava.Comparators;
@@ -34,6 +32,7 @@
 import io.druid.timeline.VersionedIntervalTimeline;
 
 import java.util.Map;
+import java.util.SortedSet;
 
 public class DruidCoordinatorCleanupOvershadowed implements DruidCoordinatorHelper
 {
@@ -55,7 +54,7 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
       DruidCluster cluster = params.getDruidCluster();
       Map<String, VersionedIntervalTimeline<String, DataSegment>> timelines = Maps.newHashMap();
 
-      for (MinMaxPriorityQueue<ServerHolder> serverHolders : cluster.getSortedHistoricalsByTier()) {
+      for (SortedSet<ServerHolder> serverHolders : cluster.getSortedHistoricalsByTier()) {
         for (ServerHolder serverHolder : serverHolders) {
           ImmutableDruidServer server = serverHolder.getServer();
 

server/src/main/java/io/druid/server/coordinator/helper/DruidCoordinatorCleanupUnneeded.java

@@ -19,8 +19,6 @@
 
 package io.druid.server.coordinator.helper;
 
-import com.google.common.collect.MinMaxPriorityQueue;
-
 import io.druid.client.ImmutableDruidDataSource;
 import io.druid.client.ImmutableDruidServer;
 import io.druid.java.util.common.logger.Logger;
@@ -34,6 +32,7 @@
 import io.druid.timeline.DataSegment;
 
 import java.util.Set;
+import java.util.SortedSet;
 
 /**
  */
@@ -64,7 +63,7 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     // This is done to prevent a race condition in which the coordinator would drop all segments if it started running
     // cleanup before it finished polling the metadata storage for available segments for the first time.
     if (!availableSegments.isEmpty()) {
-      for (MinMaxPriorityQueue<ServerHolder> serverHolders : cluster.getSortedHistoricalsByTier()) {
+      for (SortedSet<ServerHolder> serverHolders : cluster.getSortedHistoricalsByTier()) {
         for (ServerHolder serverHolder : serverHolders) {
           ImmutableDruidServer server = serverHolder.getServer();
 

server/src/main/java/io/druid/server/coordinator/helper/DruidCoordinatorLogger.java

@@ -19,7 +19,6 @@
 
 package io.druid.server.coordinator.helper;
 
-import com.google.common.collect.MinMaxPriorityQueue;
 import com.metamx.emitter.service.ServiceEmitter;
 import com.metamx.emitter.service.ServiceMetricEvent;
 import io.druid.client.DruidDataSource;
@@ -171,7 +170,7 @@ public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
     );
 
     log.info("Load Queues:");
-    for (MinMaxPriorityQueue<ServerHolder> serverHolders : cluster.getSortedHistoricalsByTier()) {
+    for (Iterable<ServerHolder> serverHolders : cluster.getSortedHistoricalsByTier()) {
       for (ServerHolder serverHolder : serverHolders) {
         ImmutableDruidServer server = serverHolder.getServer();
         LoadQueuePeon queuePeon = serverHolder.getPeon();