You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "uds5501 (via GitHub)" <gi...@apache.org> on 2025/06/20 09:55:09 UTC
[PR] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
uds5501 opened a new pull request, #18163:
URL: https://github.com/apache/druid/pull/18163
### Description
Removes worker thread calculation for `ScheduledThreadPoolExecutor` and creates an executor with a single core thread (can grow out further).
##### Key changed/added classes in this PR
* `SeekableStreamSupervisor`
<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:
- [x] 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.)
- [x] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [x] 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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2162945691
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +943,9 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
+ final int workerThreads = calculateWorkerThreads(tuningConfig, ioConfig);
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
Review Comment:
Yeah, I guess you are right. You can leave it as it is then.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2159828316
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -162,6 +162,9 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
private static final long MAX_RUN_FREQUENCY_MILLIS = 1000;
private static final int MAX_INITIALIZATION_RETRIES = 20;
+ private static final int MIN_CORE_POOL_SIZE = 2;
Review Comment:
```suggestion
private static final int MIN_WORKER_CORE_THREADS = 2;
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz merged PR #18163:
URL: https://github.com/apache/druid/pull/18163
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "uds5501 (via GitHub)" <gi...@apache.org>.
uds5501 commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2163068262
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +943,9 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
+ final int workerThreads = calculateWorkerThreads(tuningConfig, ioConfig);
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
Review Comment:
Will be keeping it as is for now then.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2161656471
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +965,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
+
+ this.workerExec = MoreExecutors.listeningDecorator(executor);
+ log.info(
+ "Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]",
Review Comment:
```suggestion
"Created worker pool with [%d] threads for supervisor[%s], dataSource[%s].",
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "uds5501 (via GitHub)" <gi...@apache.org>.
uds5501 commented on PR #18163:
URL: https://github.com/apache/druid/pull/18163#issuecomment-2999113868
Thanks for your review @kfaraz , I have addressed the final review comments as well.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2163100808
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +944,9 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
+ final int workerThreads = calculateWorkerThreads(tuningConfig, ioConfig);
Review Comment:
This line should be moved to where the `workerExec` is being initialized.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1009,6 +1010,33 @@ public Optional<TaskStatus> getTaskStatus(String id)
this.taskClient = taskClientFactory.build(dataSource, taskInfoProvider, this.tuningConfig, workerExec);
}
+ /**
+ * Calculates the number of worker threads to use for the supervisor tasks.
+ * <p>
+ * If the tuning config explicitly sets the worker thread count, that value is used.
+ * Otherwise, the value is derived from either the auto-scaler config (if enabled) or the ioConfig task count,
+ * divided by the default number of tasks per worker thread, with a minimum of {@code MIN_WORKER_CORE_THREADS}.
Review Comment:
```suggestion
* divided by the {@link DEFAULT_TASKS_PER_WORKER_THREAD}, with a minimum of {@link MIN_WORKER_CORE_THREADS}.
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1009,6 +1010,33 @@ public Optional<TaskStatus> getTaskStatus(String id)
this.taskClient = taskClientFactory.build(dataSource, taskInfoProvider, this.tuningConfig, workerExec);
}
+ /**
+ * Calculates the number of worker threads to use for the supervisor tasks.
Review Comment:
There should also be a line on what these worker threads are actually used for.
##########
processing/src/main/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutors.java:
##########
@@ -172,8 +173,30 @@ public static ScheduledExecutorFactory createFactory(final Lifecycle lifecycle)
return (corePoolSize, nameFormat) -> ExecutorServices.manageLifecycle(lifecycle, fixed(corePoolSize, nameFormat));
}
+ /**
+ * Creates a new {@link ScheduledExecutorService} with a minimum number of threads.
+ *
+ * @param corePoolSize the minimum number of threads in the pool
+ * @param nameFormat the naming format for threads created by the pool
+ * @return a new {@link ScheduledExecutorService} with the specified configuration
+ */
public static ScheduledExecutorService fixed(int corePoolSize, String nameFormat)
{
return Executors.newScheduledThreadPool(corePoolSize, Execs.makeThreadFactory(nameFormat));
}
+
+ public static ScheduledThreadPoolExecutor fixedWithKeepAliveTime(
Review Comment:
Needs a javadoc.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1009,6 +1010,33 @@ public Optional<TaskStatus> getTaskStatus(String id)
this.taskClient = taskClientFactory.build(dataSource, taskInfoProvider, this.tuningConfig, workerExec);
}
+ /**
+ * Calculates the number of worker threads to use for the supervisor tasks.
+ * <p>
+ * If the tuning config explicitly sets the worker thread count, that value is used.
+ * Otherwise, the value is derived from either the auto-scaler config (if enabled) or the ioConfig task count,
+ * divided by the default number of tasks per worker thread, with a minimum of {@code MIN_WORKER_CORE_THREADS}.
+ *
+ * @param tuningConfig the tuning configuration
+ * @param ioConfig the IO configuration
+ * @return the number of worker threads for executor
Review Comment:
These param descriptions don't really have any new info and can be omitted.
##########
processing/src/main/java/org/apache/druid/java/util/common/concurrent/ScheduledExecutors.java:
##########
@@ -172,8 +173,30 @@ public static ScheduledExecutorFactory createFactory(final Lifecycle lifecycle)
return (corePoolSize, nameFormat) -> ExecutorServices.manageLifecycle(lifecycle, fixed(corePoolSize, nameFormat));
}
+ /**
+ * Creates a new {@link ScheduledExecutorService} with a minimum number of threads.
+ *
+ * @param corePoolSize the minimum number of threads in the pool
+ * @param nameFormat the naming format for threads created by the pool
+ * @return a new {@link ScheduledExecutorService} with the specified configuration
+ */
public static ScheduledExecutorService fixed(int corePoolSize, String nameFormat)
{
return Executors.newScheduledThreadPool(corePoolSize, Execs.makeThreadFactory(nameFormat));
}
+
+ public static ScheduledThreadPoolExecutor fixedWithKeepAliveTime(
+ int corePoolSize,
+ String nameFormat,
+ long keepAliveTime,
+ TimeUnit timeUnit
Review Comment:
It might be simpler to just accept milliseconds.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1009,6 +1010,33 @@ public Optional<TaskStatus> getTaskStatus(String id)
this.taskClient = taskClientFactory.build(dataSource, taskInfoProvider, this.tuningConfig, workerExec);
}
+ /**
+ * Calculates the number of worker threads to use for the supervisor tasks.
Review Comment:
```suggestion
* Calculates the number of worker threads to use in a {@link SeekableStreamSupervisor}.
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1009,6 +1010,33 @@ public Optional<TaskStatus> getTaskStatus(String id)
this.taskClient = taskClientFactory.build(dataSource, taskInfoProvider, this.tuningConfig, workerExec);
}
+ /**
+ * Calculates the number of worker threads to use for the supervisor tasks.
+ * <p>
+ * If the tuning config explicitly sets the worker thread count, that value is used.
Review Comment:
```suggestion
* If the tuning config explicitly specifies the field {@code workerThreads}, that value is used.
```
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2675,13 +2763,28 @@ private static Map<String, Object> getProperties()
}
private static SeekableStreamSupervisorTuningConfig getTuningConfig()
+ {
+ return createSupervisorTuningConfigWithWorkerThreads(1);
+ }
+
+ private static SeekableStreamSupervisorTuningConfig createSupervisorTuningConfig()
+ {
+ return createSupervisorTuningConfigWithWorkerThreads(null);
+ }
+
+ /**
+ * Creates a {@link SeekableStreamSupervisorTuningConfig} with the given number of worker threads.
+ *
+ * @param workerThreads the number of threads to use, or null to imitate default settings.
+ */
Review Comment:
Nit: Not really needed (especially in tests) unless it's adding info that is not obvious from the method name and signature.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2159830110
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +941,8 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
-
- workerThreads = (this.tuningConfig.getWorkerThreads() != null
Review Comment:
We should use the factor only in conjunction with the other two configs, i.e. `taskCount` and `taskCountMax`.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2159835918
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -162,6 +162,9 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
private static final long MAX_RUN_FREQUENCY_MILLIS = 1000;
private static final int MAX_INITIALIZATION_RETRIES = 20;
+ private static final int MIN_CORE_POOL_SIZE = 2;
+ private static final int TASKS_PER_THREAD_FACTOR = 4;
Review Comment:
Nit:
```suggestion
private static final int DEFAULT_TASKS_PER_WORKER_THREAD = 4;
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "uds5501 (via GitHub)" <gi...@apache.org>.
uds5501 commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2161684923
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +965,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
Review Comment:
> Can this be passed in the constructor itself?
unfortunately, no. I explored the option but couldn't find any. There's no harm in setting this after object creation, the executor does nothing before tasks are pushed in its queue.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "uds5501 (via GitHub)" <gi...@apache.org>.
uds5501 commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2162938295
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +943,9 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
+ final int workerThreads = calculateWorkerThreads(tuningConfig, ioConfig);
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
Review Comment:
@kfaraz logging that in a method meant to be calculating worker threads seems a bit odd no?
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2158621245
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +941,8 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
-
- workerThreads = (this.tuningConfig.getWorkerThreads() != null
Review Comment:
I think the number of worker threads should be calculated something like this:
- If `workerThreads` are specified, honor that config.
- If nothing is specified, use a minimum of 2 to 5 (1 seems too aggressive a downscale from the old 10).
- If honoring the task count config (`ioConfig.taskCount` or `autoScale.taskCountMax()`),
core pool size could be some factor of the task count e.g. `taskCount/4` rather than a fixed value.
- For the non-core pool threads, we should perhaps also increase the timeout to a couple of seconds rather than the default of 10 ms.
If these settings continue to do well over a period of time, we can consider reducing them further.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2162950578
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +943,9 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
+ final int workerThreads = calculateWorkerThreads(tuningConfig, ioConfig);
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
Review Comment:
Yeah, I guess you are right. You can leave it as it is then.
Edit: The log line seemed closely related since it has to do with task count etc. but it is fine either way.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2162945691
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +943,9 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
+ final int workerThreads = calculateWorkerThreads(tuningConfig, ioConfig);
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
Review Comment:
Yeah, I guess you are right. You can leave it as it is then.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "cryptoe (via GitHub)" <gi...@apache.org>.
cryptoe commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2160739242
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +975,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
+
+ this.workerExec = MoreExecutors.listeningDecorator(executor);
+ log.info(
+ "Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]",
+ workerThreads, this.supervisorId, this.dataSource
);
- log.info("Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]", workerThreads, this.supervisorId, this.dataSource);
Review Comment:
We can rework the log to be more relevant instead of removing 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
Re: [PR] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "uds5501 (via GitHub)" <gi...@apache.org>.
uds5501 commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2158646655
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +941,8 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
-
- workerThreads = (this.tuningConfig.getWorkerThreads() != null
Review Comment:
> If workerThreads are specified, honor that config.
fair enough, otherwise this PR should've addressed removal of the tuning configs as well. Will accommodate these suggestions.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +941,8 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
-
- workerThreads = (this.tuningConfig.getWorkerThreads() != null
Review Comment:
> If workerThreads are specified, honor that config.
fair enough, otherwise this PR should've addressed removal of the tuning configs as well. Will accommodate these suggestions.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2159836584
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +975,16 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
);
- log.info("Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]", workerThreads, this.supervisorId, this.dataSource);
+ executor.setKeepAliveTime(KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
+
+ this.workerExec = MoreExecutors.listeningDecorator(executor);
+ log.info(
+ "Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]",
+ workerThreads, this.supervisorId, this.dataSource);
Review Comment:
```suggestion
"Created worker pool with [%d] threads for supervisor[%s], dataSource[%s].",
workerThreads, this.supervisorId, this.dataSource
);
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2161661423
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java:
##########
@@ -5209,6 +5283,97 @@ private void addMoreEvents(int numEventsPerPartition, int num_partitions) throws
}
}
+ private KafkaSupervisorTuningConfig getTuningConfig()
+ {
+ return getTuningConfig(null);
+ }
+
+ /**
+ * Use this method to create a {@link KafkaSupervisorTuningConfig} with custom thread count.
Review Comment:
```suggestion
* Creates a {@link KafkaSupervisorTuningConfig} with the given number of worker threads.
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "uds5501 (via GitHub)" <gi...@apache.org>.
uds5501 commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2160871052
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +975,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
+
+ this.workerExec = MoreExecutors.listeningDecorator(executor);
+ log.info(
+ "Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]",
+ workerThreads, this.supervisorId, this.dataSource
);
- log.info("Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]", workerThreads, this.supervisorId, this.dataSource);
Review Comment:
It's modified and there, kindly take a look
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2161664094
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java:
##########
@@ -5209,6 +5283,97 @@ private void addMoreEvents(int numEventsPerPartition, int num_partitions) throws
}
}
+ private KafkaSupervisorTuningConfig getTuningConfig()
+ {
+ return getTuningConfig(null);
+ }
+
+ /**
+ * Use this method to create a {@link KafkaSupervisorTuningConfig} with custom thread count.
+ *
+ * @param threadCount the number of threads to use, or null to imitate default settings.
+ */
+ private KafkaSupervisorTuningConfig getTuningConfig(@Nullable Integer threadCount)
+ {
+ return new KafkaSupervisorTuningConfig(
+ null,
+ 1000,
+ null,
+ null,
+ 50000,
+ null,
+ new Period("P1Y"),
+ null,
+ null,
+ null,
+ false,
+ null,
+ false,
+ null,
+ threadCount,
+ TEST_CHAT_RETRIES,
+ TEST_HTTP_TIMEOUT,
+ TEST_SHUTDOWN_TIMEOUT,
+ null,
+ null,
+ null,
+ null,
+ 10,
+ null,
+ null
+ );
+ }
+
+ private KafkaSupervisorIOConfig getIOConfig()
+ {
+ return getIOConfig(1, null);
+ }
+
+ private KafkaSupervisorIOConfig getIOConfig(int taskCount)
+ {
+ return getIOConfig(taskCount, null);
+ }
+
+ private KafkaSupervisorIOConfig getIOConfig(AutoScalerConfig autoScalerConfig)
+ {
+ return getIOConfig(1, autoScalerConfig);
+ }
Review Comment:
I think these are unnecessary, just use the method with two args.
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java:
##########
@@ -5209,6 +5283,97 @@ private void addMoreEvents(int numEventsPerPartition, int num_partitions) throws
}
}
+ private KafkaSupervisorTuningConfig getTuningConfig()
+ {
+ return getTuningConfig(null);
+ }
+
+ /**
+ * Use this method to create a {@link KafkaSupervisorTuningConfig} with custom thread count.
+ *
+ * @param threadCount the number of threads to use, or null to imitate default settings.
+ */
+ private KafkaSupervisorTuningConfig getTuningConfig(@Nullable Integer threadCount)
+ {
+ return new KafkaSupervisorTuningConfig(
+ null,
+ 1000,
+ null,
+ null,
+ 50000,
+ null,
+ new Period("P1Y"),
+ null,
+ null,
+ null,
+ false,
+ null,
+ false,
+ null,
+ threadCount,
+ TEST_CHAT_RETRIES,
+ TEST_HTTP_TIMEOUT,
+ TEST_SHUTDOWN_TIMEOUT,
+ null,
+ null,
+ null,
+ null,
+ 10,
+ null,
+ null
+ );
+ }
+
+ private KafkaSupervisorIOConfig getIOConfig()
+ {
+ return getIOConfig(1, null);
+ }
+
+ private KafkaSupervisorIOConfig getIOConfig(int taskCount)
+ {
+ return getIOConfig(taskCount, null);
+ }
+
+ private KafkaSupervisorIOConfig getIOConfig(AutoScalerConfig autoScalerConfig)
+ {
+ return getIOConfig(1, autoScalerConfig);
+ }
+
+ /**
+ * Use this method to create a {@link KafkaSupervisorIOConfig} with custom properties.
+ */
+ private KafkaSupervisorIOConfig getIOConfig(
Review Comment:
```suggestion
private KafkaSupervisorIOConfig createSupervisorIOConfig(
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2159827677
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +941,8 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
-
- workerThreads = (this.tuningConfig.getWorkerThreads() != null
Review Comment:
Thanks, @uds5501 .
Just to clarify, when `workerThreads` is specified, we should use it exactly, without dividing by a factor.
Otherwise, we are changing the semantics of this config which is not desirable and also confusing.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2159828806
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -162,6 +162,9 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
private static final long MAX_RUN_FREQUENCY_MILLIS = 1000;
private static final int MAX_INITIALIZATION_RETRIES = 20;
+ private static final int MIN_CORE_POOL_SIZE = 2;
+ private static final int TASKS_PER_THREAD_FACTOR = 4;
+ private static final int KEEPALIVE_TIME_SECONDS = 2;
Review Comment:
```suggestion
private static final int WORKER_THREAD_KEEPALIVE_TIME_SECONDS = 2;
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "uds5501 (via GitHub)" <gi...@apache.org>.
uds5501 commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2160871052
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +975,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
+
+ this.workerExec = MoreExecutors.listeningDecorator(executor);
+ log.info(
+ "Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]",
+ workerThreads, this.supervisorId, this.dataSource
);
- log.info("Created worker pool with [%d] threads for supervisor[%s] for dataSource[%s]", workerThreads, this.supervisorId, this.dataSource);
Review Comment:
It's there.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2162902024
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1009,6 +1008,22 @@ public Optional<TaskStatus> getTaskStatus(String id)
this.taskClient = taskClientFactory.build(dataSource, taskInfoProvider, this.tuningConfig, workerExec);
}
+ public static int calculateWorkerThreads(
Review Comment:
Please add a short javadoc about what these worker threads are used for and how their number is computed.
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2630,21 +2703,32 @@ private static DataSchema getDataSchema()
.build();
}
- private static SeekableStreamSupervisorIOConfig getIOConfig()
+ private static SeekableStreamSupervisorIOConfig getSupervisorIOConfig()
Review Comment:
Nit: Please use the prefix `create` instead of `get` for these new methods (you can leave the existing method names unchanged) since they are creating new objects return than getting some field that is already present in this instance.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +965,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
Review Comment:
Nit: I think we should put this in a utility method in `ScheduledExecutors` class itself. Essentially, another flavor of the `fixed` method which accepts timeout millis too.
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2563,6 +2565,77 @@ public LagStats computeLagStats()
EasyMock.verify(executorService, spec);
}
+ @Test
+ public void test_calculateWorkerThreads_shouldHonourWorkerConfig()
+ {
+ final int totalThreads = 5;
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads(totalThreads);
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(1, null);
+ Assert.assertEquals(totalThreads, SeekableStreamSupervisor.calculateWorkerThreads(tuningConfig, ioConfig));
+ }
+
+ @Test
+ public void test_calculateWorkerThreads_shouldUseDefaultWorkerThreads()
+ {
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads();
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(1, null);
+ Assert.assertEquals(
+ DEFAULT_WORKER_THREADS,
+ SeekableStreamSupervisor.calculateWorkerThreads(tuningConfig, ioConfig)
+ );
+ }
+
+ @Test
+ public void test_calculateWorkerThreads_shouldUseMinimumWorkerThreadstWithTasks()
+ {
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads();
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(7, null);
+ Assert.assertEquals(
+ DEFAULT_WORKER_THREADS,
+ SeekableStreamSupervisor.calculateWorkerThreads(tuningConfig, ioConfig)
+ );
+ }
+
+ @Test
+ public void test_calculateWorkerThreads_shouldUseFactorOfTaskCount()
+ {
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads();
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(18, null);
+ Assert.assertEquals(
+ 4,
+ SeekableStreamSupervisor.calculateWorkerThreads(tuningConfig, ioConfig)
+ );
+ }
+
+ @Test
+ public void test_calculateWorkerThreads_shouldUseAutoScalerConfig()
+ {
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads();
+ AutoScalerConfig autoScalerConfig = new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 21,
Review Comment:
Assign this value to a field `taskCountMax` so that the test is easier to follow.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -940,17 +943,9 @@ public SeekableStreamSupervisor(
spec.isSuspended()
);
- int workerThreads;
+ final int workerThreads = calculateWorkerThreads(tuningConfig, ioConfig);
if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler()) {
log.info("Running Task autoscaler for supervisor[%s] for datasource[%s]", supervisorId, dataSource);
Review Comment:
Nit: I guess the log line can move to the new calculate method itself.
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2563,6 +2565,77 @@ public LagStats computeLagStats()
EasyMock.verify(executorService, spec);
}
+ @Test
+ public void test_calculateWorkerThreads_shouldHonourWorkerConfig()
+ {
+ final int totalThreads = 5;
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads(totalThreads);
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(1, null);
+ Assert.assertEquals(totalThreads, SeekableStreamSupervisor.calculateWorkerThreads(tuningConfig, ioConfig));
+ }
+
+ @Test
+ public void test_calculateWorkerThreads_shouldUseDefaultWorkerThreads()
+ {
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads();
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(1, null);
+ Assert.assertEquals(
+ DEFAULT_WORKER_THREADS,
+ SeekableStreamSupervisor.calculateWorkerThreads(tuningConfig, ioConfig)
+ );
+ }
+
+ @Test
+ public void test_calculateWorkerThreads_shouldUseMinimumWorkerThreadstWithTasks()
+ {
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads();
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(7, null);
+ Assert.assertEquals(
+ DEFAULT_WORKER_THREADS,
+ SeekableStreamSupervisor.calculateWorkerThreads(tuningConfig, ioConfig)
+ );
+ }
+
+ @Test
+ public void test_calculateWorkerThreads_shouldUseFactorOfTaskCount()
+ {
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads();
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(18, null);
+ Assert.assertEquals(
+ 4,
+ SeekableStreamSupervisor.calculateWorkerThreads(tuningConfig, ioConfig)
+ );
+ }
+
+ @Test
+ public void test_calculateWorkerThreads_shouldUseAutoScalerConfig()
+ {
+ SeekableStreamSupervisorTuningConfig tuningConfig = getTuningConfigWithWorkerThreads();
+ AutoScalerConfig autoScalerConfig = new LagBasedAutoScalerConfig(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ 21,
+ null,
+ 5,
+ null,
+ null,
+ true,
+ null,
+ null
+ );
+ SeekableStreamSupervisorIOConfig ioConfig = getSupervisorIOConfig(1, autoScalerConfig);
+ Assert.assertEquals(
+ 5,
Review Comment:
This would then become:
```suggestion
taskCountMax / defaultTasksPerThread,
```
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2563,6 +2565,77 @@ public LagStats computeLagStats()
EasyMock.verify(executorService, spec);
}
+ @Test
+ public void test_calculateWorkerThreads_shouldHonourWorkerConfig()
+ {
+ final int totalThreads = 5;
Review Comment:
```suggestion
final int numWorkerThreads = 5;
```
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2630,21 +2703,32 @@ private static DataSchema getDataSchema()
.build();
}
- private static SeekableStreamSupervisorIOConfig getIOConfig()
+ private static SeekableStreamSupervisorIOConfig getSupervisorIOConfig()
+ {
+ return getSupervisorIOConfig(1, OBJECT_MAPPER.convertValue(getProperties(), AutoScalerConfig.class));
+ }
+
+ /**
+ * Use this method to create a {@link SeekableStreamSupervisorIOConfig} with custom properties.
+ */
Review Comment:
Not really needed if we name the method appropriately.
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
@@ -2675,13 +2759,28 @@ private static Map<String, Object> getProperties()
}
private static SeekableStreamSupervisorTuningConfig getTuningConfig()
+ {
+ return getTuningConfigWithWorkerThreads(1);
+ }
+
+ private static SeekableStreamSupervisorTuningConfig getTuningConfigWithWorkerThreads()
Review Comment:
Nit: since we are not really specifying any value for `workerThreads` in this method, it shouldn't be included in the name.
```suggestion
private static SeekableStreamSupervisorTuningConfig createSupervisorTuningConfig()
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2161693911
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +965,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
Review Comment:
Cool, thanks for the confirmation.
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2161662164
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java:
##########
@@ -5209,6 +5283,97 @@ private void addMoreEvents(int numEventsPerPartition, int num_partitions) throws
}
}
+ private KafkaSupervisorTuningConfig getTuningConfig()
+ {
+ return getTuningConfig(null);
+ }
+
+ /**
+ * Use this method to create a {@link KafkaSupervisorTuningConfig} with custom thread count.
+ *
+ * @param threadCount the number of threads to use, or null to imitate default settings.
+ */
+ private KafkaSupervisorTuningConfig getTuningConfig(@Nullable Integer threadCount)
Review Comment:
```suggestion
private KafkaSupervisorTuningConfig createTuningConfigWithWorkerThreads(@Nullable Integer numWorkerThreads)
```
--
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] Feat: Intialize ScheduledThreadPoolExecutor with default worker threads (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18163:
URL: https://github.com/apache/druid/pull/18163#discussion_r2161657631
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +965,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
Review Comment:
Can this be passed in the constructor itself? Is there any harm to setting this after the object has been constructed?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -970,13 +965,17 @@ public SeekableStreamSupervisor(
);
}
- this.workerExec = MoreExecutors.listeningDecorator(
- ScheduledExecutors.fixed(
- workerThreads,
- StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d"
- )
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(
+ workerThreads,
+ Execs.makeThreadFactory(StringUtils.encodeForFormat(supervisorTag) + "-Worker-%d")
+ );
+ executor.setKeepAliveTime(WORKER_THREAD_KEEPALIVE_TIME_SECONDS, TimeUnit.SECONDS);
Review Comment:
Can this be passed in the constructor itself? Is there any drawback in setting this after the object has been constructed?
--
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