J-HowHuang commented on code in PR #17623: URL: https://github.com/apache/pinot/pull/17623#discussion_r2851533154
########## pinot-server/src/main/java/org/apache/pinot/server/starter/helix/DefaultStateTransitionThreadPoolManager.java: ########## @@ -0,0 +1,246 @@ +/** + * 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.pinot.server.starter.helix; + +import com.google.common.util.concurrent.ThreadFactoryBuilder; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import org.apache.helix.HelixManager; +import org.apache.helix.messaging.handling.MessageTask; +import org.apache.helix.model.HelixConfigScope; +import org.apache.helix.model.Message; +import org.apache.helix.model.builder.HelixConfigScopeBuilder; +import org.apache.helix.participant.statemachine.StateModelFactory; +import org.apache.pinot.core.data.manager.SegmentOperationsTaskContext; +import org.apache.pinot.core.data.manager.SegmentOperationsTaskType; +import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.executor.DecoratorExecutorService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/** + * Default state transition thread pool manager backed by a fixed-size pool. + * + * <p>This replaces the Helix-managed state transition thread pool to provide explicit context tracking + * for segment operations. The pool size is configurable via + * {@link org.apache.pinot.spi.utils.CommonConstants.Server#CONFIG_OF_STATE_TRANSITION_THREAD_POOL_SIZE}.</p> + * + * <p><b>Migration from Helix configuration:</b> Previously, the thread pool size was configured in ZooKeeper + * using the key "STATE_TRANSITION.maxThreads" at either the participant or cluster config level. + * For backward compatibility, this implementation will read from the legacy Helix config if the new + * Pinot config is not set.</p> + * + * <p><b>Configuration precedence:</b></p> + * <ol> + * <li>Pinot server config: pinot.server.instance.stateTransitionThreadPoolSize</li> + * <li>Helix instance config: STATE_TRANSITION.maxThreads (from CONFIGS/PARTICIPANT/<instance>)</li> + * <li>Helix cluster config: STATE_TRANSITION.maxThreads (from CONFIGS/CLUSTER/<cluster>)</li> + * <li>Default value: 40</li> + * </ol> + */ +public class DefaultStateTransitionThreadPoolManager implements StateTransitionThreadPoolManager { + private static final Logger LOGGER = LoggerFactory.getLogger(DefaultStateTransitionThreadPoolManager.class); + private static final String HELIX_STATE_TRANSITION_KEY = "STATE_TRANSITION.maxThreads"; + + private final PinotConfiguration _serverConf; + private final HelixManager _helixManager; + private ExecutorService _executorService; + + /** + * Creates a state transition thread pool manager with the default pool size configuration. + * The pool size is read from the server configuration. + * + * @param serverConf the server configuration + */ + public DefaultStateTransitionThreadPoolManager(PinotConfiguration serverConf) { + this(serverConf, null); + } + + /** + * Creates a state transition thread pool manager with backward compatibility for legacy Helix config. + * The executor service is created lazily in {@link #onHelixManagerConnected()} after Helix connects, + * allowing it to read legacy Helix config if needed. + * + * @param serverConf the server configuration + * @param helixManager the Helix manager to read legacy config from later (can be null) + */ + public DefaultStateTransitionThreadPoolManager(PinotConfiguration serverConf, + @Nullable HelixManager helixManager) { + _serverConf = serverConf; + _helixManager = helixManager; + // Executor service will be created in onHelixManagerConnected() Review Comment: Makes sense. It will now adjust the pool size upon helix manager connected, as well as cluster config changed so now we can dynamically adjust the pool size, versus in the past it needed a server restart. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
