From 08ad56b4290a629ff94480435466de1d1213444e Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Thu, 12 Jun 2025 15:41:02 +0800 Subject: [PATCH 01/12] Add PreferredTierSelectorStrategy --- docs/configuration/index.md | 2 + .../AbstractTierSelectorStrategy.java | 2 +- ...ConnectionCountServerSelectorStrategy.java | 5 +- .../PreferredTieSelectorStrategyConfig.java | 70 ++++++ .../PreferredTierSelectorStrategy.java | 111 +++++++++ .../RandomServerSelectorStrategy.java | 5 +- .../selector/ServerSelectorStrategy.java | 9 +- .../client/selector/TierSelectorStrategy.java | 3 +- .../selector/TierSelectorStrategyTest.java | 216 +++++++++++++++++- .../java/org/apache/druid/cli/CliBroker.java | 2 + 10 files changed, 406 insertions(+), 19 deletions(-) create mode 100644 server/src/main/java/org/apache/druid/client/selector/PreferredTieSelectorStrategyConfig.java create mode 100644 server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java diff --git a/docs/configuration/index.md b/docs/configuration/index.md index 4899518144fc..8f4a27f0be10 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -1791,6 +1791,8 @@ These Broker configurations can be defined in the `broker/runtime.properties` fi |`druid.broker.balancer.type`|`random`, `connectionCount`|Determines how the broker balances connections to Historical processes. `random` choose randomly, `connectionCount` picks the process with the fewest number of active connections to|`random`| |`druid.broker.select.tier`|`highestPriority`, `lowestPriority`, `custom`|If segments are cross-replicated across tiers in a cluster, you can tell the broker to prefer to select segments in a tier with a certain priority.|`highestPriority`| |`druid.broker.select.tier.custom.priorities`|An array of integer priorities, such as `[-1, 0, 1, 2]`|Select servers in tiers with a custom priority list.|The config only has effect if `druid.broker.select.tier` is set to `custom`. If `druid.broker.select.tier` is set to `custom` but this config is not specified, the effect is the same as `druid.broker.select.tier` set to `highestPriority`. Any of the integers in this config can be ignored if there's no corresponding tiers with such priorities. Tiers with priorities explicitly specified in this config always have higher priority than those not and those not specified fall back to use `highestPriority` strategy among themselves.| +|`druid.broker.select.tier.preferred.tier`| The preferred tier name. E.g., `_default_tier` | A non-empty value that specifies the preferred tier in which historical servers will be picked up for queries. If there're no enough historical servers from the preferred tier, servers from other tiers might be choosed. | null | +|`druid.broker.select.tier.preferred.priority`| `high`, `low` | Optional. If there're multiple candidates in a preferred tier, specifies the priority to pick up candiates. By default, the higher priority a historical, the higher chances it will be picked up. | `high` | ##### Query prioritization and laning diff --git a/server/src/main/java/org/apache/druid/client/selector/AbstractTierSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/AbstractTierSelectorStrategy.java index 59a611df30ff..1821245cbf30 100644 --- a/server/src/main/java/org/apache/druid/client/selector/AbstractTierSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/AbstractTierSelectorStrategy.java @@ -34,7 +34,7 @@ */ public abstract class AbstractTierSelectorStrategy implements TierSelectorStrategy { - private final ServerSelectorStrategy serverSelectorStrategy; + protected final ServerSelectorStrategy serverSelectorStrategy; public AbstractTierSelectorStrategy(ServerSelectorStrategy serverSelectorStrategy) { diff --git a/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java index 262b23701df8..598141ca09d5 100644 --- a/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java @@ -26,6 +26,7 @@ import org.apache.druid.timeline.DataSegment; import javax.annotation.Nullable; +import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.List; @@ -42,13 +43,13 @@ public class ConnectionCountServerSelectorStrategy implements ServerSelectorStra @Nullable @Override - public QueryableDruidServer pick(Set servers, DataSegment segment) + public QueryableDruidServer pick(Collection servers, DataSegment segment) { return Collections.min(servers, COMPARATOR); } @Override - public List pick(Set servers, DataSegment segment, int numServersToPick) + public List pick(Collection servers, DataSegment segment, int numServersToPick) { if (servers.size() <= numServersToPick) { return ImmutableList.copyOf(servers); diff --git a/server/src/main/java/org/apache/druid/client/selector/PreferredTieSelectorStrategyConfig.java b/server/src/main/java/org/apache/druid/client/selector/PreferredTieSelectorStrategyConfig.java new file mode 100644 index 000000000000..dfa44eb66b9a --- /dev/null +++ b/server/src/main/java/org/apache/druid/client/selector/PreferredTieSelectorStrategyConfig.java @@ -0,0 +1,70 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + + +package org.apache.druid.client.selector; + + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.base.Preconditions; + +public class PreferredTieSelectorStrategyConfig +{ + @JsonProperty + private String tier; + + /** + * Only two options: high or low + */ + @JsonProperty + private String priority; + + @JsonCreator + public PreferredTieSelectorStrategyConfig( + @JsonProperty("tier") String tier, + @JsonProperty("priority") String priority + ) + { + Preconditions.checkState(tier != null && !tier.isEmpty(), + "druid.broker.select.tier.preferred.tier can't be empty"); + this.tier = tier.trim(); + this.priority = priority; + } + + public String getTier() + { + return tier; + } + + public void setTier(String tier) + { + this.tier = tier; + } + + public String getPriority() + { + return priority; + } + + public void setPriority(String priority) + { + this.priority = priority; + } +} diff --git a/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java new file mode 100644 index 000000000000..d5cdb0c19925 --- /dev/null +++ b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.client.selector; + + +import com.fasterxml.jackson.annotation.JacksonInject; +import com.google.common.base.Preconditions; +import it.unimi.dsi.fastutil.ints.Int2ObjectRBTreeMap; +import org.apache.druid.client.QueryableDruidServer; +import org.apache.druid.java.util.common.IAE; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.query.Query; +import org.apache.druid.timeline.DataSegment; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Set; + +public class PreferredTierSelectorStrategy extends AbstractTierSelectorStrategy +{ + private static final Logger log = new Logger(PreferredTierSelectorStrategy.class); + + private final String perferredTier; + private final TierSelectorStrategy priortyStrategy; + + public PreferredTierSelectorStrategy( + @JacksonInject ServerSelectorStrategy serverSelectorStrategy, + @JacksonInject PreferredTieSelectorStrategyConfig config + ) + { + super(serverSelectorStrategy); + this.perferredTier = config.getTier(); + + if (config.getPriority() == null) { + this.priortyStrategy = new HighestPriorityTierSelectorStrategy(serverSelectorStrategy); + } else { + if ("highest".equalsIgnoreCase(config.getPriority())) { + this.priortyStrategy = new HighestPriorityTierSelectorStrategy(serverSelectorStrategy); + } else if ("lowest".equalsIgnoreCase(config.getPriority())) { + this.priortyStrategy = new LowestPriorityTierSelectorStrategy(serverSelectorStrategy); + } else { + throw new IAE("druid.broker.select.tier.preferred.priority must be either 'highest' or 'lowest'"); + } + } + } + + @Override + public Comparator getComparator() + { + return priortyStrategy.getComparator(); + } + + @Override + public List pick( + Query query, + Int2ObjectRBTreeMap> prioritizedServers, + DataSegment segment, + int numServersToPick + ) + { + if (log.isDebugEnabled()) { + log.debug( + "Picking [%d] servers from preferred tier [%s] for segment [%s] with priority [%s]", + numServersToPick, perferredTier, segment.getId(), this.priortyStrategy.getClass().getSimpleName() + ); + } + + List preferred = new ArrayList<>(numServersToPick); + List nonPreferred = new ArrayList<>(numServersToPick); + for (Set priorityServers : prioritizedServers.values()) { + for (QueryableDruidServer server : priorityServers) { + if (perferredTier.equals(server.getServer().getMetadata().getTier())) { + preferred.add(server); + if (preferred.size() == numServersToPick) { + return this.serverSelectorStrategy.pick(query, preferred, segment, numServersToPick); + } + } else { + // We have to iterate through all servers even the numbers of the non-preferred servers reach the limit + // This is because we don't know whether there're preferred servers left in the next priority set + nonPreferred.add(server); + } + } + } + + // Fill with non-preferred servers if we don't have enough preferred servers + int fillSize = numServersToPick - preferred.size(); + for (int i = 0; i < fillSize && i < nonPreferred.size(); i++) { + preferred.add(nonPreferred.get(i)); + } + + return this.serverSelectorStrategy.pick(query, preferred, segment, numServersToPick); + } +} diff --git a/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java index b916a0aed303..f4133b9f3727 100644 --- a/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java @@ -26,6 +26,7 @@ import org.apache.druid.timeline.DataSegment; import javax.annotation.Nullable; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Set; @@ -35,13 +36,13 @@ public class RandomServerSelectorStrategy implements ServerSelectorStrategy { @Nullable @Override - public QueryableDruidServer pick(Set servers, DataSegment segment) + public QueryableDruidServer pick(Collection servers, DataSegment segment) { return Iterators.get(servers.iterator(), ThreadLocalRandom.current().nextInt(servers.size())); } @Override - public List pick(Set servers, DataSegment segment, int numServersToPick) + public List pick(Collection servers, DataSegment segment, int numServersToPick) { if (servers.size() <= numServersToPick) { return ImmutableList.copyOf(servers); diff --git a/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java index e93e19e75060..43d81dd85144 100644 --- a/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java @@ -27,6 +27,7 @@ import org.apache.druid.timeline.DataSegment; import javax.annotation.Nullable; +import java.util.Collection; import java.util.List; import java.util.Set; @@ -38,12 +39,12 @@ public interface ServerSelectorStrategy { @Nullable - default QueryableDruidServer pick(@Nullable Query query, Set servers, DataSegment segment) + default QueryableDruidServer pick(@Nullable Query query, Collection servers, DataSegment segment) { return Iterables.getOnlyElement(pick(query, servers, segment, 1), null); } - default List pick(@Nullable Query query, Set servers, DataSegment segment, + default List pick(@Nullable Query query, Collection servers, DataSegment segment, int numServersToPick) { return pick(servers, segment, numServersToPick); @@ -51,13 +52,13 @@ default List pick(@Nullable Query query, Set servers, DataSegment segment) + default QueryableDruidServer pick(Collection servers, DataSegment segment) { return pick(null, servers, segment); } @Deprecated - default List pick(Set servers, DataSegment segment, int numServersToPick) + default List pick(Collection servers, DataSegment segment, int numServersToPick) { return pick(null, servers, segment, numServersToPick); } diff --git a/server/src/main/java/org/apache/druid/client/selector/TierSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/TierSelectorStrategy.java index 79d541dcd499..6eae7db4f343 100644 --- a/server/src/main/java/org/apache/druid/client/selector/TierSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/TierSelectorStrategy.java @@ -37,7 +37,8 @@ @JsonSubTypes(value = { @JsonSubTypes.Type(name = "highestPriority", value = HighestPriorityTierSelectorStrategy.class), @JsonSubTypes.Type(name = "lowestPriority", value = LowestPriorityTierSelectorStrategy.class), - @JsonSubTypes.Type(name = "custom", value = CustomTierSelectorStrategy.class) + @JsonSubTypes.Type(name = "custom", value = CustomTierSelectorStrategy.class), + @JsonSubTypes.Type(name = "preferred", value = PreferredTierSelectorStrategy.class), }) public interface TierSelectorStrategy { diff --git a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java index 335b0ef7c502..d6213977e009 100644 --- a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java +++ b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java @@ -37,6 +37,7 @@ import javax.annotation.Nullable; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -45,7 +46,7 @@ public class TierSelectorStrategyTest { - + @Test public void testHighestPriorityTierSelectorStrategyRealtime() { @@ -64,7 +65,7 @@ public void testHighestPriorityTierSelectorStrategyRealtime() highPriority, lowPriority ); } - + @Test public void testHighestPriorityTierSelectorStrategy() { @@ -247,7 +248,7 @@ private void testTierSelectorStrategy( Assert.assertEquals(expectedCandidates, serverSelector.getCandidates(-1, CloneQueryMode.EXCLUDECLONES)); Assert.assertEquals(expectedCandidates.subList(0, 2), serverSelector.getCandidates(2, CloneQueryMode.EXCLUDECLONES)); } - + @Test public void testServerSelectorStrategyDefaults() { @@ -260,17 +261,214 @@ public void testServerSelectorStrategyDefaults() servers.add(p0); RandomServerSelectorStrategy strategy = new RandomServerSelectorStrategy(); Assert.assertEquals(strategy.pick(servers, EasyMock.createMock(DataSegment.class)), p0); - Assert.assertEquals(strategy.pick(EasyMock.createMock(Query.class), servers, EasyMock.createMock(DataSegment.class)), p0); - ServerSelectorStrategy defaultDeprecatedServerSelectorStrategy = new ServerSelectorStrategy() { + Assert.assertEquals( + strategy.pick( + EasyMock.createMock(Query.class), + servers, + EasyMock.createMock(DataSegment.class) + ), p0 + ); + ServerSelectorStrategy defaultDeprecatedServerSelectorStrategy = new ServerSelectorStrategy() + { @Override - public List pick(@Nullable Query query, Set servers, DataSegment segment, - int numServersToPick) + public List pick( + @Nullable Query query, Collection servers, DataSegment segment, + int numServersToPick + ) { return strategy.pick(servers, segment, numServersToPick); } }; - Assert.assertEquals(defaultDeprecatedServerSelectorStrategy.pick(servers, EasyMock.createMock(DataSegment.class)), p0); - Assert.assertEquals(defaultDeprecatedServerSelectorStrategy.pick(servers, EasyMock.createMock(DataSegment.class), 1).get(0), p0); + Assert.assertEquals( + defaultDeprecatedServerSelectorStrategy.pick(servers, EasyMock.createMock(DataSegment.class)), + p0 + ); + Assert.assertEquals( + defaultDeprecatedServerSelectorStrategy.pick(servers, EasyMock.createMock(DataSegment.class), 1) + .get(0), p0 + ); + } + + private void testPreferredTierSelectorStrategy( + PreferredTierSelectorStrategy tierSelectorStrategy, + QueryableDruidServer... expectedSelection + ) + { + final ServerSelector serverSelector = new ServerSelector( + new DataSegment( + "test", + Intervals.of("2013-01-01/2013-01-02"), + DateTimes.of("2013-01-01").toString(), + new HashMap<>(), + new ArrayList<>(), + new ArrayList<>(), + NoneShardSpec.instance(), + 0, + 0L + ), + tierSelectorStrategy, + HistoricalFilter.IDENTITY_FILTER + ); + + List servers = new ArrayList<>(Arrays.asList(expectedSelection)); + + List expectedCandidates = new ArrayList<>(); + for (QueryableDruidServer server : servers) { + expectedCandidates.add(server.getServer().getMetadata()); + } + Collections.shuffle(servers); + for (QueryableDruidServer server : servers) { + serverSelector.addServerAndUpdateSegment(server, serverSelector.getSegment()); + } + + Assert.assertEquals(expectedSelection[0], serverSelector.pick(null, CloneQueryMode.EXCLUDECLONES)); + Assert.assertEquals(expectedSelection[0], serverSelector.pick(EasyMock.createMock(Query.class), CloneQueryMode.EXCLUDECLONES)); + + // when getting all severs, the preferred tier is ignored, the returned list is sorted by priority + List allServers = new ArrayList<>(expectedCandidates); + allServers.sort((o1, o2) -> tierSelectorStrategy.getComparator().compare(o1.getPriority(), o2.getPriority())); + Assert.assertEquals(allServers, serverSelector.getCandidates(-1, CloneQueryMode.EXCLUDECLONES)); + + Assert.assertEquals(expectedCandidates.subList(0, 2), serverSelector.getCandidates(2, CloneQueryMode.EXCLUDECLONES)); + } + + @Test + public void testPreferredTierSelectorStrategyHighestPriority() + { + DirectDruidClient client = EasyMock.createMock(DirectDruidClient.class); + QueryableDruidServer preferredTierLowPriority = new QueryableDruidServer( + new DruidServer("test1", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 0), + client + ); + QueryableDruidServer preferredTierHighPriority = new QueryableDruidServer( + new DruidServer("test2", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 1), + client + ); + QueryableDruidServer nonPreferredTierHighestPriority = new QueryableDruidServer( + new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "non-preferred", 2), + client + ); + + testPreferredTierSelectorStrategy( + new PreferredTierSelectorStrategy( + new ConnectionCountServerSelectorStrategy(), + new PreferredTieSelectorStrategyConfig("preferred", "highest") + ), + preferredTierHighPriority, preferredTierLowPriority, nonPreferredTierHighestPriority + ); + } + + @Test + public void testPreferredTierSelectorStrategyLowestPriority() + { + DirectDruidClient client = EasyMock.createMock(DirectDruidClient.class); + QueryableDruidServer preferredTierLowPriority = new QueryableDruidServer( + new DruidServer("test1", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 0), + client + ); + QueryableDruidServer preferredTierHighPriority = new QueryableDruidServer( + new DruidServer("test2", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 1), + client + ); + QueryableDruidServer nonPreferredTierLowestPriority = new QueryableDruidServer( + new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "non-preferred", -1), + client + ); + + testPreferredTierSelectorStrategy( + new PreferredTierSelectorStrategy( + new ConnectionCountServerSelectorStrategy(), + new PreferredTieSelectorStrategyConfig("preferred", "lowest") + ), + preferredTierLowPriority, preferredTierHighPriority, nonPreferredTierLowestPriority + ); + } + + @Test + public void testPreferredTierSelectorStrategyWithFallback() + { + DirectDruidClient client = EasyMock.createMock(DirectDruidClient.class); + // Create only non-preferred tier servers with different priorities + QueryableDruidServer nonPreferredTierLowPriority = new QueryableDruidServer( + new DruidServer("test1", "localhost", null, 0, ServerType.HISTORICAL, "non-preferred", 0), + client + ); + QueryableDruidServer nonPreferredTierMediumPriority = new QueryableDruidServer( + new DruidServer("test2", "localhost", null, 0, ServerType.HISTORICAL, "non-preferred", 1), + client + ); + QueryableDruidServer nonPreferredTierHighPriority = new QueryableDruidServer( + new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "non-preferred", 2), + client + ); + + // Since no preferred tier servers are available, it should fall back to other servers + // based on highest priority + testPreferredTierSelectorStrategy( + new PreferredTierSelectorStrategy( + new ConnectionCountServerSelectorStrategy(), + new PreferredTieSelectorStrategyConfig("preferred", "highest") + ), + nonPreferredTierHighPriority, nonPreferredTierMediumPriority, nonPreferredTierLowPriority + ); + } + + @Test + public void testPreferredTierSelectorStrategyMixedServers() + { + DirectDruidClient client = EasyMock.createMock(DirectDruidClient.class); + QueryableDruidServer preferredTierLowPriority = new QueryableDruidServer( + new DruidServer("test1", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 0), + client + ); + QueryableDruidServer preferredTierHighPriority = new QueryableDruidServer( + new DruidServer("test2", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 1), + client + ); + QueryableDruidServer anotherTierHighPriority = new QueryableDruidServer( + new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "tier1", 2), + client + ); + QueryableDruidServer yetAnotherTierMediumPriority = new QueryableDruidServer( + new DruidServer("test4", "localhost", null, 0, ServerType.HISTORICAL, "tier2", 1), + client + ); + + // Should return preferred tier servers first, sorted by priority + testPreferredTierSelectorStrategy( + new PreferredTierSelectorStrategy( + new ConnectionCountServerSelectorStrategy(), + new PreferredTieSelectorStrategyConfig("preferred", "highest") + ), + preferredTierHighPriority, preferredTierLowPriority, anotherTierHighPriority, yetAnotherTierMediumPriority + ); } + @Test + public void testPreferredTierSelectorStrategyDefaultPriority() + { + DirectDruidClient client = EasyMock.createMock(DirectDruidClient.class); + + QueryableDruidServer preferredTierLowPriority = new QueryableDruidServer( + new DruidServer("test1", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 0), + client + ); + QueryableDruidServer preferredTierHighPriority = new QueryableDruidServer( + new DruidServer("test2", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 1), + client + ); + QueryableDruidServer nonPreferredTierHighestPriority = new QueryableDruidServer( + new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "non-preferred", 2), + client + ); + + testPreferredTierSelectorStrategy( + new PreferredTierSelectorStrategy( + new ConnectionCountServerSelectorStrategy(), + // Using null for priority should default to highest priority + new PreferredTieSelectorStrategyConfig("preferred", null) + ), + preferredTierHighPriority, preferredTierLowPriority, nonPreferredTierHighestPriority + ); + } } diff --git a/services/src/main/java/org/apache/druid/cli/CliBroker.java b/services/src/main/java/org/apache/druid/cli/CliBroker.java index 7af834942bdf..7e65f1c775d2 100644 --- a/services/src/main/java/org/apache/druid/cli/CliBroker.java +++ b/services/src/main/java/org/apache/druid/cli/CliBroker.java @@ -37,6 +37,7 @@ import org.apache.druid.client.TimelineServerView; import org.apache.druid.client.cache.CacheConfig; import org.apache.druid.client.selector.CustomTierSelectorStrategyConfig; +import org.apache.druid.client.selector.PreferredTieSelectorStrategyConfig; import org.apache.druid.client.selector.ServerSelectorStrategy; import org.apache.druid.client.selector.TierSelectorStrategy; import org.apache.druid.curator.ZkEnablementConfig; @@ -146,6 +147,7 @@ protected List getModules() JsonConfigProvider.bind(binder, "druid.broker.select", TierSelectorStrategy.class); JsonConfigProvider.bind(binder, "druid.broker.select.tier.custom", CustomTierSelectorStrategyConfig.class); + JsonConfigProvider.bind(binder, "druid.broker.select.tier.preferred", PreferredTieSelectorStrategyConfig.class); JsonConfigProvider.bind(binder, "druid.broker.balancer", ServerSelectorStrategy.class); JsonConfigProvider.bind(binder, "druid.broker.retryPolicy", RetryQueryRunnerConfig.class); JsonConfigProvider.bind(binder, "druid.broker.segment", BrokerSegmentWatcherConfig.class); From 19c9122bb0c7dfe7a1f5a91682e775bc3c1c4366 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Thu, 12 Jun 2025 15:45:41 +0800 Subject: [PATCH 02/12] Fix spelling --- .../PreferredTierSelectorStrategy.java | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java index d5cdb0c19925..06e6e8677347 100644 --- a/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java @@ -21,7 +21,6 @@ import com.fasterxml.jackson.annotation.JacksonInject; -import com.google.common.base.Preconditions; import it.unimi.dsi.fastutil.ints.Int2ObjectRBTreeMap; import org.apache.druid.client.QueryableDruidServer; import org.apache.druid.java.util.common.IAE; @@ -38,8 +37,8 @@ public class PreferredTierSelectorStrategy extends AbstractTierSelectorStrategy { private static final Logger log = new Logger(PreferredTierSelectorStrategy.class); - private final String perferredTier; - private final TierSelectorStrategy priortyStrategy; + private final String preferredTier; + private final TierSelectorStrategy priorityStrategy; public PreferredTierSelectorStrategy( @JacksonInject ServerSelectorStrategy serverSelectorStrategy, @@ -47,15 +46,15 @@ public PreferredTierSelectorStrategy( ) { super(serverSelectorStrategy); - this.perferredTier = config.getTier(); + this.preferredTier = config.getTier(); if (config.getPriority() == null) { - this.priortyStrategy = new HighestPriorityTierSelectorStrategy(serverSelectorStrategy); + this.priorityStrategy = new HighestPriorityTierSelectorStrategy(serverSelectorStrategy); } else { if ("highest".equalsIgnoreCase(config.getPriority())) { - this.priortyStrategy = new HighestPriorityTierSelectorStrategy(serverSelectorStrategy); + this.priorityStrategy = new HighestPriorityTierSelectorStrategy(serverSelectorStrategy); } else if ("lowest".equalsIgnoreCase(config.getPriority())) { - this.priortyStrategy = new LowestPriorityTierSelectorStrategy(serverSelectorStrategy); + this.priorityStrategy = new LowestPriorityTierSelectorStrategy(serverSelectorStrategy); } else { throw new IAE("druid.broker.select.tier.preferred.priority must be either 'highest' or 'lowest'"); } @@ -65,7 +64,7 @@ public PreferredTierSelectorStrategy( @Override public Comparator getComparator() { - return priortyStrategy.getComparator(); + return priorityStrategy.getComparator(); } @Override @@ -79,7 +78,7 @@ public List pick( if (log.isDebugEnabled()) { log.debug( "Picking [%d] servers from preferred tier [%s] for segment [%s] with priority [%s]", - numServersToPick, perferredTier, segment.getId(), this.priortyStrategy.getClass().getSimpleName() + numServersToPick, preferredTier, segment.getId(), this.priorityStrategy.getClass().getSimpleName() ); } @@ -87,7 +86,7 @@ public List pick( List nonPreferred = new ArrayList<>(numServersToPick); for (Set priorityServers : prioritizedServers.values()) { for (QueryableDruidServer server : priorityServers) { - if (perferredTier.equals(server.getServer().getMetadata().getTier())) { + if (preferredTier.equals(server.getServer().getMetadata().getTier())) { preferred.add(server); if (preferred.size() == numServersToPick) { return this.serverSelectorStrategy.pick(query, preferred, segment, numServersToPick); From 326f46d62a13058e366c932a45e638c01ff99297 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Thu, 12 Jun 2025 15:46:37 +0800 Subject: [PATCH 03/12] Fix class name --- .../client/selector/PreferredTierSelectorStrategy.java | 2 +- ...g.java => PreferredTierSelectorStrategyConfig.java} | 4 ++-- .../client/selector/TierSelectorStrategyTest.java | 10 +++++----- .../src/main/java/org/apache/druid/cli/CliBroker.java | 4 ++-- 4 files changed, 10 insertions(+), 10 deletions(-) rename server/src/main/java/org/apache/druid/client/selector/{PreferredTieSelectorStrategyConfig.java => PreferredTierSelectorStrategyConfig.java} (94%) diff --git a/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java index 06e6e8677347..57e1fbe65da6 100644 --- a/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java @@ -42,7 +42,7 @@ public class PreferredTierSelectorStrategy extends AbstractTierSelectorStrategy public PreferredTierSelectorStrategy( @JacksonInject ServerSelectorStrategy serverSelectorStrategy, - @JacksonInject PreferredTieSelectorStrategyConfig config + @JacksonInject PreferredTierSelectorStrategyConfig config ) { super(serverSelectorStrategy); diff --git a/server/src/main/java/org/apache/druid/client/selector/PreferredTieSelectorStrategyConfig.java b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategyConfig.java similarity index 94% rename from server/src/main/java/org/apache/druid/client/selector/PreferredTieSelectorStrategyConfig.java rename to server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategyConfig.java index dfa44eb66b9a..fda56641159c 100644 --- a/server/src/main/java/org/apache/druid/client/selector/PreferredTieSelectorStrategyConfig.java +++ b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategyConfig.java @@ -25,7 +25,7 @@ import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.base.Preconditions; -public class PreferredTieSelectorStrategyConfig +public class PreferredTierSelectorStrategyConfig { @JsonProperty private String tier; @@ -37,7 +37,7 @@ public class PreferredTieSelectorStrategyConfig private String priority; @JsonCreator - public PreferredTieSelectorStrategyConfig( + public PreferredTierSelectorStrategyConfig( @JsonProperty("tier") String tier, @JsonProperty("priority") String priority ) diff --git a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java index d6213977e009..1d213ad02462 100644 --- a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java +++ b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java @@ -352,7 +352,7 @@ public void testPreferredTierSelectorStrategyHighestPriority() testPreferredTierSelectorStrategy( new PreferredTierSelectorStrategy( new ConnectionCountServerSelectorStrategy(), - new PreferredTieSelectorStrategyConfig("preferred", "highest") + new PreferredTierSelectorStrategyConfig("preferred", "highest") ), preferredTierHighPriority, preferredTierLowPriority, nonPreferredTierHighestPriority ); @@ -378,7 +378,7 @@ public void testPreferredTierSelectorStrategyLowestPriority() testPreferredTierSelectorStrategy( new PreferredTierSelectorStrategy( new ConnectionCountServerSelectorStrategy(), - new PreferredTieSelectorStrategyConfig("preferred", "lowest") + new PreferredTierSelectorStrategyConfig("preferred", "lowest") ), preferredTierLowPriority, preferredTierHighPriority, nonPreferredTierLowestPriority ); @@ -407,7 +407,7 @@ public void testPreferredTierSelectorStrategyWithFallback() testPreferredTierSelectorStrategy( new PreferredTierSelectorStrategy( new ConnectionCountServerSelectorStrategy(), - new PreferredTieSelectorStrategyConfig("preferred", "highest") + new PreferredTierSelectorStrategyConfig("preferred", "highest") ), nonPreferredTierHighPriority, nonPreferredTierMediumPriority, nonPreferredTierLowPriority ); @@ -438,7 +438,7 @@ public void testPreferredTierSelectorStrategyMixedServers() testPreferredTierSelectorStrategy( new PreferredTierSelectorStrategy( new ConnectionCountServerSelectorStrategy(), - new PreferredTieSelectorStrategyConfig("preferred", "highest") + new PreferredTierSelectorStrategyConfig("preferred", "highest") ), preferredTierHighPriority, preferredTierLowPriority, anotherTierHighPriority, yetAnotherTierMediumPriority ); @@ -466,7 +466,7 @@ public void testPreferredTierSelectorStrategyDefaultPriority() new PreferredTierSelectorStrategy( new ConnectionCountServerSelectorStrategy(), // Using null for priority should default to highest priority - new PreferredTieSelectorStrategyConfig("preferred", null) + new PreferredTierSelectorStrategyConfig("preferred", null) ), preferredTierHighPriority, preferredTierLowPriority, nonPreferredTierHighestPriority ); diff --git a/services/src/main/java/org/apache/druid/cli/CliBroker.java b/services/src/main/java/org/apache/druid/cli/CliBroker.java index 7e65f1c775d2..7e3c11a66311 100644 --- a/services/src/main/java/org/apache/druid/cli/CliBroker.java +++ b/services/src/main/java/org/apache/druid/cli/CliBroker.java @@ -37,7 +37,7 @@ import org.apache.druid.client.TimelineServerView; import org.apache.druid.client.cache.CacheConfig; import org.apache.druid.client.selector.CustomTierSelectorStrategyConfig; -import org.apache.druid.client.selector.PreferredTieSelectorStrategyConfig; +import org.apache.druid.client.selector.PreferredTierSelectorStrategyConfig; import org.apache.druid.client.selector.ServerSelectorStrategy; import org.apache.druid.client.selector.TierSelectorStrategy; import org.apache.druid.curator.ZkEnablementConfig; @@ -147,7 +147,7 @@ protected List getModules() JsonConfigProvider.bind(binder, "druid.broker.select", TierSelectorStrategy.class); JsonConfigProvider.bind(binder, "druid.broker.select.tier.custom", CustomTierSelectorStrategyConfig.class); - JsonConfigProvider.bind(binder, "druid.broker.select.tier.preferred", PreferredTieSelectorStrategyConfig.class); + JsonConfigProvider.bind(binder, "druid.broker.select.tier.preferred", PreferredTierSelectorStrategyConfig.class); JsonConfigProvider.bind(binder, "druid.broker.balancer", ServerSelectorStrategy.class); JsonConfigProvider.bind(binder, "druid.broker.retryPolicy", RetryQueryRunnerConfig.class); JsonConfigProvider.bind(binder, "druid.broker.segment", BrokerSegmentWatcherConfig.class); From 739f46515e6906cdb10ebda1eb8f19ac67093d0f Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Thu, 12 Jun 2025 15:47:20 +0800 Subject: [PATCH 04/12] Fix doc --- docs/configuration/index.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/configuration/index.md b/docs/configuration/index.md index 8f4a27f0be10..aa575350cbc1 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -1792,7 +1792,7 @@ These Broker configurations can be defined in the `broker/runtime.properties` fi |`druid.broker.select.tier`|`highestPriority`, `lowestPriority`, `custom`|If segments are cross-replicated across tiers in a cluster, you can tell the broker to prefer to select segments in a tier with a certain priority.|`highestPriority`| |`druid.broker.select.tier.custom.priorities`|An array of integer priorities, such as `[-1, 0, 1, 2]`|Select servers in tiers with a custom priority list.|The config only has effect if `druid.broker.select.tier` is set to `custom`. If `druid.broker.select.tier` is set to `custom` but this config is not specified, the effect is the same as `druid.broker.select.tier` set to `highestPriority`. Any of the integers in this config can be ignored if there's no corresponding tiers with such priorities. Tiers with priorities explicitly specified in this config always have higher priority than those not and those not specified fall back to use `highestPriority` strategy among themselves.| |`druid.broker.select.tier.preferred.tier`| The preferred tier name. E.g., `_default_tier` | A non-empty value that specifies the preferred tier in which historical servers will be picked up for queries. If there're no enough historical servers from the preferred tier, servers from other tiers might be choosed. | null | -|`druid.broker.select.tier.preferred.priority`| `high`, `low` | Optional. If there're multiple candidates in a preferred tier, specifies the priority to pick up candiates. By default, the higher priority a historical, the higher chances it will be picked up. | `high` | +|`druid.broker.select.tier.preferred.priority`| `highest`, `lowest` | Optional. If there're multiple candidates in a preferred tier, specifies the priority to pick up candiates. By default, the higher priority a historical, the higher chances it will be picked up. | `highest` | ##### Query prioritization and laning From ed048a4eba210b8c43a6464067db3300fe58d727 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Fri, 13 Jun 2025 10:09:29 +0800 Subject: [PATCH 05/12] Fix static check --- .../client/selector/ConnectionCountServerSelectorStrategy.java | 1 - .../druid/client/selector/RandomServerSelectorStrategy.java | 1 - .../org/apache/druid/client/selector/ServerSelectorStrategy.java | 1 - 3 files changed, 3 deletions(-) diff --git a/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java index 598141ca09d5..cde7ca783d31 100644 --- a/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java @@ -30,7 +30,6 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; -import java.util.Set; import java.util.concurrent.ThreadLocalRandom; public class ConnectionCountServerSelectorStrategy implements ServerSelectorStrategy diff --git a/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java index f4133b9f3727..e25b8620723d 100644 --- a/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java @@ -29,7 +29,6 @@ import java.util.Collection; import java.util.Collections; import java.util.List; -import java.util.Set; import java.util.concurrent.ThreadLocalRandom; public class RandomServerSelectorStrategy implements ServerSelectorStrategy diff --git a/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java index 43d81dd85144..f487ade6e89b 100644 --- a/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java @@ -29,7 +29,6 @@ import javax.annotation.Nullable; import java.util.Collection; import java.util.List; -import java.util.Set; @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = RandomServerSelectorStrategy.class) @JsonSubTypes(value = { From 12baa0befaa3177016b0bbbc20c6ccd961bccd4c Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Fri, 13 Jun 2025 12:02:55 +0800 Subject: [PATCH 06/12] Fix doc --- docs/configuration/index.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/configuration/index.md b/docs/configuration/index.md index aa575350cbc1..fd85259b6223 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -1791,8 +1791,8 @@ These Broker configurations can be defined in the `broker/runtime.properties` fi |`druid.broker.balancer.type`|`random`, `connectionCount`|Determines how the broker balances connections to Historical processes. `random` choose randomly, `connectionCount` picks the process with the fewest number of active connections to|`random`| |`druid.broker.select.tier`|`highestPriority`, `lowestPriority`, `custom`|If segments are cross-replicated across tiers in a cluster, you can tell the broker to prefer to select segments in a tier with a certain priority.|`highestPriority`| |`druid.broker.select.tier.custom.priorities`|An array of integer priorities, such as `[-1, 0, 1, 2]`|Select servers in tiers with a custom priority list.|The config only has effect if `druid.broker.select.tier` is set to `custom`. If `druid.broker.select.tier` is set to `custom` but this config is not specified, the effect is the same as `druid.broker.select.tier` set to `highestPriority`. Any of the integers in this config can be ignored if there's no corresponding tiers with such priorities. Tiers with priorities explicitly specified in this config always have higher priority than those not and those not specified fall back to use `highestPriority` strategy among themselves.| -|`druid.broker.select.tier.preferred.tier`| The preferred tier name. E.g., `_default_tier` | A non-empty value that specifies the preferred tier in which historical servers will be picked up for queries. If there're no enough historical servers from the preferred tier, servers from other tiers might be choosed. | null | -|`druid.broker.select.tier.preferred.priority`| `highest`, `lowest` | Optional. If there're multiple candidates in a preferred tier, specifies the priority to pick up candiates. By default, the higher priority a historical, the higher chances it will be picked up. | `highest` | +|`druid.broker.select.tier.preferred.tier`| The preferred tier name. E.g., `_default_tier` | A non-empty value that specifies the preferred tier in which historical servers will be picked up for queries. If there are no enough historical servers from the preferred tier, servers from other tiers (if there are) will be selected. | null | +|`druid.broker.select.tier.preferred.priority`| `highest`, `lowest` | Optional. If there are multiple candidates in a preferred tier, specifies the priority to pick up candidates. By default, the higher priority a historical, the higher chances it will be picked up. | `highest` | ##### Query prioritization and laning From f94f44d0ad8e68d8caf01e1f565bbfd9af30f692 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 1 Jul 2025 11:53:00 +0800 Subject: [PATCH 07/12] Update docs/configuration/index.md Co-authored-by: Lucas Capistrant --- docs/configuration/index.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/configuration/index.md b/docs/configuration/index.md index fd85259b6223..0463a9107a2c 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -1792,7 +1792,7 @@ These Broker configurations can be defined in the `broker/runtime.properties` fi |`druid.broker.select.tier`|`highestPriority`, `lowestPriority`, `custom`|If segments are cross-replicated across tiers in a cluster, you can tell the broker to prefer to select segments in a tier with a certain priority.|`highestPriority`| |`druid.broker.select.tier.custom.priorities`|An array of integer priorities, such as `[-1, 0, 1, 2]`|Select servers in tiers with a custom priority list.|The config only has effect if `druid.broker.select.tier` is set to `custom`. If `druid.broker.select.tier` is set to `custom` but this config is not specified, the effect is the same as `druid.broker.select.tier` set to `highestPriority`. Any of the integers in this config can be ignored if there's no corresponding tiers with such priorities. Tiers with priorities explicitly specified in this config always have higher priority than those not and those not specified fall back to use `highestPriority` strategy among themselves.| |`druid.broker.select.tier.preferred.tier`| The preferred tier name. E.g., `_default_tier` | A non-empty value that specifies the preferred tier in which historical servers will be picked up for queries. If there are no enough historical servers from the preferred tier, servers from other tiers (if there are) will be selected. | null | -|`druid.broker.select.tier.preferred.priority`| `highest`, `lowest` | Optional. If there are multiple candidates in a preferred tier, specifies the priority to pick up candidates. By default, the higher priority a historical, the higher chances it will be picked up. | `highest` | +|`druid.broker.select.tier.preferred.priority`| `highest`, `lowest` | If there are multiple candidates in a preferred tier, specifies the priority to pick up candidates. By default, the higher priority a historical, the higher chances it will be picked up. This config only has effect if `druid.broker.select.tier` is set to `preferred`| `highest` | ##### Query prioritization and laning From 874976c34728aa8d2c4a706876ed952cc16f0842 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 1 Jul 2025 11:53:07 +0800 Subject: [PATCH 08/12] Update docs/configuration/index.md Co-authored-by: Lucas Capistrant --- docs/configuration/index.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/configuration/index.md b/docs/configuration/index.md index 0463a9107a2c..1fa510e10944 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -1791,7 +1791,7 @@ These Broker configurations can be defined in the `broker/runtime.properties` fi |`druid.broker.balancer.type`|`random`, `connectionCount`|Determines how the broker balances connections to Historical processes. `random` choose randomly, `connectionCount` picks the process with the fewest number of active connections to|`random`| |`druid.broker.select.tier`|`highestPriority`, `lowestPriority`, `custom`|If segments are cross-replicated across tiers in a cluster, you can tell the broker to prefer to select segments in a tier with a certain priority.|`highestPriority`| |`druid.broker.select.tier.custom.priorities`|An array of integer priorities, such as `[-1, 0, 1, 2]`|Select servers in tiers with a custom priority list.|The config only has effect if `druid.broker.select.tier` is set to `custom`. If `druid.broker.select.tier` is set to `custom` but this config is not specified, the effect is the same as `druid.broker.select.tier` set to `highestPriority`. Any of the integers in this config can be ignored if there's no corresponding tiers with such priorities. Tiers with priorities explicitly specified in this config always have higher priority than those not and those not specified fall back to use `highestPriority` strategy among themselves.| -|`druid.broker.select.tier.preferred.tier`| The preferred tier name. E.g., `_default_tier` | A non-empty value that specifies the preferred tier in which historical servers will be picked up for queries. If there are no enough historical servers from the preferred tier, servers from other tiers (if there are) will be selected. | null | +|`druid.broker.select.tier.preferred.tier`| The preferred tier name. E.g., `_default_tier` | A non-empty value that specifies the preferred tier in which historical servers will be picked up for queries. If there are not enough historical servers from the preferred tier, servers from other tiers (if there are any) will be selected. This config only has effect if `druid.broker.select.tier` is set to `preferred` | null | |`druid.broker.select.tier.preferred.priority`| `highest`, `lowest` | If there are multiple candidates in a preferred tier, specifies the priority to pick up candidates. By default, the higher priority a historical, the higher chances it will be picked up. This config only has effect if `druid.broker.select.tier` is set to `preferred`| `highest` | ##### Query prioritization and laning From 9950244c84d1251996e062877aecd07110947f7b Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 1 Jul 2025 11:53:43 +0800 Subject: [PATCH 09/12] Update docs/configuration/index.md Co-authored-by: Lucas Capistrant --- docs/configuration/index.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/configuration/index.md b/docs/configuration/index.md index 1fa510e10944..427946f9bc3f 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -1789,7 +1789,7 @@ These Broker configurations can be defined in the `broker/runtime.properties` fi |Property|Possible Values|Description|Default| |--------|---------------|-----------|-------| |`druid.broker.balancer.type`|`random`, `connectionCount`|Determines how the broker balances connections to Historical processes. `random` choose randomly, `connectionCount` picks the process with the fewest number of active connections to|`random`| -|`druid.broker.select.tier`|`highestPriority`, `lowestPriority`, `custom`|If segments are cross-replicated across tiers in a cluster, you can tell the broker to prefer to select segments in a tier with a certain priority.|`highestPriority`| +|`druid.broker.select.tier`|`highestPriority`, `lowestPriority`, `custom`, `preferred`|If segments are cross-replicated across tiers in a cluster, you can tell the broker to prefer to select segments in a tier with a certain priority.|`highestPriority`| |`druid.broker.select.tier.custom.priorities`|An array of integer priorities, such as `[-1, 0, 1, 2]`|Select servers in tiers with a custom priority list.|The config only has effect if `druid.broker.select.tier` is set to `custom`. If `druid.broker.select.tier` is set to `custom` but this config is not specified, the effect is the same as `druid.broker.select.tier` set to `highestPriority`. Any of the integers in this config can be ignored if there's no corresponding tiers with such priorities. Tiers with priorities explicitly specified in this config always have higher priority than those not and those not specified fall back to use `highestPriority` strategy among themselves.| |`druid.broker.select.tier.preferred.tier`| The preferred tier name. E.g., `_default_tier` | A non-empty value that specifies the preferred tier in which historical servers will be picked up for queries. If there are not enough historical servers from the preferred tier, servers from other tiers (if there are any) will be selected. This config only has effect if `druid.broker.select.tier` is set to `preferred` | null | |`druid.broker.select.tier.preferred.priority`| `highest`, `lowest` | If there are multiple candidates in a preferred tier, specifies the priority to pick up candidates. By default, the higher priority a historical, the higher chances it will be picked up. This config only has effect if `druid.broker.select.tier` is set to `preferred`| `highest` | From a82d427b4c0a1def51ae1c12428c0451a474eadf Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 1 Jul 2025 12:01:55 +0800 Subject: [PATCH 10/12] Add comments --- .../client/selector/TierSelectorStrategyTest.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java index 1d213ad02462..dfdf337738d5 100644 --- a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java +++ b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java @@ -289,6 +289,13 @@ public List pick( ); } + /** + * Tests the PreferredTierSelectorStrategy with various configurations and expected selections. + * It verifies + * 1. The preferred tier is respected when picking a server. + * 2. When getting all servers, the preferred tier is ignored, and the returned list is sorted by priority. + * 3. When getting a limited number of candidates, it returns the top N servers with the preferred tier first. + */ private void testPreferredTierSelectorStrategy( PreferredTierSelectorStrategy tierSelectorStrategy, QueryableDruidServer... expectedSelection @@ -321,14 +328,16 @@ private void testPreferredTierSelectorStrategy( serverSelector.addServerAndUpdateSegment(server, serverSelector.getSegment()); } + // Verify that the preferred tier is respected when picking a server Assert.assertEquals(expectedSelection[0], serverSelector.pick(null, CloneQueryMode.EXCLUDECLONES)); Assert.assertEquals(expectedSelection[0], serverSelector.pick(EasyMock.createMock(Query.class), CloneQueryMode.EXCLUDECLONES)); - // when getting all severs, the preferred tier is ignored, the returned list is sorted by priority + // Verify that when getting all severs, the preferred tier is ignored, the returned list is sorted by priority List allServers = new ArrayList<>(expectedCandidates); allServers.sort((o1, o2) -> tierSelectorStrategy.getComparator().compare(o1.getPriority(), o2.getPriority())); Assert.assertEquals(allServers, serverSelector.getCandidates(-1, CloneQueryMode.EXCLUDECLONES)); + // Verify that when getting a limited number of candidates, returns the top N servers with preferred tier first Assert.assertEquals(expectedCandidates.subList(0, 2), serverSelector.getCandidates(2, CloneQueryMode.EXCLUDECLONES)); } From 5ed6c389059d59dcd692709ff8b9a89f58cdd854 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Thu, 3 Jul 2025 14:25:11 +0800 Subject: [PATCH 11/12] Fix tests --- .../druid/client/selector/TierSelectorStrategyTest.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java index dfdf337738d5..7bf63d8d371b 100644 --- a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java +++ b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java @@ -323,7 +323,6 @@ private void testPreferredTierSelectorStrategy( for (QueryableDruidServer server : servers) { expectedCandidates.add(server.getServer().getMetadata()); } - Collections.shuffle(servers); for (QueryableDruidServer server : servers) { serverSelector.addServerAndUpdateSegment(server, serverSelector.getSegment()); } @@ -435,11 +434,11 @@ public void testPreferredTierSelectorStrategyMixedServers() client ); QueryableDruidServer anotherTierHighPriority = new QueryableDruidServer( - new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "tier1", 2), + new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "tier1", 3), client ); QueryableDruidServer yetAnotherTierMediumPriority = new QueryableDruidServer( - new DruidServer("test4", "localhost", null, 0, ServerType.HISTORICAL, "tier2", 1), + new DruidServer("test4", "localhost", null, 0, ServerType.HISTORICAL, "tier2", 2), client ); From 2fdfe606ca24c331b7391a2839d14fb579c9fa2a Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Mon, 7 Jul 2025 17:34:06 +0800 Subject: [PATCH 12/12] Fix logic --- ...ConnectionCountServerSelectorStrategy.java | 6 +- .../PreferredTierSelectorStrategy.java | 32 +++--- .../RandomServerSelectorStrategy.java | 6 +- .../selector/ServerSelectorStrategy.java | 10 +- .../selector/TierSelectorStrategyTest.java | 107 ++++++++++++++++-- 5 files changed, 127 insertions(+), 34 deletions(-) diff --git a/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java index cde7ca783d31..262b23701df8 100644 --- a/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/ConnectionCountServerSelectorStrategy.java @@ -26,10 +26,10 @@ import org.apache.druid.timeline.DataSegment; import javax.annotation.Nullable; -import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.List; +import java.util.Set; import java.util.concurrent.ThreadLocalRandom; public class ConnectionCountServerSelectorStrategy implements ServerSelectorStrategy @@ -42,13 +42,13 @@ public class ConnectionCountServerSelectorStrategy implements ServerSelectorStra @Nullable @Override - public QueryableDruidServer pick(Collection servers, DataSegment segment) + public QueryableDruidServer pick(Set servers, DataSegment segment) { return Collections.min(servers, COMPARATOR); } @Override - public List pick(Collection servers, DataSegment segment, int numServersToPick) + public List pick(Set servers, DataSegment segment, int numServersToPick) { if (servers.size() <= numServersToPick) { return ImmutableList.copyOf(servers); diff --git a/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java index 57e1fbe65da6..8287830d0c52 100644 --- a/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/PreferredTierSelectorStrategy.java @@ -30,6 +30,7 @@ import java.util.ArrayList; import java.util.Comparator; +import java.util.HashSet; import java.util.List; import java.util.Set; @@ -82,29 +83,32 @@ public List pick( ); } - List preferred = new ArrayList<>(numServersToPick); - List nonPreferred = new ArrayList<>(numServersToPick); + Int2ObjectRBTreeMap> preferred = new Int2ObjectRBTreeMap<>(priorityStrategy.getComparator()); + Int2ObjectRBTreeMap> nonPreferred = new Int2ObjectRBTreeMap<>(priorityStrategy.getComparator()); for (Set priorityServers : prioritizedServers.values()) { for (QueryableDruidServer server : priorityServers) { if (preferredTier.equals(server.getServer().getMetadata().getTier())) { - preferred.add(server); - if (preferred.size() == numServersToPick) { - return this.serverSelectorStrategy.pick(query, preferred, segment, numServersToPick); - } + preferred.computeIfAbsent(server.getServer().getPriority(), k -> new HashSet<>()) + .add(server); } else { - // We have to iterate through all servers even the numbers of the non-preferred servers reach the limit - // This is because we don't know whether there're preferred servers left in the next priority set - nonPreferred.add(server); + nonPreferred.computeIfAbsent(server.getServer().getPriority(), k -> new HashSet<>()) + .add(server); } } } - // Fill with non-preferred servers if we don't have enough preferred servers - int fillSize = numServersToPick - preferred.size(); - for (int i = 0; i < fillSize && i < nonPreferred.size(); i++) { - preferred.add(nonPreferred.get(i)); + List picks = new ArrayList<>(numServersToPick); + if (!preferred.isEmpty()) { + // If we have preferred servers, pick them first + picks.addAll(priorityStrategy.pick(query, preferred, segment, numServersToPick)); } - return this.serverSelectorStrategy.pick(query, preferred, segment, numServersToPick); + if (picks.size() < numServersToPick && !nonPreferred.isEmpty()) { + // If we don't have enough preferred servers, pick from the non-preferred ones + int remaining = numServersToPick - picks.size(); + picks.addAll(priorityStrategy.pick(query, nonPreferred, segment, remaining)); + } + + return picks; } } diff --git a/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java index e25b8620723d..b916a0aed303 100644 --- a/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/RandomServerSelectorStrategy.java @@ -26,22 +26,22 @@ import org.apache.druid.timeline.DataSegment; import javax.annotation.Nullable; -import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.Set; import java.util.concurrent.ThreadLocalRandom; public class RandomServerSelectorStrategy implements ServerSelectorStrategy { @Nullable @Override - public QueryableDruidServer pick(Collection servers, DataSegment segment) + public QueryableDruidServer pick(Set servers, DataSegment segment) { return Iterators.get(servers.iterator(), ThreadLocalRandom.current().nextInt(servers.size())); } @Override - public List pick(Collection servers, DataSegment segment, int numServersToPick) + public List pick(Set servers, DataSegment segment, int numServersToPick) { if (servers.size() <= numServersToPick) { return ImmutableList.copyOf(servers); diff --git a/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java b/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java index f487ade6e89b..e93e19e75060 100644 --- a/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java +++ b/server/src/main/java/org/apache/druid/client/selector/ServerSelectorStrategy.java @@ -27,8 +27,8 @@ import org.apache.druid.timeline.DataSegment; import javax.annotation.Nullable; -import java.util.Collection; import java.util.List; +import java.util.Set; @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = RandomServerSelectorStrategy.class) @JsonSubTypes(value = { @@ -38,12 +38,12 @@ public interface ServerSelectorStrategy { @Nullable - default QueryableDruidServer pick(@Nullable Query query, Collection servers, DataSegment segment) + default QueryableDruidServer pick(@Nullable Query query, Set servers, DataSegment segment) { return Iterables.getOnlyElement(pick(query, servers, segment, 1), null); } - default List pick(@Nullable Query query, Collection servers, DataSegment segment, + default List pick(@Nullable Query query, Set servers, DataSegment segment, int numServersToPick) { return pick(servers, segment, numServersToPick); @@ -51,13 +51,13 @@ default List pick(@Nullable Query query, Collection @Deprecated @Nullable - default QueryableDruidServer pick(Collection servers, DataSegment segment) + default QueryableDruidServer pick(Set servers, DataSegment segment) { return pick(null, servers, segment); } @Deprecated - default List pick(Collection servers, DataSegment segment, int numServersToPick) + default List pick(Set servers, DataSegment segment, int numServersToPick) { return pick(null, servers, segment, numServersToPick); } diff --git a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java index 7bf63d8d371b..a243d5d0fbfa 100644 --- a/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java +++ b/server/src/test/java/org/apache/druid/client/selector/TierSelectorStrategyTest.java @@ -19,6 +19,7 @@ package org.apache.druid.client.selector; +import com.google.common.collect.ImmutableList; import org.apache.druid.client.DirectDruidClient; import org.apache.druid.client.DruidServer; import org.apache.druid.client.QueryableDruidServer; @@ -37,12 +38,12 @@ import javax.annotation.Nullable; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.stream.Collectors; public class TierSelectorStrategyTest { @@ -272,7 +273,7 @@ public void testServerSelectorStrategyDefaults() { @Override public List pick( - @Nullable Query query, Collection servers, DataSegment segment, + @Nullable Query query, Set servers, DataSegment segment, int numServersToPick ) { @@ -334,7 +335,11 @@ private void testPreferredTierSelectorStrategy( // Verify that when getting all severs, the preferred tier is ignored, the returned list is sorted by priority List allServers = new ArrayList<>(expectedCandidates); allServers.sort((o1, o2) -> tierSelectorStrategy.getComparator().compare(o1.getPriority(), o2.getPriority())); - Assert.assertEquals(allServers, serverSelector.getCandidates(-1, CloneQueryMode.EXCLUDECLONES)); + // verify the priority only because values with same priority may return in different order + Assert.assertEquals( + allServers.stream().map(DruidServerMetadata::getPriority).collect(Collectors.toList()), + serverSelector.getCandidates(-1, CloneQueryMode.EXCLUDECLONES).stream().map(DruidServerMetadata::getPriority).collect(Collectors.toList()) + ); // Verify that when getting a limited number of candidates, returns the top N servers with preferred tier first Assert.assertEquals(expectedCandidates.subList(0, 2), serverSelector.getCandidates(2, CloneQueryMode.EXCLUDECLONES)); @@ -344,6 +349,8 @@ private void testPreferredTierSelectorStrategy( public void testPreferredTierSelectorStrategyHighestPriority() { DirectDruidClient client = EasyMock.createMock(DirectDruidClient.class); + + // Two servers that have same tier and priority QueryableDruidServer preferredTierLowPriority = new QueryableDruidServer( new DruidServer("test1", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 0), client @@ -352,17 +359,99 @@ public void testPreferredTierSelectorStrategyHighestPriority() new DruidServer("test2", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 1), client ); + + QueryableDruidServer preferredTierHighPriority2 = new QueryableDruidServer( + new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "preferred", 1), + client + ); + QueryableDruidServer nonPreferredTierHighestPriority = new QueryableDruidServer( - new DruidServer("test3", "localhost", null, 0, ServerType.HISTORICAL, "non-preferred", 2), + new DruidServer("test4", "localhost", null, 0, ServerType.HISTORICAL, "non-preferred", 2), client ); - testPreferredTierSelectorStrategy( - new PreferredTierSelectorStrategy( - new ConnectionCountServerSelectorStrategy(), - new PreferredTierSelectorStrategyConfig("preferred", "highest") + PreferredTierSelectorStrategy tierSelectorStrategy = new PreferredTierSelectorStrategy( + // Use a customized strategy that return the 2nd server + new ServerSelectorStrategy() + { + @Override + public List pick(Set servers, DataSegment segment, int numServersToPick) + { + if (servers.size() <= numServersToPick) { + return ImmutableList.copyOf(servers); + } + List list = new ArrayList<>(servers); + if (numServersToPick == 1) { + // return the server whose name is greater + return list.stream() + .sorted((o1, o2) -> o1.getServer().getName().compareTo(o2.getServer().getName())) + .skip(1) + .limit(1) + .collect(Collectors.toList()); + } else { + return list.stream().limit(numServersToPick).collect(Collectors.toList()); + } + } + }, + new PreferredTierSelectorStrategyConfig("preferred", "highest") + ); + + final ServerSelector serverSelector = new ServerSelector( + new DataSegment( + "test", + Intervals.of("2013-01-01/2013-01-02"), + DateTimes.of("2013-01-01").toString(), + new HashMap<>(), + new ArrayList<>(), + new ArrayList<>(), + NoneShardSpec.instance(), + 0, + 0L ), - preferredTierHighPriority, preferredTierLowPriority, nonPreferredTierHighestPriority + tierSelectorStrategy, + HistoricalFilter.IDENTITY_FILTER + ); + + List servers = new ArrayList<>(Arrays.asList( + preferredTierLowPriority, + preferredTierHighPriority, + preferredTierHighPriority2, + nonPreferredTierHighestPriority + )); + + List expectedCandidates = new ArrayList<>(); + for (QueryableDruidServer server : servers) { + expectedCandidates.add(server.getServer().getMetadata()); + } + for (QueryableDruidServer server : servers) { + serverSelector.addServerAndUpdateSegment(server, serverSelector.getSegment()); + } + + // Verify that the 2nd server is selected + Assert.assertEquals(preferredTierHighPriority2, serverSelector.pick(null, CloneQueryMode.EXCLUDECLONES)); + Assert.assertEquals(preferredTierHighPriority2, serverSelector.pick(EasyMock.createMock(Query.class), CloneQueryMode.EXCLUDECLONES)); + + // Verify that when getting all severs, the preferred tier is ignored, the returned list is sorted by priority + List allServers = new ArrayList<>(expectedCandidates); + allServers.sort((o1, o2) -> tierSelectorStrategy.getComparator().compare(o1.getPriority(), o2.getPriority())); + // verify the priority only because values with same priority may return in different order + Assert.assertEquals( + allServers.stream().map(DruidServerMetadata::getPriority).collect(Collectors.toList()), + serverSelector.getCandidates(-1, CloneQueryMode.EXCLUDECLONES).stream().map(DruidServerMetadata::getPriority).collect(Collectors.toList()) + ); + + // Verify that when getting 2 candidates, returns the top N servers with preferred tier first + Assert.assertEquals( + Arrays.asList( + preferredTierHighPriority.getServer().getMetadata(), + preferredTierHighPriority2.getServer().getMetadata() + ), + + serverSelector.getCandidates(2, CloneQueryMode.EXCLUDECLONES) + .stream() + // sort the name to make sure the test is stable + .sorted((o1, o2) -> o1.getName().compareTo(o2.getName())) + .collect(Collectors.toList()) ); }