fix: make BrokerDynamicConfig backwards-compatible (#19120)

Make BrokerDynamicConfig backwards-compatible. By default, brokers who poll coordinators without the broker dynamic config endpoint (and receive a 404) will return an empty BrokerDynamicConfig.

Author: jtuglu1

server/src/main/java/org/apache/druid/client/coordinator/CoordinatorClient.java

@@ -105,6 +105,7 @@ public interface CoordinatorClient
   /**
    * Gets the latest broker dynamic config from the Coordinator.
    * Brokers use this to fetch their configuration from the Coordinator on startup.
+   * Returns an empty {@link BrokerDynamicConfig} when querying older coordinators to ensure backwards-compatibility.
    * <p>
    * API: {@code GET /druid/coordinator/v1/broker/config}
    */

server/src/main/java/org/apache/druid/client/coordinator/CoordinatorClientImpl.java

@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.apache.druid.client.BootstrapSegmentsResponse;
 import org.apache.druid.client.ImmutableSegmentLoadInfo;
 import org.apache.druid.client.JsonParserIterator;
@@ -37,6 +38,7 @@
 import org.apache.druid.query.SegmentDescriptor;
 import org.apache.druid.query.lookup.LookupExtractorFactoryContainer;
 import org.apache.druid.query.lookup.LookupUtils;
+import org.apache.druid.rpc.HttpResponseException;
 import org.apache.druid.rpc.IgnoreHttpResponseHandler;
 import org.apache.druid.rpc.RequestBuilder;
 import org.apache.druid.rpc.ServiceClient;
@@ -257,16 +259,23 @@ public ListenableFuture<CoordinatorDynamicConfig> getCoordinatorDynamicConfig()
   @Override
   public ListenableFuture<BrokerDynamicConfig> getBrokerDynamicConfig()
   {
-    return FutureUtils.transform(
-        client.asyncRequest(
-            new RequestBuilder(HttpMethod.GET, "/druid/coordinator/v1/broker/config"),
-            new BytesFullResponseHandler()
+    // Older coordinators may not support this endpoint; fall back to an empty config on 404.
+    return Futures.catching(
+        FutureUtils.transform(
+            client.asyncRequest(
+                new RequestBuilder(HttpMethod.GET, "/druid/coordinator/v1/broker/config"),
+                new BytesFullResponseHandler()
+            ),
+            holder -> JacksonUtils.readValue(jsonMapper, holder.getContent(), BrokerDynamicConfig.class)
         ),
-        holder -> JacksonUtils.readValue(
-            jsonMapper,
-            holder.getContent(),
-            BrokerDynamicConfig.class
-        )
+        HttpResponseException.class,
+        e -> {
+          if (e != null && e.getResponse().getStatus().getCode() == 404) {
+            return BrokerDynamicConfig.builder().build();
+          }
+          throw new RuntimeException(e);
+        },
+        MoreExecutors.directExecutor()
     );
   }
 

server/src/test/java/org/apache/druid/client/coordinator/CoordinatorClientImplTest.java

@@ -49,6 +49,8 @@
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.RowSignature;
 import org.apache.druid.segment.metadata.DataSourceInformation;
+import org.apache.druid.server.QueryBlocklistRule;
+import org.apache.druid.server.broker.BrokerDynamicConfig;
 import org.apache.druid.server.compaction.CompactionStatusResponse;
 import org.apache.druid.server.coordination.DruidServerMetadata;
 import org.apache.druid.server.coordination.ServerType;
@@ -832,4 +834,37 @@ public void test_updateRulesForDatasource() throws Exception
 
     Assert.assertNull(coordinatorClient.updateRulesForDatasource("xyz", rules).get());
   }
+
+  @Test
+  public void test_getBrokerDynamicConfig() throws Exception
+  {
+    final BrokerDynamicConfig brokerDynamicConfig = BrokerDynamicConfig.builder().withQueryBlocklist(
+        List.of(
+            new QueryBlocklistRule("test", Set.of("dataSource"), null, null)
+        )
+    ).build();
+
+    serviceClient.expectAndRespond(
+        new RequestBuilder(HttpMethod.GET, "/druid/coordinator/v1/broker/config"),
+        HttpResponseStatus.OK,
+        ImmutableMap.of(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON),
+        jsonMapper.writeValueAsBytes(brokerDynamicConfig)
+    );
+    Assert.assertEquals(brokerDynamicConfig, coordinatorClient.getBrokerDynamicConfig().get());
+  }
+
+  @Test
+  public void test_getBrokerDynamicConfig_backwardsCompatibleCoordinatorRequest() throws Exception
+  {
+    serviceClient.expectAndThrow(
+        new RequestBuilder(HttpMethod.GET, "/druid/coordinator/v1/broker/config"),
+        new HttpResponseException(
+            new StringFullResponseHolder(
+                new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND),
+                StandardCharsets.UTF_8
+            )
+        )
+    );
+    Assert.assertEquals(BrokerDynamicConfig.builder().build(), coordinatorClient.getBrokerDynamicConfig().get());
+  }
 }