You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2019/08/21 07:49:27 UTC
[incubator-pinot] 03/08: [Controller Separation] Add logic for lead
controller resource (#4323)
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a commit to branch winedepot-0.10
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 179377c232d56cc6b26352b3034f1206167aeaa2
Author: Jialiang Li <jl...@linkedin.com>
AuthorDate: Thu Aug 15 15:26:29 2019 -0700
[Controller Separation] Add logic for lead controller resource (#4323)
* Add logic for leveraging lead controller resource
* Add logic for lead controller resource on controller side
* Remove API design out of this PR
* Move murmur2 functions to an util class in pinot-common
* Use resource config to check whether resource is enabled
* Create a customized MasterSlaveStateModelFactory instead of inherating one from Helix
* Check resource config is enabled from server
---
.../apache/pinot/common/utils/CommonConstants.java | 2 +
.../org/apache/pinot/common/utils/HashUtil.java | 47 ++++
.../common/utils/helix/LeadControllerUtils.java | 61 +++++
.../controller/ControllerLeadershipManager.java | 172 --------------
.../apache/pinot/controller/ControllerStarter.java | 162 +++++++------
.../pinot/controller/LeadControllerManager.java | 229 ++++++++++++++++++
.../PinotSegmentUploadRestletResource.java | 6 +-
.../controller/api/upload/SegmentValidator.java | 10 +-
.../controller/helix/SegmentStatusChecker.java | 8 +-
.../helix/core/PinotHelixResourceManager.java | 6 +-
.../helix/core/minion/PinotTaskManager.java | 8 +-
.../core/periodictask/ControllerPeriodicTask.java | 10 +-
.../ControllerPeriodicTaskScheduler.java | 60 -----
.../realtime/PinotLLCRealtimeSegmentManager.java | 32 ++-
.../core/realtime/PinotRealtimeSegmentManager.java | 43 +---
.../core/realtime/SegmentCompletionManager.java | 54 +++--
.../core/relocation/RealtimeSegmentRelocator.java | 8 +-
.../helix/core/retention/RetentionManager.java | 8 +-
...rollerResourceMasterSlaveStateModelFactory.java | 84 +++++++
.../helix/core/util/HelixSetupUtils.java | 17 +-
.../BrokerResourceValidationManager.java | 5 +-
.../validation/OfflineSegmentIntervalChecker.java | 13 +-
.../RealtimeSegmentValidationManager.java | 8 +-
.../controller/validation/StorageQuotaChecker.java | 17 +-
.../controller/LeadControllerManagerTest.java | 125 ++++++++++
.../pinot/controller/helix/ControllerTest.java | 14 ++
.../controller/helix/PinotControllerModeTest.java | 169 ++++++++-----
.../controller/helix/SegmentStatusCheckerTest.java | 73 +++++-
.../helix/core/PinotHelixResourceManagerTest.java | 21 +-
.../periodictask/ControllerPeriodicTaskTest.java | 7 +-
.../PinotLLCRealtimeSegmentManagerTest.java | 7 +-
.../helix/core/realtime/SegmentCompletionTest.java | 263 ++++++++++-----------
.../relocation/RealtimeSegmentRelocatorTest.java | 14 +-
.../helix/core/retention/RetentionManagerTest.java | 12 +-
.../validation/StorageQuotaCheckerTest.java | 36 +--
.../realtime/LLRealtimeSegmentDataManager.java | 4 +-
.../manager/realtime/RealtimeTableDataManager.java | 2 +-
.../realtime/SegmentBuildTimeLeaseExtender.java | 8 +-
.../data/partition/MurmurPartitionFunction.java | 41 +---
.../server/realtime/ControllerLeaderLocator.java | 153 ++++++++++--
.../ServerSegmentCompletionProtocolHandler.java | 10 +-
.../realtime/LLRealtimeSegmentDataManagerTest.java | 2 +-
.../realtime/ControllerLeaderLocatorTest.java | 99 +++++++-
...abledLeadControllerResourceIntegrationTest.java | 22 +-
.../tests/SegmentCompletionIntegrationTest.java | 2 +-
45 files changed, 1394 insertions(+), 760 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
index d17eb79..964a376 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
@@ -39,6 +39,8 @@ public class CommonConstants {
public static final String BROKER_RESOURCE_INSTANCE = "brokerResource";
public static final String LEAD_CONTROLLER_RESOURCE_NAME = "leadControllerResource";
+ public static final String LEAD_CONTROLLER_RESOURCE_ENABLED_KEY = "RESOURCE_ENABLED";
+
// More information on why these numbers are set can be found in the following doc:
// https://cwiki.apache.org/confluence/display/PINOT/Controller+Separation+between+Helix+and+Pinot
public static final int NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE = 24;
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/HashUtil.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/HashUtil.java
index 955ab84..be12cc1 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/HashUtil.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/HashUtil.java
@@ -105,4 +105,51 @@ public class HashUtil {
h ^= h >>> r;
return h;
}
+
+ /**
+ * Generates 32 bit murmur2 hash from byte array
+ * @param data byte array to hash
+ * @return 32 bit hash of the given array
+ */
+ public static int murmur2(final byte[] data) {
+ int length = data.length;
+ int seed = 0x9747b28c;
+ // 'm' and 'r' are mixing constants generated offline.
+ // They're not really 'magic', they just happen to work well.
+ final int m = 0x5bd1e995;
+ final int r = 24;
+
+ // Initialize the hash to a random value
+ int h = seed ^ length;
+ int length4 = length / 4;
+
+ for (int i = 0; i < length4; i++) {
+ final int i4 = i * 4;
+ int k =
+ (data[i4 + 0] & 0xff) + ((data[i4 + 1] & 0xff) << 8) + ((data[i4 + 2] & 0xff) << 16) + ((data[i4 + 3] & 0xff)
+ << 24);
+ k *= m;
+ k ^= k >>> r;
+ k *= m;
+ h *= m;
+ h ^= k;
+ }
+
+ // Handle the last few bytes of the input array
+ switch (length % 4) {
+ case 3:
+ h ^= (data[(length & ~3) + 2] & 0xff) << 16;
+ case 2:
+ h ^= (data[(length & ~3) + 1] & 0xff) << 8;
+ case 1:
+ h ^= data[length & ~3] & 0xff;
+ h *= m;
+ }
+
+ h ^= h >>> 13;
+ h *= m;
+ h ^= h >>> 15;
+
+ return h;
+ }
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/LeadControllerUtils.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/LeadControllerUtils.java
new file mode 100644
index 0000000..e6fffc8
--- /dev/null
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/LeadControllerUtils.java
@@ -0,0 +1,61 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.common.utils.helix;
+
+import org.apache.pinot.common.utils.CommonConstants.Helix;
+import org.apache.pinot.common.utils.HashUtil;
+import org.apache.pinot.common.utils.StringUtil;
+
+
+public class LeadControllerUtils {
+
+ /**
+ * Given a raw table name and number of partitions, returns the partition id in lead controller resource.
+ * Uses murmur2 function to get hashcode for table, ignores the most significant bit.
+ * Note: This method CANNOT be changed when lead controller resource is enabled.
+ * Otherwise it will assign different controller for the same table, which will mess up the controller periodic tasks and realtime segment completion.
+ * @param rawTableName raw table name
+ * @return partition id in lead controller resource.
+ */
+ public static int getPartitionIdForTable(String rawTableName) {
+ return (HashUtil.murmur2(StringUtil.encodeUtf8(rawTableName)) & Integer.MAX_VALUE)
+ % Helix.NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE;
+ }
+
+ /**
+ * Generates participant instance id, e.g. returns Controller_localhost_9000 given localhost as hostname and 9000 as port.
+ */
+ public static String generateParticipantInstanceId(String controllerHost, int controllerPort) {
+ return Helix.PREFIX_OF_CONTROLLER_INSTANCE + controllerHost + "_" + controllerPort;
+ }
+
+ /**
+ * Generates partition name, e.g. returns leadControllerResource_0 given 0 as partition index.
+ */
+ public static String generatePartitionName(int partitionId) {
+ return Helix.LEAD_CONTROLLER_RESOURCE_NAME + "_" + partitionId;
+ }
+
+ /**
+ * Extracts partition index from partition name, e.g. returns 0 given leadControllerResource_0 as partition name.
+ */
+ public static int extractPartitionId(String partitionName) {
+ return Integer.parseInt(partitionName.substring(partitionName.lastIndexOf('_') + 1));
+ }
+}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerLeadershipManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerLeadershipManager.java
deleted file mode 100644
index 92779d1..0000000
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/ControllerLeadershipManager.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.controller;
-
-import java.util.HashMap;
-import java.util.Map;
-import javax.annotation.concurrent.ThreadSafe;
-import org.apache.helix.HelixManager;
-import org.apache.pinot.common.metrics.ControllerGauge;
-import org.apache.pinot.common.metrics.ControllerMeter;
-import org.apache.pinot.common.metrics.ControllerMetrics;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * Single place for listening on controller changes.
- * This should be created at controller startup and everyone who wants to listen to controller changes should subscribe.
- */
-@ThreadSafe
-public class ControllerLeadershipManager {
- private static final Logger LOGGER = LoggerFactory.getLogger(ControllerLeadershipManager.class);
- private static final long CONTROLLER_LEADERSHIP_FETCH_INTERVAL = 60_000L;
-
- private final HelixManager _helixControllerManager;
- private final ControllerMetrics _controllerMetrics;
- private final Map<String, LeadershipChangeSubscriber> _subscribers = new HashMap<>();
- private final Thread _controllerLeadershipFetchingThread;
-
- private boolean _amILeader = false;
- private boolean _stopped = false;
-
- public ControllerLeadershipManager(HelixManager helixControllerManager, ControllerMetrics controllerMetrics) {
- _helixControllerManager = helixControllerManager;
- _controllerMetrics = controllerMetrics;
- _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PINOT_CONTROLLER_LEADER, 0L);
-
- // Create a thread to periodically fetch controller leadership as a work-around of Helix callback delay
- _controllerLeadershipFetchingThread = new Thread("ControllerLeadershipFetchingThread") {
- @Override
- public void run() {
- while (true) {
- try {
- synchronized (ControllerLeadershipManager.this) {
- if (_stopped) {
- return;
- }
- if (_helixControllerManager.isLeader()) {
- if (!_amILeader) {
- _amILeader = true;
- LOGGER.warn("Becoming leader without getting Helix change callback");
- _controllerMetrics
- .addMeteredGlobalValue(ControllerMeter.CONTROLLER_LEADERSHIP_CHANGE_WITHOUT_CALLBACK, 1L);
- onBecomingLeader();
- }
- } else {
- if (_amILeader) {
- _amILeader = false;
- LOGGER.warn("Losing leadership without getting Helix change callback");
- _controllerMetrics
- .addMeteredGlobalValue(ControllerMeter.CONTROLLER_LEADERSHIP_CHANGE_WITHOUT_CALLBACK, 1L);
- onBecomingNonLeader();
- }
- }
- ControllerLeadershipManager.this.wait(CONTROLLER_LEADERSHIP_FETCH_INTERVAL);
- }
- } catch (Exception e) {
- // Ignore all exceptions. The thread keeps running until ControllerLeadershipManager.stop() is invoked.
- LOGGER.error("Caught exception within controller leadership fetching thread", e);
- }
- }
- }
- };
- }
-
- /**
- * Subscribes to changes in the controller leadership.
- * <p>If controller is already leader, invoke {@link LeadershipChangeSubscriber#onBecomingLeader()}
- */
- public synchronized void subscribe(String name, LeadershipChangeSubscriber subscriber) {
- LOGGER.info("{} subscribing to leadership changes", name);
- _subscribers.put(name, subscriber);
- if (_amILeader) {
- subscriber.onBecomingLeader();
- }
- }
-
- public boolean isLeader() {
- return _amILeader;
- }
-
- /**
- * Starts the service.
- */
- public synchronized void start() {
- onControllerChange();
- _controllerLeadershipFetchingThread.start();
- }
-
- /**
- * Stops the service.
- * <p>If controller is leader, invoke {@link ControllerLeadershipManager#onBecomingNonLeader()}
- */
- public void stop() {
- synchronized (this) {
- if (_amILeader) {
- onBecomingNonLeader();
- }
- _stopped = true;
- notify();
- }
- try {
- _controllerLeadershipFetchingThread.join();
- } catch (InterruptedException e) {
- LOGGER.error("Caught InterruptedException while waiting for controller leadership fetching thread to die");
- Thread.currentThread().interrupt();
- }
- }
-
- /**
- * Callback on changes in the controller. Should be registered to the controller callback.
- */
- synchronized void onControllerChange() {
- if (_helixControllerManager.isLeader()) {
- if (!_amILeader) {
- _amILeader = true;
- LOGGER.info("Becoming leader");
- onBecomingLeader();
- }
- } else {
- if (_amILeader) {
- _amILeader = false;
- LOGGER.info("Losing leadership");
- onBecomingNonLeader();
- }
- }
- }
-
- private void onBecomingLeader() {
- long startTimeMs = System.currentTimeMillis();
- _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PINOT_CONTROLLER_LEADER, 1L);
- for (LeadershipChangeSubscriber subscriber : _subscribers.values()) {
- subscriber.onBecomingLeader();
- }
- LOGGER.info("Finished on becoming leader in {}ms", System.currentTimeMillis() - startTimeMs);
- }
-
- private void onBecomingNonLeader() {
- long startTimeMs = System.currentTimeMillis();
- _controllerMetrics.setValueOfGlobalGauge(ControllerGauge.PINOT_CONTROLLER_LEADER, 0L);
- for (LeadershipChangeSubscriber subscriber : _subscribers.values()) {
- subscriber.onBecomingNonLeader();
- }
- LOGGER.info("Finished on becoming non-leader in {}ms", System.currentTimeMillis() - startTimeMs);
- }
-}
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 7c04510..cab9783 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
@@ -19,7 +19,6 @@
package org.apache.pinot.controller;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import com.google.common.primitives.Longs;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.yammer.metrics.core.MetricsRegistry;
@@ -42,7 +41,6 @@ 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.examples.MasterSlaveStateModelFactory;
import org.apache.helix.model.MasterSlaveSMD;
import org.apache.helix.task.TaskDriver;
import org.apache.pinot.common.Utils;
@@ -54,6 +52,7 @@ import org.apache.pinot.common.segment.fetcher.SegmentFetcherFactory;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.NetUtil;
import org.apache.pinot.common.utils.ServiceStatus;
+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;
@@ -61,13 +60,13 @@ import org.apache.pinot.controller.helix.SegmentStatusChecker;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
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.periodictask.ControllerPeriodicTaskScheduler;
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.rebalance.RebalanceSegmentStrategyFactory;
import org.apache.pinot.controller.helix.core.relocation.RealtimeSegmentRelocator;
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;
@@ -75,6 +74,7 @@ import org.apache.pinot.controller.validation.OfflineSegmentIntervalChecker;
import org.apache.pinot.controller.validation.RealtimeSegmentValidationManager;
import org.apache.pinot.core.crypt.PinotCrypterFactory;
import org.apache.pinot.core.periodictask.PeriodicTask;
+import org.apache.pinot.core.periodictask.PeriodicTaskScheduler;
import org.apache.pinot.filesystem.PinotFSFactory;
import org.glassfish.hk2.utilities.binding.AbstractBinder;
import org.slf4j.Logger;
@@ -106,6 +106,7 @@ public class ControllerStarter {
private final ControllerConf.ControllerMode _controllerMode;
private HelixManager _helixControllerManager;
+ private HelixManager _helixParticipantManager;
// Can only be constructed after resource manager getting started
private OfflineSegmentIntervalChecker _offlineSegmentIntervalChecker;
@@ -115,12 +116,12 @@ public class ControllerStarter {
private RetentionManager _retentionManager;
private SegmentStatusChecker _segmentStatusChecker;
private PinotTaskManager _taskManager;
- private ControllerPeriodicTaskScheduler _controllerPeriodicTaskScheduler;
+ private PeriodicTaskScheduler _periodicTaskScheduler;
private PinotHelixTaskResourceManager _helixTaskResourceManager;
private PinotRealtimeSegmentManager _realtimeSegmentsManager;
private PinotLLCRealtimeSegmentManager _pinotLLCRealtimeSegmentManager;
private SegmentCompletionManager _segmentCompletionManager;
- private ControllerLeadershipManager _controllerLeadershipManager;
+ private LeadControllerManager _leadControllerManager;
private List<ServiceStatus.ServiceStatusCallback> _serviceStatusCallbackList;
public ControllerStarter(ControllerConf conf) {
@@ -132,8 +133,10 @@ public class ControllerStarter {
// Helix related settings.
_helixZkURL = HelixConfig.getAbsoluteZkPathForHelix(_config.getZkStr());
_helixClusterName = _config.getHelixClusterName();
- _helixControllerInstanceId = conf.getControllerHost() + "_" + conf.getControllerPort();
- _helixParticipantInstanceId = CommonConstants.Helix.PREFIX_OF_CONTROLLER_INSTANCE + _helixControllerInstanceId;
+ String host = conf.getControllerHost();
+ int port = Integer.parseInt(conf.getControllerPort());
+ _helixControllerInstanceId = host + "_" + port;
+ _helixParticipantInstanceId = LeadControllerUtils.generateParticipantInstanceId(host, port);
_isUpdateStateModel = _config.isUpdateSegmentStateModel();
_enableBatchMessageMode = _config.getEnableBatchMessageMode();
@@ -188,6 +191,10 @@ public class ControllerStarter {
return _helixControllerManager;
}
+ public LeadControllerManager getLeadControllerManager() {
+ return _leadControllerManager;
+ }
+
public OfflineSegmentIntervalChecker getOfflineSegmentIntervalChecker() {
return _offlineSegmentIntervalChecker;
}
@@ -208,10 +215,6 @@ public class ControllerStarter {
return _taskManager;
}
- public ControllerLeadershipManager getControllerLeadershipManager() {
- return _controllerLeadershipManager;
- }
-
public void start() {
LOGGER.info("Starting Pinot controller in mode: {}.", _controllerMode.name());
Utils.logVersions();
@@ -243,8 +246,8 @@ public class ControllerStarter {
private void setUpHelixController() {
// Register and connect instance as Helix controller.
LOGGER.info("Starting Helix controller");
- _helixControllerManager = HelixSetupUtils.setupHelixController(_helixClusterName, _helixZkURL,
- _helixControllerInstanceId);
+ _helixControllerManager =
+ HelixSetupUtils.setupHelixController(_helixClusterName, _helixZkURL, _helixControllerInstanceId);
// Emit helix controller metrics
_controllerMetrics.addCallbackGauge(CommonConstants.Helix.INSTANCE_CONNECTED_METRIC_NAME,
@@ -255,18 +258,10 @@ public class ControllerStarter {
() -> _controllerMetrics.addMeteredGlobalValue(ControllerMeter.HELIX_ZOOKEEPER_RECONNECTS, 1L));
_serviceStatusCallbackList.add(generateServiceStatusCallback(_helixControllerManager));
-
- LOGGER.info("Initializing controller leadership manager");
- _controllerLeadershipManager = new ControllerLeadershipManager(_helixControllerManager, _controllerMetrics);
}
private void setUpPinotController() {
- // Note: Right now we don't allow Pinot-only controller as ControllerLeadershipManager is setup in Helix controller
- // and Pinot controller relies on it
- // TODO: Remove ControllerLeadershipManager
- Preconditions.checkState(_controllerLeadershipManager != null);
-
- // Set up Pinot cluster in Helix
+ // Set up Pinot cluster in Helix if needed
HelixSetupUtils.setupPinotCluster(_helixClusterName, _helixZkURL, _isUpdateStateModel, _enableBatchMessageMode);
// Start all components
@@ -274,37 +269,41 @@ public class ControllerStarter {
initSegmentFetcherFactory();
initPinotCrypterFactory();
- LOGGER.info("Starting Helix manager as Helix participant");
- HelixManager helixParticipantManager = registerAndConnectAsHelixParticipant();
+ LOGGER.info("Initializing Helix participant manager");
+ _helixParticipantManager = HelixManagerFactory
+ .getZKHelixManager(_helixClusterName, _helixParticipantInstanceId, InstanceType.PARTICIPANT, _helixZkURL);
- LOGGER.info("Starting Pinot Helix resource manager and connecting to Zookeeper");
- _helixResourceManager.start(helixParticipantManager);
+ // 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("Registering controller leadership manager");
- // TODO: when Helix separation is completed, leadership only depends on the master in leadControllerResource, remove
- // ControllerLeadershipManager and this callback.
- _controllerLeadershipManager.start();
- helixParticipantManager.addControllerListener(
- (ControllerChangeListener) changeContext -> _controllerLeadershipManager.onControllerChange());
+ 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));
+ _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,
- _controllerLeadershipManager);
- // TODO: Need to put this inside HelixResourceManager when ControllerLeadershipManager is removed.
+ new PinotLLCRealtimeSegmentManager(_helixResourceManager, _config, _controllerMetrics, _leadControllerManager);
+ // TODO: Need to put this inside HelixResourceManager when HelixControllerLeadershipManager is removed.
_helixResourceManager.registerPinotLLCRealtimeSegmentManager(_pinotLLCRealtimeSegmentManager);
_segmentCompletionManager =
- new SegmentCompletionManager(helixParticipantManager, _pinotLLCRealtimeSegmentManager, _controllerMetrics,
- _controllerLeadershipManager, _config.getSegmentCommitTimeoutSeconds());
+ 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, _controllerLeadershipManager);
+ _realtimeSegmentsManager = new PinotRealtimeSegmentManager(_helixResourceManager, _leadControllerManager);
_realtimeSegmentsManager.start(_controllerMetrics);
} else {
LOGGER.info("Realtime tables with High Level consumers will NOT be supported");
@@ -314,12 +313,14 @@ public class ControllerStarter {
// Setting up periodic tasks
List<PeriodicTask> controllerPeriodicTasks = setupControllerPeriodicTasks();
LOGGER.info("Init controller periodic tasks scheduler");
- _controllerPeriodicTaskScheduler = new ControllerPeriodicTaskScheduler();
- _controllerPeriodicTaskScheduler.init(controllerPeriodicTasks, _controllerLeadershipManager);
+ _periodicTaskScheduler = new PeriodicTaskScheduler();
+ _periodicTaskScheduler.init(controllerPeriodicTasks);
+
+ _periodicTaskScheduler.start();
LOGGER.info("Registering rebalance segments factory");
_helixResourceManager
- .registerRebalanceSegmentStrategyFactory(new RebalanceSegmentStrategyFactory(helixParticipantManager));
+ .registerRebalanceSegmentStrategyFactory(new RebalanceSegmentStrategyFactory(_helixParticipantManager));
String accessControlFactoryClass = _config.getAccessControlFactoryClass();
LOGGER.info("Use class: {} as the AccessControlFactory", accessControlFactoryClass);
@@ -353,7 +354,7 @@ public class ControllerStarter {
bind(_controllerMetrics).to(ControllerMetrics.class);
bind(accessControlFactory).to(AccessControlFactory.class);
bind(metadataEventNotifierFactory).to(MetadataEventNotifierFactory.class);
- bind(_controllerLeadershipManager).to(ControllerLeadershipManager.class);
+ bind(_leadControllerManager).to(LeadControllerManager.class);
}
});
@@ -384,7 +385,7 @@ public class ControllerStarter {
}
});
- _serviceStatusCallbackList.add(generateServiceStatusCallback(helixParticipantManager));
+ _serviceStatusCallbackList.add(generateServiceStatusCallback(_helixParticipantManager));
}
private ServiceStatus.ServiceStatusCallback generateServiceStatusCallback(HelixManager helixManager) {
@@ -452,25 +453,38 @@ public class ControllerStarter {
}
/**
- * Register and connect to Helix cluster as PARTICIPANT role.
+ * Registers, connects to Helix cluster as PARTICIPANT role, and adds listeners.
*/
- private HelixManager registerAndConnectAsHelixParticipant() {
- HelixManager helixManager = HelixManagerFactory
- .getZKHelixManager(_helixClusterName, _helixParticipantInstanceId, InstanceType.PARTICIPANT, _helixZkURL);
-
- // Registers Master-Slave state model to state machine engine, which is for calculating participant assignment in lead controller resource.
- helixManager.getStateMachineEngine()
- .registerStateModelFactory(MasterSlaveSMD.name, new MasterSlaveStateModelFactory());
+ 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 {
- helixManager.connect();
- return helixManager;
+ _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() {
@@ -481,24 +495,29 @@ public class ControllerStarter {
protected List<PeriodicTask> setupControllerPeriodicTasks() {
LOGGER.info("Setting up periodic tasks");
List<PeriodicTask> periodicTasks = new ArrayList<>();
- _taskManager = new PinotTaskManager(_helixTaskResourceManager, _helixResourceManager, _config, _controllerMetrics);
+ _taskManager =
+ new PinotTaskManager(_helixTaskResourceManager, _helixResourceManager, _leadControllerManager, _config,
+ _controllerMetrics);
periodicTasks.add(_taskManager);
- _retentionManager = new RetentionManager(_helixResourceManager, _config, _controllerMetrics);
+ _retentionManager =
+ new RetentionManager(_helixResourceManager, _leadControllerManager, _config, _controllerMetrics);
periodicTasks.add(_retentionManager);
_offlineSegmentIntervalChecker =
- new OfflineSegmentIntervalChecker(_config, _helixResourceManager, new ValidationMetrics(_metricsRegistry),
- _controllerMetrics);
+ new OfflineSegmentIntervalChecker(_config, _helixResourceManager, _leadControllerManager,
+ new ValidationMetrics(_metricsRegistry), _controllerMetrics);
periodicTasks.add(_offlineSegmentIntervalChecker);
_realtimeSegmentValidationManager =
- new RealtimeSegmentValidationManager(_config, _helixResourceManager, _pinotLLCRealtimeSegmentManager,
- new ValidationMetrics(_metricsRegistry), _controllerMetrics);
+ new RealtimeSegmentValidationManager(_config, _helixResourceManager, _leadControllerManager,
+ _pinotLLCRealtimeSegmentManager, new ValidationMetrics(_metricsRegistry), _controllerMetrics);
periodicTasks.add(_realtimeSegmentValidationManager);
_brokerResourceValidationManager =
- new BrokerResourceValidationManager(_config, _helixResourceManager, _controllerMetrics);
+ new BrokerResourceValidationManager(_config, _helixResourceManager, _leadControllerManager, _controllerMetrics);
periodicTasks.add(_brokerResourceValidationManager);
- _segmentStatusChecker = new SegmentStatusChecker(_helixResourceManager, _config, _controllerMetrics);
+ _segmentStatusChecker =
+ new SegmentStatusChecker(_helixResourceManager, _leadControllerManager, _config, _controllerMetrics);
periodicTasks.add(_segmentStatusChecker);
- _realtimeSegmentRelocator = new RealtimeSegmentRelocator(_helixResourceManager, _config, _controllerMetrics);
+ _realtimeSegmentRelocator =
+ new RealtimeSegmentRelocator(_helixResourceManager, _leadControllerManager, _config, _controllerMetrics);
periodicTasks.add(_realtimeSegmentRelocator);
return periodicTasks;
@@ -520,15 +539,19 @@ public class ControllerStarter {
}
private void stopHelixController() {
- LOGGER.info("Disconnecting helix zk manager");
+ LOGGER.info("Disconnecting helix controller zk manager");
_helixControllerManager.disconnect();
}
private void stopPinotController() {
try {
- // Stopping ControllerLeadershipManager has to be done before stopping HelixResourceManager.
- LOGGER.info("Stopping controller leadership manager");
- _controllerLeadershipManager.stop();
+ // 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.
@@ -548,6 +571,9 @@ public class ControllerStarter {
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);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/LeadControllerManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/LeadControllerManager.java
new file mode 100644
index 0000000..b03268c
--- /dev/null
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/LeadControllerManager.java
@@ -0,0 +1,229 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.ResourceConfig;
+import org.apache.pinot.common.config.TableNameBuilder;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.common.utils.CommonConstants.Helix;
+import org.apache.pinot.common.utils.helix.LeadControllerUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Class for handling lead controller assignments given the table names. This should be created at controller startup.
+ */
+@ThreadSafe
+public class LeadControllerManager {
+ private static final Logger LOGGER = LoggerFactory.getLogger(LeadControllerManager.class);
+ private static final long CONTROLLER_LEADERSHIP_FETCH_INTERVAL_MS = 60_000L;
+
+ private final Set<Integer> _leadForPartitions;
+ private final String _instanceId;
+ private final HelixManager _helixManager;
+ private final ControllerMetrics _controllerMetrics;
+ private final Thread _controllerLeadershipFetchingThread;
+
+ private volatile boolean _isLeadControllerResourceEnabled = false;
+ private volatile boolean _amIHelixLeader = false;
+ private volatile boolean _isShuttingDown = false;
+
+ public LeadControllerManager(HelixManager helixParticipantManager, ControllerMetrics controllerMetrics) {
+ _helixManager = helixParticipantManager;
+ _controllerMetrics = controllerMetrics;
+ _instanceId = helixParticipantManager.getInstanceName();
+ _leadForPartitions = ConcurrentHashMap.newKeySet();
+
+ // Create a thread to periodically fetch controller leadership as a work-around of Helix callback delay
+ _controllerLeadershipFetchingThread = new Thread("ControllerLeadershipFetchingThread") {
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ synchronized (LeadControllerManager.this) {
+ if (_isShuttingDown) {
+ return;
+ }
+ if (isHelixLeader()) {
+ if (!_amIHelixLeader) {
+ _amIHelixLeader = true;
+ LOGGER.warn("Becoming leader without getting Helix change callback");
+ _controllerMetrics
+ .addMeteredGlobalValue(ControllerMeter.CONTROLLER_LEADERSHIP_CHANGE_WITHOUT_CALLBACK, 1L);
+ }
+ } else {
+ if (_amIHelixLeader) {
+ _amIHelixLeader = false;
+ LOGGER.warn("Losing leadership without getting Helix change callback");
+ _controllerMetrics
+ .addMeteredGlobalValue(ControllerMeter.CONTROLLER_LEADERSHIP_CHANGE_WITHOUT_CALLBACK, 1L);
+ }
+ }
+ LeadControllerManager.this.wait(CONTROLLER_LEADERSHIP_FETCH_INTERVAL_MS);
+ }
+ } catch (Exception e) {
+ // Ignore all exceptions. The thread keeps running until LeadControllerManager.stop() is invoked.
+ LOGGER.error("Caught exception within controller leadership fetching thread", e);
+ }
+ }
+ }
+ };
+ }
+
+ /**
+ * Checks whether the current controller is the leader for the given table. Return true if current controller is the leader for this table.
+ * Otherwise check whether the current controller is helix leader if the resource is disabled.
+ * @param tableName table name with/without table type.
+ */
+ public boolean isLeaderForTable(String tableName) {
+ if (_isLeadControllerResourceEnabled) {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ int partitionId = LeadControllerUtils.getPartitionIdForTable(rawTableName);
+ return _leadForPartitions.contains(partitionId);
+ } else {
+ // Checks if it's Helix leader if lead controller resource is disabled.
+ return _amIHelixLeader;
+ }
+ }
+
+ /**
+ * Given a partition name, marks current controller as lead controller for this partition by caching the partition id to current controller.
+ * @param partitionName partition name in lead controller resource, e.g. leadControllerResource_0.
+ */
+ public synchronized void addPartitionLeader(String partitionName) {
+ LOGGER.info("Add Partition: {} to LeadControllerManager", partitionName);
+ int partitionId = LeadControllerUtils.extractPartitionId(partitionName);
+ _leadForPartitions.add(partitionId);
+ }
+
+ /**
+ * Given a partition name, removes current controller as lead controller for this partition by removing the partition id from current controller.
+ * @param partitionName partition name in lead controller resource, e.g. leadControllerResource_0.
+ */
+ public synchronized void removePartitionLeader(String partitionName) {
+ LOGGER.info("Remove Partition: {} from LeadControllerManager", partitionName);
+ int partitionId = LeadControllerUtils.extractPartitionId(partitionName);
+ _leadForPartitions.remove(partitionId);
+ }
+
+ /**
+ * Checks from ZK if the current controller host is Helix cluster leader.
+ */
+ private boolean isHelixLeader() {
+ HelixDataAccessor helixDataAccessor = _helixManager.getHelixDataAccessor();
+ PropertyKey propertyKey = helixDataAccessor.keyBuilder().controllerLeader();
+ LiveInstance liveInstance = helixDataAccessor.getProperty(propertyKey);
+ if (liveInstance == null) {
+ LOGGER.warn("Helix leader ZNode is missing");
+ return false;
+ }
+ String helixLeaderInstanceId = liveInstance.getInstanceName();
+ // The instance name from Helix leader ZNode is without controller prefix.
+ // It is essential to convert to participant id for fair comparison.
+ return _instanceId.equals(Helix.PREFIX_OF_CONTROLLER_INSTANCE + helixLeaderInstanceId);
+ }
+
+ /**
+ * Checks from ZK if resource config of leadControllerResource is enabled.
+ */
+ public boolean isLeadControllerResourceEnabled() {
+ HelixDataAccessor helixDataAccessor = _helixManager.getHelixDataAccessor();
+ PropertyKey propertyKey = helixDataAccessor.keyBuilder().resourceConfig(Helix.LEAD_CONTROLLER_RESOURCE_NAME);
+ ResourceConfig resourceConfig = helixDataAccessor.getProperty(propertyKey);
+ String enableResource = resourceConfig.getSimpleConfig(Helix.LEAD_CONTROLLER_RESOURCE_ENABLED_KEY);
+ return Boolean.parseBoolean(enableResource);
+ }
+
+ /**
+ * Starts the fetching thread to actively fetch helix leadership and resource config of lead controller resource.
+ */
+ public synchronized void start() {
+ _controllerLeadershipFetchingThread.start();
+ }
+
+ /**
+ * Marks the cached indices invalid and isShuttingDown to be true.
+ * Adding the synchronized block here and in the following callback methods
+ * to make sure that {@link HelixManager} won't be closed when the callback changes happened.
+ */
+ public void stop() {
+ synchronized (this) {
+ _isShuttingDown = true;
+ notify();
+ }
+ _leadForPartitions.clear();
+
+ try {
+ _controllerLeadershipFetchingThread.join();
+ } catch (InterruptedException e) {
+ LOGGER.error("Caught InterruptedException while waiting for controller leadership fetching thread to die");
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Callback on changes in the controller. Should be registered to the controller callback. This callback is not needed when the resource 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.
+ */
+ synchronized void onHelixControllerChange() {
+ if (_isShuttingDown) {
+ return;
+ }
+ if (isHelixLeader()) {
+ if (!_amIHelixLeader) {
+ _amIHelixLeader = true;
+ LOGGER.info("Became Helix leader");
+ } else {
+ LOGGER.info("Already Helix leader. Duplicate notification");
+ }
+ } else {
+ if (_amIHelixLeader) {
+ _amIHelixLeader = false;
+ LOGGER.info("Lost Helix leadership");
+ } else {
+ LOGGER.info("Already not Helix leader. Duplicate notification");
+ }
+ }
+ }
+
+ /**
+ * Callback on changes in resource config.
+ */
+ synchronized void onResourceConfigChange() {
+ if (_isShuttingDown) {
+ return;
+ }
+ if (isLeadControllerResourceEnabled()) {
+ LOGGER.info("Lead controller resource is enabled.");
+ _isLeadControllerResourceEnabled = true;
+ } else {
+ LOGGER.info("Lead controller resource is disabled.");
+ _isLeadControllerResourceEnabled = false;
+ }
+ }
+}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadRestletResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadRestletResource.java
index 6f74f8a..b32909c 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadRestletResource.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadRestletResource.java
@@ -66,7 +66,7 @@ import org.apache.pinot.common.utils.JsonUtils;
import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.ControllerLeadershipManager;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.api.access.AccessControl;
import org.apache.pinot.controller.api.access.AccessControlFactory;
import org.apache.pinot.controller.api.upload.SegmentValidator;
@@ -113,7 +113,7 @@ public class PinotSegmentUploadRestletResource {
AccessControlFactory _accessControlFactory;
@Inject
- ControllerLeadershipManager _controllerLeadershipManager;
+ LeadControllerManager _leadControllerManager;
@GET
@Produces(MediaType.APPLICATION_JSON)
@@ -321,7 +321,7 @@ public class PinotSegmentUploadRestletResource {
// Validate segment
SegmentValidatorResponse segmentValidatorResponse =
new SegmentValidator(_pinotHelixResourceManager, _controllerConf, _executor, _connectionManager,
- _controllerMetrics, _controllerLeadershipManager)
+ _controllerMetrics, _leadControllerManager.isLeaderForTable(offlineTableName))
.validateSegment(rawTableName, segmentMetadata, tempSegmentDir);
// Zk operations
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidator.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidator.java
index bea9f74..47eb90b 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidator.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentValidator.java
@@ -34,7 +34,6 @@ import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.segment.SegmentMetadata;
import org.apache.pinot.common.utils.time.TimeUtils;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.ControllerLeadershipManager;
import org.apache.pinot.controller.api.resources.ControllerApplicationException;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.util.TableSizeReader;
@@ -55,17 +54,17 @@ public class SegmentValidator {
private final Executor _executor;
private final HttpConnectionManager _connectionManager;
private final ControllerMetrics _controllerMetrics;
- private final ControllerLeadershipManager _controllerLeadershipManager;
+ private final boolean _isLeaderForTable;
public SegmentValidator(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf controllerConf,
Executor executor, HttpConnectionManager connectionManager, ControllerMetrics controllerMetrics,
- ControllerLeadershipManager controllerLeadershipManager) {
+ boolean isLeaderForTable) {
_pinotHelixResourceManager = pinotHelixResourceManager;
_controllerConf = controllerConf;
_executor = executor;
_connectionManager = connectionManager;
_controllerMetrics = controllerMetrics;
- _controllerLeadershipManager = controllerLeadershipManager;
+ _isLeaderForTable = isLeaderForTable;
}
public SegmentValidatorResponse validateSegment(String rawTableName, SegmentMetadata segmentMetadata,
@@ -134,8 +133,7 @@ public class SegmentValidator {
TableSizeReader tableSizeReader =
new TableSizeReader(_executor, _connectionManager, _controllerMetrics, _pinotHelixResourceManager);
StorageQuotaChecker quotaChecker =
- new StorageQuotaChecker(offlineTableConfig, tableSizeReader, _controllerMetrics, _pinotHelixResourceManager,
- _controllerLeadershipManager);
+ new StorageQuotaChecker(offlineTableConfig, tableSizeReader, _controllerMetrics, _isLeaderForTable);
return quotaChecker.isSegmentStorageWithinQuota(segmentFile, metadata.getName(),
_controllerConf.getServerAdminRequestTimeoutSeconds() * 1000);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
index f40e2b3..9261420 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
@@ -29,6 +29,7 @@ import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
import org.slf4j.Logger;
@@ -56,10 +57,11 @@ public class SegmentStatusChecker extends ControllerPeriodicTask<SegmentStatusCh
* @param pinotHelixResourceManager The resource checker used to interact with Helix
* @param config The controller configuration object
*/
- public SegmentStatusChecker(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf config,
- ControllerMetrics controllerMetrics) {
+ public SegmentStatusChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics) {
super("SegmentStatusChecker", config.getStatusCheckerFrequencyInSeconds(),
- config.getStatusCheckerInitialDelayInSeconds(), pinotHelixResourceManager, controllerMetrics);
+ config.getStatusCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
_waitForPushTimeSeconds = config.getStatusCheckerWaitForPushTimeInSeconds();
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index 6714044..d4579de 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -161,7 +161,10 @@ public class PinotHelixResourceManager {
}
/**
- * Create Helix cluster if needed, and then start a Pinot controller instance.
+ * Starts a Pinot controller instance.
+ * Note: Helix instance type should be explicitly set to PARTICIPANT ONLY in ControllerStarter.
+ * Other places like PerfBenchmarkDriver which directly call {@link PinotHelixResourceManager} should NOT register as PARTICIPANT,
+ * which would be put to lead controller resource and mess up the leadership assignment. Those places should use SPECTATOR other than PARTICIPANT.
*/
public synchronized void start(HelixManager helixZkManager) {
_helixZkManager = helixZkManager;
@@ -191,7 +194,6 @@ public class PinotHelixResourceManager {
*/
public synchronized void stop() {
_segmentDeletionManager.stop();
- _helixZkManager.disconnect();
}
/**
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
index 7a09c00..649b966 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotTaskManager.java
@@ -29,6 +29,7 @@ import org.apache.pinot.common.config.TableTaskConfig;
import org.apache.pinot.common.metrics.ControllerMeter;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.minion.generator.PinotTaskGenerator;
import org.apache.pinot.controller.helix.core.minion.generator.TaskGeneratorRegistry;
@@ -51,10 +52,11 @@ public class PinotTaskManager extends ControllerPeriodicTask<Void> {
private final TaskGeneratorRegistry _taskGeneratorRegistry;
public PinotTaskManager(PinotHelixTaskResourceManager helixTaskResourceManager,
- PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
- ControllerMetrics controllerMetrics) {
+ PinotHelixResourceManager helixResourceManager, LeadControllerManager leadControllerManager,
+ ControllerConf controllerConf, ControllerMetrics controllerMetrics) {
super("PinotTaskManager", controllerConf.getTaskManagerFrequencyInSeconds(),
- controllerConf.getPinotTaskManagerInitialDelaySeconds(), helixResourceManager, controllerMetrics);
+ controllerConf.getPinotTaskManagerInitialDelaySeconds(), helixResourceManager, leadControllerManager,
+ controllerMetrics);
_helixTaskResourceManager = helixTaskResourceManager;
_clusterInfoProvider = new ClusterInfoProvider(helixResourceManager, helixTaskResourceManager, controllerConf);
_taskGeneratorRegistry = new TaskGeneratorRegistry(_clusterInfoProvider);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTask.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTask.java
index 7e764f3..a782385 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTask.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTask.java
@@ -23,6 +23,7 @@ import javax.annotation.concurrent.ThreadSafe;
import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMeter;
import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.core.periodictask.BasePeriodicTask;
import org.slf4j.Logger;
@@ -40,12 +41,15 @@ public abstract class ControllerPeriodicTask<C> extends BasePeriodicTask {
private static final Logger LOGGER = LoggerFactory.getLogger(ControllerPeriodicTask.class);
protected final PinotHelixResourceManager _pinotHelixResourceManager;
+ protected final LeadControllerManager _leadControllerManager;
protected final ControllerMetrics _controllerMetrics;
public ControllerPeriodicTask(String taskName, long runFrequencyInSeconds, long initialDelayInSeconds,
- PinotHelixResourceManager pinotHelixResourceManager, ControllerMetrics controllerMetrics) {
+ PinotHelixResourceManager pinotHelixResourceManager, LeadControllerManager leadControllerManager,
+ ControllerMetrics controllerMetrics) {
super(taskName, runFrequencyInSeconds, initialDelayInSeconds);
_pinotHelixResourceManager = pinotHelixResourceManager;
+ _leadControllerManager = leadControllerManager;
_controllerMetrics = controllerMetrics;
}
@@ -75,6 +79,10 @@ public abstract class ControllerPeriodicTask<C> extends BasePeriodicTask {
LOGGER.info("Task: {} is stopped, early terminate the task", _taskName);
break;
}
+ // Check if current controller is the leader for this table.
+ if (!_leadControllerManager.isLeaderForTable(tableNameWithType)) {
+ continue;
+ }
try {
processTable(tableNameWithType, context);
} catch (Exception e) {
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTaskScheduler.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTaskScheduler.java
deleted file mode 100644
index 8a2b63c..0000000
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTaskScheduler.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.pinot.controller.helix.core.periodictask;
-
-import java.util.List;
-import org.apache.pinot.controller.ControllerLeadershipManager;
-import org.apache.pinot.controller.LeadershipChangeSubscriber;
-import org.apache.pinot.core.periodictask.PeriodicTask;
-import org.apache.pinot.core.periodictask.PeriodicTaskScheduler;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * A {@link PeriodicTaskScheduler} for scheduling {@link ControllerPeriodicTask} according to controller leadership changes.
- * Any controllerPeriodicTasks provided during initialization, will run only on leadership, and stop when leadership lost
- */
-public class ControllerPeriodicTaskScheduler extends PeriodicTaskScheduler implements LeadershipChangeSubscriber {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(ControllerPeriodicTaskScheduler.class);
-
- /**
- * Initialize the {@link ControllerPeriodicTaskScheduler} with the list of {@link ControllerPeriodicTask} created at startup
- * This is called only once during controller startup
- * @param controllerPeriodicTasks
- * @param controllerLeadershipManager
- */
- public void init(List<PeriodicTask> controllerPeriodicTasks, ControllerLeadershipManager controllerLeadershipManager) {
- super.init(controllerPeriodicTasks);
- controllerLeadershipManager.subscribe(ControllerPeriodicTaskScheduler.class.getName(), this);
- }
-
- @Override
- public void onBecomingLeader() {
- LOGGER.info("Received callback for controller leadership gain. Starting PeriodicTaskScheduler.");
- start();
- }
-
- @Override
- public void onBecomingNonLeader() {
- LOGGER.info("Received callback for controller leadership loss. Stopping PeriodicTaskScheduler.");
- stop();
- }
-}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
index 22239f3..f515c66 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
@@ -66,7 +66,7 @@ import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.common.utils.retry.RetryPolicies;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.ControllerLeadershipManager;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.api.events.MetadataEventNotifierFactory;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.PinotHelixSegmentOnlineOfflineStateModelGenerator;
@@ -123,13 +123,13 @@ public class PinotLLCRealtimeSegmentManager {
private final TableConfigCache _tableConfigCache;
private final StreamPartitionAssignmentGenerator _streamPartitionAssignmentGenerator;
private final FlushThresholdUpdateManager _flushThresholdUpdateManager;
- private final ControllerLeadershipManager _controllerLeadershipManager;
+ private final LeadControllerManager _leadControllerManager;
private volatile boolean _isStopping = false;
private AtomicInteger _numCompletingSegments = new AtomicInteger(0);
public PinotLLCRealtimeSegmentManager(PinotHelixResourceManager helixResourceManager, ControllerConf controllerConf,
- ControllerMetrics controllerMetrics, ControllerLeadershipManager controllerLeadershipManager) {
+ ControllerMetrics controllerMetrics, LeadControllerManager leadControllerManager) {
_helixAdmin = helixResourceManager.getHelixAdmin();
_helixManager = helixResourceManager.getHelixZkManager();
_propertyStore = helixResourceManager.getPropertyStore();
@@ -145,7 +145,7 @@ public class PinotLLCRealtimeSegmentManager {
_tableConfigCache = new TableConfigCache(_propertyStore);
_streamPartitionAssignmentGenerator = new StreamPartitionAssignmentGenerator(_helixManager);
_flushThresholdUpdateManager = new FlushThresholdUpdateManager();
- _controllerLeadershipManager = controllerLeadershipManager;
+ _leadControllerManager = leadControllerManager;
}
public boolean getIsSplitCommitEnabled() {
@@ -181,8 +181,8 @@ public class PinotLLCRealtimeSegmentManager {
LOGGER.info("Wait completed: Number of completing segments = {}", _numCompletingSegments.get());
}
- protected boolean isLeader() {
- return _controllerLeadershipManager.isLeader();
+ protected boolean isLeader(String tableName) {
+ return _leadControllerManager.isLeaderForTable(tableName);
}
protected boolean isConnected() {
@@ -339,10 +339,10 @@ public class PinotLLCRealtimeSegmentManager {
URI uriToMoveTo = URIUtils.getUri(_controllerConf.getDataDir(), tableName, URIUtils.encode(segmentName));
PinotFS pinotFS = PinotFSFactory.create(tableDirURI.getScheme());
- if (!isConnected() || !isLeader()) {
+ if (!isConnected() || !isLeader(tableName)) {
// We can potentially log a different value than what we saw ....
LOGGER.warn("Lost leadership while committing segment file {}, {} for table {}: isLeader={}, isConnected={}",
- segmentName, segmentLocation, tableName, isLeader(), isConnected());
+ segmentName, segmentLocation, tableName, isLeader(tableName), isConnected());
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
return false;
}
@@ -537,17 +537,17 @@ public class PinotLLCRealtimeSegmentManager {
final String oldZnodePath =
ZKMetadataProvider.constructPropertyStorePathForSegment(realtimeTableName, committingSegmentNameStr);
- if (!isConnected() || !isLeader()) {
+ if (!isConnected() || !isLeader(realtimeTableName)) {
// We can potentially log a different value than what we saw ....
LOGGER.warn("Lost leadership while committing segment metadata for {} for table {}: isLeader={}, isConnected={}",
- committingSegmentNameStr, realtimeTableName, isLeader(), isConnected());
+ committingSegmentNameStr, realtimeTableName, isLeader(realtimeTableName), isConnected());
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
return false;
}
boolean success = writeSegmentToPropertyStore(oldZnodePath, oldZnRecord, realtimeTableName, stat.getVersion());
if (!success) {
LOGGER.warn("Fail to write old segment to property store for {} for table {}: isLeader={}, isConnected={}",
- committingSegmentNameStr, realtimeTableName, isLeader(), isConnected());
+ committingSegmentNameStr, realtimeTableName, isLeader(realtimeTableName), isConnected());
}
return success;
}
@@ -596,11 +596,11 @@ public class PinotLLCRealtimeSegmentManager {
ZKMetadataProvider.constructPropertyStorePathForSegment(realtimeTableName, newSegmentNameStr);
if (!isNewTableSetup) {
- if (!isLeader() || !isConnected()) {
+ if (!isLeader(realtimeTableName) || !isConnected()) {
// We can potentially log a different value than what we saw ....
LOGGER.warn(
"Lost leadership while committing new segment metadata for {} for table {}: isLeader={}, isConnected={}",
- newSegmentNameStr, rawTableName, isLeader(), isConnected());
+ newSegmentNameStr, rawTableName, isLeader(realtimeTableName), isConnected());
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
return false;
}
@@ -609,7 +609,7 @@ public class PinotLLCRealtimeSegmentManager {
boolean success = writeSegmentToPropertyStore(newZnodePath, newZnRecord, realtimeTableName);
if (!success) {
LOGGER.warn("Fail to write new segment to property store for {} for table {}: isLeader={}, isConnected={}",
- newSegmentNameStr, rawTableName, isLeader(), isConnected());
+ newSegmentNameStr, rawTableName, isLeader(realtimeTableName), isConnected());
}
return success;
}
@@ -1345,8 +1345,4 @@ public class PinotLLCRealtimeSegmentManager {
return idealState;
}
-
- public ControllerLeadershipManager getControllerLeadershipManager() {
- return _controllerLeadershipManager;
- }
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java
index 32642dd..0de057a 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java
@@ -52,8 +52,7 @@ import org.apache.pinot.common.utils.HLCSegmentName;
import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.common.utils.retry.RetryPolicies;
-import org.apache.pinot.controller.ControllerLeadershipManager;
-import org.apache.pinot.controller.LeadershipChangeSubscriber;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.PinotTableIdealStateBuilder;
import org.apache.pinot.core.query.utils.Pair;
@@ -66,7 +65,7 @@ import org.slf4j.LoggerFactory;
/**
* Realtime segment manager, which assigns realtime segments to server instances so that they can consume from the stream.
*/
-public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkChildListener, IZkDataListener, LeadershipChangeSubscriber {
+public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkChildListener, IZkDataListener {
private static final Logger LOGGER = LoggerFactory.getLogger(PinotRealtimeSegmentManager.class);
private static final String TABLE_CONFIG = "/CONFIGS/TABLE";
private static final String SEGMENTS_PATH = "/SEGMENTS";
@@ -80,12 +79,12 @@ public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkCh
private final PinotHelixResourceManager _pinotHelixResourceManager;
private ZkClient _zkClient;
private ControllerMetrics _controllerMetrics;
- private final ControllerLeadershipManager _controllerLeadershipManager;
+ private final LeadControllerManager _leadControllerManager;
public PinotRealtimeSegmentManager(PinotHelixResourceManager pinotManager,
- ControllerLeadershipManager controllerLeadershipManager) {
+ LeadControllerManager leadControllerManager) {
_pinotHelixResourceManager = pinotManager;
- _controllerLeadershipManager = controllerLeadershipManager;
+ _leadControllerManager = leadControllerManager;
String clusterName = _pinotHelixResourceManager.getHelixClusterName();
_propertyStorePath = PropertyPathConfig.getPath(PropertyType.PROPERTYSTORE, clusterName);
_tableConfigPath = _propertyStorePath + TABLE_CONFIG;
@@ -104,9 +103,6 @@ public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkCh
_zkClient.subscribeChildChanges(_tableConfigPath, this);
_zkClient.subscribeDataChanges(_tableConfigPath, this);
- // Subscribe to leadership changes
- _controllerLeadershipManager.subscribe(PinotLLCRealtimeSegmentManager.class.getName(), this);
-
// Setup change listeners for already existing tables, if any.
processPropertyStoreChange(_tableConfigPath);
}
@@ -128,6 +124,11 @@ public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkCh
continue;
}
+ // Skip if the current controller isn't the leader of this table
+ if (!_leadControllerManager.isLeaderForTable(realtimeTableName)) {
+ continue;
+ }
+
StreamConfig metadata = new StreamConfig(tableConfig.getIndexingConfig().getStreamConfigs());
if (metadata.hasHighLevelConsumerType()) {
idealStateMap.put(realtimeTableName, _pinotHelixResourceManager.getHelixAdmin()
@@ -273,10 +274,6 @@ public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkCh
}
}
- private boolean isLeader() {
- return _controllerLeadershipManager.isLeader();
- }
-
@Override
public synchronized void onDataChange(String path) {
LOGGER.info("PinotRealtimeSegmentManager.onDataChange: {}", path);
@@ -300,13 +297,9 @@ public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkCh
LOGGER.info("Processing change notification for path: {}", path);
refreshWatchers(path);
- if (isLeader()) {
- if (path.matches(REALTIME_SEGMENT_PROPERTY_STORE_PATH_PATTERN) || path
- .matches(REALTIME_TABLE_CONFIG_PROPERTY_STORE_PATH_PATTERN) || path.equals(CONTROLLER_LEADER_CHANGE)) {
- assignRealtimeSegmentsToServerInstancesIfNecessary();
- }
- } else {
- LOGGER.info("Not the leader of this cluster, ignoring realtime segment property store change.");
+ if (path.matches(REALTIME_SEGMENT_PROPERTY_STORE_PATH_PATTERN) || path
+ .matches(REALTIME_TABLE_CONFIG_PROPERTY_STORE_PATH_PATTERN) || path.equals(CONTROLLER_LEADER_CHANGE)) {
+ assignRealtimeSegmentsToServerInstancesIfNecessary();
}
} catch (Exception e) {
LOGGER.error("Caught exception while processing change for path {}", path, e);
@@ -415,14 +408,4 @@ public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkCh
LOGGER.info("PinotRealtimeSegmentManager.handleDataDeleted: {}", dataPath);
processPropertyStoreChange(dataPath);
}
-
- @Override
- public void onBecomingLeader() {
- processPropertyStoreChange(CONTROLLER_LEADER_CHANGE);
- }
-
- @Override
- public void onBecomingNonLeader() {
- processPropertyStoreChange(CONTROLLER_LEADER_CHANGE);
- }
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
index a4da6b4..8935c6b 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
@@ -19,7 +19,6 @@
package org.apache.pinot.controller.helix.core.realtime;
import com.google.common.annotations.VisibleForTesting;
-import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -29,7 +28,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.helix.HelixManager;
-import org.apache.helix.ZNRecord;
import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.metadata.segment.LLCRealtimeSegmentZKMetadata;
import org.apache.pinot.common.metrics.ControllerMeter;
@@ -37,12 +35,13 @@ import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.LLCSegmentName;
-import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.ControllerLeadershipManager;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.pinot.common.utils.SegmentName.SEPARATOR;
+
/**
* This is a singleton class in the controller that drives the state machines for segments that are in the
@@ -73,7 +72,7 @@ public class SegmentCompletionManager {
private final Map<String, Long> _commitTimeMap = new ConcurrentHashMap<>();
private final PinotLLCRealtimeSegmentManager _segmentManager;
private final ControllerMetrics _controllerMetrics;
- private final ControllerLeadershipManager _controllerLeadershipManager;
+ private final LeadControllerManager _leadControllerManager;
private final Lock[] _fsmLocks;
private static final int NUM_FSM_LOCKS = 20;
@@ -87,12 +86,12 @@ public class SegmentCompletionManager {
// TODO keep some history of past committed segments so that we can avoid looking up PROPERTYSTORE if some server comes in late.
public SegmentCompletionManager(HelixManager helixManager, PinotLLCRealtimeSegmentManager segmentManager,
- ControllerMetrics controllerMetrics, ControllerLeadershipManager controllerLeadershipManager,
+ ControllerMetrics controllerMetrics, LeadControllerManager leadControllerManager,
int segmentCommitTimeoutSeconds) {
_helixManager = helixManager;
_segmentManager = segmentManager;
_controllerMetrics = controllerMetrics;
- _controllerLeadershipManager = controllerLeadershipManager;
+ _leadControllerManager = leadControllerManager;
SegmentCompletionProtocol
.setMaxSegmentCommitTimeMs(TimeUnit.MILLISECONDS.convert(segmentCommitTimeoutSeconds, TimeUnit.SECONDS));
_fsmLocks = new Lock[NUM_FSM_LOCKS];
@@ -163,11 +162,12 @@ public class SegmentCompletionManager {
* that it currently has (i.e. next offset that it will consume, if it continues to consume).
*/
public SegmentCompletionProtocol.Response segmentConsumed(SegmentCompletionProtocol.Request.Params reqParams) {
- if (!isLeader() || !_helixManager.isConnected()) {
+ final String segmentNameStr = reqParams.getSegmentName();
+ final String tableName = segmentNameStr.split(SEPARATOR)[0];
+ if (!isLeader(tableName) || !_helixManager.isConnected()) {
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
return SegmentCompletionProtocol.RESP_NOT_LEADER;
}
- final String segmentNameStr = reqParams.getSegmentName();
final String instanceId = reqParams.getInstanceId();
final String stopReason = reqParams.getReason();
final long offset = reqParams.getOffset();
@@ -201,11 +201,12 @@ public class SegmentCompletionManager {
*/
public SegmentCompletionProtocol.Response segmentCommitStart(
final SegmentCompletionProtocol.Request.Params reqParams) {
- if (!isLeader() || !_helixManager.isConnected()) {
+ final String segmentNameStr = reqParams.getSegmentName();
+ final String tableName = segmentNameStr.split(SEPARATOR)[0];
+ if (!isLeader(tableName) || !_helixManager.isConnected()) {
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
return SegmentCompletionProtocol.RESP_NOT_LEADER;
}
- final String segmentNameStr = reqParams.getSegmentName();
final String instanceId = reqParams.getInstanceId();
final long offset = reqParams.getOffset();
LLCSegmentName segmentName = new LLCSegmentName(segmentNameStr);
@@ -225,11 +226,12 @@ public class SegmentCompletionManager {
}
public SegmentCompletionProtocol.Response extendBuildTime(final SegmentCompletionProtocol.Request.Params reqParams) {
- if (!isLeader() || !_helixManager.isConnected()) {
+ final String segmentNameStr = reqParams.getSegmentName();
+ final String tableName = segmentNameStr.split(SEPARATOR)[0];
+ if (!isLeader(tableName) || !_helixManager.isConnected()) {
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
return SegmentCompletionProtocol.RESP_NOT_LEADER;
}
- final String segmentNameStr = reqParams.getSegmentName();
final String instanceId = reqParams.getInstanceId();
final long offset = reqParams.getOffset();
final int extTimeSec = reqParams.getExtraTimeSec();
@@ -256,11 +258,12 @@ public class SegmentCompletionManager {
*/
public SegmentCompletionProtocol.Response segmentStoppedConsuming(
SegmentCompletionProtocol.Request.Params reqParams) {
- if (!isLeader() || !_helixManager.isConnected()) {
+ final String segmentNameStr = reqParams.getSegmentName();
+ final String tableName = segmentNameStr.split(SEPARATOR)[0];
+ if (!isLeader(tableName) || !_helixManager.isConnected()) {
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
return SegmentCompletionProtocol.RESP_NOT_LEADER;
}
- final String segmentNameStr = reqParams.getSegmentName();
final String instanceId = reqParams.getInstanceId();
final long offset = reqParams.getOffset();
final String reason = reqParams.getReason();
@@ -292,11 +295,12 @@ public class SegmentCompletionManager {
*/
public SegmentCompletionProtocol.Response segmentCommitEnd(SegmentCompletionProtocol.Request.Params reqParams,
boolean success, boolean isSplitCommit, CommittingSegmentDescriptor committingSegmentDescriptor) {
- if (!isLeader() || !_helixManager.isConnected()) {
+ final String segmentNameStr = reqParams.getSegmentName();
+ final String tableName = segmentNameStr.split(SEPARATOR)[0];
+ if (!isLeader(tableName) || !_helixManager.isConnected()) {
_controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_NOT_LEADER, 1L);
return SegmentCompletionProtocol.RESP_NOT_LEADER;
}
- final String segmentNameStr = reqParams.getSegmentName();
LLCSegmentName segmentName = new LLCSegmentName(segmentNameStr);
SegmentCompletionFSM fsm = null;
SegmentCompletionProtocol.Response response = SegmentCompletionProtocol.RESP_FAILED;
@@ -352,6 +356,7 @@ public class SegmentCompletionManager {
State _state = State.HOLDING; // Typically start off in HOLDING state.
final long _startTimeMs;
private final LLCSegmentName _segmentName;
+ private final String _realtimeTableName;
private final int _numReplicas;
private final Set<String> _excludedServerStateMap;
private final Map<String, Long> _commitStateMap;
@@ -394,6 +399,7 @@ public class SegmentCompletionManager {
private SegmentCompletionFSM(PinotLLCRealtimeSegmentManager segmentManager,
SegmentCompletionManager segmentCompletionManager, LLCSegmentName segmentName, int numReplicas) {
_segmentName = segmentName;
+ _realtimeTableName = _segmentName.getTableName();
_numReplicas = numReplicas;
_segmentManager = segmentManager;
_commitStateMap = new HashMap<>(_numReplicas);
@@ -403,8 +409,8 @@ public class SegmentCompletionManager {
_maxTimeToPickWinnerMs = _startTimeMs + MAX_TIME_TO_PICK_WINNER_MS;
_maxTimeToNotifyWinnerMs = _startTimeMs + MAX_TIME_TO_NOTIFY_WINNER_MS;
long initialCommitTimeMs =
- MAX_TIME_TO_NOTIFY_WINNER_MS + _segmentManager.getCommitTimeoutMS(_segmentName.getTableName());
- Long savedCommitTime = _segmentCompletionManager._commitTimeMap.get(segmentName.getTableName());
+ MAX_TIME_TO_NOTIFY_WINNER_MS + _segmentManager.getCommitTimeoutMS(_realtimeTableName);
+ Long savedCommitTime = _segmentCompletionManager._commitTimeMap.get(_realtimeTableName);
if (savedCommitTime != null && savedCommitTime > initialCommitTimeMs) {
initialCommitTimeMs = savedCommitTime;
}
@@ -413,7 +419,7 @@ public class SegmentCompletionManager {
// The table has a really high value configured for max commit time. Set it to a higher value than default
// and go from there.
LOGGER.info("Configured max commit time {}s too high for table {}, changing to {}s", initialCommitTimeMs / 1000,
- segmentName.getTableName(), MAX_COMMIT_TIME_FOR_ALL_SEGMENTS_SECONDS);
+ _realtimeTableName, MAX_COMMIT_TIME_FOR_ALL_SEGMENTS_SECONDS);
initialCommitTimeMs = MAX_COMMIT_TIME_FOR_ALL_SEGMENTS_SECONDS * 1000;
}
_initialCommitTimeMs = initialCommitTimeMs;
@@ -681,14 +687,14 @@ public class SegmentCompletionManager {
private SegmentCompletionProtocol.Response abortAndReturnHold(long now, String instanceId, long offset) {
_state = State.ABORTED;
_segmentCompletionManager._controllerMetrics
- .addMeteredTableValue(_segmentName.getTableName(), ControllerMeter.LLC_STATE_MACHINE_ABORTS, 1);
+ .addMeteredTableValue(_realtimeTableName, ControllerMeter.LLC_STATE_MACHINE_ABORTS, 1);
return hold(instanceId, offset);
}
private SegmentCompletionProtocol.Response abortAndReturnFailed() {
_state = State.ABORTED;
_segmentCompletionManager._controllerMetrics
- .addMeteredTableValue(_segmentName.getTableName(), ControllerMeter.LLC_STATE_MACHINE_ABORTS, 1);
+ .addMeteredTableValue(_realtimeTableName, ControllerMeter.LLC_STATE_MACHINE_ABORTS, 1);
return SegmentCompletionProtocol.RESP_FAILED;
}
@@ -1117,7 +1123,7 @@ public class SegmentCompletionManager {
}
@VisibleForTesting
- protected boolean isLeader() {
- return _controllerLeadershipManager.isLeader();
+ protected boolean isLeader(String tableName) {
+ return _leadControllerManager.isLeaderForTable(tableName);
}
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/RealtimeSegmentRelocator.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/RealtimeSegmentRelocator.java
index ea5b05b..2c2c017 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/RealtimeSegmentRelocator.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/RealtimeSegmentRelocator.java
@@ -38,6 +38,7 @@ import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.common.utils.retry.RetryPolicies;
import org.apache.pinot.common.utils.time.TimeUtils;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.PinotHelixSegmentOnlineOfflineStateModelGenerator;
import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
@@ -55,10 +56,11 @@ import org.slf4j.LoggerFactory;
public class RealtimeSegmentRelocator extends ControllerPeriodicTask<Void> {
private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeSegmentRelocator.class);
- public RealtimeSegmentRelocator(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf config,
- ControllerMetrics controllerMetrics) {
+ public RealtimeSegmentRelocator(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics) {
super("RealtimeSegmentRelocator", getRunFrequencySeconds(config.getRealtimeSegmentRelocatorFrequency()),
- config.getRealtimeSegmentRelocationInitialDelayInSeconds(), pinotHelixResourceManager, controllerMetrics);
+ config.getRealtimeSegmentRelocationInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
}
@Override
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
index 7d5182d..fecb4ac 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
@@ -35,6 +35,7 @@ import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.CommonConstants.Segment.Realtime.Status;
import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
import org.apache.pinot.controller.helix.core.retention.strategy.RetentionStrategy;
@@ -54,10 +55,11 @@ public class RetentionManager extends ControllerPeriodicTask<Void> {
private final int _deletedSegmentsRetentionInDays;
- public RetentionManager(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf config,
- ControllerMetrics controllerMetrics) {
+ public RetentionManager(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics) {
super("RetentionManager", config.getRetentionControllerFrequencyInSeconds(),
- config.getRetentionManagerInitialDelayInSeconds(), pinotHelixResourceManager, controllerMetrics);
+ config.getRetentionManagerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
_deletedSegmentsRetentionInDays = config.getDeletedSegmentsRetentionInDays();
LOGGER.info("Starting RetentionManager with runFrequencyInSeconds: {}, deletedSegmentsRetentionInDays: {}",
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/statemodel/LeadControllerResourceMasterSlaveStateModelFactory.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/statemodel/LeadControllerResourceMasterSlaveStateModelFactory.java
new file mode 100644
index 0000000..5655de8
--- /dev/null
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/statemodel/LeadControllerResourceMasterSlaveStateModelFactory.java
@@ -0,0 +1,84 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.helix.core.statemodel;
+
+import org.apache.helix.NotificationContext;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class LeadControllerResourceMasterSlaveStateModelFactory extends StateModelFactory<StateModel> {
+ private final LeadControllerManager _leadControllerManager;
+
+ public LeadControllerResourceMasterSlaveStateModelFactory(LeadControllerManager leadControllerManager) {
+ _leadControllerManager = leadControllerManager;
+ }
+
+ @Override
+ public StateModel createNewStateModel(String resourceName, String partitionName) {
+ return new LeadControllerResourceMasterSlaveStateModel(_leadControllerManager, partitionName);
+ }
+
+ @StateModelInfo(states = "{'MASTER','SLAVE', 'OFFLINE', 'DROPPED'}", initialState = "OFFLINE")
+ public static class LeadControllerResourceMasterSlaveStateModel extends StateModel {
+ private static final Logger LOGGER = LoggerFactory.getLogger(LeadControllerResourceMasterSlaveStateModel.class);
+
+ private final LeadControllerManager _leadControllerManager;
+ private final String _partitionName;
+
+ private LeadControllerResourceMasterSlaveStateModel(LeadControllerManager leadControllerManager,
+ String partitionName) {
+ _leadControllerManager = leadControllerManager;
+ _partitionName = partitionName;
+ }
+
+ @Transition(from = "OFFLINE", to = "SLAVE")
+ public void onBecomeSlaveFromOffline(Message message, NotificationContext context) {
+ LOGGER.info("Got state transition from OFFLINE to SLAVE for partition: {}", _partitionName);
+ }
+
+ @Transition(from = "SLAVE", to = "MASTER")
+ public void onBecomeMasterFromSlave(Message message, NotificationContext context) {
+ LOGGER.info("Got state transition from SLAVE to MASTER for partition: {}", _partitionName);
+ _leadControllerManager.addPartitionLeader(_partitionName);
+ }
+
+ @Transition(from = "MASTER", to = "SLAVE")
+ public void onBecomeSlaveFromMaster(Message message, NotificationContext context) {
+ LOGGER.info("Got state transition from MASTER to SLAVE for partition: {}", _partitionName);
+ _leadControllerManager.removePartitionLeader(_partitionName);
+ }
+
+ @Transition(from = "SLAVE", to = "OFFLINE")
+ public void onBecomeOfflineFromSlave(Message message, NotificationContext context) {
+ LOGGER.info("Got state transition from SLAVE to OFFLINE for partition: {}", _partitionName);
+ }
+
+ @Transition(from = "OFFLINE", to = "DROPPED")
+ public void onBecomeDroppedFromOffline(Message message, NotificationContext context) {
+ LOGGER.info("Got state transition from OFFLINE to DROPPED for partition: {}", _partitionName);
+ }
+ }
+}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
index 37dec22..8564374 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
@@ -43,6 +43,7 @@ import org.apache.helix.model.StateModelDefinition;
import org.apache.helix.model.builder.CustomModeISBuilder;
import org.apache.helix.model.builder.FullAutoModeISBuilder;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
+import org.apache.pinot.common.utils.helix.LeadControllerUtils;
import org.apache.pinot.controller.helix.core.PinotHelixBrokerResourceOnlineOfflineStateModelGenerator;
import org.apache.pinot.controller.helix.core.PinotHelixSegmentOnlineOfflineStateModelGenerator;
import org.slf4j.Logger;
@@ -148,7 +149,8 @@ public class HelixSetupUtils {
private static void createLeadControllerResourceIfNeeded(String helixClusterName, HelixAdmin helixAdmin,
ConfigAccessor configAccessor, boolean enableBatchMessageMode) {
- if (helixAdmin.getResourceIdealState(helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME) == null) {
+ IdealState idealState = helixAdmin.getResourceIdealState(helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME);
+ if (idealState == null) {
LOGGER.info("Adding resource: {}", LEAD_CONTROLLER_RESOURCE_NAME);
// FULL-AUTO Master-Slave state model with CrushED rebalance strategy
@@ -158,7 +160,7 @@ public class HelixSetupUtils {
// Initialize partitions and replicas
idealStateBuilder.setNumPartitions(NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE);
for (int i = 0; i < NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE; i++) {
- idealStateBuilder.add(LEAD_CONTROLLER_RESOURCE_NAME + "_" + i);
+ idealStateBuilder.add(LeadControllerUtils.generatePartitionName(i));
}
idealStateBuilder.setNumReplica(LEAD_CONTROLLER_RESOURCE_REPLICA_COUNT);
// The below config guarantees if active number of replicas is no less than minimum active replica, there will not be partition movements happened.
@@ -169,16 +171,19 @@ public class HelixSetupUtils {
idealStateBuilder.setRebalanceDelay(REBALANCE_DELAY_MS);
idealStateBuilder.enableDelayRebalance();
// Set instance group tag
- IdealState idealState = idealStateBuilder.build();
+ idealState = idealStateBuilder.build();
idealState.setInstanceGroupTag(CONTROLLER_INSTANCE);
// Set batch message mode
idealState.setBatchMessageMode(enableBatchMessageMode);
// Explicitly disable this resource when creating this new resource.
// When all the controllers are running the code with the logic to handle this resource, it can be enabled for backward compatibility.
// In the next major release, we can enable this resource by default, so that all the controller logic can be separated.
- idealState.enable(false);
helixAdmin.addResource(helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, idealState);
+ } else if (!idealState.isEnabled()) {
+ // Enable lead controller resource and let resource config be the only switch for enabling logic of lead controller resource.
+ idealState.enable(true);
+ helixAdmin.updateIdealState(helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, idealState);
}
// Create resource config for lead controller resource if it doesn't exist
@@ -187,8 +192,8 @@ public class HelixSetupUtils {
resourceConfig = new ResourceConfig(LEAD_CONTROLLER_RESOURCE_NAME);
}
// Set RESOURCE_ENABLED to false if it's absent in resource config
- if (resourceConfig.getSimpleConfig("RESOURCE_ENABLED") == null) {
- resourceConfig.putSimpleConfig("RESOURCE_ENABLED", Boolean.FALSE.toString());
+ if (resourceConfig.getSimpleConfig(LEAD_CONTROLLER_RESOURCE_ENABLED_KEY) == null) {
+ resourceConfig.putSimpleConfig(LEAD_CONTROLLER_RESOURCE_ENABLED_KEY, Boolean.FALSE.toString());
}
configAccessor.setResourceConfig(helixClusterName, LEAD_CONTROLLER_RESOURCE_NAME, resourceConfig);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/BrokerResourceValidationManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/BrokerResourceValidationManager.java
index 5748d3c..a82a161 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/BrokerResourceValidationManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/BrokerResourceValidationManager.java
@@ -24,6 +24,7 @@ import org.apache.helix.model.InstanceConfig;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
import org.slf4j.Logger;
@@ -37,9 +38,9 @@ public class BrokerResourceValidationManager extends ControllerPeriodicTask<Brok
private static final Logger LOGGER = LoggerFactory.getLogger(BrokerResourceValidationManager.class);
public BrokerResourceValidationManager(ControllerConf config, PinotHelixResourceManager pinotHelixResourceManager,
- ControllerMetrics controllerMetrics) {
+ LeadControllerManager leadControllerManager, ControllerMetrics controllerMetrics) {
super("BrokerResourceValidationManager", config.getBrokerResourceValidationFrequencyInSeconds(),
- config.getBrokerResourceValidationInitialDelayInSeconds(), pinotHelixResourceManager, controllerMetrics);
+ config.getBrokerResourceValidationInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager, controllerMetrics);
}
@Override
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/OfflineSegmentIntervalChecker.java b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/OfflineSegmentIntervalChecker.java
index 7f19395..0eeadc7 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/OfflineSegmentIntervalChecker.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/OfflineSegmentIntervalChecker.java
@@ -30,6 +30,7 @@ import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.metrics.ValidationMetrics;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
import org.apache.pinot.controller.util.SegmentIntervalUtils;
@@ -50,9 +51,11 @@ public class OfflineSegmentIntervalChecker extends ControllerPeriodicTask<Void>
private final ValidationMetrics _validationMetrics;
public OfflineSegmentIntervalChecker(ControllerConf config, PinotHelixResourceManager pinotHelixResourceManager,
- ValidationMetrics validationMetrics, ControllerMetrics controllerMetrics) {
+ LeadControllerManager leadControllerManager, ValidationMetrics validationMetrics,
+ ControllerMetrics controllerMetrics) {
super("OfflineSegmentIntervalChecker", config.getOfflineSegmentIntervalCheckerFrequencyInSeconds(),
- config.getOfflineSegmentIntervalCheckerInitialDelayInSeconds(), pinotHelixResourceManager, controllerMetrics);
+ config.getOfflineSegmentIntervalCheckerInitialDelayInSeconds(), pinotHelixResourceManager,
+ leadControllerManager, controllerMetrics);
_validationMetrics = validationMetrics;
}
@@ -88,12 +91,12 @@ public class OfflineSegmentIntervalChecker extends ControllerPeriodicTask<Void>
if (SegmentIntervalUtils.isValidInterval(timeInterval)) {
segmentIntervals.add(timeInterval);
} else {
- numSegmentsWithInvalidIntervals ++;
+ numSegmentsWithInvalidIntervals++;
}
}
if (numSegmentsWithInvalidIntervals > 0) {
- LOGGER.warn("Table: {} has {} segments with invalid interval", offlineTableName,
- numSegmentsWithInvalidIntervals);
+ LOGGER
+ .warn("Table: {} has {} segments with invalid interval", offlineTableName, numSegmentsWithInvalidIntervals);
}
Duration frequency = SegmentIntervalUtils.convertToDuration(validationConfig.getSegmentPushFrequency());
numMissingSegments = computeNumMissingSegments(segmentIntervals, frequency);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
index 5eb5a6f..88ad642 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
@@ -32,6 +32,7 @@ import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.HLCSegmentName;
import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
@@ -53,10 +54,11 @@ public class RealtimeSegmentValidationManager extends ControllerPeriodicTask<Rea
private long _lastUpdateRealtimeDocumentCountTimeMs = 0L;
public RealtimeSegmentValidationManager(ControllerConf config, PinotHelixResourceManager pinotHelixResourceManager,
- PinotLLCRealtimeSegmentManager llcRealtimeSegmentManager, ValidationMetrics validationMetrics,
- ControllerMetrics controllerMetrics) {
+ LeadControllerManager leadControllerManager, PinotLLCRealtimeSegmentManager llcRealtimeSegmentManager,
+ ValidationMetrics validationMetrics, ControllerMetrics controllerMetrics) {
super("RealtimeSegmentValidationManager", config.getRealtimeSegmentValidationFrequencyInSeconds(),
- config.getRealtimeSegmentValidationManagerInitialDelaySeconds(), pinotHelixResourceManager, controllerMetrics);
+ config.getRealtimeSegmentValidationManagerInitialDelaySeconds(), pinotHelixResourceManager,
+ leadControllerManager, controllerMetrics);
_llcRealtimeSegmentManager = llcRealtimeSegmentManager;
_validationMetrics = validationMetrics;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/StorageQuotaChecker.java b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/StorageQuotaChecker.java
index e895242..6a3066c 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/StorageQuotaChecker.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/StorageQuotaChecker.java
@@ -28,8 +28,6 @@ import org.apache.pinot.common.exception.InvalidConfigException;
import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.utils.DataSize;
-import org.apache.pinot.controller.ControllerLeadershipManager;
-import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.util.TableSizeReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,17 +43,14 @@ public class StorageQuotaChecker {
private final TableSizeReader _tableSizeReader;
private final TableConfig _tableConfig;
private final ControllerMetrics _controllerMetrics;
- private final PinotHelixResourceManager _pinotHelixResourceManager;
- private final ControllerLeadershipManager _controllerLeadershipManager;
+ private final boolean _isLeaderForTable;
public StorageQuotaChecker(TableConfig tableConfig, TableSizeReader tableSizeReader,
- ControllerMetrics controllerMetrics, PinotHelixResourceManager pinotHelixResourceManager,
- ControllerLeadershipManager controllerLeadershipManager) {
+ ControllerMetrics controllerMetrics, boolean isLeaderForTable) {
_tableConfig = tableConfig;
_tableSizeReader = tableSizeReader;
_controllerMetrics = controllerMetrics;
- _pinotHelixResourceManager = pinotHelixResourceManager;
- _controllerLeadershipManager = controllerLeadershipManager;
+ _isLeaderForTable = isLeaderForTable;
}
public static class QuotaCheckerResponse {
@@ -157,7 +152,7 @@ public class StorageQuotaChecker {
tableNameWithType, tableSubtypeSize.estimatedSizeInBytes, tableSubtypeSize.reportedSizeInBytes);
// Only emit the real percentage of storage quota usage by lead controller, otherwise emit 0L.
- if (isLeader()) {
+ if (_isLeaderForTable) {
long existingStorageQuotaUtilization = tableSubtypeSize.estimatedSizeInBytes * 100 / allowedStorageBytes;
_controllerMetrics.setValueOfTableGauge(tableNameWithType, ControllerGauge.TABLE_STORAGE_QUOTA_UTILIZATION,
existingStorageQuotaUtilization);
@@ -212,8 +207,4 @@ public class StorageQuotaChecker {
return failure(message);
}
}
-
- protected boolean isLeader() {
- return _controllerLeadershipManager.isLeader();
- }
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/LeadControllerManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/LeadControllerManagerTest.java
new file mode 100644
index 0000000..c9e4e74
--- /dev/null
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/LeadControllerManagerTest.java
@@ -0,0 +1,125 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller;
+
+import com.yammer.metrics.core.MetricsRegistry;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.ResourceConfig;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.common.utils.helix.LeadControllerUtils;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+public class LeadControllerManagerTest {
+ private static final String CONTROLLER_HOST = "localhost";
+ private static final int CONTROLLER_PORT = 18998;
+
+ private HelixManager _helixManager;
+ private ControllerMetrics _controllerMetrics;
+ private LiveInstance _liveInstance;
+ private ResourceConfig _resourceConfig;
+
+ @BeforeMethod
+ public void setup() {
+ _controllerMetrics = new ControllerMetrics(new MetricsRegistry());
+ _helixManager = mock(HelixManager.class);
+ HelixDataAccessor helixDataAccessor = mock(HelixDataAccessor.class);
+ when(_helixManager.getHelixDataAccessor()).thenReturn(helixDataAccessor);
+
+ PropertyKey.Builder keyBuilder = mock(PropertyKey.Builder.class);
+ when(helixDataAccessor.keyBuilder()).thenReturn(keyBuilder);
+
+ PropertyKey controllerLeader = mock(PropertyKey.class);
+ when(keyBuilder.controllerLeader()).thenReturn(controllerLeader);
+ _liveInstance = mock(LiveInstance.class);
+ when(helixDataAccessor.getProperty(controllerLeader)).thenReturn(_liveInstance);
+
+ PropertyKey resourceConfigPropertyKey = mock(PropertyKey.class);
+ when(keyBuilder.resourceConfig(any())).thenReturn(resourceConfigPropertyKey);
+ _resourceConfig = mock(ResourceConfig.class);
+ when(helixDataAccessor.getProperty(resourceConfigPropertyKey)).thenReturn(_resourceConfig);
+
+ String instanceId = LeadControllerUtils.generateParticipantInstanceId(CONTROLLER_HOST, CONTROLLER_PORT);
+ when(_helixManager.getInstanceName()).thenReturn(instanceId);
+ }
+
+ @Test
+ public void testLeadControllerManager() {
+ LeadControllerManager leadControllerManager = new LeadControllerManager(_helixManager, _controllerMetrics);
+ String tableName = "testTable";
+ int expectedPartitionIndex = LeadControllerUtils.getPartitionIdForTable(tableName);
+ String partitionName = LeadControllerUtils.generatePartitionName(expectedPartitionIndex);
+
+ becomeHelixLeader(false);
+ leadControllerManager.onHelixControllerChange();
+
+ // When there's no resource config change nor helix controller change, leadControllerManager should return false.
+ Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));
+
+ enableResourceConfig(true);
+ leadControllerManager.onResourceConfigChange();
+
+ // Even resource config is enabled, leadControllerManager should return false because no index is cached yet.
+ Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));
+ Assert.assertTrue(leadControllerManager.isLeadControllerResourceEnabled());
+
+ // After the target partition index is cached, leadControllerManager should return true.
+ leadControllerManager.addPartitionLeader(partitionName);
+ Assert.assertTrue(leadControllerManager.isLeaderForTable(tableName));
+
+ // When the target partition index is removed, leadControllerManager should return false.
+ leadControllerManager.removePartitionLeader(partitionName);
+ Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));
+
+ // When resource config is set to false, the cache should be disabled, even if the target partition index is in the cache.
+ // The leader depends on whether the current controller is helix leader.
+ enableResourceConfig(false);
+ leadControllerManager.onResourceConfigChange();
+
+ Assert.assertFalse(leadControllerManager.isLeadControllerResourceEnabled());
+ Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));
+ leadControllerManager.addPartitionLeader(partitionName);
+ Assert.assertFalse(leadControllerManager.isLeaderForTable(tableName));
+
+ // When the current controller becomes helix leader and resource is disabled, leadControllerManager should return false.
+ becomeHelixLeader(true);
+ leadControllerManager.onHelixControllerChange();
+ Assert.assertTrue(leadControllerManager.isLeaderForTable(tableName));
+ }
+
+ private void becomeHelixLeader(boolean becomeHelixLeader) {
+ if (becomeHelixLeader) {
+ when(_liveInstance.getInstanceName()).thenReturn(CONTROLLER_HOST + "_" + CONTROLLER_PORT);
+ }
+ }
+
+ private void enableResourceConfig(boolean enable) {
+ when(_resourceConfig.getSimpleConfig(anyString())).thenReturn(Boolean.toString(enable));
+ }
+}
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 ab1ea56..5f51010 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
@@ -41,6 +41,7 @@ import org.apache.commons.httpclient.methods.multipart.Part;
import org.apache.commons.httpclient.methods.multipart.StringPart;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
+import org.apache.helix.ConfigAccessor;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixManager;
import org.apache.helix.HelixManagerFactory;
@@ -50,6 +51,7 @@ import org.apache.helix.ZNRecord;
import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.HelixConfigScope;
import org.apache.helix.model.Message;
+import org.apache.helix.model.ResourceConfig;
import org.apache.helix.model.builder.HelixConfigScopeBuilder;
import org.apache.helix.participant.statemachine.StateModel;
import org.apache.helix.participant.statemachine.StateModelFactory;
@@ -73,6 +75,8 @@ import org.slf4j.LoggerFactory;
import org.testng.Assert;
import static org.apache.pinot.common.utils.CommonConstants.Helix.Instance.ADMIN_PORT_KEY;
+import static org.apache.pinot.common.utils.CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_ENABLED_KEY;
+import static org.apache.pinot.common.utils.CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME;
import static org.apache.pinot.common.utils.CommonConstants.Helix.UNTAGGED_BROKER_INSTANCE;
import static org.apache.pinot.common.utils.CommonConstants.Helix.UNTAGGED_SERVER_INSTANCE;
import static org.apache.pinot.common.utils.CommonConstants.Server.DEFAULT_ADMIN_API_PORT;
@@ -440,6 +444,16 @@ public abstract class ControllerTest {
getServerTenantRequestPayload(tenantName, numOfflineServers, numRealtimeServers));
}
+ public void enableResourceConfigForLeadControllerResource(boolean enable) {
+ ConfigAccessor configAccessor = _helixManager.getConfigAccessor();
+ ResourceConfig resourceConfig =
+ configAccessor.getResourceConfig(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
+ if (!Boolean.toString(enable).equals(resourceConfig.getSimpleConfig(LEAD_CONTROLLER_RESOURCE_ENABLED_KEY))) {
+ resourceConfig.putSimpleConfig(LEAD_CONTROLLER_RESOURCE_ENABLED_KEY, Boolean.toString(enable));
+ configAccessor.setResourceConfig(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, resourceConfig);
+ }
+ }
+
public static String sendGetRequest(String urlString)
throws IOException {
return constructResponse(new URL(urlString).openStream());
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeTest.java
index 7cd5612..c1f3b8a 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/PinotControllerModeTest.java
@@ -22,19 +22,23 @@ import java.util.Map;
import java.util.Set;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixManager;
+import org.apache.helix.manager.zk.ZkClient;
import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.IdealState;
+import org.apache.pinot.common.utils.CommonConstants.Helix;
+import org.apache.pinot.common.utils.ZkStarter;
+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.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.util.TestUtils;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-import static org.apache.pinot.common.utils.CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME;
-import static org.apache.pinot.common.utils.CommonConstants.Helix.NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE;
-
public class PinotControllerModeTest extends ControllerTest {
private static long TIMEOUT_IN_MS = 10_000L;
@@ -56,25 +60,41 @@ public class PinotControllerModeTest extends ControllerTest {
stopController();
}
- @Test(enabled = false)
+ @Test
public void testDualModeController() {
// Start the first dual-mode controller
ControllerConf firstDualModeControllerConfig = getDefaultControllerConfiguration();
firstDualModeControllerConfig.setControllerMode(ControllerConf.ControllerMode.DUAL);
startController(firstDualModeControllerConfig);
+
+ // Disable delay rebalance feature
+ IdealState idealState = _helixResourceManager.getTableIdealState(Helix.LEAD_CONTROLLER_RESOURCE_NAME);
+ idealState.setDelayRebalanceEnabled(false);
+ idealState.setMinActiveReplicas(1);
+ _helixAdmin.updateIdealState(getHelixClusterName(), Helix.LEAD_CONTROLLER_RESOURCE_NAME, idealState);
+
TestUtils.waitForCondition(aVoid -> _helixManager.isConnected(), TIMEOUT_IN_MS,
"Failed to start the first dual-mode controller");
+ // The first controller should be the MASTER for all partitions
+ checkInstanceState(_helixAdmin);
- // There should be no partition in the external view because the resource is disabled
- TestUtils.waitForCondition(aVoid -> {
- ExternalView leadControllerResourceExternalView =
- _helixAdmin.getResourceExternalView(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
- return leadControllerResourceExternalView.getPartitionSet().isEmpty();
- }, TIMEOUT_IN_MS, "There should be no partition in the disabled resource's external view");
+ final LeadControllerManager firstLeadControllerManager = _controllerStarter.getLeadControllerManager();
+ String firstTableName = "firstTableName";
+ String secondTableName = "secondTableName";
+
+ Assert.assertNotEquals(LeadControllerUtils.getPartitionIdForTable(firstTableName),
+ LeadControllerUtils.getPartitionIdForTable(secondTableName));
- // Enable the lead controller resource, and the first controller should be the MASTER for all partitions
- _helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
- checkInstanceState(_helixAdmin, "MASTER");
+ // The first controller should be the leader for both tables, which is the Helix leader, before the resource is enabled.
+ Assert.assertTrue(firstLeadControllerManager.isLeaderForTable(firstTableName));
+ Assert.assertTrue(firstLeadControllerManager.isLeaderForTable(secondTableName));
+
+ enableResourceConfigForLeadControllerResource(true);
+ TestUtils.waitForCondition(aVoid -> firstLeadControllerManager.isLeadControllerResourceEnabled(), TIMEOUT_IN_MS,
+ "Failed to mark lead controller resource as enabled");
+ // The first controller should still be the leader for both tables, which is the partition leader, after the resource is enabled.
+ Assert.assertTrue(firstLeadControllerManager.isLeaderForTable(firstTableName));
+ Assert.assertTrue(firstLeadControllerManager.isLeaderForTable(secondTableName));
// Start the second dual-mode controller
ControllerConf secondDualModeControllerConfig = getDefaultControllerConfiguration();
@@ -85,29 +105,76 @@ public class PinotControllerModeTest extends ControllerTest {
TestUtils
.waitForCondition(aVoid -> secondDualModeController.getHelixResourceManager().getHelixZkManager().isConnected(),
TIMEOUT_IN_MS, "Failed to start the second dual-mode controller");
-
// There should still be only one MASTER instance for each partition
- checkInstanceState(_helixAdmin, "MASTER");
+ checkInstanceState(_helixAdmin);
- // Disable the lead controller resource, and there should be only one OFFLINE instance for each partition
- _helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, false);
- checkInstanceState(_helixAdmin, "OFFLINE");
+ LeadControllerManager secondLeadControllerManager = secondDualModeController.getLeadControllerManager();
+ Assert.assertTrue(secondLeadControllerManager.isLeadControllerResourceEnabled());
- // Re-enable the lead controller resource, and there should be only one MASTER instance for each partition
- _helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
- checkInstanceState(_helixAdmin, "MASTER");
+ // Either one of the controllers is the only leader for a given table name.
+ TestUtils.waitForCondition(aVoid -> {
+ boolean result;
+ result = firstLeadControllerManager.isLeaderForTable(firstTableName) ^ secondLeadControllerManager
+ .isLeaderForTable(firstTableName);
+ result &= firstLeadControllerManager.isLeaderForTable(secondTableName) ^ secondLeadControllerManager
+ .isLeaderForTable(secondTableName);
+ return result;
+ }, TIMEOUT_IN_MS, "Either one of the controllers is the only leader for a given table");
// Stop the second controller, and there should still be only one MASTER instance for each partition
secondDualModeController.stop();
- checkInstanceState(_helixAdmin, "MASTER");
+ checkInstanceState(_helixAdmin);
// Stop the first controller
stopController();
+
+ // Both controllers are stopped and no one should be the partition leader for tables.
+ TestUtils.waitForCondition(aVoid -> {
+ boolean result;
+ result = !firstLeadControllerManager.isLeaderForTable(firstTableName);
+ result &= !firstLeadControllerManager.isLeaderForTable(secondTableName);
+ result &= !secondLeadControllerManager.isLeaderForTable(firstTableName);
+ result &= !secondLeadControllerManager.isLeaderForTable(secondTableName);
+ return result;
+ }, TIMEOUT_IN_MS, "No one should be the partition leader for tables");
+
+ ZkClient zkClient = new ZkClient(ZkStarter.DEFAULT_ZK_STR);
+ TestUtils
+ .waitForCondition(aVoid -> !zkClient.exists("/" + getHelixClusterName() + "/CONTROLLER/LEADER"), TIMEOUT_IN_MS,
+ "No cluster leader should be shown in Helix cluster");
+ zkClient.close();
+
+ ControllerConf thirdDualModeControllerConfig = getDefaultControllerConfiguration();
+ thirdDualModeControllerConfig.setControllerMode(ControllerConf.ControllerMode.DUAL);
+ thirdDualModeControllerConfig.setControllerPort(Integer.toString(DEFAULT_CONTROLLER_PORT + 2));
+ ControllerStarter thirdDualModeController = getControllerStarter(thirdDualModeControllerConfig);
+ thirdDualModeController.start();
+ PinotHelixResourceManager pinotHelixResourceManager = thirdDualModeController.getHelixResourceManager();
+ _helixManager = pinotHelixResourceManager.getHelixZkManager();
+ _helixAdmin = _helixManager.getClusterManagmentTool();
+ TestUtils
+ .waitForCondition(aVoid -> thirdDualModeController.getHelixResourceManager().getHelixZkManager().isConnected(),
+ TIMEOUT_IN_MS, "Failed to start the 3rd dual-mode controller");
+ // There should still be only one MASTER instance for each partition
+ checkInstanceState(_helixAdmin);
+
+ enableResourceConfigForLeadControllerResource(false);
+ final LeadControllerManager thirdLeadControllerManager = thirdDualModeController.getLeadControllerManager();
+
+ TestUtils.waitForCondition(aVoid -> !thirdLeadControllerManager.isLeadControllerResourceEnabled(), TIMEOUT_IN_MS,
+ "Lead controller resource should be disabled.");
+
+ // After disabling lead controller resource, helix leader should be the leader for both tables.
+ TestUtils.waitForCondition(
+ aVoid -> thirdLeadControllerManager.isLeaderForTable(firstTableName) && thirdLeadControllerManager
+ .isLeaderForTable(secondTableName), TIMEOUT_IN_MS,
+ "The 3rd controller should be the leader for both tables, which is the Helix leader.");
+
+ // Stop the 3rd controller
+ thirdDualModeController.stop();
}
- // TODO: enable it after removing ControllerLeadershipManager which requires both CONTROLLER and PARTICIPANT
- // HelixManager
- @Test(enabled = false)
+ @Test
public void testPinotOnlyController() {
ControllerConf firstPinotOnlyControllerConfig = getDefaultControllerConfiguration();
firstPinotOnlyControllerConfig.setControllerMode(ControllerConf.ControllerMode.PINOT_ONLY);
@@ -132,30 +199,13 @@ public class PinotControllerModeTest extends ControllerTest {
TestUtils.waitForCondition(aVoid -> helixControllerManager.isConnected(), TIMEOUT_IN_MS,
"Failed to start the Helix-only controller");
- // Enabling the lead controller resource before setting up the Pinot cluster should fail
- try {
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
- Assert.fail("Enabling the lead controller resource before setting up the Pinot cluster should fail");
- } catch (Exception e) {
- // Expected
- }
-
// Start the first Pinot-only controller
firstPinotOnlyController.start();
PinotHelixResourceManager helixResourceManager = firstPinotOnlyController.getHelixResourceManager();
TestUtils.waitForCondition(aVoid -> helixResourceManager.getHelixZkManager().isConnected(), TIMEOUT_IN_MS,
"Failed to start the first Pinot-only controller");
-
- // There should be no partition in the external view because the resource is disabled
- TestUtils.waitForCondition(aVoid -> {
- ExternalView leadControllerResourceExternalView =
- helixAdmin.getResourceExternalView(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
- return leadControllerResourceExternalView.getPartitionSet().isEmpty();
- }, TIMEOUT_IN_MS, "There should be no partition in the disabled resource's external view");
-
- // Enable the lead controller resource, and the first Pinot-only controller should be the MASTER for all partitions
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
- checkInstanceState(helixAdmin, "MASTER");
+ // The first Pinot-only controller should be the MASTER for all partitions
+ checkInstanceState(helixAdmin);
// Start the second Pinot-only controller
ControllerConf secondPinotOnlyControllerConfig = getDefaultControllerConfiguration();
@@ -166,27 +216,18 @@ public class PinotControllerModeTest extends ControllerTest {
TestUtils.waitForCondition(
aVoid -> secondPinotOnlyController.getHelixResourceManager().getHelixZkManager().isConnected(), TIMEOUT_IN_MS,
"Failed to start the second Pinot-only controller");
-
// There should still be only one MASTER instance for each partition
- checkInstanceState(helixAdmin, "MASTER");
-
- // Disable the lead controller resource, and there should be only one OFFLINE instance for each partition
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, false);
- checkInstanceState(helixAdmin, "OFFLINE");
-
- // Re-enable the lead controller resource, and there should be only one MASTER instance for each partition
- helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
- checkInstanceState(helixAdmin, "MASTER");
+ checkInstanceState(helixAdmin);
// Stop the second Pinot-only controller, and there should still be only one MASTER instance for each partition
secondPinotOnlyController.stop();
- checkInstanceState(helixAdmin, "MASTER");
+ checkInstanceState(helixAdmin);
// Stop the first Pinot-only controller, and there should be no partition in the external view
firstPinotOnlyController.stop();
TestUtils.waitForCondition(aVoid -> {
ExternalView leadControllerResourceExternalView =
- helixAdmin.getResourceExternalView(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
+ helixAdmin.getResourceExternalView(getHelixClusterName(), Helix.LEAD_CONTROLLER_RESOURCE_NAME);
return leadControllerResourceExternalView.getPartitionSet().isEmpty();
}, TIMEOUT_IN_MS, "Without live instance, there should be no partition in the external view");
@@ -194,12 +235,13 @@ public class PinotControllerModeTest extends ControllerTest {
helixOnlyController.stop();
}
- private void checkInstanceState(HelixAdmin helixAdmin, String expectedInstanceState) {
+ private void checkInstanceState(HelixAdmin helixAdmin) {
+ String expectedInstanceState = "MASTER";
TestUtils.waitForCondition(aVoid -> {
ExternalView leadControllerResourceExternalView =
- helixAdmin.getResourceExternalView(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME);
+ helixAdmin.getResourceExternalView(getHelixClusterName(), Helix.LEAD_CONTROLLER_RESOURCE_NAME);
Set<String> partitionSet = leadControllerResourceExternalView.getPartitionSet();
- if (partitionSet.size() != NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE) {
+ if (partitionSet.size() != Helix.NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE) {
return false;
}
for (String partition : partitionSet) {
@@ -212,6 +254,15 @@ public class PinotControllerModeTest extends ControllerTest {
}, TIMEOUT_IN_MS, "Failed to pick only one instance as: " + expectedInstanceState);
}
+ @AfterMethod
+ public void cleanUpCluster() {
+ ZkClient zkClient = new ZkClient(ZkStarter.DEFAULT_ZK_STR);
+ if (zkClient.exists("/" + getHelixClusterName())) {
+ zkClient.deleteRecursive("/" + getHelixClusterName());
+ }
+ zkClient.close();
+ }
+
@AfterClass
public void tearDown() {
stopZk();
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
index c235483..066040e 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
@@ -35,10 +35,12 @@ import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.testng.Assert;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -46,6 +48,7 @@ import static org.mockito.Mockito.when;
public class SegmentStatusCheckerTest {
private SegmentStatusChecker segmentStatusChecker;
private PinotHelixResourceManager helixResourceManager;
+ private LeadControllerManager leadControllerManager;
private MetricsRegistry metricsRegistry;
private ControllerMetrics controllerMetrics;
private ControllerConf config;
@@ -84,9 +87,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
segmentStatusChecker.start();
segmentStatusChecker.run();
Assert.assertEquals(
@@ -146,9 +154,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
segmentStatusChecker.start();
segmentStatusChecker.run();
Assert.assertEquals(
@@ -222,9 +235,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(0);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
segmentStatusChecker.start();
segmentStatusChecker.run();
Assert.assertEquals(
@@ -264,9 +282,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
segmentStatusChecker.start();
segmentStatusChecker.run();
Assert.assertEquals(controllerMetrics.getValueOfTableGauge(tableName, ControllerGauge.SEGMENTS_IN_ERROR_STATE), 0);
@@ -291,9 +314,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
segmentStatusChecker.start();
segmentStatusChecker.run();
Assert.assertEquals(controllerMetrics.getValueOfTableGauge(tableName, ControllerGauge.SEGMENTS_IN_ERROR_STATE),
@@ -349,9 +377,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
segmentStatusChecker.start();
segmentStatusChecker.run();
Assert.assertEquals(
@@ -390,9 +423,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
segmentStatusChecker.start();
segmentStatusChecker.run();
Assert.assertEquals(controllerMetrics.getValueOfTableGauge(tableName, ControllerGauge.SEGMENTS_IN_ERROR_STATE), 0);
@@ -429,9 +467,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
// verify state before test
Assert.assertEquals(controllerMetrics.getValueOfGlobalGauge(ControllerGauge.DISABLED_TABLE_COUNT), 0);
// update metrics
@@ -463,9 +506,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
// verify state before test
Assert.assertEquals(controllerMetrics.getValueOfGlobalGauge(ControllerGauge.DISABLED_TABLE_COUNT), 0);
// update metrics
@@ -508,9 +556,14 @@ public class SegmentStatusCheckerTest {
when(config.getStatusCheckerFrequencyInSeconds()).thenReturn(300);
when(config.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(300);
}
+ {
+ leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+ }
metricsRegistry = new MetricsRegistry();
controllerMetrics = new ControllerMetrics(metricsRegistry);
- segmentStatusChecker = new SegmentStatusChecker(helixResourceManager, config, controllerMetrics);
+ segmentStatusChecker =
+ new SegmentStatusChecker(helixResourceManager, leadControllerManager, config, controllerMetrics);
segmentStatusChecker.start();
segmentStatusChecker.run();
Assert.assertEquals(controllerMetrics.getValueOfTableGauge(tableName, ControllerGauge.SEGMENTS_IN_ERROR_STATE),
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java
index cdf3d9e..d8479dc 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerTest.java
@@ -38,6 +38,7 @@ import org.apache.helix.model.ClusterConfig;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.model.MasterSlaveSMD;
import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.config.TagNameUtils;
@@ -50,6 +51,7 @@ import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.CommonConstants.Helix.TableType;
import org.apache.pinot.common.utils.TenantRole;
+import org.apache.pinot.common.utils.helix.LeadControllerUtils;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.util.TestUtils;
@@ -61,7 +63,6 @@ import org.testng.annotations.Test;
import static org.apache.pinot.common.utils.CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME;
import static org.apache.pinot.common.utils.CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_REPLICA_COUNT;
import static org.apache.pinot.common.utils.CommonConstants.Helix.NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE;
-import static org.apache.pinot.common.utils.CommonConstants.Helix.PREFIX_OF_CONTROLLER_INSTANCE;
import static org.apache.pinot.controller.helix.core.PinotHelixResourceManager.InvalidTableConfigException;
@@ -94,7 +95,8 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
.build();
_helixResourceManager.createServerTenant(serverTenant);
- _helixAdmin.enableResource(getHelixClusterName(), LEAD_CONTROLLER_RESOURCE_NAME, true);
+ // Enable lead controller resource
+ enableResourceConfigForLeadControllerResource(true);
}
@Test
@@ -421,8 +423,9 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
for (String partition : leadControllerResourceExternalView.getPartitionSet()) {
Map<String, String> stateMap = leadControllerResourceExternalView.getStateMap(partition);
Map.Entry<String, String> entry = stateMap.entrySet().iterator().next();
- boolean result = (PREFIX_OF_CONTROLLER_INSTANCE + LOCAL_HOST + "_" + _controllerPort).equals(entry.getKey());
- result &= "MASTER".equals(entry.getValue());
+ boolean result =
+ (LeadControllerUtils.generateParticipantInstanceId(LOCAL_HOST, _controllerPort)).equals(entry.getKey());
+ result &= MasterSlaveSMD.States.MASTER.name().equals(entry.getValue());
if (!result) {
return false;
}
@@ -438,19 +441,19 @@ public class PinotHelixResourceManagerTest extends ControllerTest {
List<String> instanceNames = new ArrayList<>(nInstances);
List<Integer> ports = new ArrayList<>(nInstances);
for (int i = 0; i < nInstances; i++) {
- instanceNames.add(PREFIX_OF_CONTROLLER_INSTANCE + LOCAL_HOST + "_" + i);
+ instanceNames.add(LeadControllerUtils.generateParticipantInstanceId(LOCAL_HOST, i));
ports.add(i);
}
List<String> partitions = new ArrayList<>(NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE);
for (int i = 0; i < NUMBER_OF_PARTITIONS_IN_LEAD_CONTROLLER_RESOURCE; i++) {
- partitions.add(LEAD_CONTROLLER_RESOURCE_NAME + "_" + i);
+ partitions.add(LeadControllerUtils.generatePartitionName(i));
}
LinkedHashMap<String, Integer> states = new LinkedHashMap<>(2);
- states.put("OFFLINE", 0);
- states.put("SLAVE", LEAD_CONTROLLER_RESOURCE_REPLICA_COUNT - 1);
- states.put("MASTER", 1);
+ states.put(MasterSlaveSMD.States.OFFLINE.name(), 0);
+ states.put(MasterSlaveSMD.States.SLAVE.name(), LEAD_CONTROLLER_RESOURCE_REPLICA_COUNT - 1);
+ states.put(MasterSlaveSMD.States.MASTER.name(), 1);
CrushEdRebalanceStrategy crushEdRebalanceStrategy = new CrushEdRebalanceStrategy();
crushEdRebalanceStrategy.init(LEAD_CONTROLLER_RESOURCE_NAME, partitions, states, Integer.MAX_VALUE);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTaskTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTaskTest.java
index a928922..dcd5469 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTaskTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/periodictask/ControllerPeriodicTaskTest.java
@@ -27,10 +27,12 @@ import java.util.stream.IntStream;
import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -43,6 +45,7 @@ public class ControllerPeriodicTaskTest {
private final ControllerConf _controllerConf = new ControllerConf();
private final PinotHelixResourceManager _resourceManager = mock(PinotHelixResourceManager.class);
+ private final LeadControllerManager _leadControllerManager = mock(LeadControllerManager.class);
private final ControllerMetrics _controllerMetrics = new ControllerMetrics(new MetricsRegistry());
private final AtomicBoolean _startTaskCalled = new AtomicBoolean();
private final AtomicBoolean _stopTaskCalled = new AtomicBoolean();
@@ -52,7 +55,8 @@ public class ControllerPeriodicTaskTest {
private static final String TASK_NAME = "TestTask";
private final ControllerPeriodicTask _task = new ControllerPeriodicTask<Void>(TASK_NAME, RUN_FREQUENCY_IN_SECONDS,
- _controllerConf.getPeriodicTaskInitialDelayInSeconds(), _resourceManager, _controllerMetrics) {
+ _controllerConf.getPeriodicTaskInitialDelayInSeconds(), _resourceManager, _leadControllerManager,
+ _controllerMetrics) {
@Override
protected void setUpTask() {
@@ -81,6 +85,7 @@ public class ControllerPeriodicTaskTest {
List<String> tables = new ArrayList<>(_numTables);
IntStream.range(0, _numTables).forEach(i -> tables.add("table_" + i + " _OFFLINE"));
when(_resourceManager.getAllTables()).thenReturn(tables);
+ when(_leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
}
private void resetState() {
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
index 04e2b7b..3396066 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
@@ -56,7 +56,7 @@ import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.ControllerLeadershipManager;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.api.resources.LLCSegmentCompletionHandlers;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.PinotTableIdealStateBuilder;
@@ -66,7 +66,6 @@ import org.apache.pinot.core.indexsegment.generator.SegmentVersion;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
import org.apache.pinot.core.realtime.stream.OffsetCriteria;
import org.apache.pinot.core.realtime.stream.StreamConfig;
-import org.apache.pinot.core.realtime.stream.StreamConfigProperties;
import org.apache.pinot.core.segment.index.SegmentMetadataImpl;
import org.apache.pinot.filesystem.PinotFSFactory;
import org.apache.zookeeper.data.Stat;
@@ -1323,7 +1322,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
protected FakePinotLLCRealtimeSegmentManager(PinotHelixResourceManager pinotHelixResourceManager,
List<String> existingLLCSegments, ControllerMetrics controllerMetrics) {
super(pinotHelixResourceManager, CONTROLLER_CONF, controllerMetrics,
- new ControllerLeadershipManager(pinotHelixResourceManager.getHelixZkManager(), controllerMetrics));
+ new LeadControllerManager(pinotHelixResourceManager.getHelixZkManager(), controllerMetrics));
try {
TableConfigCache mockCache = mock(TableConfigCache.class);
@@ -1513,7 +1512,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
}
@Override
- protected boolean isLeader() {
+ protected boolean isLeader(String tableName) {
return IS_LEADER;
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionTest.java
index 51ddf67..7dd2944 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionTest.java
@@ -31,7 +31,7 @@ import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.ControllerLeadershipManager;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
import org.apache.zookeeper.data.Stat;
@@ -98,9 +98,9 @@ public class SegmentCompletionTest {
// but segment metadata is fine in zk
private void replaceSegmentCompletionManager()
throws Exception {
- long oldSecs = segmentCompletionMgr._secconds;
+ long oldSecs = segmentCompletionMgr._seconds;
segmentCompletionMgr = new MockSegmentCompletionManager(segmentManager, true, true);
- segmentCompletionMgr._secconds = oldSecs;
+ segmentCompletionMgr._seconds = oldSecs;
Field fsmMapField = SegmentCompletionManager.class.getDeclaredField("_fsmMap");
fsmMapField.setAccessible(true);
fsmMap = (Map<String, Object>) fsmMapField.get(segmentCompletionMgr);
@@ -114,17 +114,17 @@ public class SegmentCompletionTest {
final String reason = "IAmLazy";
// s1 sends offset of 20, gets HOLD at t = 5s;
- segmentCompletionMgr._secconds = 5;
+ segmentCompletionMgr._seconds = 5;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params =
new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr).withReason(reason);
response = segmentCompletionMgr.segmentStoppedConsuming(params);
@@ -135,27 +135,27 @@ public class SegmentCompletionTest {
segmentManager._stoppedInstance = null;
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
// s3 comes back with new caught up offset, it should get a HOLD, since commit is not done yet.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 executes a successful commit
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr
.segmentCommitEnd(params, true, false, CommittingSegmentDescriptor.fromSegmentCompletionReqParams(params));
@@ -165,7 +165,7 @@ public class SegmentCompletionTest {
Assert.assertFalse(fsmMap.containsKey(segmentNameStr));
// Now if s3 or s1 come back, they are asked to keep the segment they have.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.KEEP);
@@ -181,7 +181,7 @@ public class SegmentCompletionTest {
Request.Params params;
final String reason = "IAmLazy";
// s1 stops consuming at t = 5;
- segmentCompletionMgr._secconds = 5;
+ segmentCompletionMgr._seconds = 5;
params =
new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr).withReason(reason);
response = segmentCompletionMgr.segmentStoppedConsuming(params);
@@ -192,14 +192,14 @@ public class SegmentCompletionTest {
segmentManager._stoppedInstance = null;
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40, s2 should have been decided as the winner now
// since we are never expecting to hear back from s1
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
@@ -208,24 +208,24 @@ public class SegmentCompletionTest {
// s3 happens to come back (after catchup to s2offset) before s2 should get a hold since s2 has been decided as
// the winner.
// TODO Can optimize here since s2 is not notified yet.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
Assert.assertEquals(response.getOffset(), s2Offset);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
// s2 executes a successful commit
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr
.segmentCommitEnd(params, true, false, CommittingSegmentDescriptor.fromSegmentCompletionReqParams(params));
@@ -235,7 +235,7 @@ public class SegmentCompletionTest {
Assert.assertFalse(fsmMap.containsKey(segmentNameStr));
// Now if s3 or s1 come back, they are asked to keep the segment they have.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.KEEP);
@@ -250,7 +250,7 @@ public class SegmentCompletionTest {
throws Exception {
SegmentCompletionProtocol.Response response;
Request.Params params;
- segmentCompletionMgr._secconds = 5;
+ segmentCompletionMgr._seconds = 5;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withReason("some reason");
@@ -284,7 +284,7 @@ public class SegmentCompletionTest {
SegmentCompletionProtocol.Response response;
Request.Params params;
- segmentCompletionMgr._secconds = 10;
+ segmentCompletionMgr._seconds = 10;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.FAILED);
@@ -297,46 +297,46 @@ public class SegmentCompletionTest {
SegmentCompletionProtocol.Response response;
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
- segmentCompletionMgr._secconds = 5L;
+ segmentCompletionMgr._seconds = 5L;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
// TODO: Verify controller asked to do a split commit
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
// s3 comes back with new caught up offset, it should get a HOLD, since commit is not done yet.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 executes a successful commit start
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
// s2's file does not successfully commit because MockPinotLLCRealtimeSegmentManager.commitSegmentFile() returns
// false when detecting SegmentLocation == "doNotCommitMe";
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withSegmentLocation("doNotCommitMe");
response = segmentCompletionMgr
@@ -347,30 +347,30 @@ public class SegmentCompletionTest {
Assert.assertFalse(fsmMap.containsKey(segmentNameStr));
// Now s1 comes back; it is asked to hold.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// Now s3 comes back; it is asked to commit
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.COMMIT);
// Now s2 comes back; it is asked to hold
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.HOLD);
// s3 executes a successful commit start
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
// s3's file successfully commits
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withSegmentLocation("location");
response = segmentCompletionMgr
@@ -385,45 +385,45 @@ public class SegmentCompletionTest {
SegmentCompletionProtocol.Response response;
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
- segmentCompletionMgr._secconds = startTime;
+ segmentCompletionMgr._seconds = startTime;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
// TODO: Verify controller asked to do a split commit
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
// s3 comes back with new caught up offset, it should get a HOLD, since commit is not done yet.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 executes a successful commit
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withSegmentLocation("location");
response = segmentCompletionMgr
@@ -434,7 +434,7 @@ public class SegmentCompletionTest {
Assert.assertFalse(fsmMap.containsKey(segmentNameStr));
// Now if s3 or s1 come back, they are asked to keep the segment they have.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.KEEP);
@@ -450,30 +450,30 @@ public class SegmentCompletionTest {
SegmentCompletionProtocol.Response response;
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
- segmentCompletionMgr._secconds = 5L;
+ segmentCompletionMgr._seconds = 5L;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 20, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 20, gets commit
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.COMMIT);
Assert.assertEquals(response.getOffset(), s1Offset);
// s3 sends a commit start with 20
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
// s3 comes back to try to commit with a different offset
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr)
.withSegmentLocation("location");
response = segmentCompletionMgr
@@ -484,7 +484,7 @@ public class SegmentCompletionTest {
Assert.assertFalse(fsmMap.containsKey(segmentNameStr));
// Now if s2 or s1 come back, they are asked to hold.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
@@ -498,43 +498,43 @@ public class SegmentCompletionTest {
SegmentCompletionProtocol.Response response;
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
- segmentCompletionMgr._secconds = startTime;
+ segmentCompletionMgr._seconds = startTime;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
// s3 comes back with new caught up offset, it should get a HOLD, since commit is not done yet.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 executes a successful commit
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr
.segmentCommitEnd(params, true, false, CommittingSegmentDescriptor.fromSegmentCompletionReqParams(params));
@@ -544,7 +544,7 @@ public class SegmentCompletionTest {
Assert.assertFalse(fsmMap.containsKey(segmentNameStr));
// Now if s3 or s1 come back, they are asked to keep the segment they have.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.KEEP);
@@ -560,7 +560,7 @@ public class SegmentCompletionTest {
SegmentCompletionProtocol.Response response;
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
- segmentCompletionMgr._secconds = 5L;
+ segmentCompletionMgr._seconds = 5L;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr)
.withReason("rowLimit");
response = segmentCompletionMgr.segmentConsumed(params);
@@ -572,7 +572,7 @@ public class SegmentCompletionTest {
throws Exception {
SegmentCompletionProtocol.Response response;
Request.Params params;
- segmentCompletionMgr._secconds = 10L;
+ segmentCompletionMgr._seconds = 10L;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr)
.withReason(SegmentCompletionProtocol.REASON_TIME_LIMIT);
response = segmentCompletionMgr.segmentConsumed(params);
@@ -584,30 +584,30 @@ public class SegmentCompletionTest {
throws Exception {
SegmentCompletionProtocol.Response response;
Request.Params params;
- segmentCompletionMgr._secconds = 10L;
+ segmentCompletionMgr._seconds = 10L;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr)
.withReason(SegmentCompletionProtocol.REASON_ROW_LIMIT);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.COMMIT);
// S2 comes with the same offset as S1
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s1Offset).withSegmentName(segmentNameStr)
.withReason(SegmentCompletionProtocol.REASON_ROW_LIMIT);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.HOLD);
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
// S3 comes with a different offset and without row limit. we ask it to hold even though it is higher.
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr)
.withReason(SegmentCompletionProtocol.REASON_TIME_LIMIT);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.HOLD);
// S1 comes back to commit the segment
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr
.segmentCommitEnd(params, true, false, CommittingSegmentDescriptor.fromSegmentCompletionReqParams(params));
@@ -643,37 +643,37 @@ public class SegmentCompletionTest {
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
final int startTimeSecs = 5;
- segmentCompletionMgr._secconds = startTimeSecs;
+ segmentCompletionMgr._seconds = startTimeSecs;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// Now s1 comes back again, and is asked to hold
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += SegmentCompletionProtocol.MAX_HOLD_TIME_MS / 1000;
+ segmentCompletionMgr._seconds += SegmentCompletionProtocol.MAX_HOLD_TIME_MS / 1000;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
// Now s3 comes up with a better offset, but we ask it to hold, since the committer has not committed yet.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset + 10).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 commits.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withSegmentLocation("location");
response = segmentCompletionMgr.segmentCommitEnd(params, true, isSplitCommit,
@@ -683,7 +683,7 @@ public class SegmentCompletionTest {
Assert.assertFalse(fsmMap.containsKey(segmentNameStr));
// Now s3 comes back to get a discard.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset + 10).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.DISCARD);
@@ -699,30 +699,30 @@ public class SegmentCompletionTest {
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
final int startTimeSecs = 5;
- segmentCompletionMgr._secconds = startTimeSecs;
+ segmentCompletionMgr._seconds = startTimeSecs;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 is asked to catch up.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
// All servers are dead
- segmentCompletionMgr._secconds += 3600;
+ segmentCompletionMgr._seconds += 3600;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
@@ -734,7 +734,7 @@ public class SegmentCompletionTest {
Assert.assertTrue(fsmMap.containsKey(segmentNameStr));
// Now s2 is asked to commit because the max time to pick committer has passed.
- segmentCompletionMgr._secconds += 4;
+ segmentCompletionMgr._seconds += 4;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.COMMIT);
@@ -748,29 +748,29 @@ public class SegmentCompletionTest {
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
final int startTimeSecs = 5;
- segmentCompletionMgr._secconds = startTimeSecs;
+ segmentCompletionMgr._seconds = startTimeSecs;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 is asked to hold.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
// Time passes, s2 never comes back.
- segmentCompletionMgr._secconds += SegmentCompletionProtocol.MAX_HOLD_TIME_MS / 1000;
+ segmentCompletionMgr._seconds += SegmentCompletionProtocol.MAX_HOLD_TIME_MS / 1000;
// But since s1 and s3 are in HOLDING state, they should come back again. s1 is asked to catchup
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
@@ -778,7 +778,7 @@ public class SegmentCompletionTest {
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
// Max time to commit passes
- segmentCompletionMgr._secconds +=
+ segmentCompletionMgr._seconds +=
SegmentCompletionProtocol.getMaxSegmentCommitTimeMs() * SegmentCompletionProtocol.MAX_HOLD_TIME_MS / 1000;
// s1 comes back with the updated offset, since it was asked to catch up.
@@ -790,19 +790,19 @@ public class SegmentCompletionTest {
Assert.assertFalse(fsmMap.containsKey(segmentNameStr));
// s1 comes back again, a new FSM created
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 comes back
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// And winner chosen when the last one does not come back at all
- segmentCompletionMgr._secconds += 5;
+ segmentCompletionMgr._seconds += 5;
params = new Request.Params().withInstanceId(s3).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
@@ -820,28 +820,28 @@ public class SegmentCompletionTest {
final long startTime = 1509242466;
final String tableName = new LLCSegmentName(segmentNameStr).getTableName();
Assert.assertNull(commitTimeMap.get(tableName));
- segmentCompletionMgr._secconds = startTime;
+ segmentCompletionMgr._seconds = startTime;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
@@ -849,7 +849,7 @@ public class SegmentCompletionTest {
Assert.assertTrue(commitTimeSec > 0);
// Fast forward to one second before commit time, and send a lease renewal request for 20s
- segmentCompletionMgr._secconds = startTime + commitTimeSec - 1;
+ segmentCompletionMgr._seconds = startTime + commitTimeSec - 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withExtraTimeSec(20);
response = segmentCompletionMgr.extendBuildTime(params);
@@ -857,7 +857,7 @@ public class SegmentCompletionTest {
Assert.assertTrue((fsmMap.containsKey(segmentNameStr)));
// Another lease extension in 19s.
- segmentCompletionMgr._secconds += 19;
+ segmentCompletionMgr._seconds += 19;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withExtraTimeSec(20);
response = segmentCompletionMgr.extendBuildTime(params);
@@ -865,13 +865,13 @@ public class SegmentCompletionTest {
Assert.assertTrue((fsmMap.containsKey(segmentNameStr)));
// Commit in 15s
- segmentCompletionMgr._secconds += 15;
+ segmentCompletionMgr._seconds += 15;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_CONTINUE);
- long commitTimeMs = (segmentCompletionMgr._secconds - startTime) * 1000;
+ long commitTimeMs = (segmentCompletionMgr._seconds - startTime) * 1000;
Assert.assertEquals(commitTimeMap.get(tableName).longValue(), commitTimeMs);
- segmentCompletionMgr._secconds += 55;
+ segmentCompletionMgr._seconds += 55;
response = segmentCompletionMgr
.segmentCommitEnd(params, true, false, CommittingSegmentDescriptor.fromSegmentCompletionReqParams(params));
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT_SUCCESS);
@@ -887,28 +887,28 @@ public class SegmentCompletionTest {
final String tableName = new LLCSegmentName(segmentNameStr).getTableName();
// s1 sends offset of 20, gets HOLD at t = 5s;
final long startTime = 5;
- segmentCompletionMgr._secconds = startTime;
+ segmentCompletionMgr._seconds = startTime;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
@@ -916,7 +916,7 @@ public class SegmentCompletionTest {
Assert.assertTrue(commitTimeSec > 0);
// Fast forward to one second before commit time, and send a lease renewal request for 20s
- segmentCompletionMgr._secconds = startTime + commitTimeSec - 1;
+ segmentCompletionMgr._seconds = startTime + commitTimeSec - 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withExtraTimeSec(20);
response = segmentCompletionMgr.extendBuildTime(params);
@@ -924,7 +924,7 @@ public class SegmentCompletionTest {
Assert.assertTrue((fsmMap.containsKey(segmentNameStr)));
// Come back too late.
- segmentCompletionMgr._secconds += 25;
+ segmentCompletionMgr._seconds += 25;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.HOLD);
@@ -940,28 +940,28 @@ public class SegmentCompletionTest {
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
final long startTime = 5;
- segmentCompletionMgr._secconds = startTime;
+ segmentCompletionMgr._seconds = startTime;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
@@ -969,7 +969,7 @@ public class SegmentCompletionTest {
Assert.assertTrue(commitTimeSec > 0);
// Fast forward to one second before commit time, and send a lease renewal request for 20s
- segmentCompletionMgr._secconds = startTime + commitTimeSec - 1;
+ segmentCompletionMgr._seconds = startTime + commitTimeSec - 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withExtraTimeSec(20);
response = segmentCompletionMgr.extendBuildTime(params);
@@ -978,14 +978,14 @@ public class SegmentCompletionTest {
final int leaseTimeSec = 20;
// Lease will not be granted if the time taken so far plus lease time exceeds the max allowabale.
- while (segmentCompletionMgr._secconds + leaseTimeSec <= startTime + SegmentCompletionManager
+ while (segmentCompletionMgr._seconds + leaseTimeSec <= startTime + SegmentCompletionManager
.getMaxCommitTimeForAllSegmentsSeconds()) {
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr)
.withExtraTimeSec(leaseTimeSec);
response = segmentCompletionMgr.extendBuildTime(params);
Assert.assertEquals(response.getStatus(), ControllerResponseStatus.PROCESSED);
Assert.assertTrue((fsmMap.containsKey(segmentNameStr)));
- segmentCompletionMgr._secconds += leaseTimeSec;
+ segmentCompletionMgr._seconds += leaseTimeSec;
}
// Now the lease request should fail.
@@ -1002,28 +1002,28 @@ public class SegmentCompletionTest {
SegmentCompletionProtocol.Response response;
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
- segmentCompletionMgr._secconds = 5;
+ segmentCompletionMgr._seconds = 5;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
@@ -1043,7 +1043,7 @@ public class SegmentCompletionTest {
// s2 has no idea the controller failed, so it comes back with a commit,but the controller asks it to hold,
// (essentially a commit failure)
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertTrue(response.getStatus().equals(SegmentCompletionProtocol.ControllerResponseStatus.HOLD));
@@ -1062,28 +1062,28 @@ public class SegmentCompletionTest {
SegmentCompletionProtocol.Response response;
Request.Params params;
// s1 sends offset of 20, gets HOLD at t = 5s;
- segmentCompletionMgr._secconds = 5;
+ segmentCompletionMgr._seconds = 5;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s2 sends offset of 40, gets HOLD
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.HOLD);
// s3 sends offset of 30, gets catchup to 40
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s3).withOffset(s3Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
Assert.assertEquals(response.getOffset(), s2Offset);
// Now s1 comes back, and is asked to catchup.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s1).withOffset(s1Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.CATCH_UP);
// s2 is asked to commit.
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentConsumed(params);
Assert.assertEquals(response.getStatus(), SegmentCompletionProtocol.ControllerResponseStatus.COMMIT);
@@ -1103,7 +1103,7 @@ public class SegmentCompletionTest {
// s2 has no idea the controller failed, so it comes back with a commit,but the controller asks it to hold,
// (essentially a commit failure)
- segmentCompletionMgr._secconds += 1;
+ segmentCompletionMgr._seconds += 1;
params = new Request.Params().withInstanceId(s2).withOffset(s2Offset).withSegmentName(segmentNameStr);
response = segmentCompletionMgr.segmentCommitStart(params);
Assert.assertTrue(response.getStatus().equals(SegmentCompletionProtocol.ControllerResponseStatus.HOLD));
@@ -1154,7 +1154,7 @@ public class SegmentCompletionTest {
protected MockPinotLLCRealtimeSegmentManager(PinotHelixResourceManager pinotHelixResourceManager,
ControllerMetrics controllerMetrics) {
super(pinotHelixResourceManager, CONTROLLER_CONF, controllerMetrics,
- new ControllerLeadershipManager(pinotHelixResourceManager.getHelixZkManager(), controllerMetrics));
+ new LeadControllerManager(pinotHelixResourceManager.getHelixZkManager(), controllerMetrics));
}
@Override
@@ -1195,7 +1195,7 @@ public class SegmentCompletionTest {
}
public static class MockSegmentCompletionManager extends SegmentCompletionManager {
- public long _secconds;
+ public long _seconds;
private boolean _isLeader;
protected MockSegmentCompletionManager(PinotLLCRealtimeSegmentManager segmentManager, boolean isLeader,
@@ -1210,19 +1210,18 @@ public class SegmentCompletionTest {
protected MockSegmentCompletionManager(HelixManager helixManager, PinotLLCRealtimeSegmentManager segmentManager,
boolean isLeader, ControllerMetrics controllerMetrics) {
- super(helixManager, segmentManager, controllerMetrics,
- new ControllerLeadershipManager(helixManager, controllerMetrics),
+ super(helixManager, segmentManager, controllerMetrics, new LeadControllerManager(helixManager, controllerMetrics),
SegmentCompletionProtocol.getDefaultMaxSegmentCommitTimeSeconds());
_isLeader = isLeader;
}
@Override
protected long getCurrentTimeMs() {
- return _secconds * 1000L;
+ return _seconds * 1000L;
}
@Override
- protected boolean isLeader() {
+ protected boolean isLeader(String tableName) {
return _isLeader;
}
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/relocation/RealtimeSegmentRelocatorTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/relocation/RealtimeSegmentRelocatorTest.java
index 8c5c8ec..9316062 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/relocation/RealtimeSegmentRelocatorTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/relocation/RealtimeSegmentRelocatorTest.java
@@ -34,12 +34,14 @@ import org.apache.helix.model.builder.CustomModeISBuilder;
import org.apache.pinot.common.config.RealtimeTagConfig;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.PinotHelixSegmentOnlineOfflineStateModelGenerator;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -48,6 +50,7 @@ public class RealtimeSegmentRelocatorTest {
private TestRealtimeSegmentRelocator _realtimeSegmentRelocator;
private HelixManager _mockHelixManager;
+ private LeadControllerManager _leadControllerManager;
private String[] serverNames;
private String[] consumingServerNames;
@@ -69,10 +72,13 @@ public class RealtimeSegmentRelocatorTest {
PinotHelixResourceManager mockPinotHelixResourceManager = mock(PinotHelixResourceManager.class);
_mockHelixManager = mock(HelixManager.class);
when(mockPinotHelixResourceManager.getHelixZkManager()).thenReturn(_mockHelixManager);
+ LeadControllerManager mockLeadControllerManager = mock(LeadControllerManager.class);
+ when(mockLeadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
ControllerConf controllerConfig = new ControllerConf();
ControllerMetrics controllerMetrics = new ControllerMetrics(new MetricsRegistry());
_realtimeSegmentRelocator =
- new TestRealtimeSegmentRelocator(mockPinotHelixResourceManager, controllerConfig, controllerMetrics);
+ new TestRealtimeSegmentRelocator(mockPinotHelixResourceManager, mockLeadControllerManager, controllerConfig,
+ controllerMetrics);
final int maxInstances = 20;
serverNames = new String[maxInstances];
@@ -268,9 +274,9 @@ public class RealtimeSegmentRelocatorTest {
private Map<String, List<String>> tagToInstances;
- public TestRealtimeSegmentRelocator(PinotHelixResourceManager pinotHelixResourceManager, ControllerConf config,
- ControllerMetrics controllerMetrics) {
- super(pinotHelixResourceManager, config, controllerMetrics);
+ public TestRealtimeSegmentRelocator(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics) {
+ super(pinotHelixResourceManager, leadControllerManager, config, controllerMetrics);
tagToInstances = new HashedMap();
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
index bb20ca3..96915ff 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
@@ -35,6 +35,7 @@ import org.apache.pinot.common.segment.SegmentMetadata;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.PinotTableIdealStateBuilder;
import org.apache.pinot.controller.helix.core.SegmentDeletionManager;
@@ -84,6 +85,9 @@ public class RetentionManagerTest {
PinotHelixResourceManager pinotHelixResourceManager = mock(PinotHelixResourceManager.class);
setupPinotHelixResourceManager(tableConfig, removedSegments, pinotHelixResourceManager);
+ LeadControllerManager leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+
when(pinotHelixResourceManager.getTableConfig(OFFLINE_TABLE_NAME)).thenReturn(tableConfig);
when(pinotHelixResourceManager.getOfflineSegmentMetadata(OFFLINE_TABLE_NAME)).thenReturn(metadataList);
@@ -91,7 +95,8 @@ public class RetentionManagerTest {
ControllerMetrics controllerMetrics = new ControllerMetrics(new MetricsRegistry());
conf.setRetentionControllerFrequencyInSeconds(0);
conf.setDeletedSegmentsRetentionInDays(0);
- RetentionManager retentionManager = new RetentionManager(pinotHelixResourceManager, conf, controllerMetrics);
+ RetentionManager retentionManager =
+ new RetentionManager(pinotHelixResourceManager, leadControllerManager, conf, controllerMetrics);
retentionManager.start();
retentionManager.run();
@@ -210,11 +215,14 @@ public class RetentionManagerTest {
setupSegmentMetadata(tableConfig, now, initialNumSegments, removedSegments);
setupPinotHelixResourceManager(tableConfig, removedSegments, pinotHelixResourceManager);
+ LeadControllerManager leadControllerManager = mock(LeadControllerManager.class);
+ when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
+
ControllerConf conf = new ControllerConf();
ControllerMetrics controllerMetrics = new ControllerMetrics(new MetricsRegistry());
conf.setRetentionControllerFrequencyInSeconds(0);
conf.setDeletedSegmentsRetentionInDays(0);
- RetentionManager retentionManager = new RetentionManager(pinotHelixResourceManager, conf, controllerMetrics);
+ RetentionManager retentionManager = new RetentionManager(pinotHelixResourceManager, leadControllerManager, conf, controllerMetrics);
retentionManager.start();
retentionManager.run();
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/validation/StorageQuotaCheckerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/validation/StorageQuotaCheckerTest.java
index e84a947..a4a2a3d 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/validation/StorageQuotaCheckerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/validation/StorageQuotaCheckerTest.java
@@ -30,8 +30,6 @@ import org.apache.pinot.common.config.TableConfig;
import org.apache.pinot.common.exception.InvalidConfigException;
import org.apache.pinot.common.metrics.ControllerGauge;
import org.apache.pinot.common.metrics.ControllerMetrics;
-import org.apache.pinot.controller.ControllerLeadershipManager;
-import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.util.TableSizeReader;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -48,8 +46,7 @@ public class StorageQuotaCheckerTest {
private TableSizeReader _tableSizeReader;
private TableConfig _tableConfig;
private ControllerMetrics _controllerMetrics;
- private PinotHelixResourceManager _pinotHelixResourceManager;
- private ControllerLeadershipManager _controllerLeadershipManager;
+ private boolean _isLeaderForTable;
private QuotaConfig _quotaConfig;
private SegmentsValidationAndRetentionConfig _validationConfig;
private static final File TEST_DIR = new File(StorageQuotaCheckerTest.class.getName());
@@ -61,8 +58,7 @@ public class StorageQuotaCheckerTest {
_quotaConfig = mock(QuotaConfig.class);
_controllerMetrics = new ControllerMetrics(new MetricsRegistry());
_validationConfig = mock(SegmentsValidationAndRetentionConfig.class);
- _pinotHelixResourceManager = mock(PinotHelixResourceManager.class);
- _controllerLeadershipManager = mock(ControllerLeadershipManager.class);
+ _isLeaderForTable = true;
when(_tableConfig.getValidationConfig()).thenReturn(_validationConfig);
when(_validationConfig.getReplicationNumber()).thenReturn(2);
TEST_DIR.mkdirs();
@@ -77,11 +73,9 @@ public class StorageQuotaCheckerTest {
public void testNoQuota()
throws InvalidConfigException {
StorageQuotaChecker checker =
- new MockStorageQuotaChecker(_tableConfig, _tableSizeReader, _controllerMetrics, _pinotHelixResourceManager,
- _controllerLeadershipManager);
+ new MockStorageQuotaChecker(_tableConfig, _tableSizeReader, _controllerMetrics, _isLeaderForTable);
when(_tableConfig.getQuotaConfig()).thenReturn(null);
- StorageQuotaChecker.QuotaCheckerResponse res =
- checker.isSegmentStorageWithinQuota(TEST_DIR, "segment", 1000);
+ StorageQuotaChecker.QuotaCheckerResponse res = checker.isSegmentStorageWithinQuota(TEST_DIR, "segment", 1000);
Assert.assertTrue(res.isSegmentWithinQuota);
}
@@ -89,12 +83,10 @@ public class StorageQuotaCheckerTest {
public void testNoStorageQuotaConfig()
throws InvalidConfigException {
StorageQuotaChecker checker =
- new MockStorageQuotaChecker(_tableConfig, _tableSizeReader, _controllerMetrics, _pinotHelixResourceManager,
- _controllerLeadershipManager);
+ new MockStorageQuotaChecker(_tableConfig, _tableSizeReader, _controllerMetrics, _isLeaderForTable);
when(_tableConfig.getQuotaConfig()).thenReturn(_quotaConfig);
when(_quotaConfig.storageSizeBytes()).thenReturn(-1L);
- StorageQuotaChecker.QuotaCheckerResponse res =
- checker.isSegmentStorageWithinQuota(TEST_DIR, "segment", 1000);
+ StorageQuotaChecker.QuotaCheckerResponse res = checker.isSegmentStorageWithinQuota(TEST_DIR, "segment", 1000);
Assert.assertTrue(res.isSegmentWithinQuota);
}
@@ -133,10 +125,8 @@ public class StorageQuotaCheckerTest {
when(_quotaConfig.storageSizeBytes()).thenReturn(3000L);
when(_quotaConfig.getStorage()).thenReturn("3K");
StorageQuotaChecker checker =
- new MockStorageQuotaChecker(_tableConfig, _tableSizeReader, _controllerMetrics, _pinotHelixResourceManager,
- _controllerLeadershipManager);
- StorageQuotaChecker.QuotaCheckerResponse response =
- checker.isSegmentStorageWithinQuota(TEST_DIR, "segment1", 1000);
+ new MockStorageQuotaChecker(_tableConfig, _tableSizeReader, _controllerMetrics, _isLeaderForTable);
+ StorageQuotaChecker.QuotaCheckerResponse response = checker.isSegmentStorageWithinQuota(TEST_DIR, "segment1", 1000);
Assert.assertTrue(response.isSegmentWithinQuota);
Assert.assertEquals(
_controllerMetrics.getValueOfTableGauge(tableName, ControllerGauge.TABLE_STORAGE_QUOTA_UTILIZATION), 80L);
@@ -183,14 +173,8 @@ public class StorageQuotaCheckerTest {
private class MockStorageQuotaChecker extends StorageQuotaChecker {
public MockStorageQuotaChecker(TableConfig tableConfig, TableSizeReader tableSizeReader,
- ControllerMetrics controllerMetrics, PinotHelixResourceManager pinotHelixResourceManager,
- ControllerLeadershipManager controllerLeadershipManager) {
- super(tableConfig, tableSizeReader, controllerMetrics, pinotHelixResourceManager, controllerLeadershipManager);
- }
-
- @Override
- protected boolean isLeader() {
- return true;
+ ControllerMetrics controllerMetrics, boolean isLeaderForTable) {
+ super(tableConfig, tableSizeReader, controllerMetrics, isLeaderForTable);
}
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
index 04cfec7..fb1a5b3 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
@@ -1063,6 +1063,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
_segBuildSemaphore = realtimeTableDataManager.getSegmentBuildSemaphore();
_segmentZKMetadata = (LLCRealtimeSegmentZKMetadata) segmentZKMetadata;
_tableConfig = tableConfig;
+ _tableNameWithType = _tableConfig.getTableName();
_realtimeTableDataManager = realtimeTableDataManager;
_resourceDataDir = resourceDataDir;
_indexLoadingConfig = indexLoadingConfig;
@@ -1071,7 +1072,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
_segmentVersion = indexLoadingConfig.getSegmentVersion();
_instanceId = _realtimeTableDataManager.getServerInstance();
_leaseExtender = SegmentBuildTimeLeaseExtender.getLeaseExtender(_instanceId);
- _protocolHandler = new ServerSegmentCompletionProtocolHandler(_serverMetrics);
+ _protocolHandler = new ServerSegmentCompletionProtocolHandler(_serverMetrics, _tableNameWithType);
// TODO Validate configs
IndexingConfig indexingConfig = _tableConfig.getIndexingConfig();
@@ -1081,7 +1082,6 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
_segmentNameStr = _segmentZKMetadata.getSegmentName();
_segmentName = new LLCSegmentName(_segmentNameStr);
_streamPartitionId = _segmentName.getPartitionId();
- _tableNameWithType = _tableConfig.getTableName();
_timeColumnName = tableConfig.getValidationConfig().getTimeColumnName();
_metricKeyName = _tableNameWithType + "-" + _streamTopic + "-" + _streamPartitionId;
segmentLogger = LoggerFactory.getLogger(LLRealtimeSegmentDataManager.class.getName() + "_" + _segmentNameStr);
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
index 8c8d4a4..47664d2 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
@@ -89,7 +89,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
@Override
protected void doInit() {
- _leaseExtender = SegmentBuildTimeLeaseExtender.create(_instanceId, _serverMetrics);
+ _leaseExtender = SegmentBuildTimeLeaseExtender.create(_instanceId, _serverMetrics, _tableNameWithType);
File statsFile = new File(_tableDataDir, STATS_FILE_NAME);
try {
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SegmentBuildTimeLeaseExtender.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SegmentBuildTimeLeaseExtender.java
index 6676619..fcd6422 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SegmentBuildTimeLeaseExtender.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SegmentBuildTimeLeaseExtender.java
@@ -57,20 +57,20 @@ public class SegmentBuildTimeLeaseExtender {
}
public static synchronized SegmentBuildTimeLeaseExtender create(final String instanceId,
- ServerMetrics serverMetrics) {
+ ServerMetrics serverMetrics, String tableNameWithType) {
SegmentBuildTimeLeaseExtender leaseExtender = INSTANCE_TO_LEASE_EXTENDER.get(instanceId);
if (leaseExtender != null) {
LOGGER.warn("Instance already exists");
} else {
- leaseExtender = new SegmentBuildTimeLeaseExtender(instanceId, serverMetrics);
+ leaseExtender = new SegmentBuildTimeLeaseExtender(instanceId, serverMetrics, tableNameWithType);
INSTANCE_TO_LEASE_EXTENDER.put(instanceId, leaseExtender);
}
return leaseExtender;
}
- private SegmentBuildTimeLeaseExtender(String instanceId, ServerMetrics serverMetrics) {
+ private SegmentBuildTimeLeaseExtender(String instanceId, ServerMetrics serverMetrics, String tableNameWithType) {
_instanceId = instanceId;
- _protocolHandler = new ServerSegmentCompletionProtocolHandler(serverMetrics);
+ _protocolHandler = new ServerSegmentCompletionProtocolHandler(serverMetrics, tableNameWithType);
_executor = new ScheduledThreadPoolExecutor(1);
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/partition/MurmurPartitionFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/data/partition/MurmurPartitionFunction.java
index 24da4dd..7e01655 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/partition/MurmurPartitionFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/partition/MurmurPartitionFunction.java
@@ -20,6 +20,7 @@ package org.apache.pinot.core.data.partition;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import org.apache.pinot.common.utils.HashUtil;
import org.apache.pinot.common.utils.StringUtil;
@@ -64,44 +65,6 @@ public class MurmurPartitionFunction implements PartitionFunction {
*/
@VisibleForTesting
int murmur2(final byte[] data) {
- int length = data.length;
- int seed = 0x9747b28c;
- // 'm' and 'r' are mixing constants generated offline.
- // They're not really 'magic', they just happen to work well.
- final int m = 0x5bd1e995;
- final int r = 24;
-
- // Initialize the hash to a random value
- int h = seed ^ length;
- int length4 = length / 4;
-
- for (int i = 0; i < length4; i++) {
- final int i4 = i * 4;
- int k =
- (data[i4 + 0] & 0xff) + ((data[i4 + 1] & 0xff) << 8) + ((data[i4 + 2] & 0xff) << 16) + ((data[i4 + 3] & 0xff)
- << 24);
- k *= m;
- k ^= k >>> r;
- k *= m;
- h *= m;
- h ^= k;
- }
-
- // Handle the last few bytes of the input array
- switch (length % 4) {
- case 3:
- h ^= (data[(length & ~3) + 2] & 0xff) << 16;
- case 2:
- h ^= (data[(length & ~3) + 1] & 0xff) << 8;
- case 1:
- h ^= data[length & ~3] & 0xff;
- h *= m;
- }
-
- h ^= h >>> 13;
- h *= m;
- h ^= h >>> 15;
-
- return h;
+ return HashUtil.murmur2(data);
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/realtime/ControllerLeaderLocator.java b/pinot-core/src/main/java/org/apache/pinot/server/realtime/ControllerLeaderLocator.java
index 8a3b0fb..12b5eb8 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/realtime/ControllerLeaderLocator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/server/realtime/ControllerLeaderLocator.java
@@ -19,11 +19,16 @@
package org.apache.pinot.server.realtime;
import com.google.common.annotations.VisibleForTesting;
+import java.util.Map;
import org.apache.helix.AccessOption;
import org.apache.helix.BaseDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.ZNRecord;
-import org.apache.pinot.core.query.utils.Pair;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.MasterSlaveSMD;
+import org.apache.pinot.common.utils.CommonConstants;
+import org.apache.pinot.common.utils.helix.LeadControllerUtils;
+import org.apache.pinot.pql.parsers.utils.Pair;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -37,7 +42,6 @@ public class ControllerLeaderLocator {
public static final Logger LOGGER = LoggerFactory.getLogger(ControllerLeaderLocator.class);
private final HelixManager _helixManager;
- private final String _clusterName;
// Co-ordinates of the last known controller leader.
private Pair<String, Integer> _controllerLeaderHostPort = null;
@@ -51,7 +55,6 @@ public class ControllerLeaderLocator {
ControllerLeaderLocator(HelixManager helixManager) {
_helixManager = helixManager;
- _clusterName = helixManager.getClusterName();
}
/**
@@ -76,38 +79,148 @@ public class ControllerLeaderLocator {
}
/**
- * Locate the controller leader so that we can send LLC segment completion requests to it.
+ * Locates the controller leader so that we can send LLC segment completion requests to it.
* Checks the {@link ControllerLeaderLocator::_cachedControllerLeaderInvalid} flag and fetches the leader from helix if cached value is invalid
- *
- * @return The host:port string of the current controller leader.
+ * @param rawTableName table name without type.
+ * @return The host-port pair of the current controller leader.
*/
- public synchronized Pair<String, Integer> getControllerLeader() {
+ public synchronized Pair<String, Integer> getControllerLeader(String rawTableName) {
if (!_cachedControllerLeaderInvalid) {
return _controllerLeaderHostPort;
}
+ Pair<String, Integer> leaderForTable = getLeaderForTable(rawTableName);
+ if (leaderForTable == null) {
+ LOGGER.warn("Failed to find a leader for Table: {}", rawTableName);
+ _cachedControllerLeaderInvalid = true;
+ return null;
+ } else {
+ _controllerLeaderHostPort = leaderForTable;
+ _cachedControllerLeaderInvalid = false;
+ LOGGER.info("Setting controller leader to be {}:{}", _controllerLeaderHostPort.getFirst(),
+ _controllerLeaderHostPort.getSecond());
+ return _controllerLeaderHostPort;
+ }
+ }
+
+ /**
+ * Firstly checks whether lead controller resource has been enabled or not.
+ * If yes, use this as the leader for realtime segment completion once partition leader exists.
+ * Otherwise, try to use Helix leader.
+ * @param rawTableName table name without type.
+ * @return the controller leader id with hostname and port for this table, e.g. localhost_9000
+ */
+ private Pair<String, Integer> getLeaderForTable(String rawTableName) {
+ // Checks whether lead controller resource has been enabled or not.
+ if (isLeadControllerResourceEnabled()) {
+ // Gets leader from lead controller resource.
+ return getLeaderFromLeadControllerResource(rawTableName);
+ } else {
+ // Gets Helix leader to be the leader to this table, otherwise returns null.
+ return getHelixClusterLeader();
+ }
+ }
+
+ /**
+ * Checks whether lead controller resource is enabled or not. The switch is in resource config.
+ */
+ private boolean isLeadControllerResourceEnabled() {
BaseDataAccessor<ZNRecord> dataAccessor = _helixManager.getHelixDataAccessor().getBaseDataAccessor();
Stat stat = new Stat();
try {
- ZNRecord znRecord =
- dataAccessor.get("/" + _clusterName + "/CONTROLLER/LEADER", stat, AccessOption.THROW_EXCEPTION_IFNOTEXIST);
- String leader = znRecord.getId();
- int index = leader.lastIndexOf('_');
- String leaderHost = leader.substring(0, index);
- int leaderPort = Integer.valueOf(leader.substring(index + 1));
- _controllerLeaderHostPort = new Pair<>(leaderHost, leaderPort);
- _cachedControllerLeaderInvalid = false;
- LOGGER.info("Setting controller leader to be {}:{} as per znode version {}, mtime {}", leaderHost, leaderPort,
- stat.getVersion(), stat.getMtime());
- return _controllerLeaderHostPort;
+ ZNRecord znRecord = dataAccessor.get("/" + _helixManager.getClusterName() + "/CONFIGS/RESOURCE/"
+ + CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME, stat, AccessOption.THROW_EXCEPTION_IFNOTEXIST);
+ return Boolean.parseBoolean(znRecord.getSimpleField(CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_ENABLED_KEY));
} catch (Exception e) {
- LOGGER.warn("Could not locate controller leader, exception", e);
- _cachedControllerLeaderInvalid = true;
+ LOGGER.warn("Could not get whether lead controller resource is enabled or not.", e);
+ return false;
+ }
+ }
+
+ /**
+ * Gets leader from lead controller resource. Null if there is no leader.
+ * @param rawTableName raw table name.
+ * @return pair of instance hostname and port of Helix cluster leader, e.g. {localhost, 9000}.
+ */
+ private Pair<String, Integer> getLeaderFromLeadControllerResource(String rawTableName) {
+ Pair<String, Integer> leaderHostAndPortPair = getLeadControllerInstanceIdForTable(rawTableName);
+ if (leaderHostAndPortPair != null) {
+ return leaderHostAndPortPair;
+ } else {
+ LOGGER.warn("Could not locate leader for table: {}", rawTableName);
+ return null;
+ }
+ }
+
+ /**
+ * Gets Helix leader in the cluster. Null if there is no leader.
+ * @return instance id of Helix cluster leader, e.g. localhost_9000.
+ */
+ private Pair<String, Integer> getHelixClusterLeader() {
+ BaseDataAccessor<ZNRecord> dataAccessor = _helixManager.getHelixDataAccessor().getBaseDataAccessor();
+ Stat stat = new Stat();
+ try {
+ ZNRecord znRecord = dataAccessor.get("/" + _helixManager.getClusterName() + "/CONTROLLER/LEADER", stat,
+ AccessOption.THROW_EXCEPTION_IFNOTEXIST);
+ String helixLeader = znRecord.getId();
+ LOGGER.info("Getting Helix leader: {} as per znode version {}, mtime {}", helixLeader, stat.getVersion(),
+ stat.getMtime());
+ return convertToHostAndPortPair(helixLeader);
+ } catch (Exception e) {
+ LOGGER.warn("Could not locate Helix leader!", e);
return null;
}
}
/**
+ * Gets lead controller participant id for table from lead controller resource.
+ * If the resource is disabled or no controller registered as participant, there is no instance in "MASTER" state.
+ * @param rawTableName table name without type
+ * @return Helix controller instance id for partition leader, e.g. localhost_9000. Null if not found or resource is disabled.
+ */
+ private Pair<String, Integer> getLeadControllerInstanceIdForTable(String rawTableName) {
+ try {
+ ExternalView leadControllerResourceExternalView = _helixManager.getClusterManagmentTool()
+ .getResourceExternalView(_helixManager.getClusterName(), CommonConstants.Helix.LEAD_CONTROLLER_RESOURCE_NAME);
+ if (leadControllerResourceExternalView == null) {
+ LOGGER.warn("External view of lead controller resource is null!");
+ return null;
+ }
+ int partitionId = LeadControllerUtils.getPartitionIdForTable(rawTableName);
+ String partitionName = LeadControllerUtils.generatePartitionName(partitionId);
+ Map<String, String> partitionStateMap = leadControllerResourceExternalView.getStateMap(partitionName);
+
+ // Get master host from partition map. Return null if no master found.
+ for (Map.Entry<String, String> entry : partitionStateMap.entrySet()) {
+ if (MasterSlaveSMD.States.MASTER.name().equals(entry.getValue())) {
+ // Found the controller in master state.
+ // Converts participant id (with Prefix "Controller_") to controller id and assigns it as the leader,
+ // since realtime segment completion protocol doesn't need the prefix in controller instance id.
+ String participantInstanceId = entry.getKey();
+ String controllerInstanceId = participantInstanceId.substring(participantInstanceId.indexOf('_') + 1);
+ return convertToHostAndPortPair(controllerInstanceId);
+ }
+ }
+ LOGGER
+ .warn("There is no controller in MASTER state for partition: {} in lead controller resource", partitionName);
+ } catch (Exception e) {
+ LOGGER.warn("Caught exception when getting lead controller instance Id for table: {}", rawTableName, e);
+ }
+ return null;
+ }
+
+ /**
+ * Converts instance id to a pair of hostname and port.
+ * @param instanceId instance id without any prefix, e.g. localhost_9000
+ * */
+ private Pair<String, Integer> convertToHostAndPortPair(String instanceId) {
+ int index = instanceId.lastIndexOf('_');
+ String leaderHost = instanceId.substring(0, index);
+ int leaderPort = Integer.valueOf(instanceId.substring(index + 1));
+ return new Pair<>(leaderHost, leaderPort);
+ }
+
+ /**
* Invalidates the cached controller leader value by setting the {@link ControllerLeaderLocator::_cacheControllerLeadeInvalid} flag.
* This flag is always checked first by {@link ControllerLeaderLocator::getControllerLeader()} method before returning the leader. If set, leader is fetched from helix, else cached leader value is returned.
*
diff --git a/pinot-core/src/main/java/org/apache/pinot/server/realtime/ServerSegmentCompletionProtocolHandler.java b/pinot-core/src/main/java/org/apache/pinot/server/realtime/ServerSegmentCompletionProtocolHandler.java
index bcfa7ee..7c2ff68 100644
--- a/pinot-core/src/main/java/org/apache/pinot/server/realtime/ServerSegmentCompletionProtocolHandler.java
+++ b/pinot-core/src/main/java/org/apache/pinot/server/realtime/ServerSegmentCompletionProtocolHandler.java
@@ -23,13 +23,14 @@ import java.net.URI;
import java.util.Map;
import javax.net.ssl.SSLContext;
import org.apache.commons.configuration.Configuration;
+import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.common.utils.ClientSSLContextGenerator;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.FileUploadDownloadClient;
-import org.apache.pinot.core.query.utils.Pair;
+import org.apache.pinot.pql.parsers.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,6 +41,7 @@ import static org.apache.pinot.common.utils.CommonConstants.Server.SegmentComple
* A class that handles sending segment completion protocol requests to the controller and getting
* back responses
*/
+// TODO: Use exception based code to handle different types of exceptions.
public class ServerSegmentCompletionProtocolHandler {
private static Logger LOGGER = LoggerFactory.getLogger(ServerSegmentCompletionProtocolHandler.class);
private static final String HTTPS_PROTOCOL = "https";
@@ -51,6 +53,7 @@ public class ServerSegmentCompletionProtocolHandler {
private final FileUploadDownloadClient _fileUploadDownloadClient;
private final ServerMetrics _serverMetrics;
+ private final String _rawTableName;
public static void init(Configuration uploaderConfig) {
Configuration httpsConfig = uploaderConfig.subset(HTTPS_PROTOCOL);
@@ -62,9 +65,10 @@ public class ServerSegmentCompletionProtocolHandler {
uploaderConfig.getInt(CONFIG_OF_SEGMENT_UPLOAD_REQUEST_TIMEOUT_MS, DEFAULT_SEGMENT_UPLOAD_REQUEST_TIMEOUT_MS);
}
- public ServerSegmentCompletionProtocolHandler(ServerMetrics serverMetrics) {
+ public ServerSegmentCompletionProtocolHandler(ServerMetrics serverMetrics, String tableNameWithType) {
_fileUploadDownloadClient = new FileUploadDownloadClient(_sslContext);
_serverMetrics = serverMetrics;
+ _rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);
}
public SegmentCompletionProtocol.Response segmentCommitStart(SegmentCompletionProtocol.Request.Params params) {
@@ -159,7 +163,7 @@ public class ServerSegmentCompletionProtocolHandler {
private String createSegmentCompletionUrl(SegmentCompletionProtocol.Request request) {
ControllerLeaderLocator leaderLocator = ControllerLeaderLocator.getInstance();
- final Pair<String, Integer> leaderHostPort = leaderLocator.getControllerLeader();
+ final Pair<String, Integer> leaderHostPort = leaderLocator.getControllerLeader(_rawTableName);
if (leaderHostPort == null) {
LOGGER.warn("No leader found while trying to send {}", request.toString());
return null;
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java
index fd5505d..0598cd2 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java
@@ -111,7 +111,7 @@ public class LLRealtimeSegmentDataManagerTest {
private RealtimeTableDataManager createTableDataManager() {
final String instanceId = "server-1";
- SegmentBuildTimeLeaseExtender.create(instanceId, new ServerMetrics(new MetricsRegistry()));
+ SegmentBuildTimeLeaseExtender.create(instanceId, new ServerMetrics(new MetricsRegistry()), _tableName);
RealtimeTableDataManager tableDataManager = mock(RealtimeTableDataManager.class);
when(tableDataManager.getServerInstance()).thenReturn(instanceId);
RealtimeSegmentStatsHistory statsHistory = mock(RealtimeSegmentStatsHistory.class);
diff --git a/pinot-core/src/test/java/org/apache/pinot/server/realtime/ControllerLeaderLocatorTest.java b/pinot-core/src/test/java/org/apache/pinot/server/realtime/ControllerLeaderLocatorTest.java
index 34e955d..a4b94e4 100644
--- a/pinot-core/src/test/java/org/apache/pinot/server/realtime/ControllerLeaderLocatorTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/server/realtime/ControllerLeaderLocatorTest.java
@@ -18,10 +18,17 @@
*/
package org.apache.pinot.server.realtime;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
import org.apache.helix.BaseDataAccessor;
+import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
import org.apache.helix.ZNRecord;
+import org.apache.helix.model.ExternalView;
+import org.apache.pinot.common.utils.helix.LeadControllerUtils;
import org.apache.pinot.core.query.utils.Pair;
import org.apache.zookeeper.data.Stat;
import org.testng.Assert;
@@ -35,6 +42,7 @@ import static org.mockito.Mockito.when;
public class ControllerLeaderLocatorTest {
+ private String testTable = "testTable";
/**
* Tests the invalidate logic for cached controller leader
@@ -45,6 +53,7 @@ public class ControllerLeaderLocatorTest {
HelixManager helixManager = mock(HelixManager.class);
HelixDataAccessor helixDataAccessor = mock(HelixDataAccessor.class);
BaseDataAccessor<ZNRecord> baseDataAccessor = mock(BaseDataAccessor.class);
+ HelixAdmin helixAdmin = mock(HelixAdmin.class);
ZNRecord znRecord = mock(ZNRecord.class);
final String leaderHost = "host";
final int leaderPort = 12345;
@@ -54,6 +63,8 @@ public class ControllerLeaderLocatorTest {
when(znRecord.getId()).thenReturn(leaderHost + "_" + leaderPort);
when(baseDataAccessor.get(anyString(), any(), anyInt())).thenReturn(znRecord);
when(helixManager.getClusterName()).thenReturn("testCluster");
+ when(helixManager.getClusterManagmentTool()).thenReturn(helixAdmin);
+ when(helixAdmin.getResourceExternalView(anyString(), anyString())).thenReturn(null);
// Create Controller Leader Locator
FakeControllerLeaderLocator.create(helixManager);
@@ -78,7 +89,7 @@ public class ControllerLeaderLocatorTest {
Assert.assertEquals(controllerLeaderLocator.getLastCacheInvalidateMillis(), lastCacheInvalidateMillis);
// getControllerLeader, which validates the cache
- controllerLeaderLocator.getControllerLeader();
+ controllerLeaderLocator.getControllerLeader(testTable);
Assert.assertFalse(controllerLeaderLocator.isCachedControllerLeaderInvalid());
Assert.assertEquals(controllerLeaderLocator.getLastCacheInvalidateMillis(), lastCacheInvalidateMillis);
@@ -104,16 +115,19 @@ public class ControllerLeaderLocatorTest {
HelixManager helixManager = mock(HelixManager.class);
HelixDataAccessor helixDataAccessor = mock(HelixDataAccessor.class);
BaseDataAccessor<ZNRecord> baseDataAccessor = mock(BaseDataAccessor.class);
+ HelixAdmin helixAdmin = mock(HelixAdmin.class);
when(helixManager.getHelixDataAccessor()).thenReturn(helixDataAccessor);
when(helixDataAccessor.getBaseDataAccessor()).thenReturn(baseDataAccessor);
when(baseDataAccessor.get(anyString(), (Stat) any(), anyInt())).thenThrow(new RuntimeException());
+ when(helixManager.getClusterManagmentTool()).thenReturn(helixAdmin);
+ when(helixAdmin.getResourceExternalView(anyString(), anyString())).thenReturn(null);
// Create Controller Leader Locator
FakeControllerLeaderLocator.create(helixManager);
ControllerLeaderLocator controllerLeaderLocator = FakeControllerLeaderLocator.getInstance();
- Assert.assertEquals(controllerLeaderLocator.getControllerLeader(), null);
+ Assert.assertEquals(controllerLeaderLocator.getControllerLeader(testTable), null);
}
@Test
@@ -121,6 +135,7 @@ public class ControllerLeaderLocatorTest {
HelixManager helixManager = mock(HelixManager.class);
HelixDataAccessor helixDataAccessor = mock(HelixDataAccessor.class);
BaseDataAccessor<ZNRecord> baseDataAccessor = mock(BaseDataAccessor.class);
+ HelixAdmin helixAdmin = mock(HelixAdmin.class);
ZNRecord znRecord = mock(ZNRecord.class);
final String leaderHost = "host";
final int leaderPort = 12345;
@@ -130,14 +145,90 @@ public class ControllerLeaderLocatorTest {
when(znRecord.getId()).thenReturn(leaderHost + "_" + leaderPort);
when(baseDataAccessor.get(anyString(), (Stat) any(), anyInt())).thenReturn(znRecord);
when(helixManager.getClusterName()).thenReturn("myCluster");
+ when(helixManager.getClusterManagmentTool()).thenReturn(helixAdmin);
+ when(helixAdmin.getResourceExternalView(anyString(), anyString())).thenReturn(null);
// Create Controller Leader Locator
FakeControllerLeaderLocator.create(helixManager);
ControllerLeaderLocator controllerLeaderLocator = FakeControllerLeaderLocator.getInstance();
Pair<String, Integer> expectedLeaderLocation = new Pair<>(leaderHost, leaderPort);
- Assert.assertEquals(controllerLeaderLocator.getControllerLeader().getFirst(), expectedLeaderLocation.getFirst());
- Assert.assertEquals(controllerLeaderLocator.getControllerLeader().getSecond(), expectedLeaderLocation.getSecond());
+ Assert.assertEquals(controllerLeaderLocator.getControllerLeader(testTable).getFirst(),
+ expectedLeaderLocation.getFirst());
+ Assert.assertEquals(controllerLeaderLocator.getControllerLeader(testTable).getSecond(),
+ expectedLeaderLocation.getSecond());
+ }
+
+ @Test
+ public void testWhenLeadControllerResourceEnabled() {
+ HelixManager helixManager = mock(HelixManager.class);
+ HelixDataAccessor helixDataAccessor = mock(HelixDataAccessor.class);
+ BaseDataAccessor<ZNRecord> baseDataAccessor = mock(BaseDataAccessor.class);
+ HelixAdmin helixAdmin = mock(HelixAdmin.class);
+ ZNRecord znRecord = mock(ZNRecord.class);
+ final String leaderHost = "host";
+ final int leaderPort = 12345;
+
+ when(helixManager.getHelixDataAccessor()).thenReturn(helixDataAccessor);
+ when(helixDataAccessor.getBaseDataAccessor()).thenReturn(baseDataAccessor);
+ when(znRecord.getId()).thenReturn(leaderHost + "_" + leaderPort);
+ when(baseDataAccessor.get(anyString(), (Stat) any(), anyInt())).thenReturn(znRecord);
+ when(helixManager.getClusterName()).thenReturn("myCluster");
+ when(helixManager.getClusterManagmentTool()).thenReturn(helixAdmin);
+ when(helixAdmin.getResourceExternalView(anyString(), anyString())).thenReturn(null);
+
+ // Create Controller Leader Locator
+ FakeControllerLeaderLocator.create(helixManager);
+ ControllerLeaderLocator controllerLeaderLocator = FakeControllerLeaderLocator.getInstance();
+ Pair<String, Integer> expectedLeaderLocation = new Pair<>(leaderHost, leaderPort);
+
+ // Before enabling lead controller resource config, the helix leader should be used.
+ Assert.assertEquals(controllerLeaderLocator.getControllerLeader(testTable).getFirst(),
+ expectedLeaderLocation.getFirst());
+ Assert.assertEquals(controllerLeaderLocator.getControllerLeader(testTable).getSecond(),
+ expectedLeaderLocation.getSecond());
+
+ // Mock the behavior that 40 seconds have passed.
+ controllerLeaderLocator.setLastCacheInvalidateMillis(System.currentTimeMillis() - 40_000L);
+ controllerLeaderLocator.invalidateCachedControllerLeader();
+
+ // After enabling lead controller resource config, the leader in lead controller resource should be used.
+ when(znRecord.getSimpleField(anyString())).thenReturn("true");
+
+ // External view is null, should return null.
+ Assert.assertNull(controllerLeaderLocator.getControllerLeader(testTable));
+
+ ExternalView externalView = mock(ExternalView.class);
+ when(helixAdmin.getResourceExternalView(anyString(), anyString())).thenReturn(externalView);
+ Set<String> partitionSet = new HashSet<>();
+ when(externalView.getPartitionSet()).thenReturn(partitionSet);
+ Map<String, String> partitionStateMap = new HashMap<>();
+ when(externalView.getStateMap(anyString())).thenReturn(partitionStateMap);
+
+ // External view is empty, should return null.
+ Assert.assertNull(controllerLeaderLocator.getControllerLeader(testTable));
+
+ // Adding one host as master, should return the correct host-port pair.
+ partitionSet.add(LeadControllerUtils.generatePartitionName(LeadControllerUtils.getPartitionIdForTable(testTable)));
+ partitionStateMap.put(LeadControllerUtils.generateParticipantInstanceId(leaderHost, leaderPort), "MASTER");
+
+ Assert.assertEquals(controllerLeaderLocator.getControllerLeader(testTable).getFirst(),
+ expectedLeaderLocation.getFirst());
+ Assert.assertEquals(controllerLeaderLocator.getControllerLeader(testTable).getSecond(),
+ expectedLeaderLocation.getSecond());
+
+ // The participant host is in offline state, should return null.
+ partitionStateMap.put(LeadControllerUtils.generateParticipantInstanceId(leaderHost, leaderPort), "OFFLINE");
+
+ // The leader is still valid since the leader is just updated within 30 seconds.
+ Assert.assertNotNull(controllerLeaderLocator.getControllerLeader(testTable));
+
+ // Mock the behavior that 40 seconds have passed.
+ controllerLeaderLocator.setLastCacheInvalidateMillis(System.currentTimeMillis() - 40_000L);
+ controllerLeaderLocator.invalidateCachedControllerLeader();
+
+ // No controller in MASTER state, should return null.
+ Assert.assertNull(controllerLeaderLocator.getControllerLeader(testTable));
}
static class FakeControllerLeaderLocator extends ControllerLeaderLocator {
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/LeadershipChangeSubscriber.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterWithEnabledLeadControllerResourceIntegrationTest.java
similarity index 64%
rename from pinot-controller/src/main/java/org/apache/pinot/controller/LeadershipChangeSubscriber.java
rename to pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterWithEnabledLeadControllerResourceIntegrationTest.java
index d2c646f..681e235 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/LeadershipChangeSubscriber.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/LLCRealtimeClusterWithEnabledLeadControllerResourceIntegrationTest.java
@@ -16,20 +16,16 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pinot.controller;
+package org.apache.pinot.integration.tests;
-/**
- * Interface for a subscriber to the {@link ControllerLeadershipManager}
- */
-public interface LeadershipChangeSubscriber {
+import org.apache.pinot.controller.ControllerConf;
- /**
- * Callback to invoke on becoming leader
- */
- void onBecomingLeader();
- /**
- * Callback to invoke on losing leadership
- */
- void onBecomingNonLeader();
+public class LLCRealtimeClusterWithEnabledLeadControllerResourceIntegrationTest extends LLCRealtimeClusterIntegrationTest {
+ @Override
+ public void startController() {
+ ControllerConf controllerConfig = getDefaultControllerConfiguration();
+ startController(controllerConfig);
+ enableResourceConfigForLeadControllerResource(true);
+ }
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentCompletionIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentCompletionIntegrationTest.java
index 84ea948..3e04c0e 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentCompletionIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentCompletionIntegrationTest.java
@@ -139,7 +139,7 @@ public class SegmentCompletionIntegrationTest extends BaseClusterIntegrationTest
// Now report to the controller that we had to stop consumption
ServerSegmentCompletionProtocolHandler protocolHandler =
- new ServerSegmentCompletionProtocolHandler(new ServerMetrics(new MetricsRegistry()));
+ new ServerSegmentCompletionProtocolHandler(new ServerMetrics(new MetricsRegistry()), realtimeTableName);
SegmentCompletionProtocol.Request.Params params = new SegmentCompletionProtocol.Request.Params();
params.withOffset(45688L).withSegmentName(_currentSegment).withReason("RandomReason")
.withInstanceId(_serverInstance);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org