Extend server priorities to realtime indexing tasks for query isolation by abhishekrb19 · Pull Request #19040 · apache/druid

abhishekrb19 · 2026-02-20T22:44:59Z

Fixes #19018

Currently all task replicas by default answer all realtime queries regardless of the priority or routing strategy applied on the Broker.
For mixed query workloads, some queries may have higher priority than others, so isolation on the replicas helps prevent degradation for critical workloads. We've seen bad queries take down some task replicas causing noisy neighbor problems.

This extends Druid’s query prioritization and routing strategies for Peon servers by letting operators configure how many replicas per server priority to configure for the tasks, similar to how it works for Historicals and Brokers. This is done by exposing serverPriorityToReplics in the supervisor's ioConfig that operators can optionally configure with the number of replicas allocated per server priority for realtime indexing tasks.

For example, some replicas can be configured to handle queries of all priorities, while others may only respond to specific priority ranges. This would isolate certain Peon replicas for certain priorities and others for more exploratory / dashboarding usecases.

Approach:

To support this, a new serverPriorityToReplicas property is added to the supervisor’s ioConfig.
The SeekableStreamSupervisor assigns priorities to SeekableStreamIndexTasks as they are created for a group. Similarly they're removed from internal bookkeeping when the tasks terminate. The ForkingTaskRunner then passes the appropriate server priority when initializing the Peon server. In the absence of this configuration, Peons continue to run with the default priority 0.

serverPriorityToReplicas is optional and is compatible with the existing replicas property if specified.

Release note

Added serverPriorityToReplicas parameter to the streaming supervisor specs (kafka/kinesis/rabbit). This allows operators to distribute task replicas across different server priorities for realtime indexing tasks. Similar to historical tiering, this enables query isolation for mixed workload scenarios on the Peons, allowing some task replicas to handle queries of specific priorities.

This PR has:

been self-reviewed.
added documentation for new or modified features or behaviors.
a release note entry in the PR description.
added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
been tested in a test Druid cluster.

…k replicas.

…plicas

+    EasyMock.reset(spec);
+    EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
+    EasyMock.expect(spec.getSupervisorStateManagerConfig()).andReturn(supervisorConfig).anyTimes();
+    EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();


+    EasyMock.reset(spec);
+    EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
+    EasyMock.expect(spec.getSupervisorStateManagerConfig()).andReturn(supervisorConfig).anyTimes();
+    EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();


aho135 · 2026-02-26T23:34:33Z


+  @Nullable
+  @JsonProperty
+  public Map<Integer, Integer> getserverPriorityToReplicas()


Suggested change

public Map<Integer, Integer> getserverPriorityToReplicas()

public Map<Integer, Integer> getServerPriorityToReplicas()

aho135 · 2026-02-26T23:40:57Z

