You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "Pankaj260100 (via GitHub)" <gi...@apache.org> on 2025/10/15 05:31:17 UTC
[PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Pankaj260100 opened a new pull request, #18634:
URL: https://github.com/apache/druid/pull/18634
<!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
<!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
Fixes #XXXX.
<!-- Replace XXXX with the id of the issue fixed in this PR. Remove this section if there is no corresponding issue. Don't reference the issue in the title of this pull-request. -->
<!-- If you are a committer, follow the PR action item checklist for committers:
https://github.com/apache/druid/blob/master/dev/committer-instructions.md#pr-and-issue-action-item-checklist-for-committers. -->
### Description
- Previously, each datasource was managed by a single supervisor (1:1 mapping).
- [This update](https://github.com/apache/druid/pull/18082) allows multiple supervisors to be assigned to a single datasource.
- Introduced a new task distribution strategy that assigns tasks to different tiers of workers based on their associated supervisor.
- Enables more flexible and efficient task allocation across the system.
#### Fixed the bug ...
#### Renamed the class ...
#### Added a forbidden-apis entry ...
<!--
In each section, please describe design decisions made, including:
- Choice of algorithms
- Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
- Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
- Method organization and design (how the logic is split between methods, parameters and return types)
- Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
-->
<!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
<!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
<!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
#### Release note
<!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users.
If your change doesn't have end user impact, you can skip this section.
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
-->
<hr>
##### Key changed/added classes in this PR
* `MyFoo`
* `OurBar`
* `TheirBaz`
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] 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 or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "Pankaj260100 (via GitHub)" <gi...@apache.org>.
Pankaj260100 commented on code in PR #18634:
URL: https://github.com/apache/druid/pull/18634#discussion_r2442552170
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/EqualDistributionWithSupervisorCategorySpecWorkerSelectStrategy.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.indexing.overlord.setup;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.indexing.common.task.Task;
+import org.apache.druid.indexing.overlord.ImmutableWorkerInfo;
+import org.apache.druid.indexing.overlord.config.WorkerTaskRunnerConfig;
+
+import javax.annotation.Nullable;
+import java.util.Objects;
+
+public class EqualDistributionWithSupervisorCategorySpecWorkerSelectStrategy implements WorkerSelectStrategy
Review Comment:
Make sense, updated the PR.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18634:
URL: https://github.com/apache/druid/pull/18634#discussion_r2450658842
##########
indexing-service/src/test/java/org/apache/druid/indexing/overlord/setup/EqualDistributionWithCategorySpecWorkerSelectStrategyTest.java:
##########
@@ -187,4 +313,65 @@ private ImmutableWorkerInfo selectWorker(WorkerCategorySpec workerCategorySpec)
return worker;
}
+
+ /**
+ * Test implementation of SeekableStreamIndexTask for testing supervisor ID affinity
+ */
+ private static class TestSeekableStreamIndexTask extends SeekableStreamIndexTask<String, String, ByteEntity>
Review Comment:
`SeekableStreamSupervisorStateTest` also has a similar test class. Please see if it is possible to use the same one here. We can move it out into a file of its own.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -88,15 +88,19 @@ public static class CategoryConfig
private final String defaultCategory;
// key: datasource, value: category
private final Map<String, String> categoryAffinity;
+ // key: supervisorId, value: category
+ private final Map<String, String> supervisorIdCategoryAffinity;
@JsonCreator
public CategoryConfig(
@JsonProperty("defaultCategory") String defaultCategory,
- @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity
+ @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity,
+ @JsonProperty("supervisorIdCategoryAffinity") Map<String, String> supervisorIdCategoryAffinity
Review Comment:
```suggestion
@JsonProperty("supervisorIdCategoryAffinity") @Nullable Map<String, String> supervisorIdCategoryAffinity
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -88,15 +88,19 @@ public static class CategoryConfig
private final String defaultCategory;
// key: datasource, value: category
private final Map<String, String> categoryAffinity;
+ // key: supervisorId, value: category
+ private final Map<String, String> supervisorIdCategoryAffinity;
@JsonCreator
public CategoryConfig(
@JsonProperty("defaultCategory") String defaultCategory,
- @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity
+ @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity,
+ @JsonProperty("supervisorIdCategoryAffinity") Map<String, String> supervisorIdCategoryAffinity
)
{
this.defaultCategory = defaultCategory;
this.categoryAffinity = categoryAffinity == null ? Collections.emptyMap() : categoryAffinity;
+ this.supervisorIdCategoryAffinity = supervisorIdCategoryAffinity == null ? Collections.emptyMap() : supervisorIdCategoryAffinity;
Review Comment:
shorthand:
```suggestion
this.supervisorIdCategoryAffinity = Configs.valueOrDefault(supervisorIdCategoryAffinity, Map.of());
```
##########
indexing-service/src/test/java/org/apache/druid/indexing/overlord/setup/EqualDistributionWithCategorySpecWorkerSelectStrategyTest.java:
##########
@@ -174,6 +198,108 @@ public void testStrongTierSpec()
Assert.assertNull(worker);
}
+ @Test
+ public void testSupervisorIdCategoryAffinity()
+ {
+ // Test that supervisor ID affinity takes precedence over datasource affinity
+ final WorkerCategorySpec workerCategorySpec = new WorkerCategorySpec(
+ ImmutableMap.of(
+ "test_seekable_stream",
+ new WorkerCategorySpec.CategoryConfig(
+ "c1", // default category
+ ImmutableMap.of("ds1", "c1"), // datasource affinity
+ ImmutableMap.of("supervisor1", "c2") // supervisor ID affinity
+ )
+ ),
+ false
+ );
+
+ // Create a test task with supervisor ID "supervisor1"
+ final Task taskWithSupervisor = new TestSeekableStreamIndexTask("task1", "supervisor1", "ds1");
+
+ final EqualDistributionWithCategorySpecWorkerSelectStrategy strategy =
+ new EqualDistributionWithCategorySpecWorkerSelectStrategy(workerCategorySpec, null);
+
+ ImmutableWorkerInfo worker = strategy.findWorkerForTask(
+ new RemoteTaskRunnerConfig(),
+ WORKERS_FOR_TIER_TESTS,
+ taskWithSupervisor
+ );
+
+ // Should select c2 worker (localhost3) because supervisor ID affinity takes precedence
+ Assert.assertNotNull(worker);
+ Assert.assertEquals("c2", worker.getWorker().getCategory());
+ Assert.assertEquals("localhost3", worker.getWorker().getHost());
+ }
+
+ @Test
+ public void testSupervisorIdCategoryAffinityFallbackToDatasource()
+ {
+ // Test that it falls back to datasource affinity when supervisor ID affinity is not found
+ final WorkerCategorySpec workerCategorySpec = new WorkerCategorySpec(
+ ImmutableMap.of(
+ "test_seekable_stream",
+ new WorkerCategorySpec.CategoryConfig(
+ "c2", // default category
+ ImmutableMap.of("ds1", "c1"), // datasource affinity
+ ImmutableMap.of("supervisor2", "c2") // supervisor ID affinity (different supervisor)
+ )
+ ),
+ false
+ );
+
+ // Create a test task with supervisor ID "supervisor1" (not in supervisorIdCategoryAffinity map)
+ final Task taskWithSupervisor = new TestSeekableStreamIndexTask("task1", "supervisor1", "ds1");
+
+ final EqualDistributionWithCategorySpecWorkerSelectStrategy strategy =
+ new EqualDistributionWithCategorySpecWorkerSelectStrategy(workerCategorySpec, null);
+
+ ImmutableWorkerInfo worker = strategy.findWorkerForTask(
+ new RemoteTaskRunnerConfig(),
+ WORKERS_FOR_TIER_TESTS,
+ taskWithSupervisor
+ );
+
+ // Should fall back to datasource affinity and select c1 worker
+ Assert.assertNotNull(worker);
+ Assert.assertEquals("c1", worker.getWorker().getCategory());
+ Assert.assertEquals("localhost1", worker.getWorker().getHost());
+ }
+
+ @Test
+ public void testSupervisorIdCategoryAffinityFallbackToDefault()
+ {
+ // Test that it falls back to default category when neither supervisor ID nor datasource affinity is found
+ final WorkerCategorySpec workerCategorySpec = new WorkerCategorySpec(
+ ImmutableMap.of(
+ "test_seekable_stream",
+ new WorkerCategorySpec.CategoryConfig(
+ "c2", // default category
+ ImmutableMap.of("ds2", "c1"), // datasource affinity (different datasource)
+ ImmutableMap.of("supervisor2", "c1") // supervisor ID affinity (different supervisor)
Review Comment:
the comments here are not really needed.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -88,15 +88,19 @@ public static class CategoryConfig
private final String defaultCategory;
// key: datasource, value: category
private final Map<String, String> categoryAffinity;
+ // key: supervisorId, value: category
Review Comment:
It would be more useful to move this comment into a javadoc for the getter of this field.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "Pankaj260100 (via GitHub)" <gi...@apache.org>.
Pankaj260100 commented on code in PR #18634:
URL: https://github.com/apache/druid/pull/18634#discussion_r2458934558
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -88,15 +88,19 @@ public static class CategoryConfig
private final String defaultCategory;
// key: datasource, value: category
private final Map<String, String> categoryAffinity;
+ // key: supervisorId, value: category
+ private final Map<String, String> supervisorIdCategoryAffinity;
@JsonCreator
public CategoryConfig(
@JsonProperty("defaultCategory") String defaultCategory,
- @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity
+ @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity,
+ @JsonProperty("supervisorIdCategoryAffinity") Map<String, String> supervisorIdCategoryAffinity
)
{
this.defaultCategory = defaultCategory;
this.categoryAffinity = categoryAffinity == null ? Collections.emptyMap() : categoryAffinity;
+ this.supervisorIdCategoryAffinity = supervisorIdCategoryAffinity == null ? Collections.emptyMap() : supervisorIdCategoryAffinity;
Review Comment:
I updated it to the shorthand expression shared above. If we change it to null, then we will also have to update WorkerSelectUtils to check for null before the .get() operation.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18634:
URL: https://github.com/apache/druid/pull/18634#discussion_r2464343154
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -105,12 +109,31 @@ public String getDefaultCategory()
return defaultCategory;
}
+ /**
+ * Returns a map of datasource names to worker category names.
Review Comment:
```suggestion
* Returns a map from datasource name to the worker category name to be used for tasks of that datasource.
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -105,12 +109,31 @@ public String getDefaultCategory()
return defaultCategory;
}
+ /**
+ * Returns a map of datasource names to worker category names.
+ * Used to assign tasks to specific worker categories based on their datasource.
+ *
+ * @return map where key is datasource name and value is worker category name
+ */
@JsonProperty
public Map<String, String> getCategoryAffinity()
{
return categoryAffinity;
}
+ /**
+ * Returns a map of supervisor IDs to worker category names.
+ * Used to assign tasks to specific worker categories based on their supervisor ID.
+ * This takes precedence over {@link #getCategoryAffinity()} when both are configured.
+ *
+ * @return map where key is supervisor ID and value is worker category name
Review Comment:
Nit: can be omitted.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -105,12 +109,31 @@ public String getDefaultCategory()
return defaultCategory;
}
+ /**
+ * Returns a map of datasource names to worker category names.
+ * Used to assign tasks to specific worker categories based on their datasource.
+ *
+ * @return map where key is datasource name and value is worker category name
+ */
@JsonProperty
public Map<String, String> getCategoryAffinity()
{
return categoryAffinity;
}
+ /**
+ * Returns a map of supervisor IDs to worker category names.
Review Comment:
```suggestion
* Returns a map from supervisor ID to worker category name to be used for tasks of that supervisor.
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerSelectUtils.java:
##########
@@ -120,10 +121,25 @@ public static ImmutableWorkerInfo selectWorker(
if (categoryConfig != null) {
final String defaultCategory = categoryConfig.getDefaultCategory();
final Map<String, String> categoryAffinity = categoryConfig.getCategoryAffinity();
-
- String preferredCategory = categoryAffinity.get(task.getDataSource());
- // If there is no preferred category for the datasource, then using the defaultCategory. However, the defaultCategory
- // may be null too, so we need to do one more null check (see below).
+ final Map<String, String> supervisorIdCategoryAffinity = categoryConfig.getSupervisorIdCategoryAffinity();
+
+ String preferredCategory = null;
+
+ // First, check if this task has a supervisorId and if there's a category affinity for it
+ if (task instanceof SeekableStreamIndexTask) {
+ final String supervisorId = ((SeekableStreamIndexTask<?, ?, ?>) task).getSupervisorId();
+ if (supervisorId != null) {
+ preferredCategory = supervisorIdCategoryAffinity.get(supervisorId);
+ }
+ }
+
+ // If no supervisor-based category is found, fall back to datasource-based category affinity
+ if (preferredCategory == null) {
+ preferredCategory = categoryAffinity.get(task.getDataSource());
+ }
+
+ // If there is no preferred category for the supervisorId or datasource, then using the defaultCategory.
Review Comment:
```suggestion
// If there is no preferred category for the supervisorId or datasource, then use the defaultCategory.
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -105,12 +109,31 @@ public String getDefaultCategory()
return defaultCategory;
}
+ /**
+ * Returns a map of datasource names to worker category names.
+ * Used to assign tasks to specific worker categories based on their datasource.
+ *
+ * @return map where key is datasource name and value is worker category name
Review Comment:
Nit: Not needed since the first line already covers this info.
##########
indexing-service/src/test/java/org/apache/druid/indexing/overlord/setup/EqualDistributionWithCategorySpecWorkerSelectStrategyTest.java:
##########
@@ -187,4 +293,27 @@ private ImmutableWorkerInfo selectWorker(WorkerCategorySpec workerCategorySpec)
return worker;
}
+
+ /**
+ * Helper method to create a test task with supervisor ID for testing
+ */
+ @SuppressWarnings("unchecked")
+ private static Task createTestTask(String id, @Nullable String supervisorId, String datasource)
+ {
+ return new TestSeekableStreamIndexTask(
+ id,
+ supervisorId,
+ null,
+ DataSchema.builder()
+ .withDataSource(datasource)
+ .withTimestamp(new TimestampSpec(null, null, null))
+ .withDimensions(new DimensionsSpec(Collections.emptyList()))
+ .withGranularity(new ArbitraryGranularitySpec(new AllGranularity(), Collections.emptyList()))
Review Comment:
Nit: formatting seems off.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz merged PR #18634:
URL: https://github.com/apache/druid/pull/18634
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18634:
URL: https://github.com/apache/druid/pull/18634#discussion_r2431360688
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/EqualDistributionWithSupervisorCategorySpecWorkerSelectStrategy.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.indexing.overlord.setup;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.indexing.common.task.Task;
+import org.apache.druid.indexing.overlord.ImmutableWorkerInfo;
+import org.apache.druid.indexing.overlord.config.WorkerTaskRunnerConfig;
+
+import javax.annotation.Nullable;
+import java.util.Objects;
+
+public class EqualDistributionWithSupervisorCategorySpecWorkerSelectStrategy implements WorkerSelectStrategy
Review Comment:
This strategy seems to be mostly a duplicate of the existing `equalDistributionWithCategorySpec`.
Instead of this, add a new map in `CategoryConfig` class : `Map<String, String> supervisorIdCategoryAffinity`.
We will first check in this map, otherwise fallback to the datasource check.
This approach is better since there is no ambiguity regarding whether the map is keyed by datasource name or supervisor id.
Also, the `fillCapacityWithCategorySpec` strategy will be able to leverage the changes.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on code in PR #18634:
URL: https://github.com/apache/druid/pull/18634#discussion_r2433244161
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/EqualDistributionWithSupervisorCategorySpecWorkerSelectStrategy.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.indexing.overlord.setup;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.indexing.common.task.Task;
+import org.apache.druid.indexing.overlord.ImmutableWorkerInfo;
+import org.apache.druid.indexing.overlord.config.WorkerTaskRunnerConfig;
+
+import javax.annotation.Nullable;
+import java.util.Objects;
+
+public class EqualDistributionWithSupervisorCategorySpecWorkerSelectStrategy implements WorkerSelectStrategy
Review Comment:
+1 agree here with the above.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18634:
URL: https://github.com/apache/druid/pull/18634#discussion_r2459138444
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -88,15 +88,19 @@ public static class CategoryConfig
private final String defaultCategory;
// key: datasource, value: category
private final Map<String, String> categoryAffinity;
+ // key: supervisorId, value: category
+ private final Map<String, String> supervisorIdCategoryAffinity;
@JsonCreator
public CategoryConfig(
@JsonProperty("defaultCategory") String defaultCategory,
- @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity
+ @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity,
+ @JsonProperty("supervisorIdCategoryAffinity") Map<String, String> supervisorIdCategoryAffinity
)
{
this.defaultCategory = defaultCategory;
this.categoryAffinity = categoryAffinity == null ? Collections.emptyMap() : categoryAffinity;
+ this.supervisorIdCategoryAffinity = supervisorIdCategoryAffinity == null ? Collections.emptyMap() : supervisorIdCategoryAffinity;
Review Comment:
It's okay to handle the null, I wanted to avoid the empty map being serialized out in the payload unnecessarily. But this is fine too.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Adding new task distribution strategy that assigns tasks to workers based on supervisor affinity (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18634:
URL: https://github.com/apache/druid/pull/18634#discussion_r2450714521
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/setup/WorkerCategorySpec.java:
##########
@@ -88,15 +88,19 @@ public static class CategoryConfig
private final String defaultCategory;
// key: datasource, value: category
private final Map<String, String> categoryAffinity;
+ // key: supervisorId, value: category
+ private final Map<String, String> supervisorIdCategoryAffinity;
@JsonCreator
public CategoryConfig(
@JsonProperty("defaultCategory") String defaultCategory,
- @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity
+ @JsonProperty("categoryAffinity") Map<String, String> categoryAffinity,
+ @JsonProperty("supervisorIdCategoryAffinity") Map<String, String> supervisorIdCategoryAffinity
)
{
this.defaultCategory = defaultCategory;
this.categoryAffinity = categoryAffinity == null ? Collections.emptyMap() : categoryAffinity;
+ this.supervisorIdCategoryAffinity = supervisorIdCategoryAffinity == null ? Collections.emptyMap() : supervisorIdCategoryAffinity;
Review Comment:
Alternatively, we can also just leave the value as null and not assign any default value to it.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org