feat: add etag conditional write support to dynamic config

Author: jtuglu1

docs/api-reference/automatic-compaction-api.md

@@ -37,6 +37,14 @@ Instead of the automatic compaction API, you can use the supervisor API to submi
 
 In this topic, `http://ROUTER_IP:ROUTER_PORT` is a placeholder for your Router service address and port. Replace it with the information for your deployment. For example, use `http://localhost:8888` for quickstart deployments.
 
+## Concurrency control with ETag and If-Match
+
+The automatic compaction write endpoints support optimistic concurrency control using HTTP `ETag` and `If-Match`. `GET /druid/coordinator/v1/config/compaction` returns an `ETag` header whose value is a stable hash of the underlying compaction configuration document (the union of cluster-level config and all per-datasource configs).
+
+To guard a write against concurrent updates, pass the most recently observed ETag back in an `If-Match` header on `POST` or `DELETE`. The change only commits if the stored configuration still hashes to that ETag; otherwise the request fails with `412 Precondition Failed`. Re-`GET` to obtain the new value and ETag, re-apply your change, and retry. The `If-Match` header is optional; requests without it preserve the previous last-writer-wins behavior. `If-Match: *` matches any existing configuration.
+
+Endpoints with this behavior: `POST /druid/coordinator/v1/config/compaction`, `DELETE /druid/coordinator/v1/config/compaction/{dataSource}`, `POST /druid/coordinator/v1/config/compaction/taskslots`, and the unified `POST /druid/indexer/v1/compaction/config/cluster`. See the equivalent section in the [Dynamic configuration API](./dynamic-configuration-api.md#concurrency-control-with-etag-and-if-match) for the general protocol and a client-flow example.
+
 ## Manage automatic compaction
 
 ### Create or update automatic compaction configuration
@@ -51,6 +59,12 @@ Note that this endpoint returns an HTTP `200 OK` message code even if the dataso
 
 `POST` `/druid/coordinator/v1/config/compaction`
 
+#### Header parameters
+
+* `If-Match`
+  * Type: String
+  * Optional. Quoted ETag previously returned by `GET /druid/coordinator/v1/config/compaction`. When supplied, the update only commits if the stored compaction configuration still matches this ETag. Pass `*` to require only that some value is already stored. See [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
+
 #### Responses
 
 <Tabs>
@@ -60,6 +74,12 @@ Note that this endpoint returns an HTTP `200 OK` message code even if the dataso
 
 *Successfully submitted auto compaction configuration*
 
+</TabItem>
+<TabItem value="1b" label="412 PRECONDITION FAILED">
+
+
+*The `If-Match` header did not match the currently stored configuration, or another writer committed a change between this request's precondition check and write. Re-read the configuration and retry.*
+
 </TabItem>
 </Tabs>
 
@@ -140,6 +160,12 @@ Removes the automatic compaction configuration for a datasource. This updates th
 
 `DELETE` `/druid/coordinator/v1/config/compaction/{dataSource}`
 
+#### Header parameters
+
+* `If-Match`
+  * Type: String
+  * Optional. Quoted ETag previously returned by `GET /druid/coordinator/v1/config/compaction`. When supplied, the delete only commits if the stored compaction configuration still matches this ETag. See [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
+
 #### Responses
 
 <Tabs>
@@ -155,6 +181,12 @@ Removes the automatic compaction configuration for a datasource. This updates th
 
 *Datasource does not have automatic compaction or invalid datasource name*
 
+</TabItem>
+<TabItem value="5b" label="412 PRECONDITION FAILED">
+
+
+*The `If-Match` header did not match the currently stored configuration, or another writer committed a change between this request's precondition check and write. Re-read the configuration and retry.*
+
 </TabItem>
 </Tabs>
 
@@ -215,6 +247,12 @@ To limit the maximum number of compaction tasks, use the optional query paramete
   * Default: 2147483647
   * Limits the maximum number of task slots for compaction tasks.
 
+#### Header parameters
+
+* `If-Match`
+  * Type: String
+  * Optional. Quoted ETag previously returned by `GET /druid/coordinator/v1/config/compaction`. When supplied, the update only commits if the stored compaction configuration still matches this ETag. See [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
+
 #### Responses
 
 <Tabs>
@@ -230,6 +268,12 @@ To limit the maximum number of compaction tasks, use the optional query paramete
 
 *Invalid `max` value*
 
+</TabItem>
+<TabItem value="9b" label="412 PRECONDITION FAILED">
+
+
+*The `If-Match` header did not match the currently stored configuration, or another writer committed a change between this request's precondition check and write. Re-read the configuration and retry.*
+
 </TabItem>
 </Tabs>
 
@@ -270,6 +314,8 @@ Retrieves all automatic compaction configurations. Returns a `compactionConfigs`
 
 You can use this endpoint to retrieve `compactionTaskSlotRatio` and `maxCompactionTaskSlots` values for managing resource allocation of compaction tasks.
 
+The response includes an `ETag` header that you can pass back in `If-Match` on a subsequent write to detect concurrent updates; see [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
+
 #### URL
 
 `GET` `/druid/coordinator/v1/config/compaction`
@@ -281,7 +327,7 @@ You can use this endpoint to retrieve `compactionTaskSlotRatio` and `maxCompacti
 <TabItem value="12" label="200 SUCCESS">
 
 
-*Successfully retrieved automatic compaction configurations*
+*Successfully retrieved automatic compaction configurations. The `ETag` response header carries an opaque identifier for the returned configuration version.*
 
 </TabItem>
 </Tabs>
@@ -926,6 +972,12 @@ This policy specifies the datasources and intervals eligible for compaction and
 
 `POST` `/druid/indexer/v1/compaction/config/cluster`
 
+#### Header parameters
+
+* `If-Match`
+  * Type: String
+  * Optional. Quoted ETag previously returned by `GET /druid/indexer/v1/compaction/config/cluster` (or `GET /druid/coordinator/v1/config/compaction`, since both reflect the same underlying configuration). When supplied, the update only commits if the stored configuration still matches this ETag. See [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
+
 #### Responses
 
 <Tabs>
@@ -941,6 +993,12 @@ This policy specifies the datasources and intervals eligible for compaction and
 
 *Invalid `max` value*
 
+</TabItem>
+<TabItem value="9b" label="412 PRECONDITION FAILED">
+
+
+*The `If-Match` header did not match the currently stored configuration, or another writer committed a change between this request's precondition check and write. Re-read the configuration and retry.*
+
 </TabItem>
 </Tabs>
 
@@ -1002,6 +1060,8 @@ A successful request returns an HTTP `200 OK` message code and an empty response
 Retrieves cluster-level configuration for compaction tasks which applies to all datasources, unless explicitly overridden in the datasource compaction config.
 This includes all the fields listed in [Update cluster-level compaction config](#update-cluster-level-compaction-config).
 
+The response includes an `ETag` header that you can pass back in `If-Match` on a subsequent write; see [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
+
 #### URL
 
 `GET` `/druid/indexer/v1/compaction/config/cluster`
@@ -1012,7 +1072,7 @@ This includes all the fields listed in [Update cluster-level compaction config](
 
 <TabItem value="8" label="200 SUCCESS">
 
-*Successfully retrieved cluster compaction configuration*
+*Successfully retrieved cluster compaction configuration. The `ETag` response header carries an opaque identifier for the returned configuration version.*
 
 </TabItem>
 </Tabs>

docs/api-reference/dynamic-configuration-api.md

@@ -33,14 +33,40 @@ In this topic, `http://ROUTER_IP:ROUTER_PORT` is a placeholder for your Router s
 Replace it with the information for your deployment.
 For example, use `http://localhost:8888` for quickstart deployments.
 
+## Concurrency control with ETag and If-Match
+
+Without coordination, two concurrent updates to the same dynamic configuration silently overwrite each other—the last writer wins and the earlier writer has no signal that their change was lost. To prevent this, the dynamic configuration endpoints support an optimistic-concurrency-control protocol modeled on HTTP `ETag` and `If-Match`:
+
+* Every `GET` response for a dynamic configuration includes an `ETag` header. The value is a stable hash of the stored configuration bytes—identical bytes always produce the same ETag.
+* On a `POST` (or `DELETE`), supply the ETag you last observed in an `If-Match` header. The server only commits the change if the currently stored configuration still hashes to that ETag.
+* If another writer changed the configuration after your read, your request fails with `412 Precondition Failed`. Re-`GET` to obtain the new value and ETag, re-apply your change, and retry.
+* The `If-Match` header is **optional**. Requests without it preserve the previous last-writer-wins behavior.
+* `If-Match: *` is supported and matches any existing configuration (it only fails if no value has been stored yet).
+
+Typical client flow:
+
+```shell
+# Read current config and capture its ETag
+ETAG=$(curl -sD - "http://ROUTER_IP:ROUTER_PORT/druid/coordinator/v1/config" | awk '/^ETag:/ {print $2}' | tr -d '\r')
+
+# Submit an update guarded by the ETag
+curl -X POST "http://ROUTER_IP:ROUTER_PORT/druid/coordinator/v1/config" \
+  -H "Content-Type: application/json" \
+  -H "If-Match: $ETAG" \
+  -d @new-config.json
+# 200 on success, 412 if someone else updated the config in between.
+```
+
+Endpoints that support this protocol are noted individually below.
+
 ## Coordinator dynamic configuration
 
 The Coordinator has dynamic configurations to tune certain behavior on the fly, without requiring a service restart.
 For information on the supported properties, see [Coordinator dynamic configuration](../configuration/index.md#dynamic-configuration).
 
 ### Get dynamic configuration
 
-Retrieves the current Coordinator dynamic configuration. Returns a JSON object with the dynamic configuration properties.
+Retrieves the current Coordinator dynamic configuration. Returns a JSON object with the dynamic configuration properties. The response includes an `ETag` header that can be used with `If-Match` on a subsequent update; see [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
 
 #### URL
 
@@ -53,7 +79,7 @@ Retrieves the current Coordinator dynamic configuration. Returns a JSON object w
 <TabItem value="1" label="200 SUCCESS">
 
 
-*Successfully retrieved dynamic configuration*
+*Successfully retrieved dynamic configuration. The `ETag` response header carries an opaque identifier for the returned configuration version.*
 
 </TabItem>
 </Tabs>
@@ -136,6 +162,9 @@ The endpoint supports a set of optional header parameters to populate the `autho
 * `X-Druid-Comment`
   * Type: String
   * Description for the update.
+* `If-Match`
+  * Type: String
+  * Optional. Quoted ETag previously returned by `GET /druid/coordinator/v1/config`. When supplied, the update only commits if the currently stored configuration still matches this ETag. Pass `*` to require only that some value is already stored. See [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
 
 #### Responses
 
@@ -146,6 +175,12 @@ The endpoint supports a set of optional header parameters to populate the `autho
 
 *Successfully updated dynamic configuration*
 
+</TabItem>
+<TabItem value="4b" label="412 PRECONDITION FAILED">
+
+
+*The `If-Match` header did not match the currently stored configuration, or another writer committed a change between this request's precondition check and write. Re-read the configuration and retry.*
+
 </TabItem>
 </Tabs>
 
@@ -319,7 +354,7 @@ These settings control broker behavior such as query blocking rules and default
 
 ### Get broker dynamic configuration
 
-Retrieves the current Broker dynamic configuration. Returns a JSON object with the dynamic configuration properties.
+Retrieves the current Broker dynamic configuration. Returns a JSON object with the dynamic configuration properties. The response includes an `ETag` header that can be used with `If-Match` on a subsequent update; see [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
 
 #### URL
 
@@ -332,7 +367,7 @@ Retrieves the current Broker dynamic configuration. Returns a JSON object with t
 <TabItem value="25" label="200 SUCCESS">
 
 
-*Successfully retrieved broker dynamic configuration*
+*Successfully retrieved broker dynamic configuration. The `ETag` response header carries an opaque identifier for the returned configuration version.*
 
 </TabItem>
 </Tabs>
@@ -412,6 +447,10 @@ The endpoint supports a set of optional header parameters to populate the audit
   * Type: String
   * Comment describing the change.
 
+* `If-Match`
+  * Type: String
+  * Optional. Quoted ETag previously returned by `GET /druid/coordinator/v1/broker/config`. When supplied, the update only commits if the currently stored configuration still matches this ETag. Pass `*` to require only that some value is already stored. See [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
+
 #### Responses
 
 <Tabs>
@@ -421,6 +460,12 @@ The endpoint supports a set of optional header parameters to populate the audit
 
 *Successfully updated configuration*
 
+</TabItem>
+<TabItem value="28b" label="412 PRECONDITION FAILED">
+
+
+*The `If-Match` header did not match the currently stored configuration, or another writer committed a change between this request's precondition check and write. Re-read the configuration and retry.*
+
 </TabItem>
 </Tabs>
 
@@ -719,6 +764,7 @@ For information on the supported properties, see [Overlord dynamic configuration
 Retrieves the current Overlord dynamic configuration.
 Returns a JSON object with the dynamic configuration properties.
 Returns an empty response body if there is no current Overlord dynamic configuration.
+When a configuration is present, the response includes an `ETag` header that can be used with `If-Match` on a subsequent update; see [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
 
 #### URL
 
@@ -731,7 +777,7 @@ Returns an empty response body if there is no current Overlord dynamic configura
 <TabItem value="10" label="200 SUCCESS">
 
 
-*Successfully retrieved dynamic configuration*
+*Successfully retrieved dynamic configuration. When a configuration is stored, the `ETag` response header carries an opaque identifier for the returned configuration version.*
 
 </TabItem>
 </Tabs>
@@ -799,6 +845,9 @@ The endpoint supports a set of optional header parameters to populate the `autho
 * `X-Druid-Comment`
   * Type: String
   * Description for the update.
+* `If-Match`
+  * Type: String
+  * Optional. Quoted ETag previously returned by `GET /druid/indexer/v1/worker`. When supplied, the update only commits if the currently stored configuration still matches this ETag. Pass `*` to require only that some value is already stored. See [Concurrency control with ETag and If-Match](#concurrency-control-with-etag-and-if-match).
 
 #### Responses
 
@@ -809,6 +858,12 @@ The endpoint supports a set of optional header parameters to populate the `autho
 
 *Successfully updated dynamic configuration*
 
+</TabItem>
+<TabItem value="13b" label="412 PRECONDITION FAILED">
+
+
+*The `If-Match` header did not match the currently stored configuration, or another writer committed a change between this request's precondition check and write. Re-read the configuration and retry.*
+
 </TabItem>
 </Tabs>
 

extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/execution/KubernetesTaskExecutionConfigResource.java

@@ -25,8 +25,11 @@
 import org.apache.druid.audit.AuditManager;
 import org.apache.druid.common.config.ConfigManager;
 import org.apache.druid.common.config.JacksonConfigManager;
+import org.apache.druid.error.DruidException;
 import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.server.http.DynamicConfigEtagHelper;
+import org.apache.druid.server.http.ServletResourceUtils;
 import org.apache.druid.server.http.security.ConfigResourceFilter;
 import org.apache.druid.server.security.AuthorizationUtils;
 import org.joda.time.Interval;
@@ -84,23 +87,27 @@ public Response setExecutionConfig(
       @Context final HttpServletRequest req
   )
   {
-    KubernetesTaskRunnerDynamicConfig currentConfig = getDynamicConfig();
-    KubernetesTaskRunnerDynamicConfig mergedConfig = dynamicConfig;
+    try {
+      final KubernetesTaskRunnerDynamicConfig currentConfig = getDynamicConfig();
+      final KubernetesTaskRunnerDynamicConfig mergedConfig = currentConfig == null
+          ? dynamicConfig
+          : currentConfig.merge(dynamicConfig);
+      final ConfigManager.SetResult setResult = configManager.setIfMatch(
+          KubernetesTaskRunnerDynamicConfig.CONFIG_KEY,
+          DynamicConfigEtagHelper.getIfMatch(req),
+          mergedConfig,
+          AuthorizationUtils.buildAuditInfo(req)
+      );
+      if (setResult.isOk()) {
+        log.info("Updating K8s execution configs: %s", mergedConfig);
 
-    if (currentConfig != null) {
-      mergedConfig = currentConfig.merge(dynamicConfig);
+        return Response.ok().build();
+      } else {
+        return DynamicConfigEtagHelper.toErrorResponse(setResult);
+      }
     }
-    final ConfigManager.SetResult setResult = configManager.set(
-        KubernetesTaskRunnerDynamicConfig.CONFIG_KEY,
-        mergedConfig,
-        AuthorizationUtils.buildAuditInfo(req)
-    );
-    if (setResult.isOk()) {
-      log.info("Updating K8s execution configs: %s", mergedConfig);
-
-      return Response.ok().build();
-    } else {
-      return Response.status(Response.Status.BAD_REQUEST).build();
+    catch (DruidException e) {
+      return ServletResourceUtils.buildErrorResponseFrom(e);
     }
   }
 
@@ -154,7 +161,14 @@ public Response getExecutionConfigHistory(
   @ResourceFilters(ConfigResourceFilter.class)
   public Response getExecutionConfig()
   {
-    return Response.ok(getDynamicConfig()).build();
+    return DynamicConfigEtagHelper.buildReadResponseWithEtag(
+        () -> configManager.getCurrentBytes(KubernetesTaskRunnerDynamicConfig.CONFIG_KEY),
+        currentBytes -> configManager.convertByteToConfig(
+            currentBytes,
+            KubernetesTaskRunnerDynamicConfig.class,
+            null
+        )
+    );
   }
 
   private KubernetesTaskRunnerDynamicConfig getDynamicConfig()