NealSun96 commented on a change in pull request #973:
URL: https://github.com/apache/helix/pull/973#discussion_r420313966
##########
File path:
helix-core/src/main/java/org/apache/helix/task/TaskStateModelFactory.java
##########
@@ -81,6 +79,12 @@ public Thread newThread(Runnable r) {
@Override
public TaskStateModel createNewStateModel(String resourceName, String
partitionKey) {
+ if (_taskExecutor == null) {
+ int taskThreadPoolSize = getTaskThreadPoolSize();
+ updateLiveInstanceWithThreadPoolSize(taskThreadPoolSize);
Review comment:
Answering "why lazy initialization" question here as well:
When `TaskStateModelFactory` is initialized, it is usually during factory
registration. However, factory registration can occur before HelixManager
connection, or after. When it occurs before HelixManager connection,
`TaskStateModelFactory` actually cannot access the configs, therefore it cannot
create the thread pool before manager connection.
We have 2 ways to make it work: 1. we enforce `TaskStateModelFactory`
initialization to be after HelixManager connection, 2. we lazily initialize the
thread pool. I chose 2 because it is backward compatible - clients don't need
to change their existing code in which they registered `TaskStateModelFactory`
before HelixManager connection (which exists).
In terms of `createNewStateModel()`, it is called during message handling
when the state model of the message doesn't exist for the resource name and
partition key. From the code I read, I don't think "the controller needs to
know the pool size before assign the tasks", because as we can see from the
original code of this class, the task thread pool is never exposed anywhere
other than `createNewStateModel()`. In fact, it's not used anywhere other than
`createNewStateModel()` either, therefore making it safe to initialize in this
function.
Feel free to correct me. @jiajunwang
##########
File path:
helix-core/src/main/java/org/apache/helix/task/TaskStateModelFactory.java
##########
@@ -41,23 +47,15 @@
private final HelixManager _manager;
private final Map<String, TaskFactory> _taskFactoryRegistry;
- private final ScheduledExecutorService _taskExecutor;
+ private ScheduledExecutorService _taskExecutor;
private final ScheduledExecutorService _timerTaskExecutor;
private ThreadPoolExecutorMonitor _monitor;
- public final static int TASK_THREADPOOL_SIZE = 40;
public TaskStateModelFactory(HelixManager manager, Map<String, TaskFactory>
taskFactoryRegistry) {
- this(manager, taskFactoryRegistry,
- Executors.newScheduledThreadPool(TASK_THREADPOOL_SIZE, new
ThreadFactory() {
- private AtomicInteger threadId = new AtomicInteger(0);
-
- @Override
- public Thread newThread(Runnable r) {
- return new Thread(r, "TaskStateModelFactory-task_thread-" +
threadId.getAndIncrement());
- }
- }));
+ this(manager, taskFactoryRegistry, null);
Review comment:
Combined with the previous comment.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
