This is an automated email from the ASF dual-hosted git repository.
kgyrtkirk 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 63db6623bfc feat: Make Coordinator client maxAttempts configurable
(#20442)
63db6623bfc is described below
commit 63db6623bfcf17712bf00369b76f0d8e3e2bf3fe
Author: Zoltan Haindrich <[email protected]>
AuthorDate: Tue Oct 27 10:08:13 2026 +0100
feat: Make Coordinator client maxAttempts configurable (#20442)
---
docs/configuration/index.md | 1 +
.../druid/client/BrokerViewOfBrokerConfig.java | 31 -------------
.../client/BrokerViewOfCoordinatorConfig.java | 28 ------------
.../coordinator/CoordinatorClientConfig.java | 41 +++++++++++++++++
.../druid/rpc/guice/ServiceClientModule.java | 9 ++--
.../druid/rpc/guice/ServiceClientModuleTest.java | 52 +++++++++++++++++++++-
.../testing/embedded/EmbeddedServiceClient.java | 4 +-
website/.spelling | 1 +
8 files changed, 103 insertions(+), 64 deletions(-)
diff --git a/docs/configuration/index.md b/docs/configuration/index.md
index 7745e41bd00..de05cec50d4 100644
--- a/docs/configuration/index.md
+++ b/docs/configuration/index.md
@@ -707,6 +707,7 @@ All Druid components can communicate with each other over
HTTP.
|`druid.global.http.connectTimeout`|Connect timeout for the HTTP client used
for most direct RPC between Druid services. This covers, among other things,
Overlord-to-task and supervisor-to-task calls in the indexing service,
Coordinator lookup management, dynamic config sync between services, MSQ tasks
reading from data servers, and general Coordinator/Overlord/Broker service
clients. Does not affect Broker-to-Historical query dispatch (see
`druid.broker.http.connectTimeout`) or request [...]
|`druid.global.http.allocator`|Netty memory allocator used by the direct-RPC
HTTP client. Accepts `adaptive` (adaptive between `pooled` and `unpooled` based
on load), `pooled`, or `unpooled`.|`adaptive`|
|`druid.global.http.poolImplementation`|How the connection pool tracks demand,
never exceeding `numConnections` either way. With `adaptive`, a request
discards every stale or broken connection it walks past and opens a new one
only once none is left, so the pool falls back to the number of connections the
traffic actually needs. With `retaining`, the pool holds on to every connection
it has opened, replacing a stale or broken one by a fresh one, one for one, so
it stays at its high-water [...]
+|`druid.client.coordinator.maxAttempts`|Maximum number of attempts, including
the first one, that a service makes for a request to the Coordinator before
giving up on a retryable error. Must be at least 1.|`15`|
### Common endpoints configuration
diff --git
a/server/src/main/java/org/apache/druid/client/BrokerViewOfBrokerConfig.java
b/server/src/main/java/org/apache/druid/client/BrokerViewOfBrokerConfig.java
index 3129a020393..0831092b088 100644
--- a/server/src/main/java/org/apache/druid/client/BrokerViewOfBrokerConfig.java
+++ b/server/src/main/java/org/apache/druid/client/BrokerViewOfBrokerConfig.java
@@ -19,22 +19,12 @@
package org.apache.druid.client;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.annotations.VisibleForTesting;
import com.google.inject.Inject;
import jakarta.validation.constraints.NotNull;
-import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.coordinator.CoordinatorClient;
-import org.apache.druid.client.coordinator.CoordinatorClientImpl;
-import org.apache.druid.discovery.NodeRole;
-import org.apache.druid.guice.annotations.EscalatedGlobal;
-import org.apache.druid.guice.annotations.Json;
import org.apache.druid.query.DefaultQueryConfig;
import org.apache.druid.query.QueryConfigProvider;
import org.apache.druid.query.QueryContext;
-import org.apache.druid.rpc.ServiceClientFactory;
-import org.apache.druid.rpc.ServiceLocator;
-import org.apache.druid.rpc.StandardRetryPolicy;
import org.apache.druid.server.broker.BrokerDynamicConfig;
import org.apache.druid.server.broker.QueryConfigSnapshot;
@@ -63,27 +53,6 @@ public class BrokerViewOfBrokerConfig extends
BaseBrokerViewOfConfig<BrokerDynam
private volatile QueryConfigSnapshot querySnapshot;
@Inject
- public BrokerViewOfBrokerConfig(
- @Json final ObjectMapper jsonMapper,
- @EscalatedGlobal final ServiceClientFactory clientFactory,
- @Coordinator final ServiceLocator serviceLocator,
- final DefaultQueryConfig defaultQueryConfig
- )
- {
- this.defaultQueryConfig = defaultQueryConfig;
- this.querySnapshot = new
QueryConfigSnapshot(QueryContext.of(defaultQueryConfig.getContext()).asMap(),
null);
- this.coordinatorClient =
- new CoordinatorClientImpl(
- clientFactory.makeClient(
- NodeRole.COORDINATOR.getJsonName(),
- serviceLocator,
- StandardRetryPolicy.builder().maxAttempts(15).build()
- ),
- jsonMapper
- );
- }
-
- @VisibleForTesting
public BrokerViewOfBrokerConfig(
final CoordinatorClient coordinatorClient,
final DefaultQueryConfig defaultQueryConfig
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 20f25784086..404323da853 100644
---
a/server/src/main/java/org/apache/druid/client/BrokerViewOfCoordinatorConfig.java
+++
b/server/src/main/java/org/apache/druid/client/BrokerViewOfCoordinatorConfig.java
@@ -19,24 +19,14 @@
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.inject.Inject;
import it.unimi.dsi.fastutil.ints.Int2ObjectRBTreeMap;
import jakarta.validation.constraints.NotNull;
-import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.coordinator.CoordinatorClient;
-import org.apache.druid.client.coordinator.CoordinatorClientImpl;
import org.apache.druid.client.selector.HistoricalFilter;
-import org.apache.druid.discovery.NodeRole;
import org.apache.druid.error.DruidException;
-import org.apache.druid.guice.annotations.EscalatedGlobal;
-import org.apache.druid.guice.annotations.Json;
import org.apache.druid.query.CloneQueryMode;
-import org.apache.druid.rpc.ServiceClientFactory;
-import org.apache.druid.rpc.ServiceLocator;
-import org.apache.druid.rpc.StandardRetryPolicy;
import org.apache.druid.server.BrokerDynamicConfigResource;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
@@ -60,24 +50,6 @@ public class BrokerViewOfCoordinatorConfig extends
BaseBrokerViewOfConfig<Coordi
private volatile Set<String> sourceCloneServers = Set.of();
@Inject
- public BrokerViewOfCoordinatorConfig(
- @Json final ObjectMapper jsonMapper,
- @EscalatedGlobal final ServiceClientFactory clientFactory,
- @Coordinator final ServiceLocator serviceLocator
- )
- {
- this.coordinatorClient =
- new CoordinatorClientImpl(
- clientFactory.makeClient(
- NodeRole.COORDINATOR.getJsonName(),
- serviceLocator,
- StandardRetryPolicy.builder().maxAttempts(15).build()
- ),
- jsonMapper
- );
- }
-
- @VisibleForTesting
public BrokerViewOfCoordinatorConfig(CoordinatorClient coordinatorClient)
{
this.coordinatorClient = coordinatorClient;
diff --git
a/server/src/main/java/org/apache/druid/client/coordinator/CoordinatorClientConfig.java
b/server/src/main/java/org/apache/druid/client/coordinator/CoordinatorClientConfig.java
new file mode 100644
index 00000000000..448727c6931
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/client/coordinator/CoordinatorClientConfig.java
@@ -0,0 +1,41 @@
+/*
+ * 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.coordinator;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.constraints.Min;
+
+/**
+ * Configuration of the client that other services use to call the Coordinator.
+ */
+public class CoordinatorClientConfig
+{
+ /**
+ * Attempts per request to the Coordinator, the first one included, before
giving up on a retryable error.
+ */
+ @JsonProperty
+ @Min(1)
+ private int maxAttempts = 15;
+
+ public int getMaxAttempts()
+ {
+ return maxAttempts;
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/rpc/guice/ServiceClientModule.java
b/server/src/main/java/org/apache/druid/rpc/guice/ServiceClientModule.java
index 203886f99fe..1dbca72c24b 100644
--- a/server/src/main/java/org/apache/druid/rpc/guice/ServiceClientModule.java
+++ b/server/src/main/java/org/apache/druid/rpc/guice/ServiceClientModule.java
@@ -27,10 +27,12 @@ import org.apache.druid.client.broker.BrokerClient;
import org.apache.druid.client.broker.BrokerClientImpl;
import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.coordinator.CoordinatorClient;
+import org.apache.druid.client.coordinator.CoordinatorClientConfig;
import org.apache.druid.client.coordinator.CoordinatorClientImpl;
import org.apache.druid.client.indexing.IndexingService;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.guice.JsonConfigProvider;
import org.apache.druid.guice.LazySingleton;
import org.apache.druid.guice.ManageLifecycle;
import org.apache.druid.guice.annotations.EscalatedGlobal;
@@ -57,7 +59,7 @@ public class ServiceClientModule implements DruidModule
@Override
public void configure(Binder binder)
{
- // Nothing to do.
+ JsonConfigProvider.bind(binder, "druid.client.coordinator",
CoordinatorClientConfig.class);
}
@Provides
@@ -112,13 +114,14 @@ public class ServiceClientModule implements DruidModule
@Coordinator
public ServiceClient makeServiceClientForCoordinator(
@EscalatedGlobal final ServiceClientFactory clientFactory,
- @Coordinator final ServiceLocator serviceLocator
+ @Coordinator final ServiceLocator serviceLocator,
+ final CoordinatorClientConfig config
)
{
return clientFactory.makeClient(
NodeRole.COORDINATOR.getJsonName(),
serviceLocator,
- StandardRetryPolicy.builder().maxAttempts(CLIENT_MAX_ATTEMPTS).build()
+
StandardRetryPolicy.builder().maxAttempts(config.getMaxAttempts()).build()
);
}
diff --git
a/server/src/test/java/org/apache/druid/rpc/guice/ServiceClientModuleTest.java
b/server/src/test/java/org/apache/druid/rpc/guice/ServiceClientModuleTest.java
index f667cb8db5f..033485dc575 100644
---
a/server/src/test/java/org/apache/druid/rpc/guice/ServiceClientModuleTest.java
+++
b/server/src/test/java/org/apache/druid/rpc/guice/ServiceClientModuleTest.java
@@ -23,12 +23,16 @@ import com.google.common.collect.ImmutableList;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.Key;
+import com.google.inject.ProvisionException;
import org.apache.druid.client.broker.Broker;
import org.apache.druid.client.broker.BrokerClient;
import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.coordinator.CoordinatorClient;
+import org.apache.druid.client.coordinator.CoordinatorClientConfig;
import org.apache.druid.client.indexing.IndexingService;
import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.guice.ConfigModule;
import org.apache.druid.guice.DruidGuiceExtensions;
import org.apache.druid.guice.LifecycleModule;
import org.apache.druid.guice.annotations.EscalatedGlobal;
@@ -37,16 +41,22 @@ import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.rpc.ServiceClient;
import org.apache.druid.rpc.ServiceClientFactory;
import org.apache.druid.rpc.ServiceLocator;
+import org.apache.druid.rpc.ServiceRetryPolicy;
import org.apache.druid.rpc.indexing.OverlordClient;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatchers;
import org.mockito.Mock;
+import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.util.Properties;
+
@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.WARN)
public class ServiceClientModuleTest
@@ -68,9 +78,17 @@ public class ServiceClientModuleTest
@BeforeEach
public void setUp()
{
- injector = Guice.createInjector(
+ injector = makeInjector("4");
+ }
+
+ private Injector makeInjector(final String coordinatorMaxAttempts)
+ {
+ final Properties properties = new Properties();
+ properties.setProperty("druid.client.coordinator.maxAttempts",
coordinatorMaxAttempts);
+ return Guice.createInjector(
ImmutableList.of(
new DruidGuiceExtensions(),
+ new ConfigModule(),
new LifecycleModule(),
new JacksonModule(),
new ServiceClientModule(),
@@ -79,6 +97,7 @@ public class ServiceClientModuleTest
binder.bind(ServiceLocator.class).toInstance(serviceLocator);
binder.bind(DruidNodeDiscoveryProvider.class).toInstance(discoveryProvider);
binder.bind(ServiceClientFactory.class).toInstance(serviceClientFactory);
+ binder.bind(Properties.class).toInstance(properties);
}
)
);
@@ -102,6 +121,37 @@ public class ServiceClientModuleTest
Assertions.assertNotNull(injector.getInstance(CoordinatorClient.class));
}
+ @Test
+ public void testCoordinatorClientConfigRejectsNonPositiveMaxAttempts()
+ {
+ Assertions.assertThrows(
+ ProvisionException.class,
+ () -> makeInjector("0").getInstance(CoordinatorClientConfig.class)
+ );
+ Assertions.assertThrows(
+ ProvisionException.class,
+ () -> makeInjector("-1").getInstance(CoordinatorClientConfig.class)
+ );
+ }
+
+ @Test
+ public void testCoordinatorRetryPolicyUsesConfiguredMaxAttempts()
+ {
+ new ServiceClientModule().makeServiceClientForCoordinator(
+ serviceClientFactory,
+ serviceLocator,
+ injector.getInstance(CoordinatorClientConfig.class)
+ );
+
+ final ArgumentCaptor<ServiceRetryPolicy> retryPolicy =
ArgumentCaptor.forClass(ServiceRetryPolicy.class);
+ Mockito.verify(serviceClientFactory).makeClient(
+ ArgumentMatchers.eq(NodeRole.COORDINATOR.getJsonName()),
+ ArgumentMatchers.eq(serviceLocator),
+ retryPolicy.capture()
+ );
+ Assertions.assertEquals(4, retryPolicy.getValue().maxAttempts());
+ }
+
@Test
public void testGetBrokerClient()
{
diff --git
a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java
b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java
index 932b82cb41a..c8f3c24eee6 100644
---
a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java
+++
b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java
@@ -28,6 +28,7 @@ import org.apache.druid.client.broker.BrokerClient;
import org.apache.druid.client.broker.BrokerClientImpl;
import org.apache.druid.client.coordinator.Coordinator;
import org.apache.druid.client.coordinator.CoordinatorClient;
+import org.apache.druid.client.coordinator.CoordinatorClientConfig;
import org.apache.druid.client.indexing.IndexingService;
import org.apache.druid.common.guava.FutureUtils;
import org.apache.druid.discovery.NodeRole;
@@ -105,7 +106,8 @@ public class EmbeddedServiceClient
);
this.coordinatorServiceClient = module.makeServiceClientForCoordinator(
factory,
- anyServer.bindings().getInstance(ServiceLocator.class,
Coordinator.class)
+ anyServer.bindings().getInstance(ServiceLocator.class,
Coordinator.class),
+ new CoordinatorClientConfig()
);
}
diff --git a/website/.spelling b/website/.spelling
index 14b88a809d2..8c913436eb6 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -2650,6 +2650,7 @@ SRE
reachability
ownership_mismatch
giveup
+retryable
- ../docs/development/extensions-contrib/spectator-histogram.md
SpectatorHistogram
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
