yashmayya commented on code in PR #17623: URL: https://github.com/apache/pinot/pull/17623#discussion_r2855621888
########## pinot-core/src/test/java/org/apache/pinot/core/data/manager/SegmentOperationsExecutorServiceTest.java: ########## @@ -0,0 +1,270 @@ +/** + * 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.core.data.manager; + +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.Test; + +import static org.testng.Assert.*; + + +public class SegmentOperationsExecutorServiceTest { Review Comment: Do we really need this? `SegmentOperationsExecutorService` is simply implementing the existing decorator pattern which is already being used in a number of places. I'm not sure having these separate tests here is very useful. ########## pinot-server/src/main/java/org/apache/pinot/server/starter/helix/DefaultStateTransitionThreadPoolManager.java: ########## @@ -0,0 +1,300 @@ +/** + * 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.Map; +import java.util.Set; +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.config.provider.PinotClusterConfigChangeListener; +import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.executor.DecoratorExecutorService; +import org.apache.pinot.spi.utils.CommonConstants; +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>Cluster config: pinot.server.instance.stateTransitionThreadPoolSize (dynamic, via ZK cluster config)</li> + * <li>Pinot server config: pinot.server.instance.stateTransitionThreadPoolSize (static, from server properties)</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> + * + * <p>The pool size can be dynamically adjusted at runtime by updating the cluster config key + * {@code pinot.server.instance.stateTransitionThreadPoolSize}. This class implements + * {@link PinotClusterConfigChangeListener} and should be registered with the cluster config change handler.</p> + */ +public class DefaultStateTransitionThreadPoolManager + implements StateTransitionThreadPoolManager, PinotClusterConfigChangeListener { + 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 final ThreadPoolExecutor _threadPoolExecutor; + private final 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 thread pool is created immediately with the default core size so that + * {@link #getExecutorService(String)} never returns null. When {@link #onHelixManagerConnected()} is called, + * the pool size is adjusted based on configuration precedence (Pinot config > Helix config > default). + * + * @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; + int defaultSize = org.apache.pinot.spi.utils.CommonConstants.Server.DEFAULT_STATE_TRANSITION_THREAD_POOL_SIZE; Review Comment: nit: import directly ########## pinot-server/src/main/java/org/apache/pinot/server/starter/helix/DefaultStateTransitionThreadPoolManager.java: ########## @@ -0,0 +1,300 @@ +/** + * 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.Map; +import java.util.Set; +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.config.provider.PinotClusterConfigChangeListener; +import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.executor.DecoratorExecutorService; +import org.apache.pinot.spi.utils.CommonConstants; +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>Cluster config: pinot.server.instance.stateTransitionThreadPoolSize (dynamic, via ZK cluster config)</li> + * <li>Pinot server config: pinot.server.instance.stateTransitionThreadPoolSize (static, from server properties)</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> + * + * <p>The pool size can be dynamically adjusted at runtime by updating the cluster config key + * {@code pinot.server.instance.stateTransitionThreadPoolSize}. This class implements + * {@link PinotClusterConfigChangeListener} and should be registered with the cluster config change handler.</p> + */ +public class DefaultStateTransitionThreadPoolManager + implements StateTransitionThreadPoolManager, PinotClusterConfigChangeListener { + 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 final ThreadPoolExecutor _threadPoolExecutor; + private final 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 thread pool is created immediately with the default core size so that + * {@link #getExecutorService(String)} never returns null. When {@link #onHelixManagerConnected()} is called, + * the pool size is adjusted based on configuration precedence (Pinot config > Helix config > default). + * + * @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; + int defaultSize = org.apache.pinot.spi.utils.CommonConstants.Server.DEFAULT_STATE_TRANSITION_THREAD_POOL_SIZE; Review Comment: Same elsewhere -- 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]
