This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 2bf1643cddc fix: Apply cloneQueryMode prior to choosing server type.
(#20244)
2bf1643cddc is described below
commit 2bf1643cddce5e07cfe07c3687b7445413ae9b34
Author: Gian Merlino <[email protected]>
AuthorDate: Thu Sep 3 10:05:58 2026 -0700
fix: Apply cloneQueryMode prior to choosing server type. (#20244)
In "pick", ServerSelector needs to apply cloneQueryMode prior to
choosing whether to use Historical or realtime servers. It is possible
that all Historicals will be filtered out by cloneQueryMode, in which
case we need to fall back to realtime.
---
.../druid/client/selector/ServerSelector.java | 22 +++++-
.../druid/client/selector/ServerSelectorTest.java | 82 ++++++++++++++++++++++
2 files changed, 102 insertions(+), 2 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/client/selector/ServerSelector.java
b/server/src/main/java/org/apache/druid/client/selector/ServerSelector.java
index 54bf3b2e0fb..d47a5a42842 100644
--- a/server/src/main/java/org/apache/druid/client/selector/ServerSelector.java
+++ b/server/src/main/java/org/apache/druid/client/selector/ServerSelector.java
@@ -193,8 +193,11 @@ public class ServerSelector implements
Overshadowable<ServerSelector>
public <T> QueryableDruidServer pick(@Nullable Query<T> query,
CloneQueryMode cloneQueryMode)
{
synchronized (this) {
- if (!historicalServers.isEmpty()) {
- return historicalTierStrategy.pick(query,
filter.getQueryableServers(historicalServers, cloneQueryMode), segment.get());
+ final Int2ObjectRBTreeMap<Set<QueryableDruidServer>>
queryableHistoricals =
+ filter.getQueryableServers(historicalServers, cloneQueryMode);
+
+ if (hasAnyServers(queryableHistoricals)) {
+ return historicalTierStrategy.pick(query, queryableHistoricals,
segment.get());
}
return realtimeTierStrategy.pick(query, realtimeServers, segment.get());
}
@@ -253,4 +256,19 @@ public class ServerSelector implements
Overshadowable<ServerSelector>
return (!realtimeServers.isEmpty()) && historicalServers.isEmpty();
}
}
+
+ /**
+ * Whether the given priority-to-servers map holds at least one server.
+ */
+ private static boolean hasAnyServers(final
Int2ObjectRBTreeMap<Set<QueryableDruidServer>> servers)
+ {
+ for (final Set<QueryableDruidServer> priorityServers : servers.values()) {
+ if (!priorityServers.isEmpty()) {
+ return true;
+ }
+ }
+
+ // No sets, or all sets were empty.
+ return false;
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/client/selector/ServerSelectorTest.java
b/server/src/test/java/org/apache/druid/client/selector/ServerSelectorTest.java
index 6efa1c1a3a7..01a97ed4d20 100644
---
a/server/src/test/java/org/apache/druid/client/selector/ServerSelectorTest.java
+++
b/server/src/test/java/org/apache/druid/client/selector/ServerSelectorTest.java
@@ -21,18 +21,27 @@ package org.apache.druid.client.selector;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import org.apache.druid.client.BrokerViewOfCoordinatorConfig;
import org.apache.druid.client.DirectDruidClient;
import org.apache.druid.client.DruidServer;
import org.apache.druid.client.QueryableDruidServer;
+import org.apache.druid.client.coordinator.CoordinatorClient;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.query.CloneQueryMode;
import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.NoneShardSpec;
import org.apache.druid.timeline.partition.TombstoneShardSpec;
import org.easymock.EasyMock;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.List;
+import java.util.Map;
public class ServerSelectorTest
@@ -171,4 +180,77 @@ public class ServerSelectorTest
Assertions.assertTrue(selector.hasData());
}
+ @Test
+ public void testPickFallsBackToRealtimeWhenEveryHistoricalIsExcluded()
+ {
+ final ServerSelector selector = makeSelector("test_clone_and_realtime");
+
+ final QueryableDruidServer cloneTarget = addServer(selector, "clone:8083",
ServerType.HISTORICAL);
+ final QueryableDruidServer peon = addServer(selector, "peon:8100",
ServerType.INDEXER_EXECUTOR);
+
+ Assertions.assertEquals(peon, selector.pick(null,
CloneQueryMode.EXCLUDECLONES));
+ Assertions.assertEquals(
+ List.of(peon.getServer().getMetadata()),
+ selector.getCandidates(1, CloneQueryMode.EXCLUDECLONES)
+ );
+
+ // The clone target is queryable when clones are not excluded.
+ Assertions.assertEquals(cloneTarget, selector.pick(null,
CloneQueryMode.INCLUDECLONES));
+ }
+
+ @Test
+ public void
testPickReturnsNullWhenEveryHistoricalIsExcludedAndThereIsNoRealtimeServer()
+ {
+ final ServerSelector selector = makeSelector("test_clone_only");
+
+ addServer(selector, "clone:8083", ServerType.HISTORICAL);
+
+ Assertions.assertNull(selector.pick(null, CloneQueryMode.EXCLUDECLONES));
+ Assertions.assertEquals(List.of(), selector.getCandidates(1,
CloneQueryMode.EXCLUDECLONES));
+ }
+
+ @Test
+ public void
testPickPrefersHistoricalOverRealtimeWhenTheHistoricalIsNotExcluded()
+ {
+ final ServerSelector selector = makeSelector("test_source_and_realtime");
+
+ final QueryableDruidServer cloneSource = addServer(selector,
"source:8083", ServerType.HISTORICAL);
+ addServer(selector, "peon:8100", ServerType.INDEXER_EXECUTOR);
+
+ Assertions.assertEquals(cloneSource, selector.pick(null,
CloneQueryMode.EXCLUDECLONES));
+ }
+
+ /**
+ * Creates a selector whose {@link HistoricalFilter} treats "clone:8083" as
a clone of "source:8083".
+ */
+ private static ServerSelector makeSelector(final String dataSource)
+ {
+ final BrokerViewOfCoordinatorConfig filter =
+ new
BrokerViewOfCoordinatorConfig(Mockito.mock(CoordinatorClient.class));
+ filter.setDynamicConfig(
+ CoordinatorDynamicConfig.builder()
+ .withCloneServers(Map.of("clone:8083",
"source:8083"))
+ .build()
+ );
+
+ return new ServerSelector(
+
DataSegment.builder(SegmentId.dummy(dataSource)).shardSpec(NoneShardSpec.instance()).build(),
+ new HighestPriorityTierSelectorStrategy(new
RandomServerSelectorStrategy()),
+ filter
+ );
+ }
+
+ private static QueryableDruidServer addServer(
+ final ServerSelector selector,
+ final String host,
+ final ServerType serverType
+ )
+ {
+ final QueryableDruidServer server = new QueryableDruidServer(
+ new DruidServer(host, host, null, 0, null, serverType,
DruidServer.DEFAULT_TIER, 0),
+ EasyMock.createMock(DirectDruidClient.class)
+ );
+ selector.addServerAndUpdateSegment(server, selector.getSegment());
+ return server;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
