This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch 38.0.0
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/38.0.0 by this push:
new 0bc5a992913 fix: remove synchronized lock contention on broker query
planning hot path (#19787)
0bc5a992913 is described below
commit 0bc5a992913ce9813401f570c05e1f95cb0a487b
Author: Maytas Monsereenusorn <[email protected]>
AuthorDate: Wed Jul 29 04:39:00 2026 -0700
fix: remove synchronized lock contention on broker query planning hot path
(#19787)
Summary:
- Replace synchronized with volatile on
BrokerViewOfCoordinatorConfig.getCurrentServersToIgnore()
and setDynamicConfig() to eliminate monitor contention on the broker query
planning hot path
- getCurrentServersToIgnore() is called per-segment during
groupSegmentsByServer() and
computeResultLevelCachingEtag() — with hundreds of data servers and
hundreds of concurrent
queries, threads pile up on the monitor causing a convoy effect
- synchronized was unnecessary: each read touches only one ImmutableSet
reference (selected by
CloneQueryMode), and volatile provides sufficient visibility for immutable
reference swaps.
This is safe because:
- ImmutableSet.copyOf() creates fully-constructed immutable objects
- volatile guarantees visibility of the reference swap
- Readers may briefly see a stale config during setDynamicConfig(), but
that's acceptable since
setDynamicConfig() is called infrequently (config updates), and a query
seeing the old config for one
cycle is harmless
(cherry picked from commit 7abe001ab96872f7739c285f128e750d95f55e9e)
---
.../query/BrokerServerFilterBenchmark.java | 102 ++++++++++++++++++++
.../druid/client/BaseBrokerViewOfConfig.java | 11 ++-
.../client/BrokerViewOfCoordinatorConfig.java | 16 ++--
.../client/BrokerViewOfCoordinatorConfigTest.java | 106 +++++++++++++++++++++
4 files changed, 223 insertions(+), 12 deletions(-)
diff --git
a/benchmarks/src/test/java/org/apache/druid/benchmark/query/BrokerServerFilterBenchmark.java
b/benchmarks/src/test/java/org/apache/druid/benchmark/query/BrokerServerFilterBenchmark.java
new file mode 100644
index 00000000000..75791460c13
--- /dev/null
+++
b/benchmarks/src/test/java/org/apache/druid/benchmark/query/BrokerServerFilterBenchmark.java
@@ -0,0 +1,102 @@
+/*
+ * 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.benchmark.query;
+
+import it.unimi.dsi.fastutil.ints.Int2ObjectRBTreeMap;
+import org.apache.druid.client.BrokerViewOfCoordinatorConfig;
+import org.apache.druid.client.DruidServer;
+import org.apache.druid.client.QueryableDruidServer;
+import org.apache.druid.query.CloneQueryMode;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordination.TestCoordinatorClient;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Benchmarks {@link BrokerViewOfCoordinatorConfig#getQueryableServers} under
varying thread counts
+ * to measure concurrent read throughput. This is the hot path called
per-segment during
+ * query planning (groupSegmentsByServer, computeResultLevelCachingEtag).
+ *
+ * With the old synchronized implementation, throughput at 32 threads would be
roughly the same as
+ * 1 thread due to monitor contention. With volatile, it should scale linearly.
+ */
+@State(Scope.Benchmark)
+@Fork(value = 1, jvmArgsAppend = "-XX:+UseG1GC")
+@Warmup(iterations = 3)
+@Measurement(iterations = 5)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class BrokerServerFilterBenchmark
+{
+ private BrokerViewOfCoordinatorConfig config;
+ private Int2ObjectRBTreeMap<Set<QueryableDruidServer>> servers;
+
+ @Setup
+ public void setup()
+ {
+ config = new BrokerViewOfCoordinatorConfig(new TestCoordinatorClient());
+ config.start();
+
+ servers = new Int2ObjectRBTreeMap<>();
+ Set<QueryableDruidServer> serverSet = new HashSet<>();
+ for (int i = 0; i < 450; i++) {
+ String host = "historical-" + i;
+ DruidServer druidServer = new DruidServer(host, host, null, 100, null,
ServerType.HISTORICAL, "tier1", 0);
+ serverSet.add(new QueryableDruidServer(druidServer, (QueryRunner)
(queryPlus, responseContext) -> null));
+ }
+ servers.put(0, serverSet);
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void getQueryableServers_1thread(Blackhole blackhole)
+ {
+ blackhole.consume(config.getQueryableServers(servers,
CloneQueryMode.EXCLUDECLONES));
+ }
+
+ @Benchmark
+ @Threads(8)
+ public void getQueryableServers_8threads(Blackhole blackhole)
+ {
+ blackhole.consume(config.getQueryableServers(servers,
CloneQueryMode.EXCLUDECLONES));
+ }
+
+ @Benchmark
+ @Threads(32)
+ public void getQueryableServers_32threads(Blackhole blackhole)
+ {
+ blackhole.consume(config.getQueryableServers(servers,
CloneQueryMode.EXCLUDECLONES));
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/client/BaseBrokerViewOfConfig.java
b/server/src/main/java/org/apache/druid/client/BaseBrokerViewOfConfig.java
index 2f733f1a82d..658c309b105 100644
--- a/server/src/main/java/org/apache/druid/client/BaseBrokerViewOfConfig.java
+++ b/server/src/main/java/org/apache/druid/client/BaseBrokerViewOfConfig.java
@@ -19,7 +19,6 @@
package org.apache.druid.client;
-import com.google.errorprone.annotations.concurrent.GuardedBy;
import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
import org.apache.druid.java.util.common.logger.Logger;
@@ -37,8 +36,10 @@ public abstract class BaseBrokerViewOfConfig<DynamicConfig>
{
private static final Logger log = new Logger(BaseBrokerViewOfConfig.class);
- @GuardedBy("this")
- private DynamicConfig config;
+ // volatile, not synchronized: subclass fields derived from config are read
on the per-segment query
+ // hot path. synchronized causes monitor convoy under high concurrency.
volatile is sufficient since
+ // config is an immutable reference swap.
+ private volatile DynamicConfig config;
/**
* Fetch the configuration from the Coordinator via the HTTP client.
@@ -57,7 +58,7 @@ public abstract class BaseBrokerViewOfConfig<DynamicConfig>
/**
* Return the current dynamic configuration.
*/
- public synchronized DynamicConfig getDynamicConfig()
+ public DynamicConfig getDynamicConfig()
{
return config;
}
@@ -68,7 +69,7 @@ public abstract class BaseBrokerViewOfConfig<DynamicConfig>
*
* @param updatedConfig the new configuration snapshot
*/
- public synchronized void setDynamicConfig(@NotNull DynamicConfig
updatedConfig)
+ public void setDynamicConfig(@NotNull DynamicConfig updatedConfig)
{
config = updatedConfig;
log.info("Updated [%s] dynamic config to [%s]", getConfigTypeName(),
updatedConfig);
diff --git
a/server/src/main/java/org/apache/druid/client/BrokerViewOfCoordinatorConfig.java
b/server/src/main/java/org/apache/druid/client/BrokerViewOfCoordinatorConfig.java
index f09a51a06b3..abd2cda9afe 100644
---
a/server/src/main/java/org/apache/druid/client/BrokerViewOfCoordinatorConfig.java
+++
b/server/src/main/java/org/apache/druid/client/BrokerViewOfCoordinatorConfig.java
@@ -22,7 +22,6 @@ package org.apache.druid.client;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
-import com.google.errorprone.annotations.concurrent.GuardedBy;
import com.google.inject.Inject;
import it.unimi.dsi.fastutil.ints.Int2ObjectRBTreeMap;
import org.apache.druid.client.coordinator.Coordinator;
@@ -54,10 +53,11 @@ public class BrokerViewOfCoordinatorConfig extends
BaseBrokerViewOfConfig<Coordi
{
private final CoordinatorClient coordinatorClient;
- @GuardedBy("this")
- private Set<String> targetCloneServers;
- @GuardedBy("this")
- private Set<String> sourceCloneServers;
+ // volatile, not synchronized: getCurrentServersToIgnore() is called
per-segment during query planning.
+ // Under high concurrency, synchronized causes monitor convoy with 100x
throughput degradation.
+ // Each field is an immutable Set reference, so volatile provides sufficient
visibility.
+ private volatile Set<String> targetCloneServers = Set.of();
+ private volatile Set<String> sourceCloneServers = Set.of();
@Inject
public BrokerViewOfCoordinatorConfig(
@@ -100,7 +100,7 @@ public class BrokerViewOfCoordinatorConfig extends
BaseBrokerViewOfConfig<Coordi
* servers based on the new dynamic configuration.
*/
@Override
- public synchronized void setDynamicConfig(@NotNull CoordinatorDynamicConfig
updatedConfig)
+ public void setDynamicConfig(@NotNull CoordinatorDynamicConfig updatedConfig)
{
super.setDynamicConfig(updatedConfig);
final Map<String, String> cloneServers = updatedConfig.getCloneServers();
@@ -135,8 +135,10 @@ public class BrokerViewOfCoordinatorConfig extends
BaseBrokerViewOfConfig<Coordi
/**
* Get the list of servers that should not be queried based on the
cloneQueryMode parameter.
+ * Each branch reads only one volatile field, so readers may see
targetCloneServers and sourceCloneServers
+ * from different setDynamicConfig() calls — this is acceptable since no
CloneQueryMode needs both.
*/
- private synchronized Set<String> getCurrentServersToIgnore(CloneQueryMode
cloneQueryMode)
+ private Set<String> getCurrentServersToIgnore(CloneQueryMode cloneQueryMode)
{
switch (cloneQueryMode) {
case PREFERCLONES:
diff --git
a/server/src/test/java/org/apache/druid/client/BrokerViewOfCoordinatorConfigTest.java
b/server/src/test/java/org/apache/druid/client/BrokerViewOfCoordinatorConfigTest.java
index 5a014531bf5..3e0ab88a05e 100644
---
a/server/src/test/java/org/apache/druid/client/BrokerViewOfCoordinatorConfigTest.java
+++
b/server/src/test/java/org/apache/druid/client/BrokerViewOfCoordinatorConfigTest.java
@@ -20,14 +20,20 @@
package org.apache.druid.client;
import com.google.common.util.concurrent.Futures;
+import it.unimi.dsi.fastutil.ints.Int2ObjectRBTreeMap;
import org.apache.druid.client.coordinator.CoordinatorClient;
+import org.apache.druid.query.CloneQueryMode;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.server.coordination.ServerType;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
+import java.util.HashSet;
import java.util.Map;
+import java.util.Set;
public class BrokerViewOfCoordinatorConfigTest
{
@@ -55,4 +61,104 @@ public class BrokerViewOfCoordinatorConfigTest
Mockito.verify(coordinatorClient,
Mockito.times(1)).getCoordinatorDynamicConfig();
Assert.assertEquals(config, target.getDynamicConfig());
}
+
+ @Test
+ public void testExcludeClonesFiltersTargetCloneServers()
+ {
+ target.start();
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> servers =
makeServers("host1", "host2", "host3");
+
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> result =
+ target.getQueryableServers(servers, CloneQueryMode.EXCLUDECLONES);
+
+ Set<String> hosts = extractHosts(result);
+ Assert.assertFalse("target clone server host1 should be filtered",
hosts.contains("host1"));
+ Assert.assertTrue("source clone server host2 should remain",
hosts.contains("host2"));
+ Assert.assertTrue("non-clone server host3 should remain",
hosts.contains("host3"));
+ }
+
+ @Test
+ public void testPreferClonesFiltersSourceCloneServers()
+ {
+ target.start();
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> servers =
makeServers("host1", "host2", "host3");
+
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> result =
+ target.getQueryableServers(servers, CloneQueryMode.PREFERCLONES);
+
+ Set<String> hosts = extractHosts(result);
+ Assert.assertTrue("target clone server host1 should remain",
hosts.contains("host1"));
+ Assert.assertFalse("source clone server host2 should be filtered",
hosts.contains("host2"));
+ Assert.assertTrue("non-clone server host3 should remain",
hosts.contains("host3"));
+ }
+
+ @Test
+ public void testIncludeClonesReturnsAll()
+ {
+ target.start();
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> servers =
makeServers("host1", "host2", "host3");
+
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> result =
+ target.getQueryableServers(servers, CloneQueryMode.INCLUDECLONES);
+
+ Assert.assertSame("INCLUDECLONES should return the original map", servers,
result);
+ }
+
+ @Test
+ public void testConfigUpdateChangesFiltering()
+ {
+ target.start();
+
+ CoordinatorDynamicConfig newConfig = CoordinatorDynamicConfig.builder()
+
.withCloneServers(Map.of("host3", "host1"))
+ .build();
+ target.setDynamicConfig(newConfig);
+
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> servers =
makeServers("host1", "host2", "host3");
+
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> result =
+ target.getQueryableServers(servers, CloneQueryMode.EXCLUDECLONES);
+
+ Set<String> hosts = extractHosts(result);
+ Assert.assertFalse("new target clone host3 should be filtered",
hosts.contains("host3"));
+ Assert.assertTrue("host1 is now source, should remain",
hosts.contains("host1"));
+ Assert.assertTrue("host2 is unrelated, should remain",
hosts.contains("host2"));
+ }
+
+ /**
+ * Creates a priority-to-servers map with all servers at priority 0.
+ *
+ * @param hosts host names to create historical servers for
+ * @return map of priority to queryable server set, matching the structure
used by
+ * {@link BrokerViewOfCoordinatorConfig#getQueryableServers}
+ */
+ private static Int2ObjectRBTreeMap<Set<QueryableDruidServer>>
makeServers(String... hosts)
+ {
+ Int2ObjectRBTreeMap<Set<QueryableDruidServer>> map = new
Int2ObjectRBTreeMap<>();
+ Set<QueryableDruidServer> serverSet = new HashSet<>();
+ for (String host : hosts) {
+ DruidServer druidServer = new DruidServer(host, host, null, 100, null,
ServerType.HISTORICAL, "tier1", 0);
+ serverSet.add(new QueryableDruidServer(druidServer,
Mockito.mock(QueryRunner.class)));
+ }
+ map.put(0, serverSet);
+ return map;
+ }
+
+ /**
+ * Flattens the priority-to-servers map into a set of host names for easy
assertion.
+ *
+ * @param servers priority-to-servers map returned by {@link
BrokerViewOfCoordinatorConfig#getQueryabl
+ eServers}
+ * @return set of host names across all priority levels
+ */
+ private static Set<String>
extractHosts(Int2ObjectRBTreeMap<Set<QueryableDruidServer>> servers)
+ {
+ Set<String> hosts = new HashSet<>();
+ for (Set<QueryableDruidServer> serverSet : servers.values()) {
+ for (QueryableDruidServer server : serverSet) {
+ hosts.add(server.getServer().getHost());
+ }
+ }
+ return hosts;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