+    if (this.serverPriorityToReplicas != null) {
+      final int replicaCount = this.serverPriorityToReplicas.values().stream().mapToInt(Integer::intValue).sum();
+      if (replicas != null && replicas != replicaCount) {
+        throw InvalidInput.exception(


What do you think about dropping this requirement? Seems like additional configuration overhead since we set it explicitly below anyway

This would be triggered only if someone unintentionally sets conflicting values that cause ambiguity. I wanted to keep the behavior straightforward for end users rather than document any precedence rules in such cases. I've updated the error message accordingly, please let me know if that makes sense.

jtuglu1

Did a first pass – general approach LGTM.

aho135 · 2026-02-28T00:07:11Z

 |`lateMessageRejectionPeriod`|ISO 8601 period|Configures tasks to reject messages with timestamps earlier than this period before the task was created. For example, if this property is set to `PT1H` and the supervisor creates a task at `2016-01-01T12:00Z`, Druid drops messages with timestamps earlier than `2016-01-01T11:00Z`. This may help prevent concurrency issues if your data stream has late messages and you have multiple pipelines that need to operate on the same segments, such as a streaming and a nightly batch ingestion pipeline. You can specify only one of the late message rejection properties.|No||
 |`earlyMessageRejectionPeriod`|ISO 8601 period|Configures tasks to reject messages with timestamps later than this period after the task reached its task duration. For example, if this property is set to `PT1H`, the task duration is set to `PT1H` and the supervisor creates a task at `2016-01-01T12:00Z`, Druid drops messages with timestamps later than `2016-01-01T14:00Z`. Tasks sometimes run past their task duration, such as in cases of supervisor failover.|No||
 |`stopTaskCount`|Integer|Limits the number of ingestion tasks Druid can cycle at any given time. If not set, Druid can cycle all tasks at the same time. If set to a value less than `taskCount`, your cluster needs fewer available slots to run the supervisor. You can save costs by scaling down your ingestion tier, but this can lead to slower cycle times and lag. See [`stopTaskCount`](#stoptaskcount) for more information.|No|`taskCount` value|
+|`serverPriorityToReplicas`|Object (`Map<Integer, Integer>`)|Map of server priorities to the number of replicas per priority. When set, each task replica is assigned a server priority that corresponds to `druid.server.priority` on the Peon process to enable query isolation for mixed workloads using [query routing strategies](../configuration/index.md#query-routing). If not configured, the `replicas` setting applies and all task replicas are assigned a default priority of 0.<br/><br/>For example, setting `serverPriorityToReplicas` to `{"1": 2, "0": 1}` creates 2 task replicas with `druid.server.priority=1` and 1 task replica with `druid.server.priority=0` per task group. This configuration scales proportionally with `taskCount`. For example, if `taskCount` is set to 5, this results in 15 total task replicas - 10 replicas with priority 1 and 5 replicas with priority 0.|No|null|


Suggested change

|`serverPriorityToReplicas`|Object (`Map<Integer, Integer>`)|Map of server priorities to the number of replicas per priority. When set, each task replica is assigned a server priority that corresponds to `druid.server.priority` on the Peon process to enable query isolation for mixed workloads using [query routing strategies](../configuration/index.md#query-routing). If not configured, the `replicas` setting applies and all task replicas are assigned a default priority of 0. For example, setting `serverPriorityToReplicas` to `{"1": 2, "0": 1}` creates 2 task replicas with `druid.server.priority=1` and 1 task replica with `druid.server.priority=0` per task group. This configuration scales proportionally with `taskCount`. For example, if `taskCount` is set to 5, this results in 15 total task replicas - 10 replicas with priority 1 and 5 replicas with priority 0.|No|null|

|`serverPriorityToReplicas`|Object (`Map<Integer, Integer>`)|Map of server priorities to the number of replicas per priority. When set, each task replica is assigned a server priority that corresponds to `druid.server.priority` on the Peon process to enable query isolation for mixed workloads using [query routing strategies](../configuration/index.md#query-routing). If not configured, the `replicas` setting applies and all task replicas are assigned a default priority of 0. For example, setting `serverPriorityToReplicas` to `{"1": 2, "0": 1}` creates 2 task replicas with `druid.server.priority=1` and 1 task replica with `druid.server.priority=0` per task group. This configuration scales proportionally with `taskCount`. For example, if `taskCount` is set to 5, this results in 15 total tasks - 10 tasks with priority 1 and 5 tasks with priority 0.|No|null|

I thought using the word replicas here was a bit confusing since we're referring to task counts here

aho135 · 2026-02-28T00:43:01Z

    this.idleConfig = idleConfig;
+    this.serverPriorityToReplicas = serverPriorityToReplicas;
+    if (this.serverPriorityToReplicas != null) {
+      final int serverPriorityReplicas = this.serverPriorityToReplicas.values().stream().mapToInt(Integer::intValue).sum();


Something like this would also pass validation

Map.of( 1, 2, 0, -1 )

Maybe make this more strict such that any value has to be > 0

Yep, good point. I’ve made the validation check stricter to use >= 0 since the existing replicas can be set to 0

aho135 · 2026-03-02T22:19:19Z

+        topic,
+        null,
+        INPUT_FORMAT,
+        null,


Did you want to enforce replicas = sum of Replicas configured in serverPriorityToReplicas?

Yeah, it's verified in this test SeekableStreamSupervisorIOConfigTest#testMismatchBetweenReplicasAndServerPriorityReplicasThrowsException()

(In terms of structure, I added most tests to the seekable stream package for code coverage, since it serves as the base implementation and added a subset of those tests to the specific supervisor types)

abhishekrb19 · 2026-03-03T01:29:31Z

Thanks for the reviews @aho135 and @jtuglu1! Jesse, please feel free to share any feedback even after the PR is merged and I can address it in a follow-up.

jtuglu1

LGTM

jtuglu1 · 2026-03-03T02:33:40Z

+     */
+    void removeTask(Iterator<Entry<String, TaskData>> taskDataIterator, String taskId)
+    {
+      taskDataIterator.remove();


Do we not need to remove taskId from tasks? If not, can we document why not?

This is already an iterator from tasks.keySet(), so it does remove the entry from tasks. I’ve updated the javadoc to clarify that and added a test as well

…re discovery (#19405) With #19040, when ioConfig.serverPriorityToReplicas is set, a supervisor can get stuck throwing DruidException from computeUnassignedServerPriorities, preventing any new task replicas with destined replicas from being created until the old tasks rollover. The exception is as follows: Found unassignedServerPriorities[[]] of size[0] < total replicas[1] for taskGroupId[0]. Task server priorities[[1, 0]] have already been assigned to tasks[[foo]]. The supervisor can remain in this unhealthy state unable to create additional task replicas until tasks eventually rollover. This can happen when a task spuriously fails after it's created but before it's discovered in the runInternal() loop. This patch removes the additional writer, leaving discoverTasks and removeTask as the sole mutators. group.taskIdToServerPriority should now stay in sync with group.tasks. The added unit tests fail on master without the fix.

…re discovery (apache#19405) With apache#19040, when ioConfig.serverPriorityToReplicas is set, a supervisor can get stuck throwing DruidException from computeUnassignedServerPriorities, preventing any new task replicas with destined replicas from being created until the old tasks rollover. The exception is as follows: Found unassignedServerPriorities[[]] of size[0] < total replicas[1] for taskGroupId[0]. Task server priorities[[1, 0]] have already been assigned to tasks[[foo]]. The supervisor can remain in this unhealthy state unable to create additional task replicas until tasks eventually rollover. This can happen when a task spuriously fails after it's created but before it's discovered in the runInternal() loop. This patch removes the additional writer, leaving discoverTasks and removeTask as the sole mutators. group.taskIdToServerPriority should now stay in sync with group.tasks. The added unit tests fail on master without the fix.

github-actions Bot added Area - Streaming Ingestion Area - Ingestion labels Feb 20, 2026

github-advanced-security AI found potential problems Feb 20, 2026

View reviewed changes

Comment thread .../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java Dismissed

Add serverPriorities to supervisor spec to control priorities for tas…

eed82ea

…k replicas.

abhishekrb19 force-pushed the peon_server_priorities branch from 8e47ed3 to eed82ea Compare February 20, 2026 23:27

Tests

e617cd5

abhishekrb19 changed the title ~~Extend server priorities to Peon servers~~ Extend server priorities to realtime indexing tasks for query isolation Feb 21, 2026

abhishekrb19 added 2 commits February 22, 2026 18:57

Cleanup and test

458a020

Rename config from serverPriorityToReplicaCount -> serverPriorityToRe…

a2497d2

…plicas

github-actions Bot added the Area - Documentation label Feb 23, 2026

Docs

6a42a18

abhishekrb19 force-pushed the peon_server_priorities branch from 85b9135 to 6a42a18 Compare February 23, 2026 04:46

Coverage

ad0abca

abhishekrb19 force-pushed the peon_server_priorities branch from 7829e01 to ad0abca Compare February 23, 2026 07:44

github-advanced-security AI found potential problems Feb 23, 2026

View reviewed changes

Merge branch 'master' into peon_server_priorities

a05aa73

jtuglu1 self-requested a review February 24, 2026 06:32

Change and tests

aa1774c

github-advanced-security AI found potential problems Feb 26, 2026

View reviewed changes

aho135 reviewed Feb 26, 2026

View reviewed changes

abhishekrb19 added 3 commits February 26, 2026 17:24

Fix casing and variable name

9abee83

Merge branch 'master' into peon_server_priorities

47515cd

Better error messages

0b67dd4

jtuglu1 reviewed Feb 27, 2026

View reviewed changes

abhishekrb19 added 4 commits February 27, 2026 08:30

Docs and additional validation

b31f799

 to 

2ac1ad4

Wire in serverPriorityToReplicas for Kinesis supervisor

22cf631

Wire in serverPriorityToReplicas for RabbitStreamSupervisor

e6bd4ba

aho135 reviewed Feb 28, 2026

View reviewed changes

abhishekrb19 added 4 commits March 2, 2026 09:49

Review suggestions

aeaa98a

Add helper for removeTask()

8634409

Merge branch 'master' into peon_server_priorities

0ad6aac

Checkstyle

98cfead

aho135 reviewed Mar 2, 2026

View reviewed changes

Comment thread docs/ingestion/supervisor.md Outdated

aho135 reviewed Mar 2, 2026

View reviewed changes

aho135 approved these changes Mar 2, 2026

View reviewed changes

Update doc wording

ca3fb2a

jtuglu1 approved these changes Mar 3, 2026

View reviewed changes

Update javadoc to clarify iterator and add test

d4952a4

abhishekrb19 merged commit f17b4f4 into apache:master Mar 3, 2026
62 of 63 checks passed

abhishekrb19 deleted the peon_server_priorities branch March 3, 2026 06:49

cecemei added this to the 37.0.0 milestone Apr 8, 2026

abhishekrb19 mentioned this pull request May 4, 2026

fix: Fix unhealthy supervisor when a priority-assigned task dies before discovery #19405

Merged

4 tasks

	public Map<Integer, Integer> getserverPriorityToReplicas()
	public Map<Integer, Integer> getServerPriorityToReplicas()

Uh oh!

Conversation

abhishekrb19 commented Feb 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Approach:

Release note

Uh oh!

Uh oh!

Check notice

Uh oh!

Check notice

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jtuglu1 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aho135 Feb 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

abhishekrb19 commented Mar 3, 2026

Uh oh!

jtuglu1 left a comment

Choose a reason for hiding this comment

Uh oh!

jtuglu1 Mar 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

abhishekrb19 commented Feb 20, 2026 •

edited

Loading

aho135 Feb 28, 2026 •

edited

Loading

jtuglu1 Mar 3, 2026 •

edited

Loading