You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ne...@apache.org on 2021/06/22 16:27:49 UTC
[incubator-pinot] branch master updated: Startable class from
config (#7075)
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 80fda68 Startable class from config (#7075)
80fda68 is described below
commit 80fda686272d52f74057588d95faefbc016bca9f
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Tue Jun 22 09:27:31 2021 -0700
Startable class from config (#7075)
---
...ixBrokerStarter.java => BaseBrokerStarter.java} | 97 +--
.../broker/broker/helix/HelixBrokerStarter.java | 391 +-----------
.../broker/broker/HelixBrokerStarterTest.java | 3 +-
...llerStarter.java => BaseControllerStarter.java} | 130 ++--
.../apache/pinot/controller/ControllerStarter.java | 664 +-------------------
.../pinot/controller/ControllerTestUtils.java | 13 +-
...ControllerPeriodicTaskStarterStatelessTest.java | 12 +-
.../pinot/controller/helix/ControllerTest.java | 14 +-
.../helix/PinotControllerModeStatelessTest.java | 27 +-
.../helix/core/minion/PinotTaskManagerTest.java | 3 +-
.../pinot/integration/tests/ClusterTest.java | 22 +-
...tlyOnceKafkaRealtimeClusterIntegrationTest.java | 3 +-
.../tests/LLCRealtimeClusterIntegrationTest.java | 3 +-
...rDownloadLLCRealtimeClusterIntegrationTest.java | 5 +-
.../tests/ServerStarterIntegrationTest.java | 6 +-
.../UpsertTableSegmentUploadIntegrationTest.java | 3 +-
.../ControllerLeaderLocatorIntegrationTest.java | 15 +-
.../{MinionStarter.java => BaseMinionStarter.java} | 37 +-
.../org/apache/pinot/minion/MinionStarter.java | 257 +-------
.../SegmentGenerationAndPushTaskGeneratorTest.java | 3 +-
...ixServerStarter.java => BaseServerStarter.java} | 76 +--
.../server/starter/helix/HelixServerStarter.java | 666 +--------------------
.../pinot/spi/services/ServiceStartable.java | 3 +
.../apache/pinot/spi/utils/CommonConstants.java | 5 +
.../pinot/tools/perf/PerfBenchmarkDriver.java | 28 +-
.../pinot/tools/service/PinotServiceManager.java | 63 +-
26 files changed, 307 insertions(+), 2242 deletions(-)
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
similarity index 86%
copy from pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
copy to pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
index 03d39b6..7178961 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
@@ -25,7 +25,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import javax.annotation.Nullable;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixConstants.ChangeType;
import org.apache.helix.HelixDataAccessor;
@@ -70,63 +69,39 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Base class for broker startable implementations
+ */
@SuppressWarnings("unused")
-public class HelixBrokerStarter implements ServiceStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(HelixBrokerStarter.class);
-
- private final PinotConfiguration _brokerConf;
- private final List<ListenerConfig> _listenerConfigs;
- private final String _clusterName;
- private final String _zkServers;
- private final String _brokerId;
-
- private final List<ClusterChangeHandler> _externalViewChangeHandlers = new ArrayList<>();
- private final List<ClusterChangeHandler> _instanceConfigChangeHandlers = new ArrayList<>();
- private final List<ClusterChangeHandler> _liveInstanceChangeHandlers = new ArrayList<>();
-
+public abstract class BaseBrokerStarter implements ServiceStartable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BaseBrokerStarter.class);
+
+ protected PinotConfiguration _brokerConf;
+ protected List<ListenerConfig> _listenerConfigs;
+ protected String _clusterName;
+ protected String _zkServers;
+ protected String _brokerId;
+ protected final List<ClusterChangeHandler> _externalViewChangeHandlers = new ArrayList<>();
+ protected final List<ClusterChangeHandler> _instanceConfigChangeHandlers = new ArrayList<>();
+ protected final List<ClusterChangeHandler> _liveInstanceChangeHandlers = new ArrayList<>();
// Spectator Helix manager handles the custom change listeners, properties read/write
- private HelixManager _spectatorHelixManager;
- private HelixAdmin _helixAdmin;
- private ZkHelixPropertyStore<ZNRecord> _propertyStore;
- private HelixDataAccessor _helixDataAccessor;
-
- private PinotMetricsRegistry _metricsRegistry;
- private BrokerMetrics _brokerMetrics;
- private RoutingManager _routingManager;
- private AccessControlFactory _accessControlFactory;
- private BrokerRequestHandler _brokerRequestHandler;
- private BrokerAdminApiApplication _brokerAdminApplication;
- private ClusterChangeMediator _clusterChangeMediator;
-
+ protected HelixManager _spectatorHelixManager;
+ protected HelixAdmin _helixAdmin;
+ protected ZkHelixPropertyStore<ZNRecord> _propertyStore;
+ protected HelixDataAccessor _helixDataAccessor;
+ protected PinotMetricsRegistry _metricsRegistry;
+ protected BrokerMetrics _brokerMetrics;
+ protected RoutingManager _routingManager;
+ protected AccessControlFactory _accessControlFactory;
+ protected BrokerRequestHandler _brokerRequestHandler;
+ protected BrokerAdminApiApplication _brokerAdminApplication;
+ protected ClusterChangeMediator _clusterChangeMediator;
// Participant Helix manager handles Helix functionality such as state transitions and messages
- private HelixManager _participantHelixManager;
+ protected HelixManager _participantHelixManager;
- @Deprecated
- public HelixBrokerStarter(PinotConfiguration brokerConf, String clusterName, String zkServer)
- throws Exception {
- this(brokerConf, clusterName, zkServer, null);
- }
-
- @Deprecated
- public HelixBrokerStarter(PinotConfiguration brokerConf, String clusterName, String zkServer,
- @Nullable String brokerHost)
+ @Override
+ public void init(PinotConfiguration brokerConf)
throws Exception {
- this(applyBrokerConfigs(brokerConf, clusterName, zkServer, brokerHost));
- }
-
- @Deprecated
- private static PinotConfiguration applyBrokerConfigs(PinotConfiguration brokerConf, String clusterName, String zkServers, @Nullable String brokerHost) {
- brokerConf.setProperty(Helix.CONFIG_OF_CLUSTER_NAME, clusterName);
- brokerConf.setProperty(Helix.CONFIG_OF_ZOOKEEPR_SERVER, zkServers);
- if (brokerHost == null) {
- brokerConf.clearProperty(Broker.CONFIG_OF_BROKER_HOSTNAME);
- } else {
- brokerConf.setProperty(Broker.CONFIG_OF_BROKER_HOSTNAME, brokerHost);
- }
- return brokerConf;
- }
-
- public HelixBrokerStarter(PinotConfiguration brokerConf) throws Exception {
_brokerConf = brokerConf;
_listenerConfigs = ListenerConfigUtil.buildBrokerConfigs(brokerConf);
setupHelixSystemProperties();
@@ -430,20 +405,4 @@ public class HelixBrokerStarter implements ServiceStartable {
public BrokerRequestHandler getBrokerRequestHandler() {
return _brokerRequestHandler;
}
-
- public static HelixBrokerStarter getDefault()
- throws Exception {
- Map<String, Object> properties = new HashMap<>();
-
- properties.put(Helix.KEY_OF_BROKER_QUERY_PORT, 5001);
- properties.put(Broker.CONFIG_OF_BROKER_TIMEOUT_MS, 60 * 1000L);
- properties.put(Helix.CONFIG_OF_CLUSTER_NAME, "quickstart");
- properties.put(Helix.CONFIG_OF_ZOOKEEPR_SERVER, "localhost:2122");
- return new HelixBrokerStarter(new PinotConfiguration(properties));
- }
-
- public static void main(String[] args)
- throws Exception {
- getDefault().start();
- }
}
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
index 03d39b6..8b9f622 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
@@ -18,100 +18,35 @@
*/
package org.apache.pinot.broker.broker.helix;
-import com.google.common.collect.ImmutableList;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
-import org.apache.helix.HelixAdmin;
-import org.apache.helix.HelixConstants.ChangeType;
-import org.apache.helix.HelixDataAccessor;
-import org.apache.helix.HelixManager;
-import org.apache.helix.HelixManagerFactory;
-import org.apache.helix.InstanceType;
-import org.apache.helix.SystemPropertyKeys;
-import org.apache.helix.ZNRecord;
-import org.apache.helix.model.HelixConfigScope;
-import org.apache.helix.model.IdealState;
-import org.apache.helix.model.InstanceConfig;
-import org.apache.helix.model.Message;
-import org.apache.helix.model.builder.HelixConfigScopeBuilder;
-import org.apache.helix.store.zk.ZkHelixPropertyStore;
-import org.apache.pinot.broker.broker.AccessControlFactory;
-import org.apache.pinot.broker.broker.BrokerAdminApiApplication;
-import org.apache.pinot.broker.queryquota.HelixExternalViewBasedQueryQuotaManager;
-import org.apache.pinot.broker.requesthandler.BrokerRequestHandler;
-import org.apache.pinot.broker.requesthandler.SingleConnectionBrokerRequestHandler;
-import org.apache.pinot.broker.routing.RoutingManager;
-import org.apache.pinot.common.Utils;
-import org.apache.pinot.common.function.FunctionRegistry;
-import org.apache.pinot.common.metadata.ZKMetadataProvider;
-import org.apache.pinot.common.metrics.BrokerMeter;
-import org.apache.pinot.common.metrics.BrokerMetrics;
-import org.apache.pinot.common.metrics.PinotMetricUtils;
-import org.apache.pinot.common.utils.ServiceStatus;
-import org.apache.pinot.common.utils.config.TagNameUtils;
-import org.apache.pinot.common.utils.helix.TableCache;
-import org.apache.pinot.core.transport.ListenerConfig;
-import org.apache.pinot.core.transport.TlsConfig;
-import org.apache.pinot.core.util.ListenerConfigUtil;
-import org.apache.pinot.core.util.TlsUtils;
import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.metrics.PinotMetricsRegistry;
-import org.apache.pinot.spi.services.ServiceRole;
-import org.apache.pinot.spi.services.ServiceStartable;
import org.apache.pinot.spi.utils.CommonConstants.Broker;
import org.apache.pinot.spi.utils.CommonConstants.Helix;
-import org.apache.pinot.spi.utils.NetUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+/**
+ * Startable implementation for Pinot broker.
+ * Contains methods to start and stop a Pinot broker
+ */
@SuppressWarnings("unused")
-public class HelixBrokerStarter implements ServiceStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(HelixBrokerStarter.class);
-
- private final PinotConfiguration _brokerConf;
- private final List<ListenerConfig> _listenerConfigs;
- private final String _clusterName;
- private final String _zkServers;
- private final String _brokerId;
-
- private final List<ClusterChangeHandler> _externalViewChangeHandlers = new ArrayList<>();
- private final List<ClusterChangeHandler> _instanceConfigChangeHandlers = new ArrayList<>();
- private final List<ClusterChangeHandler> _liveInstanceChangeHandlers = new ArrayList<>();
-
- // Spectator Helix manager handles the custom change listeners, properties read/write
- private HelixManager _spectatorHelixManager;
- private HelixAdmin _helixAdmin;
- private ZkHelixPropertyStore<ZNRecord> _propertyStore;
- private HelixDataAccessor _helixDataAccessor;
-
- private PinotMetricsRegistry _metricsRegistry;
- private BrokerMetrics _brokerMetrics;
- private RoutingManager _routingManager;
- private AccessControlFactory _accessControlFactory;
- private BrokerRequestHandler _brokerRequestHandler;
- private BrokerAdminApiApplication _brokerAdminApplication;
- private ClusterChangeMediator _clusterChangeMediator;
+public class HelixBrokerStarter extends BaseBrokerStarter {
- // Participant Helix manager handles Helix functionality such as state transitions and messages
- private HelixManager _participantHelixManager;
+ public HelixBrokerStarter() {
+ }
@Deprecated
public HelixBrokerStarter(PinotConfiguration brokerConf, String clusterName, String zkServer)
throws Exception {
- this(brokerConf, clusterName, zkServer, null);
+ init(applyBrokerConfigs(brokerConf, clusterName, zkServer, null));
}
@Deprecated
public HelixBrokerStarter(PinotConfiguration brokerConf, String clusterName, String zkServer,
@Nullable String brokerHost)
throws Exception {
- this(applyBrokerConfigs(brokerConf, clusterName, zkServer, brokerHost));
+ init(applyBrokerConfigs(brokerConf, clusterName, zkServer, brokerHost));
}
@Deprecated
@@ -126,309 +61,9 @@ public class HelixBrokerStarter implements ServiceStartable {
return brokerConf;
}
+ @Deprecated
public HelixBrokerStarter(PinotConfiguration brokerConf) throws Exception {
- _brokerConf = brokerConf;
- _listenerConfigs = ListenerConfigUtil.buildBrokerConfigs(brokerConf);
- setupHelixSystemProperties();
-
- _clusterName = brokerConf.getProperty(Helix.CONFIG_OF_CLUSTER_NAME);
-
- // Remove all white-spaces from the list of zkServers (if any).
- _zkServers = brokerConf.getProperty(Helix.CONFIG_OF_ZOOKEEPR_SERVER).replaceAll("\\s+", "");
- String brokerHost = brokerConf.getProperty(Broker.CONFIG_OF_BROKER_HOSTNAME);
- if (brokerHost == null) {
- brokerHost = _brokerConf.getProperty(Helix.SET_INSTANCE_ID_TO_HOSTNAME_KEY, false) ? NetUtils
- .getHostnameOrAddress() : NetUtils.getHostAddress();
- }
-
- _brokerId = _brokerConf.getProperty(Helix.Instance.INSTANCE_ID_KEY,
- Helix.PREFIX_OF_BROKER_INSTANCE + brokerHost + "_" + _listenerConfigs.get(0).getPort());
-
- _brokerConf.addProperty(Broker.CONFIG_OF_BROKER_ID, _brokerId);
- }
-
- private void setupHelixSystemProperties() {
- // NOTE: Helix will disconnect the manager and disable the instance if it detects flapping (too frequent disconnect
- // from ZooKeeper). Setting flapping time window to a small value can avoid this from happening. Helix ignores the
- // non-positive value, so set the default value as 1.
- System.setProperty(SystemPropertyKeys.FLAPPING_TIME_WINDOW,
- _brokerConf.getProperty(Helix.CONFIG_OF_BROKER_FLAPPING_TIME_WINDOW_MS, Helix.DEFAULT_FLAPPING_TIME_WINDOW_MS));
- }
-
- /**
- * Adds an external view change handler to handle Helix external view change callbacks.
- * <p>NOTE: all change handlers will be run in a single thread, so any slow change handler can block other change
- * handlers from running. For slow change handler, make it asynchronous.
- */
- public void addExternalViewChangeHandler(ClusterChangeHandler externalViewChangeHandler) {
- _externalViewChangeHandlers.add(externalViewChangeHandler);
- }
-
- /**
- * Adds an instance config change handler to handle Helix instance config change callbacks.
- * <p>NOTE: all change handlers will be run in a single thread, so any slow change handler can block other change
- * handlers from running. For slow change handler, make it asynchronous.
- */
- public void addInstanceConfigChangeHandler(ClusterChangeHandler instanceConfigChangeHandler) {
- _instanceConfigChangeHandlers.add(instanceConfigChangeHandler);
- }
-
- /**
- * Adds a live instance change handler to handle Helix live instance change callbacks.
- * <p>NOTE: all change handlers will be run in a single thread, so any slow change handler can block other change
- * handlers from running. For slow change handler, make it asynchronous.
- */
- public void addLiveInstanceChangeHandler(ClusterChangeHandler liveInstanceChangeHandler) {
- _liveInstanceChangeHandlers.add(liveInstanceChangeHandler);
- }
-
- @Override
- public ServiceRole getServiceRole() {
- return ServiceRole.BROKER;
- }
-
- @Override
- public String getInstanceId() {
- return _brokerId;
- }
-
- @Override
- public PinotConfiguration getConfig() {
- return _brokerConf;
- }
-
- @Override
- public void start()
- throws Exception {
- LOGGER.info("Starting Pinot broker");
- Utils.logVersions();
-
- LOGGER.info("Connecting spectator Helix manager");
- _spectatorHelixManager =
- HelixManagerFactory.getZKHelixManager(_clusterName, _brokerId, InstanceType.SPECTATOR, _zkServers);
- _spectatorHelixManager.connect();
- _helixAdmin = _spectatorHelixManager.getClusterManagmentTool();
- _propertyStore = _spectatorHelixManager.getHelixPropertyStore();
- _helixDataAccessor = _spectatorHelixManager.getHelixDataAccessor();
-
- // Fetch cluster level config from ZK
- HelixConfigScope helixConfigScope =
- new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER).forCluster(_clusterName).build();
- Map<String, String> configMap = _helixAdmin.getConfig(helixConfigScope, Arrays
- .asList(Helix.ENABLE_CASE_INSENSITIVE_KEY, Helix.DEPRECATED_ENABLE_CASE_INSENSITIVE_KEY,
- Broker.CONFIG_OF_ENABLE_QUERY_LIMIT_OVERRIDE, Helix.DEFAULT_HYPERLOGLOG_LOG2M_KEY,
- Helix.ENABLE_DISTINCT_COUNT_BITMAP_OVERRIDE_KEY));
-
- boolean caseInsensitive = Boolean.parseBoolean(configMap.get(Helix.ENABLE_CASE_INSENSITIVE_KEY)) || Boolean
- .parseBoolean(configMap.get(Helix.DEPRECATED_ENABLE_CASE_INSENSITIVE_KEY));
-
- String log2mStr = configMap.get(Helix.DEFAULT_HYPERLOGLOG_LOG2M_KEY);
- if (log2mStr != null) {
- try {
- _brokerConf.setProperty(Helix.DEFAULT_HYPERLOGLOG_LOG2M_KEY, Integer.parseInt(log2mStr));
- } catch (NumberFormatException e) {
- LOGGER.warn("Invalid config of '{}': '{}', using: {} as the default log2m for HyperLogLog",
- Helix.DEFAULT_HYPERLOGLOG_LOG2M_KEY, log2mStr, Helix.DEFAULT_HYPERLOGLOG_LOG2M);
- }
- }
-
- if (Boolean.parseBoolean(configMap.get(Broker.CONFIG_OF_ENABLE_QUERY_LIMIT_OVERRIDE))) {
- _brokerConf.setProperty(Broker.CONFIG_OF_ENABLE_QUERY_LIMIT_OVERRIDE, true);
- }
-
- if (Boolean.parseBoolean(configMap.get(Helix.ENABLE_DISTINCT_COUNT_BITMAP_OVERRIDE_KEY))) {
- _brokerConf.setProperty(Helix.ENABLE_DISTINCT_COUNT_BITMAP_OVERRIDE_KEY, true);
- }
-
- LOGGER.info("Setting up broker request handler");
- // Set up metric registry and broker metrics
- PinotConfiguration metricsConfiguration = _brokerConf.subset(Broker.METRICS_CONFIG_PREFIX);
- PinotMetricUtils.init(metricsConfiguration);
- _metricsRegistry = PinotMetricUtils.getPinotMetricsRegistry();
- _brokerMetrics = new BrokerMetrics(
- _brokerConf.getProperty(Broker.CONFIG_OF_METRICS_NAME_PREFIX, Broker.DEFAULT_METRICS_NAME_PREFIX),
- _metricsRegistry,
- _brokerConf.getProperty(Broker.CONFIG_OF_ENABLE_TABLE_LEVEL_METRICS, Broker.DEFAULT_ENABLE_TABLE_LEVEL_METRICS),
- _brokerConf.getProperty(Broker.CONFIG_OF_ALLOWED_TABLES_FOR_EMITTING_METRICS, Collections.emptyList()));
- _brokerMetrics.initializeGlobalMeters();
- // Set up request handling classes
- _routingManager = new RoutingManager(_brokerMetrics);
- _routingManager.init(_spectatorHelixManager);
- _accessControlFactory = AccessControlFactory.loadFactory(_brokerConf.subset(Broker.ACCESS_CONTROL_CONFIG_PREFIX));
- HelixExternalViewBasedQueryQuotaManager queryQuotaManager =
- new HelixExternalViewBasedQueryQuotaManager(_brokerMetrics, _brokerId);
- queryQuotaManager.init(_spectatorHelixManager);
- // Initialize FunctionRegistry before starting the broker request handler
- FunctionRegistry.init();
- TableCache tableCache = new TableCache(_propertyStore, caseInsensitive);
- // Configure TLS for netty connection to server
- TlsConfig tlsDefaults = TlsUtils.extractTlsConfig(_brokerConf, Broker.BROKER_TLS_PREFIX);
-
- if (_brokerConf.getProperty(Broker.BROKER_NETTYTLS_ENABLED, false)) {
- _brokerRequestHandler = new SingleConnectionBrokerRequestHandler(_brokerConf, _routingManager,
- _accessControlFactory, queryQuotaManager, tableCache, _brokerMetrics, tlsDefaults);
- } else {
- _brokerRequestHandler = new SingleConnectionBrokerRequestHandler(_brokerConf, _routingManager,
- _accessControlFactory, queryQuotaManager, tableCache, _brokerMetrics, null);
- }
-
- LOGGER.info("Starting broker admin application on: {}", ListenerConfigUtil.toString(_listenerConfigs));
- _brokerAdminApplication =
- new BrokerAdminApiApplication(_routingManager, _brokerRequestHandler, _brokerMetrics, _brokerConf);
- _brokerAdminApplication.start(_listenerConfigs);
-
- LOGGER.info("Initializing cluster change mediator");
- for (ClusterChangeHandler externalViewChangeHandler : _externalViewChangeHandlers) {
- externalViewChangeHandler.init(_spectatorHelixManager);
- }
- _externalViewChangeHandlers.add(_routingManager);
- _externalViewChangeHandlers.add(queryQuotaManager);
- for (ClusterChangeHandler instanceConfigChangeHandler : _instanceConfigChangeHandlers) {
- instanceConfigChangeHandler.init(_spectatorHelixManager);
- }
- _instanceConfigChangeHandlers.add(_routingManager);
- _instanceConfigChangeHandlers.add(queryQuotaManager);
- for (ClusterChangeHandler liveInstanceChangeHandler : _liveInstanceChangeHandlers) {
- liveInstanceChangeHandler.init(_spectatorHelixManager);
- }
- Map<ChangeType, List<ClusterChangeHandler>> clusterChangeHandlersMap = new HashMap<>();
- clusterChangeHandlersMap.put(ChangeType.EXTERNAL_VIEW, _externalViewChangeHandlers);
- clusterChangeHandlersMap.put(ChangeType.INSTANCE_CONFIG, _instanceConfigChangeHandlers);
- if (!_liveInstanceChangeHandlers.isEmpty()) {
- clusterChangeHandlersMap.put(ChangeType.LIVE_INSTANCE, _liveInstanceChangeHandlers);
- }
- _clusterChangeMediator = new ClusterChangeMediator(clusterChangeHandlersMap, _brokerMetrics);
- _clusterChangeMediator.start();
- _spectatorHelixManager.addExternalViewChangeListener(_clusterChangeMediator);
- _spectatorHelixManager.addInstanceConfigChangeListener(_clusterChangeMediator);
- if (!_liveInstanceChangeHandlers.isEmpty()) {
- _spectatorHelixManager.addLiveInstanceChangeListener(_clusterChangeMediator);
- }
-
- LOGGER.info("Connecting participant Helix manager");
- _participantHelixManager =
- HelixManagerFactory.getZKHelixManager(_clusterName, _brokerId, InstanceType.PARTICIPANT, _zkServers);
- // Register state model factory
- _participantHelixManager.getStateMachineEngine()
- .registerStateModelFactory(BrokerResourceOnlineOfflineStateModelFactory.getStateModelDef(),
- new BrokerResourceOnlineOfflineStateModelFactory(_propertyStore, _helixDataAccessor, _routingManager,
- queryQuotaManager));
- // Register user-define message handler factory
- _participantHelixManager.getMessagingService()
- .registerMessageHandlerFactory(Message.MessageType.USER_DEFINE_MSG.toString(),
- new BrokerUserDefinedMessageHandlerFactory(_routingManager, queryQuotaManager));
- _participantHelixManager.connect();
- addInstanceTagIfNeeded();
- _brokerMetrics
- .addCallbackGauge(Helix.INSTANCE_CONNECTED_METRIC_NAME, () -> _participantHelixManager.isConnected() ? 1L : 0L);
- _participantHelixManager
- .addPreConnectCallback(() -> _brokerMetrics.addMeteredGlobalValue(BrokerMeter.HELIX_ZOOKEEPER_RECONNECTS, 1L));
-
- // Register the service status handler
- registerServiceStatusHandler();
-
- LOGGER.info("Finish starting Pinot broker");
- }
-
- /**
- * Fetches the resources to monitor and registers the {@link org.apache.pinot.common.utils.ServiceStatus.ServiceStatusCallback}s
- */
- private void registerServiceStatusHandler() {
- List<String> resourcesToMonitor = new ArrayList<>(1);
- IdealState brokerResourceIdealState =
- _helixAdmin.getResourceIdealState(_clusterName, Helix.BROKER_RESOURCE_INSTANCE);
- if (brokerResourceIdealState != null && brokerResourceIdealState.isEnabled()) {
- for (String partitionName : brokerResourceIdealState.getPartitionSet()) {
- if (brokerResourceIdealState.getInstanceSet(partitionName).contains(_brokerId)) {
- resourcesToMonitor.add(Helix.BROKER_RESOURCE_INSTANCE);
- break;
- }
- }
- }
-
- double minResourcePercentForStartup = _brokerConf
- .getProperty(Broker.CONFIG_OF_BROKER_MIN_RESOURCE_PERCENT_FOR_START,
- Broker.DEFAULT_BROKER_MIN_RESOURCE_PERCENT_FOR_START);
-
- LOGGER.info("Registering service status handler");
- ServiceStatus.setServiceStatusCallback(_brokerId, new ServiceStatus.MultipleCallbackServiceStatusCallback(
- ImmutableList.of(new ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_participantHelixManager,
- _clusterName, _brokerId, resourcesToMonitor, minResourcePercentForStartup),
- new ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_participantHelixManager,
- _clusterName, _brokerId, resourcesToMonitor, minResourcePercentForStartup))));
- }
-
- private void addInstanceTagIfNeeded() {
- InstanceConfig instanceConfig =
- _helixDataAccessor.getProperty(_helixDataAccessor.keyBuilder().instanceConfig(_brokerId));
- List<String> instanceTags = instanceConfig.getTags();
- if (instanceTags == null || instanceTags.isEmpty()) {
- if (ZKMetadataProvider.getClusterTenantIsolationEnabled(_propertyStore)) {
- _helixAdmin.addInstanceTag(_clusterName, _brokerId, TagNameUtils.getBrokerTagForTenant(null));
- } else {
- _helixAdmin.addInstanceTag(_clusterName, _brokerId, Helix.UNTAGGED_BROKER_INSTANCE);
- }
- }
- }
-
- @Override
- public void stop() {
- LOGGER.info("Shutting down Pinot broker");
-
- LOGGER.info("Disconnecting participant Helix manager");
- _participantHelixManager.disconnect();
-
- LOGGER.info("Stopping cluster change mediator");
- _clusterChangeMediator.stop();
-
- // Delay shutdown of request handler so that the pending queries can be finished. The participant Helix manager has
- // been disconnected, so instance should disappear from ExternalView soon and stop getting new queries.
- long delayShutdownTimeMs =
- _brokerConf.getProperty(Broker.CONFIG_OF_DELAY_SHUTDOWN_TIME_MS, Broker.DEFAULT_DELAY_SHUTDOWN_TIME_MS);
- LOGGER
- .info("Wait for {}ms before shutting down request handler to finish the pending queries", delayShutdownTimeMs);
- try {
- Thread.sleep(delayShutdownTimeMs);
- } catch (Exception e) {
- LOGGER.error("Caught exception while waiting for shutdown delay of {}ms", delayShutdownTimeMs, e);
- }
-
- LOGGER.info("Shutting down request handler and broker admin application");
- _brokerRequestHandler.shutDown();
- _brokerAdminApplication.stop();
-
- LOGGER.info("Disconnecting spectator Helix manager");
- _spectatorHelixManager.disconnect();
-
- LOGGER.info("Deregistering service status handler");
- ServiceStatus.removeServiceStatusCallback(_brokerId);
- LOGGER.info("Shutdown Broker Metrics Registry");
- _metricsRegistry.shutdown();
- LOGGER.info("Finish shutting down Pinot broker for {}", _brokerId);
- }
-
- public HelixManager getSpectatorHelixManager() {
- return _spectatorHelixManager;
- }
-
- public PinotMetricsRegistry getMetricsRegistry() {
- return _metricsRegistry;
- }
-
- public BrokerMetrics getBrokerMetrics() {
- return _brokerMetrics;
- }
-
- public RoutingManager getRoutingManager() {
- return _routingManager;
- }
-
- public AccessControlFactory getAccessControlFactory() {
- return _accessControlFactory;
- }
-
- public BrokerRequestHandler getBrokerRequestHandler() {
- return _brokerRequestHandler;
+ init(brokerConf);
}
public static HelixBrokerStarter getDefault()
@@ -439,7 +74,9 @@ public class HelixBrokerStarter implements ServiceStartable {
properties.put(Broker.CONFIG_OF_BROKER_TIMEOUT_MS, 60 * 1000L);
properties.put(Helix.CONFIG_OF_CLUSTER_NAME, "quickstart");
properties.put(Helix.CONFIG_OF_ZOOKEEPR_SERVER, "localhost:2122");
- return new HelixBrokerStarter(new PinotConfiguration(properties));
+ HelixBrokerStarter helixBrokerStarter = new HelixBrokerStarter();
+ helixBrokerStarter.init(new PinotConfiguration(properties));
+ return helixBrokerStarter;
}
public static void main(String[] args)
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/broker/HelixBrokerStarterTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/HelixBrokerStarterTest.java
index 6e21311..658ada7 100644
--- a/pinot-broker/src/test/java/org/apache/pinot/broker/broker/HelixBrokerStarterTest.java
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/broker/HelixBrokerStarterTest.java
@@ -79,7 +79,8 @@ public class HelixBrokerStarterTest extends ControllerTest {
properties.put(Helix.CONFIG_OF_CLUSTER_NAME, getHelixClusterName());
properties.put(Helix.CONFIG_OF_ZOOKEEPR_SERVER, getZkUrl());
- _brokerStarter = new HelixBrokerStarter(new PinotConfiguration(properties));
+ _brokerStarter = new HelixBrokerStarter();
+ _brokerStarter.init(new PinotConfiguration(properties));
_brokerStarter.start();
addFakeBrokerInstancesToAutoJoinHelixCluster(NUM_BROKERS - 1, true);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java b/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
similarity index 88%
copy from pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
copy to pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
index 6995c02..d3bb073 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
@@ -96,8 +96,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class ControllerStarter implements ServiceStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(ControllerStarter.class);
+/**
+ * Base class for controller startables
+ */
+public abstract class BaseControllerStarter implements ServiceStartable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BaseControllerStarter.class);
private static final String METRICS_REGISTRY_NAME = "pinot.controller.metrics";
private static final Long DATA_DIRECTORY_MISSING_VALUE = 1000000L;
@@ -105,55 +108,52 @@ public class ControllerStarter implements ServiceStartable {
private static final String METADATA_EVENT_NOTIFIER_PREFIX = "metadata.event.notifier";
private static final String MAX_STATE_TRANSITIONS_PER_INSTANCE = "MaxStateTransitionsPerInstance";
- private final ControllerConf _config;
- private final List<ListenerConfig> _listenerConfigs;
- private final ControllerAdminApiApplication _adminApp;
+ protected ControllerConf _config;
+ protected List<ListenerConfig> _listenerConfigs;
+ protected ControllerAdminApiApplication _adminApp;
// TODO: rename this variable once it's full separated with Helix controller.
- private final PinotHelixResourceManager _helixResourceManager;
- private final ExecutorService _executorService;
-
- private final String _helixZkURL;
- private final String _helixClusterName;
- private final String _helixControllerInstanceId;
- private final String _helixParticipantInstanceId;
- private final boolean _isUpdateStateModel;
- private final boolean _enableBatchMessageMode;
- private final ControllerConf.ControllerMode _controllerMode;
-
- private HelixManager _helixControllerManager;
- private HelixManager _helixParticipantManager;
- private PinotMetricsRegistry _metricsRegistry;
- private ControllerMetrics _controllerMetrics;
-
+ protected PinotHelixResourceManager _helixResourceManager;
+ protected ExecutorService _executorService;
+ protected String _helixZkURL;
+ protected String _helixClusterName;
+ protected String _helixControllerInstanceId;
+ protected String _helixParticipantInstanceId;
+ protected boolean _isUpdateStateModel;
+ protected boolean _enableBatchMessageMode;
+ protected ControllerConf.ControllerMode _controllerMode;
+ protected HelixManager _helixControllerManager;
+ protected HelixManager _helixParticipantManager;
+ protected PinotMetricsRegistry _metricsRegistry;
+ protected ControllerMetrics _controllerMetrics;
// Can only be constructed after resource manager getting started
- private OfflineSegmentIntervalChecker _offlineSegmentIntervalChecker;
- private RealtimeSegmentValidationManager _realtimeSegmentValidationManager;
- private BrokerResourceValidationManager _brokerResourceValidationManager;
- private SegmentRelocator _segmentRelocator;
- private RetentionManager _retentionManager;
- private SegmentStatusChecker _segmentStatusChecker;
- private PinotTaskManager _taskManager;
- private PeriodicTaskScheduler _periodicTaskScheduler;
- private PinotHelixTaskResourceManager _helixTaskResourceManager;
- private PinotRealtimeSegmentManager _realtimeSegmentsManager;
- private PinotLLCRealtimeSegmentManager _pinotLLCRealtimeSegmentManager;
- private SegmentCompletionManager _segmentCompletionManager;
- private LeadControllerManager _leadControllerManager;
- private List<ServiceStatus.ServiceStatusCallback> _serviceStatusCallbackList;
- private MinionInstancesCleanupTask _minionInstancesCleanupTask;
-
- public ControllerStarter(ControllerConf conf) {
- _config = conf;
+ protected OfflineSegmentIntervalChecker _offlineSegmentIntervalChecker;
+ protected RealtimeSegmentValidationManager _realtimeSegmentValidationManager;
+ protected BrokerResourceValidationManager _brokerResourceValidationManager;
+ protected SegmentRelocator _segmentRelocator;
+ protected RetentionManager _retentionManager;
+ protected SegmentStatusChecker _segmentStatusChecker;
+ protected PinotTaskManager _taskManager;
+ protected PeriodicTaskScheduler _periodicTaskScheduler;
+ protected PinotHelixTaskResourceManager _helixTaskResourceManager;
+ protected PinotRealtimeSegmentManager _realtimeSegmentsManager;
+ protected PinotLLCRealtimeSegmentManager _pinotLLCRealtimeSegmentManager;
+ protected SegmentCompletionManager _segmentCompletionManager;
+ protected LeadControllerManager _leadControllerManager;
+ protected List<ServiceStatus.ServiceStatusCallback> _serviceStatusCallbackList;
+ protected MinionInstancesCleanupTask _minionInstancesCleanupTask;
+
+ @Override
+ public void init(PinotConfiguration pinotConfiguration)
+ throws Exception {
+ _config = new ControllerConf(pinotConfiguration.toMap());
inferHostnameIfNeeded(_config);
setupHelixSystemProperties();
-
- _controllerMode = conf.getControllerMode();
+ _controllerMode = _config.getControllerMode();
// Helix related settings.
_helixZkURL = HelixConfig.getAbsoluteZkPathForHelix(_config.getZkStr());
_helixClusterName = _config.getHelixClusterName();
_listenerConfigs = ListenerConfigUtil.buildControllerConfigs(_config);
-
- String host = conf.getControllerHost();
+ String host = _config.getControllerHost();
int port = _listenerConfigs.get(0).getPort();
_helixControllerInstanceId = host + "_" + port;
@@ -170,7 +170,7 @@ public class ControllerStarter implements ServiceStartable {
// Initialize FunctionRegistry before starting the admin application (PinotQueryResource requires it to compile
// queries)
FunctionRegistry.init();
- _adminApp = new ControllerAdminApiApplication(conf);
+ _adminApp = new ControllerAdminApiApplication(_config);
// Do not use this before the invocation of {@link PinotHelixResourceManager::start()}, which happens in {@link ControllerStarter::start()}
_helixResourceManager = new PinotHelixResourceManager(_config);
_executorService =
@@ -670,46 +670,4 @@ public class ControllerStarter implements ServiceStartable {
public ControllerMetrics getControllerMetrics() {
return _controllerMetrics;
}
-
- public static ControllerStarter startDefault() {
- return startDefault(null);
- }
-
- public static ControllerStarter startDefault(File webappPath) {
- final ControllerConf conf = new ControllerConf();
- conf.setControllerHost("localhost");
- conf.setControllerPort("9000");
- conf.setDataDir("/tmp/PinotController");
- conf.setZkStr("localhost:2122");
- conf.setHelixClusterName("quickstart");
- if (webappPath == null) {
- String path = ControllerStarter.class.getClassLoader().getResource("webapp").getFile();
- if (!path.startsWith("file://")) {
- path = "file://" + path;
- }
- conf.setQueryConsolePath(path);
- } else {
- conf.setQueryConsolePath("file://" + webappPath.getAbsolutePath());
- }
-
- conf.setControllerVipHost("localhost");
- conf.setControllerVipProtocol(CommonConstants.HTTP_PROTOCOL);
- conf.setRetentionControllerFrequencyInSeconds(3600 * 6);
- conf.setOfflineSegmentIntervalCheckerFrequencyInSeconds(3600);
- conf.setRealtimeSegmentValidationFrequencyInSeconds(3600);
- conf.setBrokerResourceValidationFrequencyInSeconds(3600);
- conf.setStatusCheckerFrequencyInSeconds(5 * 60);
- conf.setSegmentRelocatorFrequencyInSeconds(3600);
- conf.setStatusCheckerWaitForPushTimeInSeconds(10 * 60);
- conf.setTenantIsolationEnabled(true);
-
- final ControllerStarter starter = new ControllerStarter(conf);
-
- starter.start();
- return starter;
- }
-
- public static void main(String[] args) {
- startDefault();
- }
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
index 6995c02..29b3ce5 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerStarter.java
@@ -18,664 +18,32 @@
*/
package org.apache.pinot.controller;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.primitives.Longs;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import org.apache.commons.httpclient.HttpConnectionManager;
-import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.helix.HelixManager;
-import org.apache.helix.HelixManagerFactory;
-import org.apache.helix.InstanceType;
-import org.apache.helix.SystemPropertyKeys;
-import org.apache.helix.api.listeners.ControllerChangeListener;
-import org.apache.helix.model.ClusterConstraints;
-import org.apache.helix.model.ConstraintItem;
-import org.apache.helix.model.MasterSlaveSMD;
-import org.apache.helix.model.Message;
-import org.apache.helix.task.TaskDriver;
-import org.apache.pinot.common.Utils;
-import org.apache.pinot.common.function.FunctionRegistry;
-import org.apache.pinot.common.metrics.ControllerMeter;
-import org.apache.pinot.common.metrics.ControllerMetrics;
-import org.apache.pinot.common.metrics.PinotMetricUtils;
-import org.apache.pinot.common.metrics.ValidationMetrics;
-import org.apache.pinot.common.utils.ServiceStatus;
-import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
-import org.apache.pinot.common.utils.helix.LeadControllerUtils;
-import org.apache.pinot.controller.api.ControllerAdminApiApplication;
-import org.apache.pinot.controller.api.access.AccessControlFactory;
-import org.apache.pinot.controller.api.events.MetadataEventNotifierFactory;
-import org.apache.pinot.controller.api.resources.ControllerFilePathProvider;
-import org.apache.pinot.controller.api.resources.InvalidControllerConfigException;
-import org.apache.pinot.controller.helix.SegmentStatusChecker;
-import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
-import org.apache.pinot.controller.helix.core.minion.MinionInstancesCleanupTask;
-import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager;
-import org.apache.pinot.controller.helix.core.minion.PinotTaskManager;
-import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
-import org.apache.pinot.controller.helix.core.realtime.PinotRealtimeSegmentManager;
-import org.apache.pinot.controller.helix.core.realtime.SegmentCompletionManager;
-import org.apache.pinot.controller.helix.core.relocation.SegmentRelocator;
-import org.apache.pinot.controller.helix.core.retention.RetentionManager;
-import org.apache.pinot.controller.helix.core.statemodel.LeadControllerResourceMasterSlaveStateModelFactory;
-import org.apache.pinot.controller.helix.core.util.HelixSetupUtils;
-import org.apache.pinot.controller.helix.starter.HelixConfig;
-import org.apache.pinot.controller.validation.BrokerResourceValidationManager;
-import org.apache.pinot.controller.validation.OfflineSegmentIntervalChecker;
-import org.apache.pinot.controller.validation.RealtimeSegmentValidationManager;
-import org.apache.pinot.core.periodictask.PeriodicTask;
-import org.apache.pinot.core.periodictask.PeriodicTaskScheduler;
-import org.apache.pinot.core.transport.ListenerConfig;
-import org.apache.pinot.core.transport.TlsConfig;
-import org.apache.pinot.core.util.ListenerConfigUtil;
-import org.apache.pinot.core.util.TlsUtils;
-import org.apache.pinot.spi.crypt.PinotCrypterFactory;
import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.filesystem.PinotFSFactory;
-import org.apache.pinot.spi.metrics.PinotMetricsRegistry;
-import org.apache.pinot.spi.services.ServiceRole;
-import org.apache.pinot.spi.services.ServiceStartable;
import org.apache.pinot.spi.utils.CommonConstants;
-import org.apache.pinot.spi.utils.NetUtils;
-import org.glassfish.hk2.utilities.binding.AbstractBinder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-public class ControllerStarter implements ServiceStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(ControllerStarter.class);
-
- private static final String METRICS_REGISTRY_NAME = "pinot.controller.metrics";
- private static final Long DATA_DIRECTORY_MISSING_VALUE = 1000000L;
- private static final Long DATA_DIRECTORY_EXCEPTION_VALUE = 1100000L;
- private static final String METADATA_EVENT_NOTIFIER_PREFIX = "metadata.event.notifier";
- private static final String MAX_STATE_TRANSITIONS_PER_INSTANCE = "MaxStateTransitionsPerInstance";
-
- private final ControllerConf _config;
- private final List<ListenerConfig> _listenerConfigs;
- private final ControllerAdminApiApplication _adminApp;
- // TODO: rename this variable once it's full separated with Helix controller.
- private final PinotHelixResourceManager _helixResourceManager;
- private final ExecutorService _executorService;
-
- private final String _helixZkURL;
- private final String _helixClusterName;
- private final String _helixControllerInstanceId;
- private final String _helixParticipantInstanceId;
- private final boolean _isUpdateStateModel;
- private final boolean _enableBatchMessageMode;
- private final ControllerConf.ControllerMode _controllerMode;
-
- private HelixManager _helixControllerManager;
- private HelixManager _helixParticipantManager;
- private PinotMetricsRegistry _metricsRegistry;
- private ControllerMetrics _controllerMetrics;
-
- // Can only be constructed after resource manager getting started
- private OfflineSegmentIntervalChecker _offlineSegmentIntervalChecker;
- private RealtimeSegmentValidationManager _realtimeSegmentValidationManager;
- private BrokerResourceValidationManager _brokerResourceValidationManager;
- private SegmentRelocator _segmentRelocator;
- private RetentionManager _retentionManager;
- private SegmentStatusChecker _segmentStatusChecker;
- private PinotTaskManager _taskManager;
- private PeriodicTaskScheduler _periodicTaskScheduler;
- private PinotHelixTaskResourceManager _helixTaskResourceManager;
- private PinotRealtimeSegmentManager _realtimeSegmentsManager;
- private PinotLLCRealtimeSegmentManager _pinotLLCRealtimeSegmentManager;
- private SegmentCompletionManager _segmentCompletionManager;
- private LeadControllerManager _leadControllerManager;
- private List<ServiceStatus.ServiceStatusCallback> _serviceStatusCallbackList;
- private MinionInstancesCleanupTask _minionInstancesCleanupTask;
-
- public ControllerStarter(ControllerConf conf) {
- _config = conf;
- inferHostnameIfNeeded(_config);
- setupHelixSystemProperties();
-
- _controllerMode = conf.getControllerMode();
- // Helix related settings.
- _helixZkURL = HelixConfig.getAbsoluteZkPathForHelix(_config.getZkStr());
- _helixClusterName = _config.getHelixClusterName();
- _listenerConfigs = ListenerConfigUtil.buildControllerConfigs(_config);
-
- String host = conf.getControllerHost();
- int port = _listenerConfigs.get(0).getPort();
-
- _helixControllerInstanceId = host + "_" + port;
- _helixParticipantInstanceId = LeadControllerUtils.generateParticipantInstanceId(host, port);
- _isUpdateStateModel = _config.isUpdateSegmentStateModel();
- _enableBatchMessageMode = _config.getEnableBatchMessageMode();
-
- _serviceStatusCallbackList = new ArrayList<>();
- if (_controllerMode == ControllerConf.ControllerMode.HELIX_ONLY) {
- _adminApp = null;
- _helixResourceManager = null;
- _executorService = null;
- } else {
- // Initialize FunctionRegistry before starting the admin application (PinotQueryResource requires it to compile
- // queries)
- FunctionRegistry.init();
- _adminApp = new ControllerAdminApiApplication(conf);
- // Do not use this before the invocation of {@link PinotHelixResourceManager::start()}, which happens in {@link ControllerStarter::start()}
- _helixResourceManager = new PinotHelixResourceManager(_config);
- _executorService =
- Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("restapi-multiget-thread-%d").build());
- }
- }
-
- private void inferHostnameIfNeeded(ControllerConf config) {
- if (config.getControllerHost() == null) {
- if (config.getProperty(CommonConstants.Helix.SET_INSTANCE_ID_TO_HOSTNAME_KEY, false)) {
- final String inferredHostname = NetUtils.getHostnameOrAddress();
- if (inferredHostname != null) {
- config.setControllerHost(inferredHostname);
- } else {
- throw new RuntimeException(
- "Failed to infer controller hostname, please set controller instanceId explicitly in config file.");
- }
- }
- }
- }
-
- private void setupHelixSystemProperties() {
- // NOTE: Helix will disconnect the manager and disable the instance if it detects flapping (too frequent disconnect
- // from ZooKeeper). Setting flapping time window to a small value can avoid this from happening. Helix ignores the
- // non-positive value, so set the default value as 1.
- System.setProperty(SystemPropertyKeys.FLAPPING_TIME_WINDOW, _config
- .getProperty(CommonConstants.Helix.CONFIG_OF_CONTROLLER_FLAPPING_TIME_WINDOW_MS,
- CommonConstants.Helix.DEFAULT_FLAPPING_TIME_WINDOW_MS));
- }
-
- private void setupHelixClusterConstraints() {
- String maxStateTransitions = _config
- .getProperty(CommonConstants.Helix.CONFIG_OF_HELIX_INSTANCE_MAX_STATE_TRANSITIONS,
- CommonConstants.Helix.DEFAULT_HELIX_INSTANCE_MAX_STATE_TRANSITIONS);
- Map<ClusterConstraints.ConstraintAttribute, String> constraintAttributes = new HashMap<>();
- constraintAttributes.put(ClusterConstraints.ConstraintAttribute.INSTANCE, ".*");
- constraintAttributes
- .put(ClusterConstraints.ConstraintAttribute.MESSAGE_TYPE, Message.MessageType.STATE_TRANSITION.name());
- ConstraintItem constraintItem = new ConstraintItem(constraintAttributes, maxStateTransitions);
-
- _helixControllerManager.getClusterManagmentTool()
- .setConstraint(_helixClusterName, ClusterConstraints.ConstraintType.MESSAGE_CONSTRAINT,
- MAX_STATE_TRANSITIONS_PER_INSTANCE, constraintItem);
- }
-
- public PinotHelixResourceManager getHelixResourceManager() {
- return _helixResourceManager;
- }
-
- /**
- * Gets the Helix Manager connected as Helix controller.
- */
- public HelixManager getHelixControllerManager() {
- return _helixControllerManager;
- }
-
- public LeadControllerManager getLeadControllerManager() {
- return _leadControllerManager;
- }
-
- public OfflineSegmentIntervalChecker getOfflineSegmentIntervalChecker() {
- return _offlineSegmentIntervalChecker;
- }
-
- public RealtimeSegmentValidationManager getRealtimeSegmentValidationManager() {
- return _realtimeSegmentValidationManager;
- }
-
- public BrokerResourceValidationManager getBrokerResourceValidationManager() {
- return _brokerResourceValidationManager;
- }
-
- public PinotHelixTaskResourceManager getHelixTaskResourceManager() {
- return _helixTaskResourceManager;
- }
-
- public PinotTaskManager getTaskManager() {
- return _taskManager;
- }
-
- public MinionInstancesCleanupTask getMinionInstancesCleanupTask() {
- return _minionInstancesCleanupTask;
- }
-
- @Override
- public ServiceRole getServiceRole() {
- return ServiceRole.CONTROLLER;
- }
-
- @Override
- public String getInstanceId() {
- return _helixParticipantInstanceId;
- }
-
- @Override
- public PinotConfiguration getConfig() {
- return _config;
- }
-
- @Override
- public void start() {
- LOGGER.info("Starting Pinot controller in mode: {}.", _controllerMode.name());
- Utils.logVersions();
-
- // Set up controller metrics
- initControllerMetrics();
-
- switch (_controllerMode) {
- case DUAL:
- setUpHelixController();
- setUpPinotController();
- break;
- case PINOT_ONLY:
- setUpPinotController();
- break;
- case HELIX_ONLY:
- setUpHelixController();
- break;
- default:
- LOGGER.error("Invalid mode: " + _controllerMode);
- }
-
- ServiceStatus.setServiceStatusCallback(_helixParticipantInstanceId,
- new ServiceStatus.MultipleCallbackServiceStatusCallback(_serviceStatusCallbackList));
- }
-
- private void setUpHelixController() {
- // Register and connect instance as Helix controller.
- LOGGER.info("Starting Helix controller");
- _helixControllerManager =
- HelixSetupUtils.setupHelixController(_helixClusterName, _helixZkURL, _helixControllerInstanceId);
-
- // Emit helix controller metrics
- _controllerMetrics.addCallbackGauge(CommonConstants.Helix.INSTANCE_CONNECTED_METRIC_NAME,
- () -> _helixControllerManager.isConnected() ? 1L : 0L);
- // Deprecated, since getting the leadership of Helix does not mean Helix has been ready for pinot.
- _controllerMetrics.addCallbackGauge("helix.leader", () -> _helixControllerManager.isLeader() ? 1L : 0L);
- _helixControllerManager.addPreConnectCallback(
- () -> _controllerMetrics.addMeteredGlobalValue(ControllerMeter.HELIX_ZOOKEEPER_RECONNECTS, 1L));
-
- _serviceStatusCallbackList.add(generateServiceStatusCallback(_helixControllerManager));
-
- // setup up constraint
- setupHelixClusterConstraints();
- }
-
- private void setUpPinotController() {
- // install default SSL context if necessary (even if not force-enabled everywhere)
- TlsConfig tlsDefaults = TlsUtils.extractTlsConfig(_config, ControllerConf.CONTROLLER_TLS_PREFIX);
- if (StringUtils.isNotBlank(tlsDefaults.getKeyStorePath()) || StringUtils
- .isNotBlank(tlsDefaults.getTrustStorePath())) {
- LOGGER.info("Installing default SSL context for any client requests");
- TlsUtils.installDefaultSSLSocketFactory(tlsDefaults);
- }
-
- // Set up Pinot cluster in Helix if needed
- HelixSetupUtils.setupPinotCluster(_helixClusterName, _helixZkURL, _isUpdateStateModel, _enableBatchMessageMode,
- _config.getLeadControllerResourceRebalanceStrategy());
-
- // Start all components
- initPinotFSFactory();
- initControllerFilePathProvider();
- initSegmentFetcherFactory();
- initPinotCrypterFactory();
-
- LOGGER.info("Initializing Helix participant manager");
- _helixParticipantManager = HelixManagerFactory
- .getZKHelixManager(_helixClusterName, _helixParticipantInstanceId, InstanceType.PARTICIPANT, _helixZkURL);
-
- // LeadControllerManager needs to be initialized before registering as Helix participant.
- LOGGER.info("Initializing lead controller manager");
- _leadControllerManager = new LeadControllerManager(_helixParticipantManager, _controllerMetrics);
-
- LOGGER.info("Registering and connecting Helix participant manager as Helix Participant role");
- registerAndConnectAsHelixParticipant();
-
- // LeadControllerManager needs to be started after the connection
- // as it can check Helix leadership and resource config only after connecting to Helix cluster.
- LOGGER.info("Starting lead controller manager");
- _leadControllerManager.start();
-
- LOGGER.info("Starting Pinot Helix resource manager and connecting to Zookeeper");
- _helixResourceManager.start(_helixParticipantManager);
-
- LOGGER.info("Starting task resource manager");
- _helixTaskResourceManager = new PinotHelixTaskResourceManager(new TaskDriver(_helixParticipantManager));
-
- // Helix resource manager must be started in order to create PinotLLCRealtimeSegmentManager
- LOGGER.info("Starting realtime segment manager");
- _pinotLLCRealtimeSegmentManager =
- new PinotLLCRealtimeSegmentManager(_helixResourceManager, _config, _controllerMetrics);
- // TODO: Need to put this inside HelixResourceManager when HelixControllerLeadershipManager is removed.
- _helixResourceManager.registerPinotLLCRealtimeSegmentManager(_pinotLLCRealtimeSegmentManager);
- _segmentCompletionManager =
- new SegmentCompletionManager(_helixParticipantManager, _pinotLLCRealtimeSegmentManager, _controllerMetrics,
- _leadControllerManager, _config.getSegmentCommitTimeoutSeconds());
-
- if (_config.getHLCTablesAllowed()) {
- LOGGER.info("Realtime tables with High Level consumers will be supported");
- _realtimeSegmentsManager = new PinotRealtimeSegmentManager(_helixResourceManager, _leadControllerManager);
- _realtimeSegmentsManager.start(_controllerMetrics);
- } else {
- LOGGER.info("Realtime tables with High Level consumers will NOT be supported");
- _realtimeSegmentsManager = null;
- }
-
- // Setting up periodic tasks
- List<PeriodicTask> controllerPeriodicTasks = setupControllerPeriodicTasks();
- LOGGER.info("Init controller periodic tasks scheduler");
- _periodicTaskScheduler = new PeriodicTaskScheduler();
- _periodicTaskScheduler.init(controllerPeriodicTasks);
- _periodicTaskScheduler.start();
-
- String accessControlFactoryClass = _config.getAccessControlFactoryClass();
- LOGGER.info("Use class: {} as the AccessControlFactory", accessControlFactoryClass);
- final AccessControlFactory accessControlFactory;
- try {
- accessControlFactory = (AccessControlFactory) Class.forName(accessControlFactoryClass).newInstance();
- accessControlFactory.init(_config);
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while creating new AccessControlFactory instance", e);
- }
-
- final MetadataEventNotifierFactory metadataEventNotifierFactory =
- MetadataEventNotifierFactory.loadFactory(_config.subset(METADATA_EVENT_NOTIFIER_PREFIX));
-
- LOGGER.info("Controller download url base: {}", _config.generateVipUrl());
- LOGGER.info("Injecting configuration and resource managers to the API context");
- final MultiThreadedHttpConnectionManager connectionManager = new MultiThreadedHttpConnectionManager();
- connectionManager.getParams().setConnectionTimeout(_config.getServerAdminRequestTimeoutSeconds() * 1000);
- // register all the controller objects for injection to jersey resources
- _adminApp.registerBinder(new AbstractBinder() {
- @Override
- protected void configure() {
- bind(_config).to(ControllerConf.class);
- bind(_helixResourceManager).to(PinotHelixResourceManager.class);
- bind(_helixTaskResourceManager).to(PinotHelixTaskResourceManager.class);
- bind(_segmentCompletionManager).to(SegmentCompletionManager.class);
- bind(_taskManager).to(PinotTaskManager.class);
- bind(connectionManager).to(HttpConnectionManager.class);
- bind(_executorService).to(Executor.class);
- bind(_controllerMetrics).to(ControllerMetrics.class);
- bind(accessControlFactory).to(AccessControlFactory.class);
- bind(metadataEventNotifierFactory).to(MetadataEventNotifierFactory.class);
- bind(_leadControllerManager).to(LeadControllerManager.class);
- }
- });
-
- LOGGER.info("Starting controller admin application on: {}", ListenerConfigUtil.toString(_listenerConfigs));
- _adminApp.start(_listenerConfigs);
-
- _controllerMetrics.addCallbackGauge("dataDir.exists", () -> new File(_config.getDataDir()).exists() ? 1L : 0L);
- _controllerMetrics.addCallbackGauge("dataDir.fileOpLatencyMs", () -> {
- File dataDir = new File(_config.getDataDir());
- if (dataDir.exists()) {
- try {
- long startTime = System.currentTimeMillis();
- File testFile = new File(dataDir, _config.getControllerHost());
- try (OutputStream outputStream = new FileOutputStream(testFile, false)) {
- outputStream.write(Longs.toByteArray(System.currentTimeMillis()));
- }
- FileUtils.deleteQuietly(testFile);
- return System.currentTimeMillis() - startTime;
- } catch (IOException e) {
- LOGGER.warn("Caught exception while checking the data directory operation latency", e);
- return DATA_DIRECTORY_EXCEPTION_VALUE;
- }
- } else {
- return DATA_DIRECTORY_MISSING_VALUE;
- }
- });
-
- _serviceStatusCallbackList.add(generateServiceStatusCallback(_helixParticipantManager));
- }
-
- private ServiceStatus.ServiceStatusCallback generateServiceStatusCallback(HelixManager helixManager) {
- return new ServiceStatus.ServiceStatusCallback() {
- private boolean _isStarted = false;
- private String _statusDescription = "Helix ZK Not connected as " + helixManager.getInstanceType();
-
- @Override
- public ServiceStatus.Status getServiceStatus() {
- if (_isStarted) {
- // If we've connected to Helix at some point, the instance status depends on being connected to ZK
- if (helixManager.isConnected()) {
- return ServiceStatus.Status.GOOD;
- } else {
- return ServiceStatus.Status.BAD;
- }
- }
-
- // Return starting until zk is connected
- if (!helixManager.isConnected()) {
- return ServiceStatus.Status.STARTING;
- } else {
- _isStarted = true;
- _statusDescription = ServiceStatus.STATUS_DESCRIPTION_NONE;
- return ServiceStatus.Status.GOOD;
- }
- }
-
- @Override
- public String getStatusDescription() {
- return _statusDescription;
- }
- };
- }
-
- private void initControllerMetrics() {
- PinotConfiguration metricsConfiguration = _config.subset(METRICS_REGISTRY_NAME);
- PinotMetricUtils.init(metricsConfiguration);
- _metricsRegistry = PinotMetricUtils.getPinotMetricsRegistry();
- _controllerMetrics = new ControllerMetrics(_config.getMetricsPrefix(), _metricsRegistry);
- _controllerMetrics.initializeGlobalMeters();
- }
-
- private void initPinotFSFactory() {
- LOGGER.info("Initializing PinotFSFactory");
-
- PinotFSFactory.init(_config.subset(CommonConstants.Controller.PREFIX_OF_CONFIG_OF_PINOT_FS_FACTORY));
- }
-
- private void initControllerFilePathProvider() {
- LOGGER.info("Initializing ControllerFilePathProvider");
- try {
- ControllerFilePathProvider.init(_config);
- } catch (InvalidControllerConfigException e) {
- throw new RuntimeException("Caught exception while initializing ControllerFilePathProvider", e);
- }
- }
-
- private void initSegmentFetcherFactory() {
- PinotConfiguration segmentFetcherFactoryConfig =
- _config.subset(CommonConstants.Controller.PREFIX_OF_CONFIG_OF_SEGMENT_FETCHER_FACTORY);
- LOGGER.info("Initializing SegmentFetcherFactory");
- try {
- SegmentFetcherFactory.init(segmentFetcherFactoryConfig);
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while initializing SegmentFetcherFactory", e);
- }
- }
-
- private void initPinotCrypterFactory() {
- PinotConfiguration pinotCrypterConfig =
- _config.subset(CommonConstants.Controller.PREFIX_OF_CONFIG_OF_PINOT_CRYPTER);
- LOGGER.info("Initializing PinotCrypterFactory");
- try {
- PinotCrypterFactory.init(pinotCrypterConfig);
- } catch (Exception e) {
- throw new RuntimeException("Caught exception while initializing PinotCrypterFactory", e);
- }
- }
-
- /**
- * Registers, connects to Helix cluster as PARTICIPANT role, and adds listeners.
- */
- private void registerAndConnectAsHelixParticipant() {
- // Registers customized Master-Slave state model to state machine engine, which is for calculating participant assignment in lead controller resource.
- _helixParticipantManager.getStateMachineEngine().registerStateModelFactory(MasterSlaveSMD.name,
- new LeadControllerResourceMasterSlaveStateModelFactory(_leadControllerManager));
-
- // Connects to cluster.
- try {
- _helixParticipantManager.connect();
- } catch (Exception e) {
- String errorMsg = String.format("Exception when connecting the instance %s as Participant role to Helix.",
- _helixParticipantInstanceId);
- LOGGER.error(errorMsg, e);
- throw new RuntimeException(errorMsg);
- }
-
- LOGGER.info("Registering helix controller listener");
- // This registration is not needed when the leadControllerResource is enabled.
- // However, the resource can be disabled sometime while the cluster is in operation, so we keep it here. Plus, it does not add much overhead.
- // At some point in future when we stop supporting the disabled resource, we will remove this line altogether and the logic that goes with it.
- _helixParticipantManager.addControllerListener(
- (ControllerChangeListener) changeContext -> _leadControllerManager.onHelixControllerChange());
-
- LOGGER.info("Registering resource config listener");
- try {
- _helixParticipantManager.addResourceConfigChangeListener(
- (resourceConfigList, changeContext) -> _leadControllerManager.onResourceConfigChange());
- } catch (Exception e) {
- throw new RuntimeException(
- "Error registering resource config listener for " + CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME, e);
- }
- }
-
- public ControllerConf.ControllerMode getControllerMode() {
- return _controllerMode;
- }
-
- @VisibleForTesting
- protected List<PeriodicTask> setupControllerPeriodicTasks() {
- LOGGER.info("Setting up periodic tasks");
- List<PeriodicTask> periodicTasks = new ArrayList<>();
- _taskManager =
- new PinotTaskManager(_helixTaskResourceManager, _helixResourceManager, _leadControllerManager, _config,
- _controllerMetrics);
- periodicTasks.add(_taskManager);
- _retentionManager =
- new RetentionManager(_helixResourceManager, _leadControllerManager, _config, _controllerMetrics);
- periodicTasks.add(_retentionManager);
- _offlineSegmentIntervalChecker =
- new OfflineSegmentIntervalChecker(_config, _helixResourceManager, _leadControllerManager,
- new ValidationMetrics(_metricsRegistry), _controllerMetrics);
- periodicTasks.add(_offlineSegmentIntervalChecker);
- _realtimeSegmentValidationManager =
- new RealtimeSegmentValidationManager(_config, _helixResourceManager, _leadControllerManager,
- _pinotLLCRealtimeSegmentManager, new ValidationMetrics(_metricsRegistry), _controllerMetrics);
- periodicTasks.add(_realtimeSegmentValidationManager);
- _brokerResourceValidationManager =
- new BrokerResourceValidationManager(_config, _helixResourceManager, _leadControllerManager, _controllerMetrics);
- periodicTasks.add(_brokerResourceValidationManager);
- _segmentStatusChecker =
- new SegmentStatusChecker(_helixResourceManager, _leadControllerManager, _config, _controllerMetrics);
- periodicTasks.add(_segmentStatusChecker);
- _segmentRelocator = new SegmentRelocator(_helixResourceManager, _leadControllerManager, _config, _controllerMetrics,
- _executorService);
- periodicTasks.add(_segmentRelocator);
- _minionInstancesCleanupTask = new MinionInstancesCleanupTask(_helixResourceManager, _config, _controllerMetrics);
- periodicTasks.add(_minionInstancesCleanupTask);
-
- return periodicTasks;
- }
-
- @Override
- public void stop() {
- switch (_controllerMode) {
- case DUAL:
- stopPinotController();
- stopHelixController();
- break;
- case PINOT_ONLY:
- stopPinotController();
- break;
- case HELIX_ONLY:
- stopHelixController();
- break;
- }
- LOGGER.info("Deregistering service status handler");
- ServiceStatus.removeServiceStatusCallback(_helixParticipantInstanceId);
- LOGGER.info("Shutdown Controller Metrics Registry");
- _metricsRegistry.shutdown();
- LOGGER.info("Finish shutting down Pinot controller for {}", _helixParticipantInstanceId);
- }
-
- private void stopHelixController() {
- LOGGER.info("Disconnecting helix controller zk manager");
- _helixControllerManager.disconnect();
- }
-
- private void stopPinotController() {
- try {
- // Stopping periodic tasks has to be done before stopping HelixResourceManager.
- // Stop controller periodic task.
- LOGGER.info("Stopping controller periodic tasks");
- _periodicTaskScheduler.stop();
-
- LOGGER.info("Stopping lead controller manager");
- _leadControllerManager.stop();
-
- // Stop PinotLLCSegmentManager before stopping Jersey API. It is possible that stopping Jersey API
- // may interrupt the handlers waiting on an I/O.
- _pinotLLCRealtimeSegmentManager.stop();
-
- LOGGER.info("Closing PinotFS classes");
- PinotFSFactory.shutdown();
-
- LOGGER.info("Stopping Jersey admin API");
- _adminApp.stop();
-
- if (_realtimeSegmentsManager != null) {
- LOGGER.info("Stopping realtime segment manager");
- _realtimeSegmentsManager.stop();
- }
-
- LOGGER.info("Stopping resource manager");
- _helixResourceManager.stop();
-
- LOGGER.info("Disconnecting helix participant zk manager");
- _helixParticipantManager.disconnect();
-
- LOGGER.info("Shutting down executor service");
- _executorService.shutdownNow();
- _executorService.awaitTermination(10L, TimeUnit.SECONDS);
- } catch (final Exception e) {
- LOGGER.error("Caught exception while shutting down", e);
- }
- }
+/**
+ * Default controller startable implementation. Contains methods to start and stop the controller
+ */
+public class ControllerStarter extends BaseControllerStarter {
- public PinotMetricsRegistry getMetricsRegistry() {
- return _metricsRegistry;
+ public ControllerStarter() {
}
- @VisibleForTesting
- public ControllerMetrics getControllerMetrics() {
- return _controllerMetrics;
+ @Deprecated
+ public ControllerStarter(PinotConfiguration pinotConfiguration)
+ throws Exception {
+ init(pinotConfiguration);
}
- public static ControllerStarter startDefault() {
+ public static ControllerStarter startDefault()
+ throws Exception {
return startDefault(null);
}
- public static ControllerStarter startDefault(File webappPath) {
+ public static ControllerStarter startDefault(File webappPath)
+ throws Exception {
final ControllerConf conf = new ControllerConf();
conf.setControllerHost("localhost");
conf.setControllerPort("9000");
@@ -703,13 +71,15 @@ public class ControllerStarter implements ServiceStartable {
conf.setStatusCheckerWaitForPushTimeInSeconds(10 * 60);
conf.setTenantIsolationEnabled(true);
- final ControllerStarter starter = new ControllerStarter(conf);
+ final ControllerStarter starter = new ControllerStarter();
+ starter.init(conf);
starter.start();
return starter;
}
- public static void main(String[] args) {
+ public static void main(String[] args)
+ throws Exception {
startDefault();
}
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/ControllerTestUtils.java b/pinot-controller/src/test/java/org/apache/pinot/controller/ControllerTestUtils.java
index 72c1c83..adb9437 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/ControllerTestUtils.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/ControllerTestUtils.java
@@ -74,6 +74,7 @@ import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.MetricFieldSpec;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -153,17 +154,19 @@ public abstract class ControllerTestUtils {
return properties;
}
- public static void startController(Map<String, Object> properties) {
+ public static void startController(Map<String, Object> properties)
+ throws Exception {
Preconditions.checkState(_controllerStarter == null);
_controllerConfig = new ControllerConf(properties);
- _controllerPort = Integer.valueOf(_controllerConfig.getControllerPort());
+ _controllerPort = Integer.parseInt(_controllerConfig.getControllerPort());
_controllerBaseApiUrl = "http://localhost:" + _controllerPort;
_controllerRequestURLBuilder = ControllerRequestURLBuilder.baseUrl(_controllerBaseApiUrl);
_controllerDataDir = _controllerConfig.getDataDir();
- _controllerStarter = getControllerStarter(_controllerConfig);
+ _controllerStarter = new ControllerStarter();
+ _controllerStarter.init(new PinotConfiguration(properties));
_controllerStarter.start();
_helixResourceManager = _controllerStarter.getHelixResourceManager();
_helixManager = _controllerStarter.getHelixControllerManager();
@@ -194,10 +197,6 @@ public abstract class ControllerTestUtils {
configAccessor.set(scope, CommonConstants.Helix.DEFAULT_HYPERLOGLOG_LOG2M_KEY, Integer.toString(12));
}
- protected static ControllerStarter getControllerStarter(ControllerConf config) {
- return new ControllerStarter(config);
- }
-
public static ControllerConf getControllerConfig() {
return _controllerConfig;
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerPeriodicTaskStarterStatelessTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerPeriodicTaskStarterStatelessTest.java
index 9c75980..1288516 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerPeriodicTaskStarterStatelessTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerPeriodicTaskStarterStatelessTest.java
@@ -19,7 +19,6 @@
package org.apache.pinot.controller.helix;
import java.util.List;
-import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.ControllerStarter;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.core.periodictask.PeriodicTask;
@@ -40,7 +39,8 @@ public class ControllerPeriodicTaskStarterStatelessTest extends ControllerTest {
* Test that controller starts up and helixResourceManager is non null before initiating periodic tasks
*/
@Test
- public void testHelixResourceManagerDuringControllerStart() {
+ public void testHelixResourceManagerDuringControllerStart()
+ throws Exception {
startController();
stopController();
}
@@ -51,15 +51,15 @@ public class ControllerPeriodicTaskStarterStatelessTest extends ControllerTest {
}
@Override
- protected ControllerStarter getControllerStarter(ControllerConf config) {
- return new MockControllerStarter(config);
+ protected ControllerStarter getControllerStarter() {
+ return new MockControllerStarter();
}
private class MockControllerStarter extends ControllerStarter {
private static final int NUM_PERIODIC_TASKS = 8;
- public MockControllerStarter(ControllerConf conf) {
- super(conf);
+ public MockControllerStarter() {
+ super();
}
@Override
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
index f206f6b..563182d 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/ControllerTest.java
@@ -75,6 +75,7 @@ import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.MetricFieldSpec;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.NetUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
@@ -150,11 +151,13 @@ public abstract class ControllerTest {
return properties;
}
- protected void startController() {
+ protected void startController()
+ throws Exception {
startController(getDefaultControllerConfiguration());
}
- protected void startController(Map<String, Object> properties) {
+ protected void startController(Map<String, Object> properties)
+ throws Exception {
Preconditions.checkState(_controllerStarter == null);
ControllerConf config = new ControllerConf(properties);
@@ -173,7 +176,8 @@ public abstract class ControllerTest {
_controllerRequestURLBuilder = ControllerRequestURLBuilder.baseUrl(_controllerBaseApiUrl);
_controllerDataDir = config.getDataDir();
- _controllerStarter = getControllerStarter(config);
+ _controllerStarter = getControllerStarter();
+ _controllerStarter.init(new PinotConfiguration(properties));
_controllerStarter.start();
_helixResourceManager = _controllerStarter.getHelixResourceManager();
_helixManager = _controllerStarter.getHelixControllerManager();
@@ -204,8 +208,8 @@ public abstract class ControllerTest {
configAccessor.set(scope, CommonConstants.Helix.DEFAULT_HYPERLOGLOG_LOG2M_KEY, Integer.toString(12));
}
- protected ControllerStarter getControllerStarter(ControllerConf config) {
- return new ControllerStarter(config);
+ protected ControllerStarter getControllerStarter() {
+ return new ControllerStarter();
}
protected int getControllerPort() {
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeStatelessTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeStatelessTest.java
index 4dfbbdb..9c28be8 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeStatelessTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeStatelessTest.java
@@ -33,6 +33,7 @@ import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.ControllerStarter;
import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Helix;
import org.apache.pinot.util.TestUtils;
@@ -44,7 +45,7 @@ import org.testng.annotations.Test;
public class PinotControllerModeStatelessTest extends ControllerTest {
- private static long TIMEOUT_IN_MS = 10_000L;
+ private static final long TIMEOUT_IN_MS = 10_000L;
@BeforeClass
public void setUp() {
@@ -52,7 +53,8 @@ public class PinotControllerModeStatelessTest extends ControllerTest {
}
@Test
- public void testHelixOnlyController() {
+ public void testHelixOnlyController()
+ throws Exception {
// Start a Helix-only controller
Map<String, Object> properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.CONTROLLER_MODE, ControllerConf.ControllerMode.HELIX_ONLY);
@@ -65,7 +67,8 @@ public class PinotControllerModeStatelessTest extends ControllerTest {
}
@Test
- public void testDualModeController() {
+ public void testDualModeController()
+ throws Exception {
// Start the first dual-mode controller
Map<String, Object> properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.CONTROLLER_MODE, ControllerConf.ControllerMode.DUAL);
@@ -113,7 +116,8 @@ public class PinotControllerModeStatelessTest extends ControllerTest {
// Start the second dual-mode controller
properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.CONTROLLER_MODE, ControllerConf.ControllerMode.DUAL);
- ControllerStarter secondDualModeController = getControllerStarter(new ControllerConf(properties));
+ ControllerStarter secondDualModeController = new ControllerStarter();
+ secondDualModeController.init(new PinotConfiguration(properties));
secondDualModeController.start();
TestUtils
.waitForCondition(aVoid -> secondDualModeController.getHelixResourceManager().getHelixZkManager().isConnected(),
@@ -160,7 +164,8 @@ public class PinotControllerModeStatelessTest extends ControllerTest {
properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.CONTROLLER_MODE, ControllerConf.ControllerMode.DUAL);
- ControllerStarter thirdDualModeController = getControllerStarter(new ControllerConf(properties));
+ ControllerStarter thirdDualModeController = new ControllerStarter();
+ thirdDualModeController.init(new PinotConfiguration(properties));
thirdDualModeController.start();
PinotHelixResourceManager pinotHelixResourceManager = thirdDualModeController.getHelixResourceManager();
_helixManager = pinotHelixResourceManager.getHelixZkManager();
@@ -195,11 +200,13 @@ public class PinotControllerModeStatelessTest extends ControllerTest {
}
@Test
- public void testPinotOnlyController() {
+ public void testPinotOnlyController()
+ throws Exception {
Map<String, Object> properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.CONTROLLER_MODE, ControllerConf.ControllerMode.PINOT_ONLY);
- ControllerStarter firstPinotOnlyController = getControllerStarter(new ControllerConf(properties));
+ ControllerStarter firstPinotOnlyController = new ControllerStarter();
+ firstPinotOnlyController.init(new PinotConfiguration(properties));
// Starting Pinot-only controller without Helix controller should fail
try {
@@ -213,7 +220,8 @@ public class PinotControllerModeStatelessTest extends ControllerTest {
properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.CONTROLLER_MODE, ControllerConf.ControllerMode.HELIX_ONLY);
- ControllerStarter helixOnlyController = new ControllerStarter(new ControllerConf(properties));
+ ControllerStarter helixOnlyController = new ControllerStarter();
+ helixOnlyController.init(new PinotConfiguration(properties));
helixOnlyController.start();
HelixManager helixControllerManager = helixOnlyController.getHelixControllerManager();
HelixAdmin helixAdmin = helixControllerManager.getClusterManagmentTool();
@@ -232,7 +240,8 @@ public class PinotControllerModeStatelessTest extends ControllerTest {
properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.CONTROLLER_MODE, ControllerConf.ControllerMode.PINOT_ONLY);
- ControllerStarter secondPinotOnlyController = getControllerStarter(new ControllerConf(properties));
+ ControllerStarter secondPinotOnlyController = new ControllerStarter();
+ secondPinotOnlyController.init(new PinotConfiguration(properties));
secondPinotOnlyController.start();
TestUtils.waitForCondition(
aVoid -> secondPinotOnlyController.getHelixResourceManager().getHelixZkManager().isConnected(), TIMEOUT_IN_MS,
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerTest.java
index a05a8f8..d9558a3 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManagerTest.java
@@ -54,7 +54,8 @@ public class PinotTaskManagerTest extends ControllerTest {
}
@Test
- public void testDefaultPinotTaskManagerNoScheduler() {
+ public void testDefaultPinotTaskManagerNoScheduler()
+ throws Exception {
startController();
PinotTaskManager taskManager = _controllerStarter.getTaskManager();
Assert.assertNull(taskManager.getScheduler());
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
index 0c53f81..7bd0f57 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
@@ -124,7 +124,8 @@ public abstract class ClusterTest extends ControllerTest {
PinotConfiguration configuration = new PinotConfiguration(properties);
overrideBrokerConf(configuration);
- HelixBrokerStarter brokerStarter = new HelixBrokerStarter(configuration);
+ HelixBrokerStarter brokerStarter = new HelixBrokerStarter();
+ brokerStarter.init(configuration);
brokerStarter.start();
_brokerStarters.add(brokerStarter);
}
@@ -140,11 +141,15 @@ public abstract class ClusterTest extends ControllerTest {
properties.put(Broker.CONFIG_OF_BROKER_TIMEOUT_MS, 60 * 1000L);
properties.put(Broker.CONFIG_OF_DELAY_SHUTDOWN_TIME_MS, 0);
+ properties.put(Helix.CONFIG_OF_CLUSTER_NAME, getHelixClusterName());
+ properties.put(Helix.CONFIG_OF_ZOOKEEPR_SERVER, getZkUrl());
+ properties.put(Broker.CONFIG_OF_BROKER_HOSTNAME, LOCAL_HOST);
+
PinotConfiguration configuration = new PinotConfiguration(properties);
overrideBrokerConf(configuration);
- HelixBrokerStarter brokerStarter =
- new HelixBrokerStarter(configuration, getHelixClusterName(), getZkUrl(), LOCAL_HOST);
+ HelixBrokerStarter brokerStarter = new HelixBrokerStarter();
+ brokerStarter.init(configuration);
brokerStarter.start();
_brokerStarters.add(brokerStarter);
@@ -209,7 +214,8 @@ public abstract class ClusterTest extends ControllerTest {
// Thread time measurement is disabled by default, enable it in integration tests.
// TODO: this can be removed when we eventually enable thread time measurement by default.
configuration.setProperty(Server.CONFIG_OF_ENABLE_THREAD_CPU_TIME_MEASUREMENT, true);
- HelixServerStarter helixServerStarter = new HelixServerStarter(configuration);
+ HelixServerStarter helixServerStarter = new HelixServerStarter();
+ helixServerStarter.init(configuration);
helixServerStarter.start();
_serverStarters.add(helixServerStarter);
}
@@ -223,12 +229,15 @@ public abstract class ClusterTest extends ControllerTest {
_serverStarters = new ArrayList<>();
Map<String, Object> properties = getDefaultServerConfiguration().toMap();
+ properties.put(Helix.CONFIG_OF_CLUSTER_NAME, getHelixClusterName());
+ properties.put(Helix.CONFIG_OF_ZOOKEEPR_SERVER, getZkUrl());
PinotConfiguration configuration = new PinotConfiguration(properties);
overrideServerConf(configuration);
try {
- HelixServerStarter helixServerStarter = new HelixServerStarter(getHelixClusterName(), getZkUrl(), configuration);
+ HelixServerStarter helixServerStarter = new HelixServerStarter();
+ helixServerStarter.init(configuration);
_serverStarters.add(helixServerStarter);
helixServerStarter.start();
} catch (Exception e) {
@@ -248,7 +257,8 @@ public abstract class ClusterTest extends ControllerTest {
PinotConfiguration minionConf = getDefaultMinionConfiguration();
minionConf.setProperty(Helix.CONFIG_OF_CLUSTER_NAME, getHelixClusterName());
minionConf.setProperty(Helix.CONFIG_OF_ZOOKEEPR_SERVER, getZkUrl());
- _minionStarter = new MinionStarter(minionConf);
+ _minionStarter = new MinionStarter();
+ _minionStarter.init(minionConf);
_minionStarter.start();
} catch (Exception e) {
throw new RuntimeException(e);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java
index 5792e7a..c900935 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ExactlyOnceKafkaRealtimeClusterIntegrationTest.java
@@ -43,7 +43,8 @@ public class ExactlyOnceKafkaRealtimeClusterIntegrationTest extends RealtimeClus
}
@Override
- public void startController() {
+ public void startController()
+ throws Exception {
Map<String, Object> properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.ALLOW_HLC_TABLES, false);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
index 6c238c0..aebbce6 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterIntegrationTest.java
@@ -73,7 +73,8 @@ public class LLCRealtimeClusterIntegrationTest extends RealtimeClusterIntegratio
}
@Override
- public void startController() {
+ public void startController()
+ throws Exception {
Map<String, Object> properties = getDefaultControllerConfiguration();
properties.put(ControllerConf.ALLOW_HLC_TABLES, false);
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PeerDownloadLLCRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PeerDownloadLLCRealtimeClusterIntegrationTest.java
index ba7734f..543645c 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PeerDownloadLLCRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/PeerDownloadLLCRealtimeClusterIntegrationTest.java
@@ -132,7 +132,8 @@ public class PeerDownloadLLCRealtimeClusterIntegrationTest extends RealtimeClust
@Override
- public void startController() {
+ public void startController()
+ throws Exception {
Map<String, Object> controllerConfig = getDefaultControllerConfiguration();
controllerConfig.put(ALLOW_HLC_TABLES, false);
controllerConfig.put(ENABLE_SPLIT_COMMIT, _enableSplitCommit);
@@ -384,4 +385,4 @@ public class PeerDownloadLLCRealtimeClusterIntegrationTest extends RealtimeClust
}
}
}
-}
\ No newline at end of file
+}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ServerStarterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ServerStarterIntegrationTest.java
index 9974dec..f3430c8 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ServerStarterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ServerStarterIntegrationTest.java
@@ -40,7 +40,8 @@ public class ServerStarterIntegrationTest extends ControllerTest {
private static final int CUSTOM_PORT = 10001;
@BeforeClass
- public void setUp() {
+ public void setUp()
+ throws Exception {
startZk();
startController();
}
@@ -56,7 +57,8 @@ public class ServerStarterIntegrationTest extends ControllerTest {
throws Exception {
serverConf.setProperty(CONFIG_OF_CLUSTER_NAME, getHelixClusterName());
serverConf.setProperty(CONFIG_OF_ZOOKEEPR_SERVER, getZkUrl());
- HelixServerStarter helixServerStarter = new HelixServerStarter(serverConf);
+ HelixServerStarter helixServerStarter = new HelixServerStarter();
+ helixServerStarter.init(serverConf);
helixServerStarter.start();
helixServerStarter.stop();
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentUploadIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentUploadIntegrationTest.java
index ac4ae79..922f285 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentUploadIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentUploadIntegrationTest.java
@@ -147,7 +147,8 @@ public class UpsertTableSegmentUploadIntegrationTest extends BaseClusterIntegrat
}
@Override
- protected void startController() {
+ protected void startController()
+ throws Exception {
Map<String, Object> controllerConfig = getDefaultControllerConfiguration();
// Perform realtime segment validation every second with 1 second initial delay.
controllerConfig
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/server/realtime/ControllerLeaderLocatorIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/server/realtime/ControllerLeaderLocatorIntegrationTest.java
index a635e4a..9f052cf 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/server/realtime/ControllerLeaderLocatorIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/server/realtime/ControllerLeaderLocatorIntegrationTest.java
@@ -25,10 +25,10 @@ import java.util.Random;
import java.util.Set;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.utils.helix.LeadControllerUtils;
-import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.ControllerStarter;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.pql.parsers.utils.Pair;
+import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants.Helix;
import org.apache.pinot.util.TestUtils;
import org.testng.Assert;
@@ -38,11 +38,12 @@ import org.testng.annotations.Test;
public class ControllerLeaderLocatorIntegrationTest extends ControllerTest {
- private static long TIMEOUT_IN_MS = 10_000L;
+ private static final long TIMEOUT_IN_MS = 10_000L;
private HashMap<Integer, String> _partitionToTableMap;
@BeforeClass
- public void setUp() {
+ public void setUp()
+ throws Exception {
startZk();
startController();
@@ -52,7 +53,8 @@ public class ControllerLeaderLocatorIntegrationTest extends ControllerTest {
}
@Test
- public void testControllerLeaderLocator() {
+ public void testControllerLeaderLocator()
+ throws Exception {
Set<String> resultSet = new HashSet<>();
FakeControllerLeaderLocator controllerLeaderLocator = FakeControllerLeaderLocator.getInstance();
@@ -66,7 +68,8 @@ public class ControllerLeaderLocatorIntegrationTest extends ControllerTest {
validateResultSet(controllerLeaderLocator, resultSet, 1, "Failed to get only one pair of controller");
Map<String, Object> properties = getDefaultControllerConfiguration();
- ControllerStarter secondControllerStarter = new ControllerStarter(new ControllerConf(properties));
+ ControllerStarter secondControllerStarter = new ControllerStarter();
+ secondControllerStarter.init(new PinotConfiguration(properties));
secondControllerStarter.start();
TestUtils
@@ -164,4 +167,4 @@ public class ControllerLeaderLocatorIntegrationTest extends ControllerTest {
_currentTimeMs = currentTimeMs;
}
}
-}
\ No newline at end of file
+}
diff --git a/pinot-minion/src/main/java/org/apache/pinot/minion/MinionStarter.java b/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
similarity index 89%
copy from pinot-minion/src/main/java/org/apache/pinot/minion/MinionStarter.java
copy to pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
index 00158dd..8d21021 100644
--- a/pinot-minion/src/main/java/org/apache/pinot/minion/MinionStarter.java
+++ b/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
@@ -62,36 +62,23 @@ import org.slf4j.LoggerFactory;
/**
- * The class <code>MinionStarter</code> provides methods to start and stop the Pinot Minion.
- * <p>Pinot Minion will automatically join the given Helix cluster as a participant.
+ * Base class for minion starter
*/
-public class MinionStarter implements ServiceStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(MinionStarter.class);
+public abstract class BaseMinionStarter implements ServiceStartable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BaseMinionStarter.class);
private static final String HTTPS_ENABLED = "enabled";
- private final PinotConfiguration _config;
- private final String _instanceId;
- private final HelixManager _helixManager;
- private final TaskExecutorFactoryRegistry _taskExecutorFactoryRegistry;
- private final EventObserverFactoryRegistry _eventObserverFactoryRegistry;
- private MinionAdminApiApplication _minionAdminApplication;
- private final List<ListenerConfig> _listenerConfigs;
+ protected PinotConfiguration _config;
+ protected String _instanceId;
+ protected HelixManager _helixManager;
+ protected TaskExecutorFactoryRegistry _taskExecutorFactoryRegistry;
+ protected EventObserverFactoryRegistry _eventObserverFactoryRegistry;
+ protected MinionAdminApiApplication _minionAdminApplication;
+ protected List<ListenerConfig> _listenerConfigs;
- @Deprecated
- public MinionStarter(String clusterName, String zkServers, PinotConfiguration minionConfig)
- throws Exception {
- this(applyMinionConfigs(minionConfig, clusterName, zkServers));
- }
-
- @Deprecated
- private static PinotConfiguration applyMinionConfigs(PinotConfiguration minionConfig, String clusterName, String zkServers) {
- minionConfig.setProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, clusterName);
- minionConfig.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER, zkServers);
- return minionConfig;
- }
-
- public MinionStarter(PinotConfiguration config)
+ @Override
+ public void init(PinotConfiguration config)
throws Exception {
_config = config;
String helixClusterName = _config.getProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
diff --git a/pinot-minion/src/main/java/org/apache/pinot/minion/MinionStarter.java b/pinot-minion/src/main/java/org/apache/pinot/minion/MinionStarter.java
index 00158dd..cc0f2a2 100644
--- a/pinot-minion/src/main/java/org/apache/pinot/minion/MinionStarter.java
+++ b/pinot-minion/src/main/java/org/apache/pinot/minion/MinionStarter.java
@@ -18,277 +18,36 @@
*/
package org.apache.pinot.minion;
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import javax.net.ssl.SSLContext;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.helix.HelixAdmin;
-import org.apache.helix.HelixManager;
-import org.apache.helix.InstanceType;
-import org.apache.helix.SystemPropertyKeys;
-import org.apache.helix.manager.zk.ZKHelixManager;
-import org.apache.helix.model.InstanceConfig;
-import org.apache.helix.task.TaskStateModelFactory;
-import org.apache.pinot.common.Utils;
-import org.apache.pinot.common.metrics.PinotMetricUtils;
-import org.apache.pinot.common.utils.ClientSSLContextGenerator;
-import org.apache.pinot.common.utils.ServiceStatus;
-import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
-import org.apache.pinot.core.transport.ListenerConfig;
-import org.apache.pinot.core.transport.TlsConfig;
-import org.apache.pinot.core.util.ListenerConfigUtil;
-import org.apache.pinot.core.util.TlsUtils;
-import org.apache.pinot.minion.event.EventObserverFactoryRegistry;
-import org.apache.pinot.minion.event.MinionEventObserverFactory;
-import org.apache.pinot.minion.executor.MinionTaskZkMetadataManager;
-import org.apache.pinot.minion.executor.PinotTaskExecutorFactory;
-import org.apache.pinot.minion.executor.TaskExecutorFactoryRegistry;
-import org.apache.pinot.minion.metrics.MinionMeter;
-import org.apache.pinot.minion.metrics.MinionMetrics;
-import org.apache.pinot.minion.taskfactory.TaskFactoryRegistry;
-import org.apache.pinot.spi.crypt.PinotCrypterFactory;
import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.filesystem.PinotFSFactory;
-import org.apache.pinot.spi.metrics.PinotMetricsRegistry;
-import org.apache.pinot.spi.services.ServiceRole;
-import org.apache.pinot.spi.services.ServiceStartable;
import org.apache.pinot.spi.utils.CommonConstants;
-import org.apache.pinot.spi.utils.NetUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* The class <code>MinionStarter</code> provides methods to start and stop the Pinot Minion.
* <p>Pinot Minion will automatically join the given Helix cluster as a participant.
*/
-public class MinionStarter implements ServiceStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(MinionStarter.class);
+public class MinionStarter extends BaseMinionStarter {
- private static final String HTTPS_ENABLED = "enabled";
-
- private final PinotConfiguration _config;
- private final String _instanceId;
- private final HelixManager _helixManager;
- private final TaskExecutorFactoryRegistry _taskExecutorFactoryRegistry;
- private final EventObserverFactoryRegistry _eventObserverFactoryRegistry;
- private MinionAdminApiApplication _minionAdminApplication;
- private final List<ListenerConfig> _listenerConfigs;
+ public MinionStarter() {
+ }
@Deprecated
public MinionStarter(String clusterName, String zkServers, PinotConfiguration minionConfig)
throws Exception {
- this(applyMinionConfigs(minionConfig, clusterName, zkServers));
+ init(applyMinionConfigs(minionConfig, clusterName, zkServers));
}
@Deprecated
- private static PinotConfiguration applyMinionConfigs(PinotConfiguration minionConfig, String clusterName, String zkServers) {
+ private static PinotConfiguration applyMinionConfigs(PinotConfiguration minionConfig, String clusterName,
+ String zkServers) {
minionConfig.setProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, clusterName);
minionConfig.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER, zkServers);
return minionConfig;
}
+ @Deprecated
public MinionStarter(PinotConfiguration config)
throws Exception {
- _config = config;
- String helixClusterName = _config.getProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
- String zkAddress = _config.getProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
- String host = _config.getProperty(CommonConstants.Helix.KEY_OF_MINION_HOST,
- _config.getProperty(CommonConstants.Helix.SET_INSTANCE_ID_TO_HOSTNAME_KEY, false) ? NetUtils
- .getHostnameOrAddress() : NetUtils.getHostAddress());
- int port = _config.getProperty(CommonConstants.Helix.KEY_OF_MINION_PORT, CommonConstants.Minion.DEFAULT_HELIX_PORT);
- _instanceId = _config.getProperty(CommonConstants.Helix.Instance.INSTANCE_ID_KEY,
- CommonConstants.Helix.PREFIX_OF_MINION_INSTANCE + host + "_" + port);
- _listenerConfigs = ListenerConfigUtil.buildMinionAdminConfigs(_config);
- setupHelixSystemProperties();
- _helixManager = new ZKHelixManager(helixClusterName, _instanceId, InstanceType.PARTICIPANT, zkAddress);
- MinionTaskZkMetadataManager minionTaskZkMetadataManager = new MinionTaskZkMetadataManager(_helixManager);
- _taskExecutorFactoryRegistry = new TaskExecutorFactoryRegistry(minionTaskZkMetadataManager);
- _eventObserverFactoryRegistry = new EventObserverFactoryRegistry(minionTaskZkMetadataManager);
- }
-
- private void setupHelixSystemProperties() {
- // NOTE: Helix will disconnect the manager and disable the instance if it detects flapping (too frequent disconnect
- // from ZooKeeper). Setting flapping time window to a small value can avoid this from happening. Helix ignores the
- // non-positive value, so set the default value as 1.
- System.setProperty(SystemPropertyKeys.FLAPPING_TIME_WINDOW, _config
- .getProperty(CommonConstants.Helix.CONFIG_OF_MINION_FLAPPING_TIME_WINDOW_MS,
- CommonConstants.Helix.DEFAULT_FLAPPING_TIME_WINDOW_MS));
- }
-
- /**
- * Registers a task executor factory.
- * <p>This is for pluggable task executor factories.
- */
- public void registerTaskExecutorFactory(PinotTaskExecutorFactory taskExecutorFactory) {
- _taskExecutorFactoryRegistry.registerTaskExecutorFactory(taskExecutorFactory);
- }
-
- /**
- * Registers an event observer factory.
- * <p>This is for pluggable event observer factories.
- */
- public void registerEventObserverFactory(MinionEventObserverFactory eventObserverFactory) {
- _eventObserverFactoryRegistry.registerEventObserverFactory(eventObserverFactory);
- }
-
- @Override
- public ServiceRole getServiceRole() {
- return ServiceRole.MINION;
- }
-
- @Override
- public String getInstanceId() {
- return _instanceId;
- }
-
- @Override
- public PinotConfiguration getConfig() {
- return _config;
- }
-
- /**
- * Starts the Pinot Minion instance.
- * <p>Should be called after all classes of task executor get registered.
- */
- @Override
- public void start()
- throws Exception {
- LOGGER.info("Starting Pinot minion: {}", _instanceId);
- Utils.logVersions();
- MinionContext minionContext = MinionContext.getInstance();
-
- // Initialize data directory
- LOGGER.info("Initializing data directory");
- File dataDir = new File(_config
- .getProperty(CommonConstants.Helix.Instance.DATA_DIR_KEY, CommonConstants.Minion.DEFAULT_INSTANCE_DATA_DIR));
- if (dataDir.exists()) {
- FileUtils.cleanDirectory(dataDir);
- } else {
- FileUtils.forceMkdir(dataDir);
- }
- minionContext.setDataDir(dataDir);
-
- // Initialize metrics
- LOGGER.info("Initializing metrics");
- // TODO: put all the metrics related configs down to "pinot.server.metrics"
- PinotConfiguration metricsConfiguration = _config;
- PinotMetricUtils.init(metricsConfiguration);
- PinotMetricsRegistry metricsRegistry = PinotMetricUtils.getPinotMetricsRegistry();
-
- MinionMetrics minionMetrics = new MinionMetrics(_config
- .getProperty(CommonConstants.Minion.CONFIG_OF_METRICS_PREFIX_KEY,
- CommonConstants.Minion.CONFIG_OF_METRICS_PREFIX), metricsRegistry);
- minionMetrics.initializeGlobalMeters();
- minionContext.setMinionMetrics(minionMetrics);
-
- // Install default SSL context if necessary (even if not force-enabled everywhere)
- TlsConfig tlsDefaults = TlsUtils.extractTlsConfig(_config, CommonConstants.Minion.MINION_TLS_PREFIX);
- if (StringUtils.isNotBlank(tlsDefaults.getKeyStorePath()) || StringUtils
- .isNotBlank(tlsDefaults.getTrustStorePath())) {
- LOGGER.info("Installing default SSL context for any client requests");
- TlsUtils.installDefaultSSLSocketFactory(tlsDefaults);
- }
-
- // initialize authentication
- minionContext.setTaskAuthToken(_config.getProperty(CommonConstants.Minion.CONFIG_OF_TASK_AUTH_TOKEN));
-
- // Start all components
- LOGGER.info("Initializing PinotFSFactory");
- PinotConfiguration pinotFSConfig = _config.subset(CommonConstants.Minion.PREFIX_OF_CONFIG_OF_PINOT_FS_FACTORY);
- PinotFSFactory.init(pinotFSConfig);
-
- LOGGER.info("Initializing segment fetchers for all protocols");
- PinotConfiguration segmentFetcherFactoryConfig =
- _config.subset(CommonConstants.Minion.PREFIX_OF_CONFIG_OF_SEGMENT_FETCHER_FACTORY);
- SegmentFetcherFactory.init(segmentFetcherFactoryConfig);
-
- LOGGER.info("Initializing pinot crypter");
- PinotConfiguration pinotCrypterConfig = _config.subset(CommonConstants.Minion.PREFIX_OF_CONFIG_OF_PINOT_CRYPTER);
- PinotCrypterFactory.init(pinotCrypterConfig);
-
- // Need to do this before we start receiving state transitions.
- LOGGER.info("Initializing ssl context for segment uploader");
- PinotConfiguration httpsConfig = _config.subset(CommonConstants.Minion.PREFIX_OF_CONFIG_OF_SEGMENT_UPLOADER)
- .subset(CommonConstants.HTTPS_PROTOCOL);
- if (httpsConfig.getProperty(HTTPS_ENABLED, false)) {
- SSLContext sslContext =
- new ClientSSLContextGenerator(httpsConfig.subset(CommonConstants.PREFIX_OF_SSL_SUBSET)).generate();
- minionContext.setSSLContext(sslContext);
- }
-
- // Join the Helix cluster
- LOGGER.info("Joining the Helix cluster");
- _helixManager.getStateMachineEngine().registerStateModelFactory("Task", new TaskStateModelFactory(_helixManager,
- new TaskFactoryRegistry(_taskExecutorFactoryRegistry, _eventObserverFactoryRegistry).getTaskFactoryRegistry()));
- _helixManager.connect();
- addInstanceTagIfNeeded();
- minionContext.setHelixPropertyStore(_helixManager.getHelixPropertyStore());
-
- LOGGER.info("Starting minion admin application on: {}", ListenerConfigUtil.toString(_listenerConfigs));
- _minionAdminApplication = new MinionAdminApiApplication(_config);
- _minionAdminApplication.start(_listenerConfigs);
-
- // Initialize health check callback
- LOGGER.info("Initializing health check callback");
- ServiceStatus.setServiceStatusCallback(_instanceId, new ServiceStatus.ServiceStatusCallback() {
- @Override
- public ServiceStatus.Status getServiceStatus() {
- // TODO: add health check here
- minionMetrics.addMeteredGlobalValue(MinionMeter.HEALTH_CHECK_GOOD_CALLS, 1L);
- return ServiceStatus.Status.GOOD;
- }
-
- @Override
- public String getStatusDescription() {
- return ServiceStatus.STATUS_DESCRIPTION_NONE;
- }
- });
-
- LOGGER.info("Pinot minion started");
- }
-
- /**
- * Stops the Pinot Minion instance.
- */
- @Override
- public void stop() {
- try {
- LOGGER.info("Closing PinotFS classes");
- PinotFSFactory.shutdown();
- } catch (IOException e) {
- LOGGER.warn("Caught exception closing PinotFS classes", e);
- }
- LOGGER.info("Shutting down admin application");
- _minionAdminApplication.stop();
-
- LOGGER.info("Stopping Pinot minion: " + _instanceId);
- _helixManager.disconnect();
- LOGGER.info("Deregistering service status handler");
- ServiceStatus.removeServiceStatusCallback(_instanceId);
- LOGGER.info("Clean up Minion data directory");
- try {
- FileUtils.cleanDirectory(MinionContext.getInstance().getDataDir());
- } catch (IOException e) {
- LOGGER.warn("Failed to clean up Minion data directory: {}", MinionContext.getInstance().getDataDir(), e);
- }
- LOGGER.info("Pinot minion stopped");
- }
-
- /**
- * Tags Pinot Minion instance if needed.
- */
- private void addInstanceTagIfNeeded() {
- HelixAdmin helixAdmin = _helixManager.getClusterManagmentTool();
- String clusterName = _helixManager.getClusterName();
- InstanceConfig instanceConfig = helixAdmin.getInstanceConfig(clusterName, _instanceId);
- List<String> instanceTags = instanceConfig == null ? new ArrayList<>(0) : instanceConfig.getTags();
- if (instanceTags.isEmpty()) {
- LOGGER.info("Adding default Helix tag: {} to Pinot minion", CommonConstants.Helix.UNTAGGED_MINION_INSTANCE);
- helixAdmin.addInstanceTag(clusterName, _instanceId, CommonConstants.Helix.UNTAGGED_MINION_INSTANCE);
- }
+ init(config);
}
}
diff --git a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/segment_generation_and_push/SegmentGenerationAndPushTaskGeneratorTest.java b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/segment_generation_and_push/SegmentGenerationAndPushTaskGeneratorTest.java
index 1518c97..5fba979 100644
--- a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/segment_generation_and_push/SegmentGenerationAndPushTaskGeneratorTest.java
+++ b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/segment_generation_and_push/SegmentGenerationAndPushTaskGeneratorTest.java
@@ -38,7 +38,8 @@ public class SegmentGenerationAndPushTaskGeneratorTest extends ControllerTest {
SegmentGenerationAndPushTaskGenerator _generator;
@BeforeClass
- public void setup() {
+ public void setup()
+ throws Exception {
int zkPort = 2171;
startZk(zkPort);
Map<String, Object> properties = getDefaultControllerConfiguration();
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
similarity index 93%
copy from pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
copy to pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
index ef239fe..138dcd7 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
@@ -107,39 +107,28 @@ import org.slf4j.LoggerFactory;
* </li>
* </ul>
*/
-public class HelixServerStarter implements ServiceStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(HelixServerStarter.class);
-
- private final String _helixClusterName;
- private final String _zkAddress;
- private final PinotConfiguration _serverConf;
- private final List<ListenerConfig> _listenerConfigs;
- private final String _host;
- private final int _port;
- private final String _instanceId;
- private final HelixConfigScope _instanceConfigScope;
- private HelixManager _helixManager;
- private HelixAdmin _helixAdmin;
- private ServerInstance _serverInstance;
- private AdminApiApplication _adminApiApplication;
- private ServerQueriesDisabledTracker _serverQueriesDisabledTracker;
- private RealtimeLuceneIndexRefreshState _realtimeLuceneIndexRefreshState;
- private PinotEnvironmentProvider _pinotEnvironmentProvider;
-
- @Deprecated
- public HelixServerStarter(String helixClusterName, String zkAddress, PinotConfiguration serverConf)
- throws Exception {
- this(applyServerConfig(serverConf, helixClusterName, zkAddress));
- }
-
- @Deprecated
- private static PinotConfiguration applyServerConfig(PinotConfiguration serverConf, String helixClusterName, String zkAddress) {
- serverConf.setProperty(Helix.CONFIG_OF_CLUSTER_NAME, helixClusterName);
- serverConf.setProperty(Helix.CONFIG_OF_ZOOKEEPR_SERVER, zkAddress);
- return serverConf;
- }
+public abstract class BaseServerStarter implements ServiceStartable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(BaseServerStarter.class);
+
+ protected String _helixClusterName;
+ protected String _zkAddress;
+ protected PinotConfiguration _serverConf;
+ protected List<ListenerConfig> _listenerConfigs;
+ protected String _host;
+ protected int _port;
+ protected String _instanceId;
+ protected HelixConfigScope _instanceConfigScope;
+ protected HelixManager _helixManager;
+ protected HelixAdmin _helixAdmin;
+ protected ServerInstance _serverInstance;
+ protected AdminApiApplication _adminApiApplication;
+ protected ServerQueriesDisabledTracker _serverQueriesDisabledTracker;
+ protected RealtimeLuceneIndexRefreshState _realtimeLuceneIndexRefreshState;
+ protected PinotEnvironmentProvider _pinotEnvironmentProvider;
- public HelixServerStarter(PinotConfiguration serverConf) throws Exception {
+ @Override
+ public void init(PinotConfiguration serverConf)
+ throws Exception {
// Make a clone so that changes to the config won't propagate to the caller
_serverConf = serverConf.clone();
_listenerConfigs = ListenerConfigUtil.buildServerAdminConfigs(_serverConf);
@@ -717,27 +706,4 @@ public class HelixServerStarter implements ServiceStartable {
public PinotConfiguration getConfig() {
return _serverConf;
}
-
- /**
- * This method is for reference purpose only.
- */
- public static HelixServerStarter startDefault()
- throws Exception {
- Map<String, Object> properties = new HashMap<>();
- int port = 8003;
- properties.put(Helix.CONFIG_OF_CLUSTER_NAME, "quickstart");
- properties.put(Helix.CONFIG_OF_ZOOKEEPR_SERVER, "localhost:2191");
- properties.put(Helix.KEY_OF_SERVER_NETTY_PORT, port);
- properties.put(Server.CONFIG_OF_INSTANCE_DATA_DIR, "/tmp/PinotServer/test" + port + "/index");
- properties.put(Server.CONFIG_OF_INSTANCE_SEGMENT_TAR_DIR, "/tmp/PinotServer/test" + port + "/segmentTar");
-
- HelixServerStarter serverStarter = new HelixServerStarter(new PinotConfiguration(properties));
- serverStarter.start();
- return serverStarter;
- }
-
- public static void main(String[] args)
- throws Exception {
- startDefault();
- }
}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
index ef239fe..3bd4038 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
@@ -18,74 +18,11 @@
*/
package org.apache.pinot.server.starter.helix;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import javax.annotation.Nullable;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.helix.HelixAdmin;
-import org.apache.helix.HelixDataAccessor;
-import org.apache.helix.HelixManager;
-import org.apache.helix.HelixManagerFactory;
-import org.apache.helix.InstanceType;
-import org.apache.helix.SystemPropertyKeys;
-import org.apache.helix.manager.zk.ZKHelixAdmin;
-import org.apache.helix.model.ExternalView;
-import org.apache.helix.model.HelixConfigScope;
-import org.apache.helix.model.HelixConfigScope.ConfigScopeProperty;
-import org.apache.helix.model.IdealState;
-import org.apache.helix.model.InstanceConfig;
-import org.apache.helix.model.Message;
-import org.apache.helix.model.builder.HelixConfigScopeBuilder;
-import org.apache.helix.participant.statemachine.StateModelFactory;
-import org.apache.pinot.common.Utils;
-import org.apache.pinot.common.metadata.ZKMetadataProvider;
-import org.apache.pinot.common.metrics.ServerMeter;
-import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.common.utils.ServiceStatus;
-import org.apache.pinot.common.utils.ServiceStatus.Status;
-import org.apache.pinot.common.utils.config.TagNameUtils;
-import org.apache.pinot.core.common.datatable.DataTableBuilder;
-import org.apache.pinot.core.data.manager.InstanceDataManager;
-import org.apache.pinot.core.query.request.context.ThreadTimer;
-import org.apache.pinot.core.transport.ListenerConfig;
-import org.apache.pinot.core.transport.TlsConfig;
-import org.apache.pinot.core.util.ListenerConfigUtil;
-import org.apache.pinot.core.util.TlsUtils;
-import org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeLuceneIndexRefreshState;
-import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
-import org.apache.pinot.server.api.access.AccessControlFactory;
-import org.apache.pinot.server.conf.ServerConf;
-import org.apache.pinot.server.realtime.ControllerLeaderLocator;
-import org.apache.pinot.server.realtime.ServerSegmentCompletionProtocolHandler;
-import org.apache.pinot.server.starter.ServerInstance;
-import org.apache.pinot.server.starter.ServerQueriesDisabledTracker;
import org.apache.pinot.spi.env.PinotConfiguration;
-import org.apache.pinot.spi.environmentprovider.PinotEnvironmentProvider;
-import org.apache.pinot.spi.environmentprovider.PinotEnvironmentProviderFactory;
-import org.apache.pinot.spi.filesystem.PinotFSFactory;
-import org.apache.pinot.spi.plugin.PluginManager;
-import org.apache.pinot.spi.services.ServiceRole;
-import org.apache.pinot.spi.services.ServiceStartable;
-import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Helix;
-import org.apache.pinot.spi.utils.CommonConstants.Helix.Instance;
-import org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel;
import org.apache.pinot.spi.utils.CommonConstants.Server;
-import org.apache.pinot.spi.utils.CommonConstants.Server.SegmentCompletionProtocol;
-import org.apache.pinot.spi.utils.NetUtils;
-import org.apache.pinot.spi.utils.builder.TableNameBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
@@ -107,29 +44,15 @@ import org.slf4j.LoggerFactory;
* </li>
* </ul>
*/
-public class HelixServerStarter implements ServiceStartable {
- private static final Logger LOGGER = LoggerFactory.getLogger(HelixServerStarter.class);
+public class HelixServerStarter extends BaseServerStarter {
- private final String _helixClusterName;
- private final String _zkAddress;
- private final PinotConfiguration _serverConf;
- private final List<ListenerConfig> _listenerConfigs;
- private final String _host;
- private final int _port;
- private final String _instanceId;
- private final HelixConfigScope _instanceConfigScope;
- private HelixManager _helixManager;
- private HelixAdmin _helixAdmin;
- private ServerInstance _serverInstance;
- private AdminApiApplication _adminApiApplication;
- private ServerQueriesDisabledTracker _serverQueriesDisabledTracker;
- private RealtimeLuceneIndexRefreshState _realtimeLuceneIndexRefreshState;
- private PinotEnvironmentProvider _pinotEnvironmentProvider;
+ public HelixServerStarter() {
+ }
@Deprecated
public HelixServerStarter(String helixClusterName, String zkAddress, PinotConfiguration serverConf)
throws Exception {
- this(applyServerConfig(serverConf, helixClusterName, zkAddress));
+ init(applyServerConfig(serverConf, helixClusterName, zkAddress));
}
@Deprecated
@@ -139,583 +62,9 @@ public class HelixServerStarter implements ServiceStartable {
return serverConf;
}
+ @Deprecated
public HelixServerStarter(PinotConfiguration serverConf) throws Exception {
- // Make a clone so that changes to the config won't propagate to the caller
- _serverConf = serverConf.clone();
- _listenerConfigs = ListenerConfigUtil.buildServerAdminConfigs(_serverConf);
- _helixClusterName = _serverConf.getProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME);
- _zkAddress = _serverConf.getProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER);
-
- _host = _serverConf.getProperty(Helix.KEY_OF_SERVER_NETTY_HOST,
- _serverConf.getProperty(Helix.SET_INSTANCE_ID_TO_HOSTNAME_KEY, false) ? NetUtils.getHostnameOrAddress()
- : NetUtils.getHostAddress());
- _port = _serverConf.getProperty(Helix.KEY_OF_SERVER_NETTY_PORT, Helix.DEFAULT_SERVER_NETTY_PORT);
-
- String instanceId = _serverConf.getProperty(Server.CONFIG_OF_INSTANCE_ID);
- if (instanceId == null) {
- instanceId = Helix.PREFIX_OF_SERVER_INSTANCE + _host + "_" + _port;
- _serverConf.addProperty(Server.CONFIG_OF_INSTANCE_ID, instanceId);
- }
- _instanceId = instanceId;
-
- _instanceConfigScope =
- new HelixConfigScopeBuilder(ConfigScopeProperty.PARTICIPANT, _helixClusterName).forParticipant(_instanceId)
- .build();
-
- // Initialize Pinot Environment Provider
- _pinotEnvironmentProvider = initializePinotEnvironmentProvider();
-
- // Enable/disable thread CPU time measurement through instance config.
- ThreadTimer.setThreadCpuTimeMeasurementEnabled(_serverConf
- .getProperty(Server.CONFIG_OF_ENABLE_THREAD_CPU_TIME_MEASUREMENT,
- Server.DEFAULT_ENABLE_THREAD_CPU_TIME_MEASUREMENT));
-
- // Set data table version send to broker.
- DataTableBuilder.setCurrentDataTableVersion(_serverConf
- .getProperty(Server.CONFIG_OF_CURRENT_DATA_TABLE_VERSION,
- Server.DEFAULT_CURRENT_DATA_TABLE_VERSION));
- }
-
- /**
- * Invoke pinot environment provider factory's init method to register the environment provider &
- * return the instantiated environment provider.
- */
- @Nullable
- private PinotEnvironmentProvider initializePinotEnvironmentProvider() {
- PinotConfiguration environmentProviderConfigs = _serverConf.subset(Server.PREFIX_OF_CONFIG_OF_ENVIRONMENT_PROVIDER_FACTORY);
- if (environmentProviderConfigs.toMap().isEmpty()) {
- LOGGER.info("No environment provider config values provided for server property: {}",
- Server.PREFIX_OF_CONFIG_OF_ENVIRONMENT_PROVIDER_FACTORY);
- return null;
- }
-
- // Invoke pinot environment provider factory's init method
- PinotEnvironmentProviderFactory.init(environmentProviderConfigs);
-
- String environmentProviderClassName = _serverConf.getProperty(Server.ENVIRONMENT_PROVIDER_CLASS_NAME);
- if (environmentProviderClassName == null) {
- LOGGER.info("No className value provided for property: {}", Server.ENVIRONMENT_PROVIDER_CLASS_NAME);
- return null;
- }
-
- // Fetch environment provider instance
- return PinotEnvironmentProviderFactory.getEnvironmentProvider(environmentProviderClassName.toLowerCase());
- }
-
- /**
- * Fetches the resources to monitor and registers the {@link org.apache.pinot.common.utils.ServiceStatus.ServiceStatusCallback}s
- */
- private void registerServiceStatusHandler() {
- double minResourcePercentForStartup = _serverConf
- .getProperty(Server.CONFIG_OF_SERVER_MIN_RESOURCE_PERCENT_FOR_START,
- Server.DEFAULT_SERVER_MIN_RESOURCE_PERCENT_FOR_START);
- int realtimeConsumptionCatchupWaitMs = _serverConf
- .getProperty(Server.CONFIG_OF_STARTUP_REALTIME_CONSUMPTION_CATCHUP_WAIT_MS,
- Server.DEFAULT_STARTUP_REALTIME_CONSUMPTION_CATCHUP_WAIT_MS);
-
- // collect all resources which have this instance in the ideal state
- List<String> resourcesToMonitor = new ArrayList<>();
- // if even 1 resource has this instance in ideal state with state CONSUMING, set this to true
- boolean foundConsuming = false;
- boolean checkRealtime = realtimeConsumptionCatchupWaitMs > 0;
-
- for (String resourceName : _helixAdmin.getResourcesInCluster(_helixClusterName)) {
- // Only monitor table resources
- if (!TableNameBuilder.isTableResource(resourceName)) {
- continue;
- }
-
- // Only monitor enabled resources
- IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, resourceName);
- if (idealState.isEnabled()) {
-
- for (String partitionName : idealState.getPartitionSet()) {
- if (idealState.getInstanceSet(partitionName).contains(_instanceId)) {
- resourcesToMonitor.add(resourceName);
- break;
- }
- }
- if (checkRealtime && !foundConsuming && TableNameBuilder.isRealtimeTableResource(resourceName)) {
- for (String partitionName : idealState.getPartitionSet()) {
- if (StateModel.SegmentStateModel.CONSUMING
- .equals(idealState.getInstanceStateMap(partitionName).get(_instanceId))) {
- foundConsuming = true;
- break;
- }
- }
- }
- }
- }
-
- ImmutableList.Builder<ServiceStatus.ServiceStatusCallback> serviceStatusCallbackListBuilder =
- new ImmutableList.Builder<>();
- serviceStatusCallbackListBuilder.add(
- new ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_helixManager, _helixClusterName,
- _instanceId, resourcesToMonitor, minResourcePercentForStartup));
- serviceStatusCallbackListBuilder.add(
- new ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_helixManager, _helixClusterName,
- _instanceId, resourcesToMonitor, minResourcePercentForStartup));
- if (checkRealtime && foundConsuming) {
- serviceStatusCallbackListBuilder.add(
- new ServiceStatus.RealtimeConsumptionCatchupServiceStatusCallback(_helixManager, _helixClusterName,
- _instanceId, realtimeConsumptionCatchupWaitMs));
- }
- LOGGER.info("Registering service status handler");
- ServiceStatus.setServiceStatusCallback(_instanceId,
- new ServiceStatus.MultipleCallbackServiceStatusCallback(serviceStatusCallbackListBuilder.build()));
- }
-
- private void updateInstanceConfigIfNeeded(String host, int port) {
- InstanceConfig instanceConfig = _helixAdmin.getInstanceConfig(_helixClusterName, _instanceId);
- boolean needToUpdateInstanceConfig = false;
-
- // Add default instance tags if not exist
- List<String> instanceTags = instanceConfig.getTags();
- if (instanceTags == null || instanceTags.size() == 0) {
- if (ZKMetadataProvider.getClusterTenantIsolationEnabled(_helixManager.getHelixPropertyStore())) {
- instanceConfig.addTag(TagNameUtils.getOfflineTagForTenant(null));
- instanceConfig.addTag(TagNameUtils.getRealtimeTagForTenant(null));
- } else {
- instanceConfig.addTag(Helix.UNTAGGED_SERVER_INSTANCE);
- }
- needToUpdateInstanceConfig = true;
- }
-
- // Update host and port if needed
- if (!host.equals(instanceConfig.getHostName())) {
- instanceConfig.setHostName(host);
- needToUpdateInstanceConfig = true;
- }
- String portStr = Integer.toString(port);
- if (!portStr.equals(instanceConfig.getPort())) {
- instanceConfig.setPort(portStr);
- needToUpdateInstanceConfig = true;
- }
-
- // Update instance config with environment properties
- if (_pinotEnvironmentProvider != null) {
- // Retrieve failure domain information and add to the environment properties map
- String failureDomain = _pinotEnvironmentProvider.getFailureDomain();
- Map<String, String> environmentProperties = new HashMap<>();
- environmentProperties.put(CommonConstants.INSTANCE_FAILURE_DOMAIN, failureDomain);
-
- // Fetch existing environment properties map from instance configs
- Map<String, String> existingEnvironmentConfigsMap = instanceConfig.getRecord().getMapField(
- CommonConstants.ENVIRONMENT_IDENTIFIER);
-
- if (existingEnvironmentConfigsMap == null || !existingEnvironmentConfigsMap.equals(environmentProperties)) {
- instanceConfig.getRecord().setMapField(CommonConstants.ENVIRONMENT_IDENTIFIER, environmentProperties);
- LOGGER.info("Adding environment properties: {} for instance: {}", environmentProperties, _instanceId);
- needToUpdateInstanceConfig = true;
- }
- }
-
- if (needToUpdateInstanceConfig) {
- LOGGER.info("Updating instance config for instance: {} with instance tags: {}, host: {}, port: {}", _instanceId,
- instanceTags, host, port);
- } else {
- LOGGER.info("Instance config for instance: {} has instance tags: {}, host: {}, port: {}, no need to update",
- _instanceId, instanceTags, host, port);
- return;
- }
-
- // NOTE: Use HelixDataAccessor.setProperty() instead of HelixAdmin.setInstanceConfig() because the latter explicitly
- // forbids instance host/port modification
- HelixDataAccessor helixDataAccessor = _helixManager.getHelixDataAccessor();
- Preconditions.checkState(
- helixDataAccessor.setProperty(helixDataAccessor.keyBuilder().instanceConfig(_instanceId), instanceConfig),
- "Failed to update instance config");
- }
-
- private void setupHelixSystemProperties() {
- // NOTE: Helix will disconnect the manager and disable the instance if it detects flapping (too frequent disconnect
- // from ZooKeeper). Setting flapping time window to a small value can avoid this from happening. Helix ignores the
- // non-positive value, so set the default value as 1.
- System.setProperty(SystemPropertyKeys.FLAPPING_TIME_WINDOW,
- _serverConf.getProperty(Helix.CONFIG_OF_SERVER_FLAPPING_TIME_WINDOW_MS, Helix.DEFAULT_FLAPPING_TIME_WINDOW_MS));
- }
-
- /**
- * When the server starts, check if the service status turns GOOD.
- *
- * @param endTimeMs Timeout for the check
- */
- private void startupServiceStatusCheck(long endTimeMs) {
- LOGGER.info("Starting startup service status check");
- long startTimeMs = System.currentTimeMillis();
- long checkIntervalMs = _serverConf.getProperty(Server.CONFIG_OF_STARTUP_SERVICE_STATUS_CHECK_INTERVAL_MS,
- Server.DEFAULT_STARTUP_SERVICE_STATUS_CHECK_INTERVAL_MS);
-
- while (System.currentTimeMillis() < endTimeMs) {
- Status serviceStatus = ServiceStatus.getServiceStatus();
- long currentTimeMs = System.currentTimeMillis();
- if (serviceStatus == Status.GOOD) {
- LOGGER.info("Service status is GOOD after {}ms", currentTimeMs - startTimeMs);
- return;
- } else if (serviceStatus == Status.BAD) {
- throw new IllegalStateException("Service status is BAD");
- }
- long sleepTimeMs = Math.min(checkIntervalMs, endTimeMs - currentTimeMs);
- if (sleepTimeMs > 0) {
- LOGGER.info("Sleep for {}ms as service status has not turned GOOD: {}", sleepTimeMs,
- ServiceStatus.getStatusDescription());
- try {
- Thread.sleep(sleepTimeMs);
- } catch (InterruptedException e) {
- LOGGER.warn("Got interrupted while checking service status", e);
- Thread.currentThread().interrupt();
- break;
- }
- }
- }
-
- LOGGER.warn("Service status has not turned GOOD within {}ms: {}", System.currentTimeMillis() - startTimeMs,
- ServiceStatus.getStatusDescription());
- }
-
- @Override
- public ServiceRole getServiceRole() {
- return ServiceRole.SERVER;
- }
-
- @Override
- public void start()
- throws Exception {
- LOGGER.info("Starting Pinot server");
- long startTimeMs = System.currentTimeMillis();
-
- // install default SSL context if necessary (even if not force-enabled everywhere)
- TlsConfig tlsDefaults = TlsUtils.extractTlsConfig(_serverConf, Server.SERVER_TLS_PREFIX);
- if (StringUtils.isNotBlank(tlsDefaults.getKeyStorePath()) || StringUtils
- .isNotBlank(tlsDefaults.getTrustStorePath())) {
- LOGGER.info("Installing default SSL context for any client requests");
- TlsUtils.installDefaultSSLSocketFactory(tlsDefaults);
- }
-
- LOGGER.info("Initializing Helix manager with zkAddress: {}, clusterName: {}, instanceId: {}", _zkAddress,
- _helixClusterName, _instanceId);
- setupHelixSystemProperties();
- _helixManager =
- HelixManagerFactory.getZKHelixManager(_helixClusterName, _instanceId, InstanceType.PARTICIPANT, _zkAddress);
-
- LOGGER.info("Initializing server instance and registering state model factory");
- Utils.logVersions();
- ControllerLeaderLocator.create(_helixManager);
- ServerSegmentCompletionProtocolHandler
- .init(_serverConf.subset(SegmentCompletionProtocol.PREFIX_OF_CONFIG_OF_SEGMENT_UPLOADER));
- ServerConf serverInstanceConfig = DefaultHelixStarterServerConfig.getDefaultHelixServerConfig(_serverConf);
- _serverInstance = new ServerInstance(serverInstanceConfig, _helixManager);
- ServerMetrics serverMetrics = _serverInstance.getServerMetrics();
- InstanceDataManager instanceDataManager = _serverInstance.getInstanceDataManager();
- SegmentFetcherAndLoader fetcherAndLoader =
- new SegmentFetcherAndLoader(_serverConf, instanceDataManager, serverMetrics);
- StateModelFactory<?> stateModelFactory =
- new SegmentOnlineOfflineStateModelFactory(_instanceId, instanceDataManager, fetcherAndLoader);
- _helixManager.getStateMachineEngine()
- .registerStateModelFactory(SegmentOnlineOfflineStateModelFactory.getStateModelName(), stateModelFactory);
- // Start the server instance as a pre-connect callback so that it starts after connecting to the ZK in order to
- // access the property store, but before receiving state transitions
- _helixManager.addPreConnectCallback(_serverInstance::start);
-
- LOGGER.info("Connecting Helix manager");
- _helixManager.connect();
- _helixAdmin = _helixManager.getClusterManagmentTool();
- updateInstanceConfigIfNeeded(_host, _port);
-
- // Start restlet server for admin API endpoint
- String accessControlFactoryClass =
- _serverConf.getProperty(Server.ACCESS_CONTROL_FACTORY_CLASS, Server.DEFAULT_ACCESS_CONTROL_FACTORY_CLASS);
- LOGGER.info("Using class: {} as the AccessControlFactory", accessControlFactoryClass);
- final AccessControlFactory accessControlFactory;
- try {
- accessControlFactory = PluginManager.get().createInstance(accessControlFactoryClass);
- } catch (Exception e) {
- throw new RuntimeException(
- "Caught exception while creating new AccessControlFactory instance using class '" + accessControlFactoryClass
- + "'", e);
- }
-
- // Update admin API port
- LOGGER.info("Starting server admin application on: {}", ListenerConfigUtil.toString(_listenerConfigs));
- _adminApiApplication = new AdminApiApplication(_serverInstance, accessControlFactory, _serverConf);
- _adminApiApplication.start(_listenerConfigs);
-
- // Update http admin port
- Optional<ListenerConfig> adminApiHttp = _listenerConfigs.stream()
- .filter(listener -> CommonConstants.HTTP_PROTOCOL.equals(listener.getProtocol())).findFirst();
- if (adminApiHttp.isPresent()) {
- _helixAdmin.setConfig(_instanceConfigScope,
- Collections.singletonMap(Instance.ADMIN_PORT_KEY, String.valueOf(adminApiHttp.get().getPort())));
- } else {
- _helixAdmin.removeConfig(_instanceConfigScope, Collections.singletonList(Instance.ADMIN_PORT_KEY));
- }
-
- // Update https admin port
- Optional<ListenerConfig> adminApiHttps = _listenerConfigs.stream()
- .filter(listener -> CommonConstants.HTTPS_PROTOCOL.equals(listener.getProtocol())).findFirst();
- if (adminApiHttps.isPresent()) {
- _helixAdmin.setConfig(_instanceConfigScope,
- Collections.singletonMap(Instance.ADMIN_HTTPS_PORT_KEY, String.valueOf(adminApiHttps.get().getPort())));
- } else {
- _helixAdmin.removeConfig(_instanceConfigScope, Collections.singletonList(Instance.ADMIN_HTTPS_PORT_KEY));
- }
-
- // Update nettytls port
- if (serverInstanceConfig.isNettyTlsServerEnabled()) {
- _helixAdmin.setConfig(_instanceConfigScope,
- Collections.singletonMap(Instance.NETTYTLS_PORT_KEY, String.valueOf(serverInstanceConfig.getNettyTlsPort())));
- } else {
- _helixAdmin.removeConfig(_instanceConfigScope, Collections.singletonList(Instance.NETTYTLS_PORT_KEY));
- }
-
- // Update gRPC port
- if (serverInstanceConfig.isEnableGrpcServer()) {
- _helixAdmin.setConfig(_instanceConfigScope,
- Collections.singletonMap(Instance.GRPC_PORT_KEY, String.valueOf(serverInstanceConfig.getGrpcPort())));
- } else {
- _helixAdmin.removeConfig(_instanceConfigScope, Collections.singletonList(Instance.GRPC_PORT_KEY));
- }
-
- // Register message handler factory
- SegmentMessageHandlerFactory messageHandlerFactory =
- new SegmentMessageHandlerFactory(fetcherAndLoader, instanceDataManager, serverMetrics);
- _helixManager.getMessagingService()
- .registerMessageHandlerFactory(Message.MessageType.USER_DEFINE_MSG.toString(), messageHandlerFactory);
-
- serverMetrics.addCallbackGauge(Helix.INSTANCE_CONNECTED_METRIC_NAME, () -> _helixManager.isConnected() ? 1L : 0L);
- _helixManager
- .addPreConnectCallback(() -> serverMetrics.addMeteredGlobalValue(ServerMeter.HELIX_ZOOKEEPER_RECONNECTS, 1L));
-
- // Register the service status handler
- registerServiceStatusHandler();
-
- if (_serverConf.getProperty(Server.CONFIG_OF_STARTUP_ENABLE_SERVICE_STATUS_CHECK,
- Server.DEFAULT_STARTUP_ENABLE_SERVICE_STATUS_CHECK)) {
- long endTimeMs =
- startTimeMs + _serverConf.getProperty(Server.CONFIG_OF_STARTUP_TIMEOUT_MS, Server.DEFAULT_STARTUP_TIMEOUT_MS);
- startupServiceStatusCheck(endTimeMs);
- }
- _helixAdmin.setConfig(_instanceConfigScope,
- Collections.singletonMap(Helix.IS_SHUTDOWN_IN_PROGRESS, Boolean.toString(false)));
- LOGGER.info("Pinot server ready");
-
- // Create metrics for mmap stuff
- serverMetrics.addCallbackGauge("memory.directBufferCount", PinotDataBuffer::getDirectBufferCount);
- serverMetrics.addCallbackGauge("memory.directBufferUsage", PinotDataBuffer::getDirectBufferUsage);
- serverMetrics.addCallbackGauge("memory.mmapBufferCount", PinotDataBuffer::getMmapBufferCount);
- serverMetrics.addCallbackGauge("memory.mmapBufferUsage", PinotDataBuffer::getMmapBufferUsage);
- serverMetrics.addCallbackGauge("memory.allocationFailureCount", PinotDataBuffer::getAllocationFailureCount);
-
- // Track metric for queries disabled
- _serverQueriesDisabledTracker = new ServerQueriesDisabledTracker(_helixClusterName, _instanceId, _helixManager, serverMetrics);
- _serverQueriesDisabledTracker.start();
-
- _realtimeLuceneIndexRefreshState = RealtimeLuceneIndexRefreshState.getInstance();
- _realtimeLuceneIndexRefreshState.start();
- }
-
- @Override
- public void stop() {
- LOGGER.info("Shutting down Pinot server");
- long startTimeMs = System.currentTimeMillis();
-
- try {
- LOGGER.info("Closing PinotFS classes");
- PinotFSFactory.shutdown();
- } catch (IOException e) {
- LOGGER.warn("Caught exception closing PinotFS classes", e);
- }
- _adminApiApplication.stop();
- _helixAdmin.setConfig(_instanceConfigScope,
- Collections.singletonMap(Helix.IS_SHUTDOWN_IN_PROGRESS, Boolean.toString(true)));
-
- long endTimeMs =
- startTimeMs + _serverConf.getProperty(Server.CONFIG_OF_SHUTDOWN_TIMEOUT_MS, Server.DEFAULT_SHUTDOWN_TIMEOUT_MS);
- if (_serverConf
- .getProperty(Server.CONFIG_OF_SHUTDOWN_ENABLE_QUERY_CHECK, Server.DEFAULT_SHUTDOWN_ENABLE_QUERY_CHECK)) {
- shutdownQueryCheck(endTimeMs);
- }
- _helixManager.disconnect();
- _serverInstance.shutDown();
- if (_serverConf
- .getProperty(Server.CONFIG_OF_SHUTDOWN_ENABLE_RESOURCE_CHECK, Server.DEFAULT_SHUTDOWN_ENABLE_RESOURCE_CHECK)) {
- shutdownResourceCheck(endTimeMs);
- }
- _serverQueriesDisabledTracker.stop();
- _realtimeLuceneIndexRefreshState.stop();
- LOGGER.info("Deregistering service status handler");
- ServiceStatus.removeServiceStatusCallback(_instanceId);
- LOGGER.info("Finish shutting down Pinot server for {}", _instanceId);
- }
-
- /**
- * When shutting down the server, drains the queries (no incoming queries and all existing queries finished).
- *
- * @param endTimeMs Timeout for the check
- */
- private void shutdownQueryCheck(long endTimeMs) {
- LOGGER.info("Starting shutdown query check");
- long startTimeMs = System.currentTimeMillis();
-
- long maxQueryTimeMs =
- _serverConf.getProperty(Server.CONFIG_OF_QUERY_EXECUTOR_TIMEOUT, Server.DEFAULT_QUERY_EXECUTOR_TIMEOUT_MS);
- long noQueryThresholdMs = _serverConf.getProperty(Server.CONFIG_OF_SHUTDOWN_NO_QUERY_THRESHOLD_MS, maxQueryTimeMs);
-
- // Wait until no incoming queries
- boolean noIncomingQueries = false;
- long currentTimeMs;
- while ((currentTimeMs = System.currentTimeMillis()) < endTimeMs) {
- long noQueryTimeMs = currentTimeMs - _serverInstance.getLatestQueryTime();
- if (noQueryTimeMs >= noQueryThresholdMs) {
- LOGGER.info("No query received within {}ms (larger than the threshold: {}ms), mark it as no incoming queries",
- noQueryTimeMs, noQueryThresholdMs);
- noIncomingQueries = true;
- break;
- }
- long sleepTimeMs = Math.min(noQueryThresholdMs - noQueryTimeMs, endTimeMs - currentTimeMs);
- LOGGER.info(
- "Sleep for {}ms as there are still incoming queries (no query time: {}ms is smaller than the threshold: {}ms)",
- sleepTimeMs, noQueryTimeMs, noQueryThresholdMs);
- try {
- Thread.sleep(sleepTimeMs);
- } catch (InterruptedException e) {
- LOGGER.warn("Got interrupted while waiting for no incoming queries", e);
- Thread.currentThread().interrupt();
- break;
- }
- }
- if (noIncomingQueries) {
- // Ensure all the existing queries are finished
- long latestQueryFinishTimeMs = _serverInstance.getLatestQueryTime() + maxQueryTimeMs;
- if (latestQueryFinishTimeMs > currentTimeMs) {
- long sleepTimeMs = latestQueryFinishTimeMs - currentTimeMs;
- LOGGER.info("Sleep for {}ms to ensure all the existing queries are finished", sleepTimeMs);
- try {
- Thread.sleep(sleepTimeMs);
- } catch (InterruptedException e) {
- LOGGER.warn("Got interrupted while waiting for all the existing queries to be finished", e);
- Thread.currentThread().interrupt();
- }
- }
- LOGGER.info("Finished draining queries after {}ms", System.currentTimeMillis() - startTimeMs);
- } else {
- LOGGER.warn("Failed to drain queries within {}ms", System.currentTimeMillis() - startTimeMs);
- }
- }
-
- /**
- * When shutting down the server, waits for all the resources turn OFFLINE (all partitions served by the server are
- * neither ONLINE or CONSUMING).
- *
- * @param endTimeMs Timeout for the check
- */
- private void shutdownResourceCheck(long endTimeMs) {
- LOGGER.info("Starting shutdown resource check");
- long startTimeMs = System.currentTimeMillis();
-
- if (startTimeMs >= endTimeMs) {
- LOGGER.warn("Skipping shutdown resource check because shutdown timeout is already reached");
- return;
- }
-
- HelixAdmin helixAdmin = null;
- try {
- helixAdmin = new ZKHelixAdmin(_zkAddress);
-
- // Monitor all enabled table resources that the server serves
- Set<String> resourcesToMonitor = new HashSet<>();
- for (String resourceName : helixAdmin.getResourcesInCluster(_helixClusterName)) {
- if (TableNameBuilder.isTableResource(resourceName)) {
- IdealState idealState = helixAdmin.getResourceIdealState(_helixClusterName, resourceName);
- if (idealState == null || !idealState.isEnabled()) {
- continue;
- }
- for (String partition : idealState.getPartitionSet()) {
- if (idealState.getInstanceSet(partition).contains(_instanceId)) {
- resourcesToMonitor.add(resourceName);
- break;
- }
- }
- }
- }
-
- long checkIntervalMs = _serverConf.getProperty(Server.CONFIG_OF_SHUTDOWN_RESOURCE_CHECK_INTERVAL_MS,
- Server.DEFAULT_SHUTDOWN_RESOURCE_CHECK_INTERVAL_MS);
- while (System.currentTimeMillis() < endTimeMs) {
- Iterator<String> iterator = resourcesToMonitor.iterator();
- String currentResource = null;
- while (iterator.hasNext()) {
- currentResource = iterator.next();
- if (isResourceOffline(helixAdmin, currentResource)) {
- iterator.remove();
- } else {
- // Do not check remaining resources if one resource is not OFFLINE
- break;
- }
- }
- long currentTimeMs = System.currentTimeMillis();
- if (resourcesToMonitor.isEmpty()) {
- LOGGER.info("All resources are OFFLINE after {}ms", currentTimeMs - startTimeMs);
- return;
- }
- long sleepTimeMs = Math.min(checkIntervalMs, endTimeMs - currentTimeMs);
- if (sleepTimeMs > 0) {
- LOGGER.info("Sleep for {}ms as some resources [{}, ...] are still ONLINE", sleepTimeMs, currentResource);
- try {
- Thread.sleep(sleepTimeMs);
- } catch (InterruptedException e) {
- LOGGER.warn("Got interrupted while waiting for all resources OFFLINE", e);
- Thread.currentThread().interrupt();
- break;
- }
- }
- }
-
- // Check all remaining resources
- Iterator<String> iterator = resourcesToMonitor.iterator();
- while (iterator.hasNext()) {
- if (isResourceOffline(helixAdmin, iterator.next())) {
- iterator.remove();
- }
- }
- long currentTimeMs = System.currentTimeMillis();
- if (resourcesToMonitor.isEmpty()) {
- LOGGER.info("All resources are OFFLINE after {}ms", currentTimeMs - startTimeMs);
- } else {
- LOGGER.warn("There are still {} resources ONLINE within {}ms: {}", resourcesToMonitor.size(),
- currentTimeMs - startTimeMs, resourcesToMonitor);
- }
- } finally {
- if (helixAdmin != null) {
- helixAdmin.close();
- }
- }
- }
-
- private boolean isResourceOffline(HelixAdmin helixAdmin, String resource) {
- ExternalView externalView = helixAdmin.getResourceExternalView(_helixClusterName, resource);
- // Treat deleted resource as OFFLINE
- if (externalView == null) {
- return true;
- }
- for (String partition : externalView.getPartitionSet()) {
- Map<String, String> instanceStateMap = externalView.getStateMap(partition);
- String state = instanceStateMap.get(_instanceId);
- if (StateModel.SegmentStateModel.ONLINE.equals(state) || StateModel.SegmentStateModel.CONSUMING.equals(state)) {
- return false;
- }
- }
- return true;
- }
-
- @Override
- public String getInstanceId() {
- return _instanceId;
- }
-
- @Override
- public PinotConfiguration getConfig() {
- return _serverConf;
+ init(serverConf);
}
/**
@@ -731,7 +80,8 @@ public class HelixServerStarter implements ServiceStartable {
properties.put(Server.CONFIG_OF_INSTANCE_DATA_DIR, "/tmp/PinotServer/test" + port + "/index");
properties.put(Server.CONFIG_OF_INSTANCE_SEGMENT_TAR_DIR, "/tmp/PinotServer/test" + port + "/segmentTar");
- HelixServerStarter serverStarter = new HelixServerStarter(new PinotConfiguration(properties));
+ HelixServerStarter serverStarter = new HelixServerStarter();
+ serverStarter.init(new PinotConfiguration(properties));
serverStarter.start();
return serverStarter;
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/services/ServiceStartable.java b/pinot-spi/src/main/java/org/apache/pinot/spi/services/ServiceStartable.java
index 0c54ed8..7afc07a 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/services/ServiceStartable.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/services/ServiceStartable.java
@@ -27,6 +27,9 @@ import org.apache.pinot.spi.env.PinotConfiguration;
*
*/
public interface ServiceStartable {
+ void init(PinotConfiguration conf)
+ throws Exception;
+
ServiceRole getServiceRole();
String getInstanceId();
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 65ff65f..bc77eb2 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -152,6 +152,11 @@ public class CommonConstants {
public static final String PINOT_SERVICE_ROLE = "pinot.service.role";
public static final String CONFIG_OF_CLUSTER_NAME = "pinot.cluster.name";
public static final String CONFIG_OF_ZOOKEEPR_SERVER = "pinot.zk.server";
+
+ public static final String CONFIG_OF_PINOT_CONTROLLER_STARTABLE_CLASS = "pinot.controller.startable.class";
+ public static final String CONFIG_OF_PINOT_BROKER_STARTABLE_CLASS = "pinot.broker.startable.class";
+ public static final String CONFIG_OF_PINOT_SERVER_STARTABLE_CLASS = "pinot.server.startable.class";
+ public static final String CONFIG_OF_PINOT_MINION_STARTABLE_CLASS = "pinot.minion.startable.class";
}
public static class Broker {
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java b/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
index 2e92806..cbfea90 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
@@ -188,17 +188,32 @@ public class PerfBenchmarkDriver {
ZkStarter.startLocalZkServer(zkPort);
}
- private void startController() {
+ private void startController()
+ throws Exception {
if (!_conf.shouldStartController()) {
LOGGER.info("Skipping start controller step. Assumes controller is already started.");
return;
}
- ControllerConf conf = getControllerConf();
+
LOGGER.info("Starting controller at {}", _controllerAddress);
- _controllerStarter = new ControllerStarter(conf);
+ _controllerStarter = new ControllerStarter();
+ _controllerStarter.init(new PinotConfiguration(getControllerProperties()));
_controllerStarter.start();
}
+ private Map<String, Object> getControllerProperties() {
+ Map<String, Object> properties = new HashMap<>();
+ properties.put(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, _clusterName);
+ properties.put(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER, _zkAddress);
+ properties.put(ControllerConf.CONTROLLER_HOST, _controllerHost);
+ properties.put(ControllerConf.CONTROLLER_PORT, String.valueOf(_controllerPort));
+ properties.put(ControllerConf.DATA_DIR, _controllerDataDir);
+ properties.put(ControllerConf.CLUSTER_TENANT_ISOLATION_ENABLE, false);
+ properties.put(ControllerConf.CONTROLLER_VIP_HOST, "localhost");
+ properties.put(ControllerConf.CONTROLLER_VIP_PROTOCOL, CommonConstants.HTTP_PROTOCOL);
+ return properties;
+ }
+
private ControllerConf getControllerConf() {
ControllerConf conf = new ControllerConf();
conf.setHelixClusterName(_clusterName);
@@ -228,7 +243,9 @@ public class PerfBenchmarkDriver {
LOGGER.info("Starting broker instance: {}", brokerInstanceName);
- new HelixBrokerStarter(new PinotConfiguration(properties)).start();
+ HelixBrokerStarter helixBrokerStarter = new HelixBrokerStarter();
+ helixBrokerStarter.init(new PinotConfiguration(properties));
+ helixBrokerStarter.start();
}
private void startServer()
@@ -251,7 +268,8 @@ public class PerfBenchmarkDriver {
LOGGER.info("Starting server instance: {}", _serverInstanceName);
- HelixServerStarter helixServerStarter = new HelixServerStarter(new PinotConfiguration(properties));
+ HelixServerStarter helixServerStarter = new HelixServerStarter();
+ helixServerStarter.init(new PinotConfiguration(properties));
helixServerStarter.start();
}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java b/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
index bd5bb28..9a39e7b 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/service/PinotServiceManager.java
@@ -19,12 +19,12 @@
package org.apache.pinot.tools.service;
import com.google.common.collect.ImmutableList;
+import java.lang.reflect.InvocationTargetException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.pinot.broker.broker.helix.HelixBrokerStarter;
import org.apache.pinot.common.utils.ServiceStatus;
-import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.ControllerStarter;
import org.apache.pinot.minion.MinionStarter;
import org.apache.pinot.server.starter.helix.HelixServerStarter;
@@ -87,31 +87,39 @@ public class PinotServiceManager {
throws Exception {
switch (role) {
case CONTROLLER:
- return startController(new ControllerConf(properties));
+ String controllerStarterClassName = (String) properties
+ .getOrDefault(CommonConstants.Helix.CONFIG_OF_PINOT_CONTROLLER_STARTABLE_CLASS,
+ ControllerStarter.class.getName());
+ return startController(controllerStarterClassName, new PinotConfiguration(properties));
case BROKER:
- return startBroker(new PinotConfiguration(properties));
+ String brokerStarterClassName = (String) properties
+ .getOrDefault(CommonConstants.Helix.CONFIG_OF_PINOT_BROKER_STARTABLE_CLASS,
+ HelixBrokerStarter.class.getName());
+ return startBroker(brokerStarterClassName, new PinotConfiguration(properties));
case SERVER:
- return startServer(new PinotConfiguration(properties));
+ String serverStarterClassName = (String) properties
+ .getOrDefault(CommonConstants.Helix.CONFIG_OF_PINOT_SERVER_STARTABLE_CLASS,
+ HelixServerStarter.class.getName());
+ return startServer(serverStarterClassName, new PinotConfiguration(properties));
case MINION:
- return startMinion(new PinotConfiguration(properties));
+ String minionStarterClassName = (String) properties
+ .getOrDefault(CommonConstants.Helix.CONFIG_OF_PINOT_MINION_STARTABLE_CLASS, MinionStarter.class.getName());
+ return startMinion(minionStarterClassName, new PinotConfiguration(properties));
}
return null;
}
- public String startController(ControllerConf controllerConf)
+ public String startController(String controllerStarterClassName, PinotConfiguration controllerConf)
throws Exception {
LOGGER.info("Trying to start Pinot Controller...");
- if (controllerConf.getHelixClusterName() == null) {
- controllerConf.setHelixClusterName(_clusterName);
+ if (!controllerConf.containsKey(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME)) {
+ controllerConf.setProperty(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME, _clusterName);
}
- try {
- if (controllerConf.getZkStr() == null) {
- controllerConf.setZkStr(_zkAddress);
- }
- } catch (Exception e) {
- controllerConf.setZkStr(_zkAddress);
+ if (!controllerConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER)) {
+ controllerConf.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER, _zkAddress);
}
- ControllerStarter controllerStarter = new ControllerStarter(controllerConf);
+ ServiceStartable controllerStarter = getServiceStartable(controllerStarterClassName);
+ controllerStarter.init(controllerConf);
controllerStarter.start();
String instanceId = controllerStarter.getInstanceId();
_runningInstanceMap.put(instanceId, controllerStarter);
@@ -119,7 +127,7 @@ public class PinotServiceManager {
return instanceId;
}
- public String startBroker(PinotConfiguration brokerConf)
+ public String startBroker(String brokerStarterClassName, PinotConfiguration brokerConf)
throws Exception {
LOGGER.info("Trying to start Pinot Broker...");
if (!brokerConf.containsKey(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME)) {
@@ -128,9 +136,10 @@ public class PinotServiceManager {
if (!brokerConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER)) {
brokerConf.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER, _zkAddress);
}
- HelixBrokerStarter brokerStarter;
+ ServiceStartable brokerStarter;
try {
- brokerStarter = new HelixBrokerStarter(brokerConf);
+ brokerStarter = getServiceStartable(brokerStarterClassName);
+ brokerStarter.init(brokerConf);
} catch (Exception e) {
LOGGER.error("Failed to initialize Pinot Broker Starter", e);
throw e;
@@ -147,7 +156,7 @@ public class PinotServiceManager {
return instanceId;
}
- public String startServer(PinotConfiguration serverConf)
+ public String startServer(String serverStarterClassName, PinotConfiguration serverConf)
throws Exception {
LOGGER.info("Trying to start Pinot Server...");
@@ -158,7 +167,8 @@ public class PinotServiceManager {
if (!serverConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER)) {
serverConf.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER, _zkAddress);
}
- HelixServerStarter serverStarter = new HelixServerStarter(serverConf);
+ ServiceStartable serverStarter = getServiceStartable(serverStarterClassName);
+ serverStarter.init(serverConf);
serverStarter.start();
String instanceId = serverStarter.getInstanceId();
@@ -167,7 +177,7 @@ public class PinotServiceManager {
return instanceId;
}
- public String startMinion(PinotConfiguration minionConf)
+ public String startMinion(String minionStarterClassName, PinotConfiguration minionConf)
throws Exception {
LOGGER.info("Trying to start Pinot Minion...");
if (!minionConf.containsKey(CommonConstants.Helix.CONFIG_OF_CLUSTER_NAME)) {
@@ -176,7 +186,8 @@ public class PinotServiceManager {
if (!minionConf.containsKey(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER)) {
minionConf.setProperty(CommonConstants.Helix.CONFIG_OF_ZOOKEEPR_SERVER, _zkAddress);
}
- MinionStarter minionStarter = new MinionStarter(minionConf);
+ ServiceStartable minionStarter = getServiceStartable(minionStarterClassName);
+ minionStarter.init(minionConf);
minionStarter.start();
String instanceId = minionStarter.getInstanceId();
@@ -261,4 +272,12 @@ public class PinotServiceManager {
public boolean stopPinotInstanceById(String instanceName) {
return stopPinotInstance(_runningInstanceMap.get(instanceName));
}
+
+ private ServiceStartable getServiceStartable(String serviceStartableClassName) {
+ try {
+ return (ServiceStartable) Class.forName(serviceStartableClassName).getDeclaredConstructor().newInstance();
+ } catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException | ClassNotFoundException e) {
+ throw new RuntimeException("Failed to instantiate ServiceStartable " + serviceStartableClassName, e);
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org