feat: Make Coordinator client maxAttempts configurable

docs/configuration/index.md

@@ -706,6 +706,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 forwarding (see `clientConnectTimeout`).|`PT10S`|
 |`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 mark.|`retaining`|
+|`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
 

server/src/main/java/org/apache/druid/client/BrokerViewOfBrokerConfig.java

@@ -19,23 +19,13 @@
 
 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.query.QueryContexts;
-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;
 
@@ -64,27 +54,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

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;

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;
+  }
+}

server/src/main/java/org/apache/druid/rpc/guice/ServiceClientModule.java

@@ -27,10 +27,12 @@
 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 ServiceLocator makeCoordinatorServiceLocator(final DruidNodeDiscoveryProv
   @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()
     );
   }
 

server/src/test/java/org/apache/druid/rpc/guice/ServiceClientModuleTest.java

@@ -23,12 +23,16 @@
 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.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 void setUp()
             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 void testGetCoordinatorClient()
     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()
   {

services/src/test/java/org/apache/druid/testing/embedded/EmbeddedServiceClient.java

@@ -28,6 +28,7 @@
 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 @@ private EmbeddedServiceClient(EmbeddedDruidCluster cluster, Escalator escalator)
     );
     this.coordinatorServiceClient = module.makeServiceClientForCoordinator(
         factory,
-        anyServer.bindings().getInstance(ServiceLocator.class, Coordinator.class)
+        anyServer.bindings().getInstance(ServiceLocator.class, Coordinator.class),
+        new CoordinatorClientConfig()
     );
   }
 

website/.spelling

@@ -2630,6 +2630,7 @@ SRE
 reachability
 ownership_mismatch
 giveup
+retryable
 
 - ../docs/development/extensions-contrib/spectator-histogram.md
 SpectatorHistogram