This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 62ce57f Add RealtimeConsumptionCatchupServiceCallback (#4218)
62ce57f is described below
commit 62ce57f4886e36daaeee438c9af7c6dc279b4844
Author: Neha Pawar <[email protected]>
AuthorDate: Thu May 30 13:06:41 2019 -0700
Add RealtimeConsumptionCatchupServiceCallback (#4218)
Creating a ServiceCallback within ServiceStatus, to check that realtime
consuming segments have caught up. This will help with latency issues which can
happen during server startup, because the server has to catchup on a lot of
consumption.
For the initial version, this only adds a static wait time (configurable).
More smarts can be added later (monitor stability or idleness of stream
consumption rate)
---
.../broker/broker/helix/HelixBrokerStarter.java | 34 +++++--
.../pinot/common/config/TableNameBuilder.java | 7 ++
.../apache/pinot/common/utils/CommonConstants.java | 3 +
.../apache/pinot/common/utils/ServiceStatus.java | 107 ++++++++++++---------
.../tests/OfflineClusterIntegrationTest.java | 38 ++++++--
.../server/starter/helix/HelixServerStarter.java | 73 ++++++++++++--
6 files changed, 190 insertions(+), 72 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
index 10d34fb..57beed3 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/HelixBrokerStarter.java
@@ -35,6 +35,7 @@ import org.apache.helix.HelixManagerFactory;
import org.apache.helix.InstanceType;
import org.apache.helix.SystemPropertyKeys;
import org.apache.helix.ZNRecord;
+import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.Message;
import org.apache.helix.participant.StateMachineEngine;
@@ -237,16 +238,35 @@ public class HelixBrokerStarter {
.addPreConnectCallback(() ->
brokerMetrics.addMeteredGlobalValue(BrokerMeter.HELIX_ZOOKEEPER_RECONNECTS,
1L));
// Register the service status handler
- LOGGER.info("Registering service status handler");
+ registerServiceStatusHandler();
+
+ LOGGER.info("Finish starting Pinot broker");
+ }
+
+ /**
+ * Fetches the resources to monitor and registers the {@link
org.apache.pinot.common.utils.ServiceStatus.ServiceStatusCallback}s
+ */
+ private void registerServiceStatusHandler() {
+ List<String> resourcesToMonitor = new ArrayList<>(1);
+ IdealState brokerResourceIdealState =
_helixAdmin.getResourceIdealState(_clusterName, Helix.BROKER_RESOURCE_INSTANCE);
+ if (brokerResourceIdealState != null &&
brokerResourceIdealState.isEnabled()) {
+ for (String partitionName : brokerResourceIdealState.getPartitionSet()) {
+ if
(brokerResourceIdealState.getInstanceSet(partitionName).contains(_brokerId)) {
+ resourcesToMonitor.add(Helix.BROKER_RESOURCE_INSTANCE);
+ break;
+ }
+ }
+ }
+
double minResourcePercentForStartup =
_brokerConf.getDouble(Broker.CONFIG_OF_BROKER_MIN_RESOURCE_PERCENT_FOR_START,
Broker.DEFAULT_BROKER_MIN_RESOURCE_PERCENT_FOR_START);
- ServiceStatus.setServiceStatusCallback(new
ServiceStatus.MultipleCallbackServiceStatusCallback(ImmutableList
- .of(new
ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_participantHelixManager,
- _clusterName, _brokerId, minResourcePercentForStartup),
- new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_participantHelixManager,
- _clusterName, _brokerId, minResourcePercentForStartup))));
- LOGGER.info("Finish starting Pinot broker");
+ LOGGER.info("Registering service status handler");
+ ServiceStatus.setServiceStatusCallback(new
ServiceStatus.MultipleCallbackServiceStatusCallback(ImmutableList.of(
+ new
ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_participantHelixManager,
_clusterName,
+ _brokerId, resourcesToMonitor, minResourcePercentForStartup),
+ new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_participantHelixManager,
_clusterName,
+ _brokerId, resourcesToMonitor, minResourcePercentForStartup))));
}
private void addInstanceTagIfNeeded() {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/config/TableNameBuilder.java
b/pinot-common/src/main/java/org/apache/pinot/common/config/TableNameBuilder.java
index 5ecac01..8ec01ff 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/config/TableNameBuilder.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/config/TableNameBuilder.java
@@ -129,4 +129,11 @@ public class TableNameBuilder {
public static boolean isOfflineTableResource(@Nonnull String resourceName) {
return OFFLINE.tableHasTypeSuffix(resourceName);
}
+
+ /**
+ * Return whether the given resource name represents a realtime table
resource.
+ */
+ public static boolean isRealtimeTableResource(@Nonnull String resourceName) {
+ return REALTIME.tableHasTypeSuffix(resourceName);
+ }
}
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 de9a3d0..915e96e 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
@@ -187,6 +187,9 @@ public class CommonConstants {
public static final String CONFIG_OF_SERVER_MIN_RESOURCE_PERCENT_FOR_START
=
"pinot.server.startup.minResourcePercent";
public static final double DEFAULT_SERVER_MIN_RESOURCE_PERCENT_FOR_START =
100.0;
+ public static final String
CONFIG_OF_STARTUP_REALTIME_CONSUMPTION_CATCHUP_WAIT_MS =
+ "pinot.server.starter.realtimeConsumptionCatchupWaitMs";
+ public static final int
DEFAULT_STARTUP_REALTIME_CONSUMPTION_CATCHUP_WAIT_MS = 0;
public static final int DEFAULT_ADMIN_API_PORT = 8097;
public static final String DEFAULT_READ_MODE = "heap";
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStatus.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStatus.java
index b231ac1..9cf0e21 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStatus.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStatus.java
@@ -18,12 +18,15 @@
*/
package org.apache.pinot.common.utils;
+import com.google.common.collect.Lists;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
@@ -129,14 +132,64 @@ public class ServiceStatus {
}
/**
+ * Service status callback that checks whether realtime consumption has
caught up
+ * TODO: In this initial version, we are simply adding a configurable static
wait time
+ * This can be made smarter:
+ * 1) Keep track of average consumption rate for table in server stats
+ * 2) Monitor consumption rate during startup, report GOOD when it
stabilizes to average rate
+ * 3) Monitor consumption rate during startup, report GOOD if it is idle
+ */
+ public static class RealtimeConsumptionCatchupServiceStatusCallback
implements ServiceStatusCallback {
+
+ String _statusDescription = STATUS_DESCRIPTION_INIT;
+ private Status _serviceStatus = Status.STARTING;
+ private final long _endWaitTime;
+
+ /**
+ * Realtime consumption catchup service which adds a static wait time for
consuming segments to catchup
+ */
+ public RealtimeConsumptionCatchupServiceStatusCallback(HelixManager
helixManager, String clusterName,
+ String instanceName, long realtimeConsumptionCatchupWaitMs) {
+
+ // A consuming segment will actually be ready to serve queries after
(time of creation of partition consumer) + (configured max time to catchup)
+ // We are approximating it to (time of server startup) + (configured max
time to catch up)
+ _endWaitTime = System.currentTimeMillis() +
realtimeConsumptionCatchupWaitMs;
+ LOGGER.info("Monitoring realtime consumption catchup. Will allow {} ms
before marking status GOOD",
+ realtimeConsumptionCatchupWaitMs);
+ }
+
+ @Override
+ public synchronized Status getServiceStatus() {
+ if (_serviceStatus.equals(Status.GOOD)) {
+ return _serviceStatus;
+ }
+ long now = System.currentTimeMillis();
+ if (now < _endWaitTime) {
+ _statusDescription =
+ String.format("Waiting for consuming segments to catchup,
timeRemaining=%dms", _endWaitTime - now);
+ return Status.STARTING;
+ }
+ _statusDescription = String.format("Consuming segments status GOOD since
%dms", _endWaitTime);
+ return Status.GOOD;
+ }
+
+ @Override
+ public synchronized String getStatusDescription() {
+ return _statusDescription;
+ }
+ }
+
+ /**
* Service status callback that compares ideal state with another Helix
state. Used to share most of the logic between
* the ideal state/external view comparison and ideal state/current state
comparison.
*/
- private static abstract class IdealStateMatchServiceStatusCallback<T extends
HelixProperty> implements ServiceStatusCallback {
- protected final String _clusterName;
- protected final String _instanceName;
- protected final HelixAdmin _helixAdmin;
- protected final HelixDataAccessor _helixDataAccessor;
+ private static abstract class IdealStateMatchServiceStatusCallback<T extends
HelixProperty>
+ implements ServiceStatusCallback {
+
+ final String _clusterName;
+ final String _instanceName;
+ final HelixAdmin _helixAdmin;
+ final HelixDataAccessor _helixDataAccessor;
private final Set<String> _resourcesToMonitor;
private final int _numTotalResourcesToMonitor;
@@ -146,39 +199,7 @@ public class ServiceStatus {
private String _statusDescription = STATUS_DESCRIPTION_INIT;
- public IdealStateMatchServiceStatusCallback(HelixManager helixManager,
String clusterName, String instanceName,
- double minResourcesStartPercent) {
- _clusterName = clusterName;
- _instanceName = instanceName;
- _helixAdmin = helixManager.getClusterManagmentTool();
- _helixDataAccessor = helixManager.getHelixDataAccessor();
-
- _resourcesToMonitor = new HashSet<>();
- for (String resourceName :
_helixAdmin.getResourcesInCluster(_clusterName)) {
- // Only monitor table resources and broker resource
- if (!TableNameBuilder.isTableResource(resourceName) && !resourceName
- .equals(CommonConstants.Helix.BROKER_RESOURCE_INSTANCE)) {
- continue;
- }
- // Only monitor enabled resources
- IdealState idealState = getResourceIdealState(resourceName);
- if (idealState.isEnabled()) {
- for (String partitionName : idealState.getPartitionSet()) {
- if
(idealState.getInstanceSet(partitionName).contains(_instanceName)) {
- _resourcesToMonitor.add(resourceName);
- break;
- }
- }
- }
- }
- _numTotalResourcesToMonitor = _resourcesToMonitor.size();
- _minResourcesStartCount = (int) Math.ceil(minResourcesStartPercent *
_numTotalResourcesToMonitor / 100);
-
- LOGGER.info("Monitoring {} resources: {} for start up of instance {}",
_numTotalResourcesToMonitor,
- getResourceListAsString(), _instanceName);
- }
-
- public IdealStateMatchServiceStatusCallback(HelixManager helixManager,
String clusterName, String instanceName,
+ IdealStateMatchServiceStatusCallback(HelixManager helixManager, String
clusterName, String instanceName,
List<String> resourcesToMonitor, double minResourcesStartPercent) {
_clusterName = clusterName;
_instanceName = instanceName;
@@ -361,11 +382,6 @@ public class ServiceStatus {
private static final String MATCH_NAME = "CurrentStateMatch";
public IdealStateAndCurrentStateMatchServiceStatusCallback(HelixManager
helixManager, String clusterName,
- String instanceName, double minResourcesStartPercent) {
- super(helixManager, clusterName, instanceName, minResourcesStartPercent);
- }
-
- public IdealStateAndCurrentStateMatchServiceStatusCallback(HelixManager
helixManager, String clusterName,
String instanceName, List<String> resourcesToMonitor, double
minResourcesStartPercent) {
super(helixManager, clusterName, instanceName, resourcesToMonitor,
minResourcesStartPercent);
}
@@ -398,11 +414,6 @@ public class ServiceStatus {
private static final String MATCH_NAME = "ExternalViewMatch";
public IdealStateAndExternalViewMatchServiceStatusCallback(HelixManager
helixManager, String clusterName,
- String instanceName, double minResourcesStartPercent) {
- super(helixManager, clusterName, instanceName, minResourcesStartPercent);
- }
-
- public IdealStateAndExternalViewMatchServiceStatusCallback(HelixManager
helixManager, String clusterName,
String instanceName, List<String> resourcesToMonitor, double
minResourcesStartPercent) {
super(helixManager, clusterName, instanceName, resourcesToMonitor,
minResourcesStartPercent);
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
index 2dc48c5..ac6a445 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineClusterIntegrationTest.java
@@ -32,7 +32,9 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
+import org.apache.helix.model.IdealState;
import org.apache.pinot.common.config.TableConfig;
+import org.apache.pinot.common.config.TableNameBuilder;
import org.apache.pinot.common.metadata.segment.OfflineSegmentZKMetadata;
import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.JsonUtils;
@@ -125,20 +127,36 @@ public class OfflineClusterIntegrationTest extends
BaseClusterIntegrationTestSet
// Set up service status callbacks
// NOTE: put this step after creating the table and uploading all segments
so that brokers and servers can find the
// resources to monitor
+ registerCallbackHandlers();
+
+ // Wait for all documents loaded
+ waitForAllDocsLoaded(600_000L);
+ }
+
+ private void registerCallbackHandlers() {
List<String> instances = _helixAdmin.getInstancesInCluster(_clusterName);
+ instances.removeIf(instance ->
(!instance.startsWith(CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE) &&
!instance.startsWith(
+ CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE)));
+ List<String> resourcesInCluster =
_helixAdmin.getResourcesInCluster(_clusterName);
+ resourcesInCluster.removeIf(resource ->
(!TableNameBuilder.isTableResource(resource)
+ && !CommonConstants.Helix.BROKER_RESOURCE_INSTANCE.equals(resource)));
for (String instance : instances) {
- if (instance.startsWith(CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE)
|| instance
- .startsWith(CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE)) {
- _serviceStatusCallbacks.add(new
ServiceStatus.MultipleCallbackServiceStatusCallback(ImmutableList
- .of(new
ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_helixManager,
_clusterName,
- instance, 100.0),
- new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_helixManager,
_clusterName,
- instance, 100.0))));
+ List<String> resourcesToMonitor = new ArrayList<>();
+ for (String resourceName : resourcesInCluster) {
+ IdealState idealState =
_helixAdmin.getResourceIdealState(_clusterName, resourceName);
+ for (String partitionName : idealState.getPartitionSet()) {
+ if (idealState.getInstanceSet(partitionName).contains(instance)) {
+ resourcesToMonitor.add(resourceName);
+ break;
+ }
+ }
}
+ _serviceStatusCallbacks.add(new
ServiceStatus.MultipleCallbackServiceStatusCallback(ImmutableList.of(
+ new
ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_helixManager,
_clusterName, instance,
+ resourcesToMonitor, 100.0),
+ new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_helixManager,
_clusterName, instance,
+ resourcesToMonitor, 100.0))));
}
-
- // Wait for all documents loaded
- waitForAllDocsLoaded(600_000L);
}
@Test
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
index a8a6190..4e8c20f 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixServerStarter.java
@@ -20,6 +20,7 @@ package org.apache.pinot.server.starter.helix;
import com.google.common.collect.ImmutableList;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -51,6 +52,7 @@ import org.apache.pinot.common.config.TagNameUtils;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.utils.CommonConstants;
import org.apache.pinot.common.utils.NetUtil;
import org.apache.pinot.common.utils.ServiceStatus;
import org.apache.pinot.common.utils.ServiceStatus.Status;
@@ -194,13 +196,7 @@ public class HelixServerStarter {
.addPreConnectCallback(() ->
serverMetrics.addMeteredGlobalValue(ServerMeter.HELIX_ZOOKEEPER_RECONNECTS,
1L));
// Register the service status handler
- double minResourcePercentForStartup = _serverConf
- .getDouble(CONFIG_OF_SERVER_MIN_RESOURCE_PERCENT_FOR_START,
DEFAULT_SERVER_MIN_RESOURCE_PERCENT_FOR_START);
- ServiceStatus.setServiceStatusCallback(new
ServiceStatus.MultipleCallbackServiceStatusCallback(ImmutableList
- .of(new
ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_helixManager,
_helixClusterName,
- _instanceId, minResourcePercentForStartup),
- new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_helixManager,
_helixClusterName,
- _instanceId, minResourcePercentForStartup))));
+ registerServiceStatusHandler();
ControllerLeaderLocator.create(_helixManager);
@@ -220,6 +216,69 @@ public class HelixServerStarter {
serverMetrics.addCallbackGauge("memory.allocationFailureCount",
PinotDataBuffer::getAllocationFailureCount);
}
+ /**
+ * Fetches the resources to monitor and registers the {@link
org.apache.pinot.common.utils.ServiceStatus.ServiceStatusCallback}s
+ */
+ private void registerServiceStatusHandler() {
+
+ double minResourcePercentForStartup = _serverConf
+ .getDouble(CONFIG_OF_SERVER_MIN_RESOURCE_PERCENT_FOR_START,
DEFAULT_SERVER_MIN_RESOURCE_PERCENT_FOR_START);
+ int realtimeConsumptionCatchupWaitMs = _serverConf.getInt(
+
CommonConstants.Server.CONFIG_OF_STARTUP_REALTIME_CONSUMPTION_CATCHUP_WAIT_MS,
+
CommonConstants.Server.DEFAULT_STARTUP_REALTIME_CONSUMPTION_CATCHUP_WAIT_MS);
+
+ // collect all resources which have this instance in the ideal state
+ List<String> resourcesToMonitor = new ArrayList<>();
+ // if even 1 resource has this instance in ideal state with state
CONSUMING, set this to true
+ boolean foundConsuming = false;
+ boolean checkRealtime = realtimeConsumptionCatchupWaitMs > 0;
+
+ for (String resourceName :
_helixAdmin.getResourcesInCluster(_helixClusterName)) {
+ // Only monitor table resources
+ if (!TableNameBuilder.isTableResource(resourceName)) {
+ continue;
+ }
+
+ // Only monitor enabled resources
+ IdealState idealState =
_helixAdmin.getResourceIdealState(_helixClusterName, resourceName);
+ if (idealState.isEnabled()) {
+
+ for (String partitionName : idealState.getPartitionSet()) {
+ if (idealState.getInstanceSet(partitionName).contains(_instanceId)) {
+ resourcesToMonitor.add(resourceName);
+ break;
+ }
+ }
+ if (checkRealtime && !foundConsuming &&
TableNameBuilder.isRealtimeTableResource(resourceName)) {
+ for (String partitionName : idealState.getPartitionSet()) {
+ if
(CommonConstants.Helix.StateModel.RealtimeSegmentOnlineOfflineStateModel.CONSUMING.equals(
+
idealState.getInstanceStateMap(partitionName).get(_instanceId))) {
+ foundConsuming = true;
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ ImmutableList.Builder<ServiceStatus.ServiceStatusCallback>
serviceStatusCallbackListBuilder =
+ new ImmutableList.Builder<>();
+ serviceStatusCallbackListBuilder.add(
+ new
ServiceStatus.IdealStateAndCurrentStateMatchServiceStatusCallback(_helixManager,
_helixClusterName,
+ _instanceId, resourcesToMonitor, minResourcePercentForStartup));
+ serviceStatusCallbackListBuilder.add(
+ new
ServiceStatus.IdealStateAndExternalViewMatchServiceStatusCallback(_helixManager,
_helixClusterName,
+ _instanceId, resourcesToMonitor, minResourcePercentForStartup));
+ if (checkRealtime && foundConsuming) {
+ serviceStatusCallbackListBuilder.add(
+ new
ServiceStatus.RealtimeConsumptionCatchupServiceStatusCallback(_helixManager,
_helixClusterName,
+ _instanceId, realtimeConsumptionCatchupWaitMs));
+ }
+ LOGGER.info("Registering service status handler");
+ ServiceStatus.setServiceStatusCallback(
+ new
ServiceStatus.MultipleCallbackServiceStatusCallback(serviceStatusCallbackListBuilder.build()));
+ }
+
private void setAdminApiPort(int adminApiPort) {
Map<String, String> propToUpdate = new HashMap<>();
propToUpdate.put(Instance.ADMIN_PORT_KEY, String.valueOf(adminApiPort));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
