FrankChen021 opened a new issue, #20802:
URL: https://github.com/apache/druid/issues/20802
## Background
Druid currently has six dynamic configurations. All of them are stored as
JSON in the `druid_config` table and read/written through
`JacksonConfigManager`, but each one has its own REST resource, path
convention, propagation mechanism and authorization check.
| Config | Storage key | Served by | Public API | Propagation |
|---|---|---|---|---|
| Coordinator dynamic config | `coordinator.config` | Coordinator |
`GET/POST /druid/coordinator/v1/config`, `/history`, `/syncedBrokers`,
`/cloneStatus` | Pushed to Brokers (`cloneServers`) |
| Broker dynamic config | `broker.config` | Coordinator | `GET/POST
/druid/coordinator/v1/broker/config`, `/history` | Pushed to Brokers |
| Compaction config | `coordinator.compaction.config` | Coordinator |
`/druid/coordinator/v1/config/compaction` (+ `/taskslots`, `/{dataSource}`,
`/{dataSource}/history`, `DELETE /{dataSource}`) | Polled from DB |
| Worker behavior config | `worker.config` | Overlord | `GET/POST
/druid/indexer/v1/worker`, `/worker/history` | Polled from DB |
| K8s task runner config | `k8s.taskrunner.config` | Overlord (extension) |
`GET/POST /druid/indexer/v1/k8s/taskrunner/executionconfig`, `/history` |
Polled from DB |
| Lookups | `lookupsConfig` | Coordinator |
`/druid/coordinator/v1/lookups/...` | Coordinator pushes to
`/druid/listen/v1/lookups` |
Pushing configs to Brokers goes through `BaseDynamicConfigSyncer`, which
calls `POST /druid-internal/v1/config/{coordinator|broker}`
(`BrokerDynamicConfigResource`) with the escalated identity.
Concurrent writes are handled differently by each config:
| Config | Write path | Behavior under concurrent updates |
|---|---|---|
| Compaction (`CoordinatorConfigManager.getAndUpdateCompactionConfig`) |
Reads the current bytes from the DB, applies the operator, compare-and-swaps
against those bytes; `updateConfigHelper` retries | Concurrent per-datasource
edits are both kept, but a client cannot make a write conditional on what it
last read |
| Coordinator dynamic config | Builder merges the request onto the **watched
in-memory copy** (up to `pollDuration` stale), then writes unconditionally |
Lost updates are possible |
| Broker dynamic config | Same as the Coordinator dynamic config | Same |
| Worker config | Full replace, unconditional | Last writer wins |
| K8s execution config | Full replace, unconditional | Last writer wins |
### Problems
- **Every config needs a new API.** Adding a dynamic config means writing a
new JAX-RS resource, a history endpoint, audit code, docs and a web-console
view. Paths are inconsistent: some live under `/coordinator/v1/config`, some
under `/indexer/v1/...`, and one in an extension.
- **Coarse authorization.** `ConfigResourceFilter` checks a single global
`CONFIG` resource, so a user cannot be allowed to change the compaction config
without also being allowed to change the worker config.
- **No optimistic concurrency.** No API returns a version or accepts
`If-Match`, and several write paths merge onto a stale cached copy, so
concurrent edits can silently overwrite each other.
- **Opaque writes.** `POST` returns no body, so the user cannot see what was
actually stored, especially when the request is merged onto the existing value.
- **One-off propagation.** Pushing configs to nodes without metadata-store
access is implemented only for the two Broker-facing configs.
- **Unreliable "last modified by".** The latest author and comment exist
only in `druid_audit`, which is written in a separate transaction from the
config value (see the comment in `JacksonConfigManager.set`). It may be absent
when `druid.audit.manager.type=log`, or removed by audit cleanup.
### Motivating case: global Kafka consumer properties
For high-throughput topics, Kafka ingestion can be bound by fetch size
rather than CPU. The consumer defaults (`max.partition.fetch.bytes` = 1 MiB,
`receive.buffer.bytes` = 64 KiB) cap how much data each fetch returns, so tasks
spend their time waiting on the network while the ingestion thread is mostly
idle. Today the only fix is to edit `consumerProperties` on every supervisor.
Operators need a cluster-wide default for consumer properties that can be
changed without restarting the Overlord. Precedence, from lowest to highest:
1. the global default;
2. the supervisor's `consumerProperties`;
3. the fixed settings in `KafkaConsumerConfigs`.
The merged result is written into the task `ioConfig`, so each task records
the values it ran with.
Under the current model this would be a seventh bespoke resource. Under this
proposal it is just a descriptor named `kafka.consumerProperties`.
## Goals
- One framework for all dynamic configs: one REST API, one storage model,
one propagation mechanism, one authorization model.
- Adding a dynamic config means registering a descriptor, not writing a
resource.
- Existing endpoints and storage keys keep working.
Lookups and the per-datasource compaction sub-resources are collections
rather than single values. They are out of scope for the first phase and could
later be modelled as collection descriptors.
## Proposal
### 1. Descriptor
```java
public interface DynamicConfigDescriptor<T>
{
String name(); // public name, e.g.
"kafka.consumerProperties"
String storageKey(); // druid_config.name; existing configs keep
their key
Class<T> type();
T defaultValue();
UpdateMode updateMode(); // REPLACE | MERGE (JSON merge-patch)
Set<NodeRole> syncTo(); // roles without metadata-store access that
need the value pushed
default void validate(T value) {}
}
```
Descriptors are registered through a Guice multibinder, so extensions can
contribute their own.
### 2. Storage
- Values stay in `druid_config`. Add nullable `version`, `author`, `comment`
and `updated_at` columns through an `ALTER TABLE` migration. Older nodes ignore
the extra columns, so rolling back is safe.
- The value and its update details are written in the same transaction.
`version` is a hash of the payload.
- `druid_audit` remains the full change history behind `/history`.
### 3. Concurrency
- **Server-side compare-and-swap for every write.** Read the current bytes
from the metadata store (not the watched copy), apply the merge or replace,
compare-and-swap, and retry on conflict. This generalizes
`CoordinatorConfigManager.updateConfigHelper` and fixes the lost-update issue
in today's Coordinator and Broker dynamic config endpoints.
- **Optional `If-Match`.** When the client sends a version, the write is
conditional on it and fails with `409` instead of retrying. It stays optional
so existing scripts and the web console keep working; the web console can
always send it.
### 4. Public REST API
Served by both the Coordinator and the Overlord. The Router routes
`/druid/v1/config/*` to the Coordinator by default.
| Method | Path | Description |
|---|---|---|
| `GET` | `/druid/v1/config` | All configs the caller can read, with full
details |
| `GET` | `/druid/v1/config/{name}` | A single config |
| `POST` | `/druid/v1/config/{name}` | Update a config. Optional `If-Match:
<version>`. Comment from the `X-Druid-Comment` header or a `comment` parameter |
| `GET` | `/druid/v1/config/{name}/history` | Audit history (`interval`,
`count`) |
| `GET` | `/druid/v1/config/{name}/syncStatus` | Per-node push status, for
configs with `syncTo` |
Both `GET` and `POST` return this config object:
```json
{
"name": "kafka.consumerProperties",
"value": { "max.partition.fetch.bytes": 16777216 },
"isDefault": false,
"version": "d41f9a",
"updateMode": "MERGE",
"syncTo": [],
"lastUpdated": { "author": "alice", "comment": "raise fetch size", "time":
"2026-10-30T08:12:00Z" }
}
```
`POST` does the following, in order: parse, merge or replace, `validate`,
compare-and-swap, read the stored row back, return it.
What it returns:
- The full stored value. For MERGE configs this is the merged result.
Defaults are filled in and unknown fields dropped, so users see exactly what
the server kept.
- The new `version`, which can be used as the next `If-Match`.
- For configs with `syncTo`: `200` with `syncStatus` if every push finishes
within a short timeout, otherwise `202` with a `syncStatus` link.
Error codes:
- `400`: validation failed.
- `409`: version mismatch.
- `404`: unknown name.
The existing endpoints are kept and delegate to the same engine locally.
### 5. Propagation
**Nodes with metadata-store access (Coordinator, Overlord)**
- They poll with `ConfigManager.watch` every
`druid.manager.config.pollDuration`.
- The node that accepted a write updates its own copy immediately.
**Nodes listed in `syncTo`**
- After each write, plus a periodic re-sync, the Coordinator pushes the
value to `POST /druid-internal/v1/config/{name}`. This generalizes
`BaseDynamicConfigSyncer`.
- On startup, each node fetches its configs from `GET /druid/v1/config`.
- A push carrying a version the node already holds is ignored.
### 6. Authorization
- Public API: each config is its own `CONFIG` resource, named after the
config. `READ` is required for `GET`, `WRITE` for `POST`. A grant on the
existing global `CONFIG` resource still covers every config, for compatibility.
- Internal push endpoint: requires a dedicated internal permission that only
the escalated system identity holds.
- The audit author is always the authenticated caller.
### 7. Future work: `sys.dynamic_configurations`
A read-only system table with one row per config the caller can read:
| Column | Type |
|---|---|
| `name` | VARCHAR |
| `value` | VARCHAR (JSON) |
| `is_default` | BOOLEAN |
| `version` | VARCHAR |
| `update_mode` | VARCHAR |
| `sync_to` | VARCHAR |
| `in_sync` | BOOLEAN |
| `last_author` | VARCHAR |
| `last_comment` | VARCHAR |
| `last_updated` | TIMESTAMP |
The Broker builds the table from a single `GET /druid/v1/config` call to the
Coordinator and checks `CONFIG` `READ` for each row.
## Delivery plan
1. Engine, descriptor registry, storage migration, public API, and
`kafka.consumerProperties`.
2. Migrate the existing configs onto descriptors, with the legacy endpoints
delegating to the engine.
3. Generalize push propagation (`syncTo`, `/syncStatus`) and replace
`BaseDynamicConfigSyncer`.
4. `sys.dynamic_configurations`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
