You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2022/04/07 22:11:29 UTC
[GitHub] [pinot] Jackie-Jiang opened a new pull request, #8491: Add connection based FailureDetector
Jackie-Jiang opened a new pull request, #8491:
URL: https://github.com/apache/pinot/pull/8491
For issue #8490
## Description
- Add FailureDetector module to the broker
- Add QueryResponse interface
- Add ConnectionFailureDetector to detect server connection failures
## Release Notes
Added the following failure detector related configs to the broker config:
- `pinot.broker.failure.detector.class`
- `pinot.broker.failure.detector.retry.initial.delay.ms` (5000 by default)
- `pinot.broker.failure.detector.retry.delay.factor` (2 by default)
- `pinot.broker.failure.detector.max.retries` (10 by default)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849747904
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/BrokerRoutingManager.java:
##########
@@ -214,87 +219,178 @@ private void processInstanceConfigChange() {
long startTimeMs = System.currentTimeMillis();
List<ZNRecord> instanceConfigZNRecords =
- _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT,
- CommonConstants.Helix.ZkClient.RETRY_COUNT, CommonConstants.Helix.ZkClient.RETRY_INTERVAL_MS);
+ _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT, Helix.ZkClient.RETRY_COUNT,
+ Helix.ZkClient.RETRY_INTERVAL_MS);
long fetchInstanceConfigsEndTimeMs = System.currentTimeMillis();
- // Calculate new enabled and disabled instances
- Set<String> enabledInstances = new HashSet<>();
- List<String> newEnabledInstances = new ArrayList<>();
+ // Calculate new enabled and disabled servers
+ Set<String> enabledServers = new HashSet<>();
+ List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
- String instance = instanceConfigZNRecord.getId();
- if (isInstanceEnabled(instanceConfigZNRecord)) {
- enabledInstances.add(instance);
+ String instanceId = instanceConfigZNRecord.getId();
+ if (isEnabledServer(instanceConfigZNRecord)) {
+ enabledServers.add(instanceId);
// Always refresh the server instance with the latest instance config in case it changes
ServerInstance serverInstance = new ServerInstance(new InstanceConfig(instanceConfigZNRecord));
- if (_enabledServerInstanceMap.put(instance, serverInstance) == null) {
- newEnabledInstances.add(instance);
+ if (_enabledServerInstanceMap.put(instanceId, serverInstance) == null) {
+ newEnabledServers.add(instanceId);
+
+ // NOTE: Remove new enabled server from excluded servers because the server is likely being restarted
+ if (_excludedServers.remove(instanceId)) {
+ LOGGER.info("Got excluded server: {} re-enabled, including it into the routing", instanceId);
+ }
}
}
}
- List<String> newDisabledInstances = new ArrayList<>();
+ List<String> newDisabledServers = new ArrayList<>();
for (String instance : _enabledServerInstanceMap.keySet()) {
- if (!enabledInstances.contains(instance)) {
- newDisabledInstances.add(instance);
+ if (!enabledServers.contains(instance)) {
+ newDisabledServers.add(instance);
}
}
- List<String> changedInstances = new ArrayList<>(newEnabledInstances.size() + newDisabledInstances.size());
- changedInstances.addAll(newEnabledInstances);
- changedInstances.addAll(newDisabledInstances);
- long calculateChangedInstancesEndTimeMs = System.currentTimeMillis();
-
- // Early terminate if there is no instance changed
- if (changedInstances.isEmpty()) {
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms) without instance change", calculateChangedInstancesEndTimeMs - startTimeMs,
- instanceConfigZNRecords.size(), fetchInstanceConfigsEndTimeMs - startTimeMs,
- calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs);
+
+ // Calculate the routable servers and the changed routable servers
+ List<String> changedServers = new ArrayList<>(newEnabledServers.size() + newDisabledServers.size());
+ if (_excludedServers.isEmpty()) {
+ _routableServers = enabledServers;
+ changedServers.addAll(newEnabledServers);
+ changedServers.addAll(newDisabledServers);
+ } else {
+ enabledServers.removeAll(_excludedServers);
+ _routableServers = enabledServers;
+ // NOTE: All new enabled servers are routable
+ changedServers.addAll(newEnabledServers);
+ for (String newDisabledServer : newDisabledServers) {
+ if (_excludedServers.contains(newDisabledServer)) {
+ changedServers.add(newDisabledServer);
+ }
+ }
+ }
+ long calculateChangedServersEndTimeMs = System.currentTimeMillis();
+
+ // Early terminate if there is no changed servers
+ if (changedServers.isEmpty()) {
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms) without instance change",
+ calculateChangedServersEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
+ fetchInstanceConfigsEndTimeMs - startTimeMs,
+ calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs);
return;
}
// Update routing entry for all tables
for (RoutingEntry routingEntry : _routingEntryMap.values()) {
try {
- routingEntry.onInstancesChange(enabledInstances, changedInstances);
+ routingEntry.onInstancesChange(_routableServers, changedServers);
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry on instances change for table: {}",
routingEntry.getTableNameWithType(), e);
}
}
long updateRoutingEntriesEndTimeMs = System.currentTimeMillis();
- // Remove new disabled instances from _enabledServerInstanceMap after updating all routing entries to ensure it
- // always contains the selected instances
- for (String newDisabledInstance : newDisabledInstances) {
+ // Remove new disabled servers from _enabledServerInstanceMap after updating all routing entries to ensure it
+ // always contains the selected servers
+ for (String newDisabledInstance : newDisabledServers) {
_enabledServerInstanceMap.remove(newDisabledInstance);
}
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms, update {} routing entries: {}ms), new enabled instances: {}, new disabled instances: {}",
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms, update {} routing entries: {}ms), "
+ + "new enabled servers: {}, new disabled servers: {}, excluded servers: {}",
updateRoutingEntriesEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
- fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs,
- _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedInstancesEndTimeMs,
- newEnabledInstances, newDisabledInstances);
+ fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs,
+ _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedServersEndTimeMs, newEnabledServers,
+ newDisabledServers, _excludedServers);
}
- private static boolean isInstanceEnabled(ZNRecord instanceConfigZNRecord) {
+ private static boolean isEnabledServer(ZNRecord instanceConfigZNRecord) {
+ String instanceId = instanceConfigZNRecord.getId();
+ if (!instanceId.startsWith(Helix.PREFIX_OF_SERVER_INSTANCE)) {
+ // NOTE: Some legacy configs might not contain the instance type prefix
+ if (instanceId.startsWith(Helix.PREFIX_OF_CONTROLLER_INSTANCE) || instanceId.startsWith(
+ Helix.PREFIX_OF_BROKER_INSTANCE) || instanceId.startsWith(Helix.PREFIX_OF_MINION_INSTANCE)) {
+ return false;
+ }
+ }
if ("false".equalsIgnoreCase(
instanceConfigZNRecord.getSimpleField(InstanceConfig.InstanceConfigProperty.HELIX_ENABLED.name()))) {
return false;
}
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.IS_SHUTDOWN_IN_PROGRESS))) {
return false;
}
//noinspection RedundantIfStatement
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.QUERIES_DISABLED))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.QUERIES_DISABLED))) {
return false;
}
return true;
}
+ /**
+ * Excludes a server from the routing.
+ */
+ public synchronized void excludeServerFromRouting(String instanceId) {
+ LOGGER.warn("Excluding server: {} from routing", instanceId);
+ if (!_excludedServers.add(instanceId)) {
+ LOGGER.info("Server: {} is already excluded from routing, skipping updating the routing", instanceId);
+ return;
+ }
+ if (!_routableServers.contains(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.remove(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when excluding server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Excluded server: {} from routing in {}ms (updated {} routing entries)", instanceId,
+ System.currentTimeMillis() - startTimeMs, _routingEntryMap.size());
+ }
+
+ /**
+ * Includes a previous excluded server to the routing.
+ */
+ public synchronized void includeServerToRouting(String instanceId) {
+ LOGGER.info("Including server: {} to routing", instanceId);
+ if (!_excludedServers.remove(instanceId)) {
+ LOGGER.info("Server: {} is not previously excluded, skipping updating the routing", instanceId);
+ }
+ if (!_enabledServerInstanceMap.containsKey(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.add(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when including server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Included server: {} to routing in {}ms (updated {} routing entries)", instanceId,
Review Comment:
There's no difference, they both call `clock_gettime`, `System.currentTimeMillis` with `CLOCK_REALTIME` and `System.nanoTime` with `CLOCK_MONOTONIC`, both clocks are in vDSO. I don't know what we need to do to settle this. Run a multithreaded microbenchmark on a range of servers?
The durations are incomparable because `System.currentTimeMillis` is subject to adjustment but `System.nanoTime` isn't. So if you record a number of milliseconds with `System.currentTimeMillis` it can be negative, smaller, the same, or larger than the same duration recorded with `System.nanoTime`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849794389
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -181,24 +191,43 @@ private void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryRespo
}
}
- private void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryResponse,
- ServerRoutingInstance serverRoutingInstance, byte[] requestBytes)
- throws Exception {
+ void connectWithoutLocking()
+ throws InterruptedException {
if (_channel == null || !_channel.isActive()) {
long startTime = System.currentTimeMillis();
_channel = _bootstrap.connect().sync().channel();
_brokerMetrics.setValueOfGlobalGauge(BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS,
System.currentTimeMillis() - startTime);
}
- long sendRequestStartTimeMs = System.currentTimeMillis();
+ }
+
+ void sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse asyncQueryResponse,
+ ServerRoutingInstance serverRoutingInstance, byte[] requestBytes) {
+ long startTimeMs = System.currentTimeMillis();
_channel.writeAndFlush(Unpooled.wrappedBuffer(requestBytes)).addListener(f -> {
- long requestSentLatencyMs = System.currentTimeMillis() - sendRequestStartTimeMs;
+ long requestSentLatencyMs = System.currentTimeMillis() - startTimeMs;
_brokerMetrics.addTimedTableValue(rawTableName, BrokerTimer.NETTY_CONNECTION_SEND_REQUEST_LATENCY,
requestSentLatencyMs, TimeUnit.MILLISECONDS);
asyncQueryResponse.markRequestSent(serverRoutingInstance, requestSentLatencyMs);
});
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_REQUESTS_SENT, 1);
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_BYTES_SENT, requestBytes.length);
}
+
+ boolean connect() {
+ try {
+ if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
+ try {
+ connectWithoutLocking();
+ return true;
+ } finally {
+ _channelLock.unlock();
+ }
+ }
+ } catch (Exception e) {
+ // Ignored
Review Comment:
Changed it to throw the exception and let the caller log the exception to have consistent behavior as `sendRequest()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849788623
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
+ protected final String _name = getClass().getSimpleName();
+ protected final List<Listener> _listeners = new ArrayList<>();
+ protected final ConcurrentHashMap<String, RetryInfo> _unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
+
+ protected BrokerMetrics _brokerMetrics;
+ protected long _retryInitialDelayNs;
+ protected double _retryDelayFactor;
+ protected int _maxRetries;
+ protected Thread _retryThread;
+
+ protected volatile boolean _running;
+
+ @Override
+ public void init(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ _brokerMetrics = brokerMetrics;
+ long retryInitialDelayMs = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
+ Broker.FailureDetector.DEFAULT_RETRY_INITIAL_DELAY_MS);
+ _retryInitialDelayNs = TimeUnit.MILLISECONDS.toNanos(retryInitialDelayMs);
+ _retryDelayFactor = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
+ Broker.FailureDetector.DEFAULT_RETRY_DELAY_FACTOR);
+ _maxRetries =
+ config.getProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES, Broker.FailureDetector.DEFAULT_MAX_RETIRES);
+ _logger.info("Initialized {} with retry initial delay: {}ms, exponential backoff factor: {}, max " + "retries: {}",
+ _name, retryInitialDelayMs, _retryDelayFactor, _maxRetries);
+ }
+
+ @Override
+ public void register(Listener listener) {
+ _listeners.add(listener);
+ }
+
+ @Override
+ public void start() {
+ _logger.info("Starting {}", _name);
+ _running = true;
+
+ _retryThread = new Thread(() -> {
Review Comment:
And regarding overhead, I really wouldn’t worry about it at all, it’s a third or fourth order differential
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847754858
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/BrokerRoutingManager.java:
##########
@@ -214,87 +219,178 @@ private void processInstanceConfigChange() {
long startTimeMs = System.currentTimeMillis();
List<ZNRecord> instanceConfigZNRecords =
- _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT,
- CommonConstants.Helix.ZkClient.RETRY_COUNT, CommonConstants.Helix.ZkClient.RETRY_INTERVAL_MS);
+ _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT, Helix.ZkClient.RETRY_COUNT,
+ Helix.ZkClient.RETRY_INTERVAL_MS);
long fetchInstanceConfigsEndTimeMs = System.currentTimeMillis();
- // Calculate new enabled and disabled instances
- Set<String> enabledInstances = new HashSet<>();
- List<String> newEnabledInstances = new ArrayList<>();
+ // Calculate new enabled and disabled servers
+ Set<String> enabledServers = new HashSet<>();
+ List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
- String instance = instanceConfigZNRecord.getId();
- if (isInstanceEnabled(instanceConfigZNRecord)) {
- enabledInstances.add(instance);
+ String instanceId = instanceConfigZNRecord.getId();
+ if (isEnabledServer(instanceConfigZNRecord)) {
+ enabledServers.add(instanceId);
// Always refresh the server instance with the latest instance config in case it changes
ServerInstance serverInstance = new ServerInstance(new InstanceConfig(instanceConfigZNRecord));
- if (_enabledServerInstanceMap.put(instance, serverInstance) == null) {
- newEnabledInstances.add(instance);
+ if (_enabledServerInstanceMap.put(instanceId, serverInstance) == null) {
+ newEnabledServers.add(instanceId);
+
+ // NOTE: Remove new enabled server from excluded servers because the server is likely being restarted
+ if (_excludedServers.remove(instanceId)) {
+ LOGGER.info("Got excluded server: {} re-enabled, including it into the routing", instanceId);
+ }
}
}
}
- List<String> newDisabledInstances = new ArrayList<>();
+ List<String> newDisabledServers = new ArrayList<>();
for (String instance : _enabledServerInstanceMap.keySet()) {
- if (!enabledInstances.contains(instance)) {
- newDisabledInstances.add(instance);
+ if (!enabledServers.contains(instance)) {
+ newDisabledServers.add(instance);
}
}
- List<String> changedInstances = new ArrayList<>(newEnabledInstances.size() + newDisabledInstances.size());
- changedInstances.addAll(newEnabledInstances);
- changedInstances.addAll(newDisabledInstances);
- long calculateChangedInstancesEndTimeMs = System.currentTimeMillis();
-
- // Early terminate if there is no instance changed
- if (changedInstances.isEmpty()) {
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms) without instance change", calculateChangedInstancesEndTimeMs - startTimeMs,
- instanceConfigZNRecords.size(), fetchInstanceConfigsEndTimeMs - startTimeMs,
- calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs);
+
+ // Calculate the routable servers and the changed routable servers
+ List<String> changedServers = new ArrayList<>(newEnabledServers.size() + newDisabledServers.size());
+ if (_excludedServers.isEmpty()) {
+ _routableServers = enabledServers;
+ changedServers.addAll(newEnabledServers);
+ changedServers.addAll(newDisabledServers);
+ } else {
+ enabledServers.removeAll(_excludedServers);
+ _routableServers = enabledServers;
+ // NOTE: All new enabled servers are routable
+ changedServers.addAll(newEnabledServers);
+ for (String newDisabledServer : newDisabledServers) {
+ if (_excludedServers.contains(newDisabledServer)) {
+ changedServers.add(newDisabledServer);
+ }
+ }
+ }
+ long calculateChangedServersEndTimeMs = System.currentTimeMillis();
+
+ // Early terminate if there is no changed servers
+ if (changedServers.isEmpty()) {
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms) without instance change",
+ calculateChangedServersEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
+ fetchInstanceConfigsEndTimeMs - startTimeMs,
+ calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs);
return;
}
// Update routing entry for all tables
for (RoutingEntry routingEntry : _routingEntryMap.values()) {
try {
- routingEntry.onInstancesChange(enabledInstances, changedInstances);
+ routingEntry.onInstancesChange(_routableServers, changedServers);
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry on instances change for table: {}",
routingEntry.getTableNameWithType(), e);
}
}
long updateRoutingEntriesEndTimeMs = System.currentTimeMillis();
- // Remove new disabled instances from _enabledServerInstanceMap after updating all routing entries to ensure it
- // always contains the selected instances
- for (String newDisabledInstance : newDisabledInstances) {
+ // Remove new disabled servers from _enabledServerInstanceMap after updating all routing entries to ensure it
+ // always contains the selected servers
+ for (String newDisabledInstance : newDisabledServers) {
_enabledServerInstanceMap.remove(newDisabledInstance);
}
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms, update {} routing entries: {}ms), new enabled instances: {}, new disabled instances: {}",
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms, update {} routing entries: {}ms), "
+ + "new enabled servers: {}, new disabled servers: {}, excluded servers: {}",
updateRoutingEntriesEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
- fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs,
- _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedInstancesEndTimeMs,
- newEnabledInstances, newDisabledInstances);
+ fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs,
+ _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedServersEndTimeMs, newEnabledServers,
+ newDisabledServers, _excludedServers);
}
- private static boolean isInstanceEnabled(ZNRecord instanceConfigZNRecord) {
+ private static boolean isEnabledServer(ZNRecord instanceConfigZNRecord) {
+ String instanceId = instanceConfigZNRecord.getId();
+ if (!instanceId.startsWith(Helix.PREFIX_OF_SERVER_INSTANCE)) {
+ // NOTE: Some legacy configs might not contain the instance type prefix
+ if (instanceId.startsWith(Helix.PREFIX_OF_CONTROLLER_INSTANCE) || instanceId.startsWith(
+ Helix.PREFIX_OF_BROKER_INSTANCE) || instanceId.startsWith(Helix.PREFIX_OF_MINION_INSTANCE)) {
+ return false;
+ }
+ }
if ("false".equalsIgnoreCase(
instanceConfigZNRecord.getSimpleField(InstanceConfig.InstanceConfigProperty.HELIX_ENABLED.name()))) {
return false;
}
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.IS_SHUTDOWN_IN_PROGRESS))) {
return false;
}
//noinspection RedundantIfStatement
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.QUERIES_DISABLED))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.QUERIES_DISABLED))) {
return false;
}
return true;
}
+ /**
+ * Excludes a server from the routing.
+ */
+ public synchronized void excludeServerFromRouting(String instanceId) {
+ LOGGER.warn("Excluding server: {} from routing", instanceId);
+ if (!_excludedServers.add(instanceId)) {
+ LOGGER.info("Server: {} is already excluded from routing, skipping updating the routing", instanceId);
+ return;
+ }
+ if (!_routableServers.contains(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.remove(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when excluding server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Excluded server: {} from routing in {}ms (updated {} routing entries)", instanceId,
+ System.currentTimeMillis() - startTimeMs, _routingEntryMap.size());
+ }
+
+ /**
+ * Includes a previous excluded server to the routing.
+ */
+ public synchronized void includeServerToRouting(String instanceId) {
+ LOGGER.info("Including server: {} to routing", instanceId);
+ if (!_excludedServers.remove(instanceId)) {
+ LOGGER.info("Server: {} is not previously excluded, skipping updating the routing", instanceId);
+ }
+ if (!_enabledServerInstanceMap.containsKey(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.add(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when including server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Included server: {} to routing in {}ms (updated {} routing entries)", instanceId,
Review Comment:
Why record durations with millitime here and nanotime elsewhere? They are implemented using different clocks which makes the durations incomparable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847761220
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerRoutingInstance.java:
##########
@@ -87,17 +100,20 @@ public boolean equals(Object o) {
if (this == o) {
return true;
}
- if (o == null || getClass() != o.getClass()) {
+ if (!(o instanceof ServerRoutingInstance)) {
return false;
}
ServerRoutingInstance that = (ServerRoutingInstance) o;
- return _tlsEnabled == that._tlsEnabled && _port == that._port && _hostname.equals(that._hostname)
- && _tableType == that._tableType;
+ // NOTE: Only check hostname, port and tableType for performance concern because they can identify a routing
+ // instance within the same query
+ return _hostname.equals(that._hostname) && _port == that._port && _tableType == that._tableType;
}
@Override
public int hashCode() {
- return Objects.hash(_tlsEnabled, _hostname, _port, _tableType);
+ // NOTE: Only check hostname, port and tableType for performance concern because they can identify a routing
+ // instance within the same query
+ return Objects.hash(_hostname, _port, _tableType);
Review Comment:
If you care about performance here, don’t use Objects.hash - it boxes its parameters. Instead calculate the polynomial manually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849781662
##########
pinot-broker/src/test/java/org/apache/pinot/broker/failuredetector/ConnectionFailureDetectorTest.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.metrics.PinotMetricUtils;
+import org.apache.pinot.core.transport.QueryResponse;
+import org.apache.pinot.core.transport.ServerRoutingInstance;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.apache.pinot.util.TestUtils;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+public class ConnectionFailureDetectorTest {
+ private static final String INSTANCE_ID = "Server_localhost_1234";
+
+ private BrokerMetrics _brokerMetrics;
+ private FailureDetector _failureDetector;
+ private Listener _listener;
+
+ @BeforeClass
+ public void setUp() {
+ PinotConfiguration config = new PinotConfiguration();
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME, ConnectionFailureDetector.class.getName());
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS, 100);
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR, 1);
+ _brokerMetrics = new BrokerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
+ _failureDetector = FailureDetectorFactory.getFailureDetector(config, _brokerMetrics);
+ assertTrue(_failureDetector instanceof ConnectionFailureDetector);
+ _listener = new Listener();
+ _failureDetector.register(_listener);
+ _failureDetector.start();
+ }
+
+ @Test
+ public void testConnectionFailure() {
Review Comment:
It must run in sequence, and the result is accumulative. But I agree the logic is kind of repeated, so move the common part into a separate method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] kkrugler commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
kkrugler commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r848975615
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
Review Comment:
Thanks for clarifying...but why do you want classes that extend `BaseExponentialBackoffRetryFailureDetector` to use its logger? Is there some issue with the concrete class having its own logger?
Just asking because I see 25+ instances of abstract classes with private static loggers, but only one other class `BaseSegmentJob` that does the same thing as here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849776928
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
+ protected final String _name = getClass().getSimpleName();
+ protected final List<Listener> _listeners = new ArrayList<>();
+ protected final ConcurrentHashMap<String, RetryInfo> _unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
+
+ protected BrokerMetrics _brokerMetrics;
+ protected long _retryInitialDelayNs;
+ protected double _retryDelayFactor;
+ protected int _maxRetries;
+ protected Thread _retryThread;
+
+ protected volatile boolean _running;
+
+ @Override
+ public void init(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ _brokerMetrics = brokerMetrics;
+ long retryInitialDelayMs = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
+ Broker.FailureDetector.DEFAULT_RETRY_INITIAL_DELAY_MS);
+ _retryInitialDelayNs = TimeUnit.MILLISECONDS.toNanos(retryInitialDelayMs);
+ _retryDelayFactor = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
+ Broker.FailureDetector.DEFAULT_RETRY_DELAY_FACTOR);
+ _maxRetries =
+ config.getProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES, Broker.FailureDetector.DEFAULT_MAX_RETIRES);
+ _logger.info("Initialized {} with retry initial delay: {}ms, exponential backoff factor: {}, max " + "retries: {}",
+ _name, retryInitialDelayMs, _retryDelayFactor, _maxRetries);
+ }
+
+ @Override
+ public void register(Listener listener) {
+ _listeners.add(listener);
+ }
+
+ @Override
+ public void start() {
+ _logger.info("Starting {}", _name);
+ _running = true;
+
+ _retryThread = new Thread(() -> {
+ while (_running) {
+ try {
+ long earliestRetryTimeNs = System.nanoTime() + _retryInitialDelayNs;
+ for (Map.Entry<String, RetryInfo> entry : _unhealthyServerRetryInfoMap.entrySet()) {
+ RetryInfo retryInfo = entry.getValue();
+ if (System.nanoTime() > retryInfo._retryTimeNs) {
+ String instanceId = entry.getKey();
+ if (retryInfo._numRetires == _maxRetries) {
+ _logger.warn(
+ "Unhealthy server: {} already reaches the max retries: {}, do not retry again and treat it as "
+ + "healthy so that the listeners do not lose track of the server", instanceId, _maxRetries);
+ markServerHealthy(instanceId);
+ continue;
+ }
+ _logger.info("Retry unhealthy server: {}", instanceId);
+ for (Listener listener : _listeners) {
+ listener.retryUnhealthyServer(instanceId, this);
+ }
+ // Update the retry info
+ retryInfo._retryDelayNs = (long) (retryInfo._retryDelayNs * _retryDelayFactor);
+ retryInfo._retryTimeNs = System.nanoTime() + retryInfo._retryDelayNs;
+ retryInfo._numRetires++;
+ } else {
+ earliestRetryTimeNs = Math.min(earliestRetryTimeNs, retryInfo._retryTimeNs);
+ }
+ }
+ //noinspection BusyWait
+ Thread.sleep(TimeUnit.NANOSECONDS.toMillis(earliestRetryTimeNs - System.nanoTime()));
Review Comment:
Good point, this can also avoid scanning the map each time it wakes up. Applied the changes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847755763
##########
pinot-broker/src/test/java/org/apache/pinot/broker/failuredetector/ConnectionFailureDetectorTest.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.metrics.PinotMetricUtils;
+import org.apache.pinot.core.transport.QueryResponse;
+import org.apache.pinot.core.transport.ServerRoutingInstance;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.apache.pinot.util.TestUtils;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+public class ConnectionFailureDetectorTest {
+ private static final String INSTANCE_ID = "Server_localhost_1234";
+
+ private BrokerMetrics _brokerMetrics;
+ private FailureDetector _failureDetector;
+ private Listener _listener;
+
+ @BeforeClass
+ public void setUp() {
+ PinotConfiguration config = new PinotConfiguration();
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME, ConnectionFailureDetector.class.getName());
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS, 100);
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR, 1);
+ _brokerMetrics = new BrokerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
+ _failureDetector = FailureDetectorFactory.getFailureDetector(config, _brokerMetrics);
+ assertTrue(_failureDetector instanceof ConnectionFailureDetector);
+ _listener = new Listener();
+ _failureDetector.register(_listener);
+ _failureDetector.start();
+ }
+
+ @Test
+ public void testConnectionFailure() {
Review Comment:
The logic in this test is repeated. Why not try factoring out the parameters into a data source so it’s more declarative and it’s easier to see what’s being asserted?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang merged pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang merged PR #8491:
URL: https://github.com/apache/pinot/pull/8491
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849812538
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/FailureDetectorFactory.java:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.broker.failuredetector;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class FailureDetectorFactory {
+ private FailureDetectorFactory() {
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FailureDetectorFactory.class);
+
+ public static FailureDetector getFailureDetector(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ String className = config.getProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME);
+ if (StringUtils.isEmpty(className)) {
+ LOGGER.info("Class name is not configured, falling back to NoOpFailureDetector");
+ return new NoOpFailureDetector();
+ } else {
+ LOGGER.info("Initializing failure detector with class: {}", className);
+ try {
+ FailureDetector failureDetector =
+ (FailureDetector) Class.forName(className).getDeclaredConstructor().newInstance();
Review Comment:
I feel changing the model of how user plug in a custom implementation is out of the scope of this PR (currently we plug in custom implementation via `PluginManager` and class name). We may revisit this in a separate discussion just for the plug in model
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849785673
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/FailureDetectorFactory.java:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.broker.failuredetector;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class FailureDetectorFactory {
+ private FailureDetectorFactory() {
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FailureDetectorFactory.class);
+
+ public static FailureDetector getFailureDetector(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ String className = config.getProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME);
+ if (StringUtils.isEmpty(className)) {
+ LOGGER.info("Class name is not configured, falling back to NoOpFailureDetector");
+ return new NoOpFailureDetector();
+ } else {
+ LOGGER.info("Initializing failure detector with class: {}", className);
+ try {
+ FailureDetector failureDetector =
+ (FailureDetector) Class.forName(className).getDeclaredConstructor().newInstance();
Review Comment:
The point is exposing the class name at all prevents from changing the class name, which isn’t good. If you have a separate configuration setting and map that to the class you want to use, without exposing the class name, you can refactor if you want.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r848983278
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/BrokerRoutingManager.java:
##########
@@ -214,87 +219,178 @@ private void processInstanceConfigChange() {
long startTimeMs = System.currentTimeMillis();
List<ZNRecord> instanceConfigZNRecords =
- _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT,
- CommonConstants.Helix.ZkClient.RETRY_COUNT, CommonConstants.Helix.ZkClient.RETRY_INTERVAL_MS);
+ _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT, Helix.ZkClient.RETRY_COUNT,
+ Helix.ZkClient.RETRY_INTERVAL_MS);
long fetchInstanceConfigsEndTimeMs = System.currentTimeMillis();
- // Calculate new enabled and disabled instances
- Set<String> enabledInstances = new HashSet<>();
- List<String> newEnabledInstances = new ArrayList<>();
+ // Calculate new enabled and disabled servers
+ Set<String> enabledServers = new HashSet<>();
+ List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
- String instance = instanceConfigZNRecord.getId();
- if (isInstanceEnabled(instanceConfigZNRecord)) {
- enabledInstances.add(instance);
+ String instanceId = instanceConfigZNRecord.getId();
+ if (isEnabledServer(instanceConfigZNRecord)) {
+ enabledServers.add(instanceId);
// Always refresh the server instance with the latest instance config in case it changes
ServerInstance serverInstance = new ServerInstance(new InstanceConfig(instanceConfigZNRecord));
- if (_enabledServerInstanceMap.put(instance, serverInstance) == null) {
- newEnabledInstances.add(instance);
+ if (_enabledServerInstanceMap.put(instanceId, serverInstance) == null) {
+ newEnabledServers.add(instanceId);
+
+ // NOTE: Remove new enabled server from excluded servers because the server is likely being restarted
+ if (_excludedServers.remove(instanceId)) {
+ LOGGER.info("Got excluded server: {} re-enabled, including it into the routing", instanceId);
+ }
}
}
}
- List<String> newDisabledInstances = new ArrayList<>();
+ List<String> newDisabledServers = new ArrayList<>();
for (String instance : _enabledServerInstanceMap.keySet()) {
- if (!enabledInstances.contains(instance)) {
- newDisabledInstances.add(instance);
+ if (!enabledServers.contains(instance)) {
+ newDisabledServers.add(instance);
}
}
- List<String> changedInstances = new ArrayList<>(newEnabledInstances.size() + newDisabledInstances.size());
- changedInstances.addAll(newEnabledInstances);
- changedInstances.addAll(newDisabledInstances);
- long calculateChangedInstancesEndTimeMs = System.currentTimeMillis();
-
- // Early terminate if there is no instance changed
- if (changedInstances.isEmpty()) {
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms) without instance change", calculateChangedInstancesEndTimeMs - startTimeMs,
- instanceConfigZNRecords.size(), fetchInstanceConfigsEndTimeMs - startTimeMs,
- calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs);
+
+ // Calculate the routable servers and the changed routable servers
+ List<String> changedServers = new ArrayList<>(newEnabledServers.size() + newDisabledServers.size());
+ if (_excludedServers.isEmpty()) {
+ _routableServers = enabledServers;
+ changedServers.addAll(newEnabledServers);
+ changedServers.addAll(newDisabledServers);
+ } else {
+ enabledServers.removeAll(_excludedServers);
+ _routableServers = enabledServers;
+ // NOTE: All new enabled servers are routable
+ changedServers.addAll(newEnabledServers);
+ for (String newDisabledServer : newDisabledServers) {
+ if (_excludedServers.contains(newDisabledServer)) {
+ changedServers.add(newDisabledServer);
+ }
+ }
+ }
+ long calculateChangedServersEndTimeMs = System.currentTimeMillis();
+
+ // Early terminate if there is no changed servers
+ if (changedServers.isEmpty()) {
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms) without instance change",
+ calculateChangedServersEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
+ fetchInstanceConfigsEndTimeMs - startTimeMs,
+ calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs);
return;
}
// Update routing entry for all tables
for (RoutingEntry routingEntry : _routingEntryMap.values()) {
try {
- routingEntry.onInstancesChange(enabledInstances, changedInstances);
+ routingEntry.onInstancesChange(_routableServers, changedServers);
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry on instances change for table: {}",
routingEntry.getTableNameWithType(), e);
}
}
long updateRoutingEntriesEndTimeMs = System.currentTimeMillis();
- // Remove new disabled instances from _enabledServerInstanceMap after updating all routing entries to ensure it
- // always contains the selected instances
- for (String newDisabledInstance : newDisabledInstances) {
+ // Remove new disabled servers from _enabledServerInstanceMap after updating all routing entries to ensure it
+ // always contains the selected servers
+ for (String newDisabledInstance : newDisabledServers) {
_enabledServerInstanceMap.remove(newDisabledInstance);
}
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms, update {} routing entries: {}ms), new enabled instances: {}, new disabled instances: {}",
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms, update {} routing entries: {}ms), "
+ + "new enabled servers: {}, new disabled servers: {}, excluded servers: {}",
updateRoutingEntriesEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
- fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs,
- _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedInstancesEndTimeMs,
- newEnabledInstances, newDisabledInstances);
+ fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs,
+ _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedServersEndTimeMs, newEnabledServers,
+ newDisabledServers, _excludedServers);
}
- private static boolean isInstanceEnabled(ZNRecord instanceConfigZNRecord) {
+ private static boolean isEnabledServer(ZNRecord instanceConfigZNRecord) {
+ String instanceId = instanceConfigZNRecord.getId();
+ if (!instanceId.startsWith(Helix.PREFIX_OF_SERVER_INSTANCE)) {
+ // NOTE: Some legacy configs might not contain the instance type prefix
+ if (instanceId.startsWith(Helix.PREFIX_OF_CONTROLLER_INSTANCE) || instanceId.startsWith(
+ Helix.PREFIX_OF_BROKER_INSTANCE) || instanceId.startsWith(Helix.PREFIX_OF_MINION_INSTANCE)) {
+ return false;
+ }
+ }
if ("false".equalsIgnoreCase(
instanceConfigZNRecord.getSimpleField(InstanceConfig.InstanceConfigProperty.HELIX_ENABLED.name()))) {
return false;
}
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.IS_SHUTDOWN_IN_PROGRESS))) {
return false;
}
//noinspection RedundantIfStatement
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.QUERIES_DISABLED))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.QUERIES_DISABLED))) {
return false;
}
return true;
}
+ /**
+ * Excludes a server from the routing.
+ */
+ public synchronized void excludeServerFromRouting(String instanceId) {
+ LOGGER.warn("Excluding server: {} from routing", instanceId);
+ if (!_excludedServers.add(instanceId)) {
+ LOGGER.info("Server: {} is already excluded from routing, skipping updating the routing", instanceId);
+ return;
+ }
+ if (!_routableServers.contains(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.remove(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when excluding server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Excluded server: {} from routing in {}ms (updated {} routing entries)", instanceId,
+ System.currentTimeMillis() - startTimeMs, _routingEntryMap.size());
+ }
+
+ /**
+ * Includes a previous excluded server to the routing.
+ */
+ public synchronized void includeServerToRouting(String instanceId) {
+ LOGGER.info("Including server: {} to routing", instanceId);
+ if (!_excludedServers.remove(instanceId)) {
+ LOGGER.info("Server: {} is not previously excluded, skipping updating the routing", instanceId);
+ }
+ if (!_enabledServerInstanceMap.containsKey(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.add(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when including server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Included server: {} to routing in {}ms (updated {} routing entries)", instanceId,
Review Comment:
My personal preference is to use nanotime to measure the duration, but there are some on-going discussion around which one should be favorable for performance perspective, and there is no conclusion yet. Since we already use millis in this class, I want to keep it consistent. Why would it cause incomparable durations? All the time we took here are in millis.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847750591
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/FailureDetectorFactory.java:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.broker.failuredetector;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class FailureDetectorFactory {
+ private FailureDetectorFactory() {
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FailureDetectorFactory.class);
+
+ public static FailureDetector getFailureDetector(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ String className = config.getProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME);
+ if (StringUtils.isEmpty(className)) {
+ LOGGER.info("Class name is not configured, falling back to NoOpFailureDetector");
+ return new NoOpFailureDetector();
+ } else {
+ LOGGER.info("Initializing failure detector with class: {}", className);
+ try {
+ FailureDetector failureDetector =
+ (FailureDetector) Class.forName(className).getDeclaredConstructor().newInstance();
Review Comment:
(You can’t even move the class to another package without breaking users as things stand…)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847746972
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
+ protected final String _name = getClass().getSimpleName();
+ protected final List<Listener> _listeners = new ArrayList<>();
+ protected final ConcurrentHashMap<String, RetryInfo> _unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
+
+ protected BrokerMetrics _brokerMetrics;
+ protected long _retryInitialDelayNs;
+ protected double _retryDelayFactor;
+ protected int _maxRetries;
+ protected Thread _retryThread;
+
+ protected volatile boolean _running;
+
+ @Override
+ public void init(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ _brokerMetrics = brokerMetrics;
+ long retryInitialDelayMs = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
+ Broker.FailureDetector.DEFAULT_RETRY_INITIAL_DELAY_MS);
+ _retryInitialDelayNs = TimeUnit.MILLISECONDS.toNanos(retryInitialDelayMs);
+ _retryDelayFactor = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
+ Broker.FailureDetector.DEFAULT_RETRY_DELAY_FACTOR);
+ _maxRetries =
+ config.getProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES, Broker.FailureDetector.DEFAULT_MAX_RETIRES);
+ _logger.info("Initialized {} with retry initial delay: {}ms, exponential backoff factor: {}, max " + "retries: {}",
+ _name, retryInitialDelayMs, _retryDelayFactor, _maxRetries);
+ }
+
+ @Override
+ public void register(Listener listener) {
+ _listeners.add(listener);
+ }
+
+ @Override
+ public void start() {
+ _logger.info("Starting {}", _name);
+ _running = true;
+
+ _retryThread = new Thread(() -> {
Review Comment:
Why don’t you use a single threaded ExecutorService here instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847750216
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/FailureDetectorFactory.java:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.broker.failuredetector;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class FailureDetectorFactory {
+ private FailureDetectorFactory() {
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FailureDetectorFactory.class);
+
+ public static FailureDetector getFailureDetector(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ String className = config.getProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME);
+ if (StringUtils.isEmpty(className)) {
+ LOGGER.info("Class name is not configured, falling back to NoOpFailureDetector");
+ return new NoOpFailureDetector();
+ } else {
+ LOGGER.info("Initializing failure detector with class: {}", className);
+ try {
+ FailureDetector failureDetector =
+ (FailureDetector) Class.forName(className).getDeclaredConstructor().newInstance();
Review Comment:
This isn’t a good configuration interface because it coupled users with the implementation. Instead have a configuration name and map it to an implementation class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r848974745
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/FailureDetectorFactory.java:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.broker.failuredetector;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class FailureDetectorFactory {
+ private FailureDetectorFactory() {
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FailureDetectorFactory.class);
+
+ public static FailureDetector getFailureDetector(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ String className = config.getProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME);
+ if (StringUtils.isEmpty(className)) {
+ LOGGER.info("Class name is not configured, falling back to NoOpFailureDetector");
+ return new NoOpFailureDetector();
+ } else {
+ LOGGER.info("Initializing failure detector with class: {}", className);
+ try {
+ FailureDetector failureDetector =
+ (FailureDetector) Class.forName(className).getDeclaredConstructor().newInstance();
Review Comment:
I don't fully follow the comment here, but changed it to use `PluginManager.get().createInstance(className)` so that we can take more plug-in classes. Does the problem still exist within the `PluginManager`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849776599
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
+ protected final String _name = getClass().getSimpleName();
+ protected final List<Listener> _listeners = new ArrayList<>();
+ protected final ConcurrentHashMap<String, RetryInfo> _unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
+
+ protected BrokerMetrics _brokerMetrics;
+ protected long _retryInitialDelayNs;
+ protected double _retryDelayFactor;
+ protected int _maxRetries;
+ protected Thread _retryThread;
+
+ protected volatile boolean _running;
+
+ @Override
+ public void init(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ _brokerMetrics = brokerMetrics;
+ long retryInitialDelayMs = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
+ Broker.FailureDetector.DEFAULT_RETRY_INITIAL_DELAY_MS);
+ _retryInitialDelayNs = TimeUnit.MILLISECONDS.toNanos(retryInitialDelayMs);
+ _retryDelayFactor = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
+ Broker.FailureDetector.DEFAULT_RETRY_DELAY_FACTOR);
+ _maxRetries =
+ config.getProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES, Broker.FailureDetector.DEFAULT_MAX_RETIRES);
+ _logger.info("Initialized {} with retry initial delay: {}ms, exponential backoff factor: {}, max " + "retries: {}",
+ _name, retryInitialDelayMs, _retryDelayFactor, _maxRetries);
+ }
+
+ @Override
+ public void register(Listener listener) {
+ _listeners.add(listener);
+ }
+
+ @Override
+ public void start() {
+ _logger.info("Starting {}", _name);
+ _running = true;
+
+ _retryThread = new Thread(() -> {
Review Comment:
Since we will only run one thread through the lifetime of this component, using a single threaded `ExecutorService` will only add overhead but not provide extra value (the code won't be simplified as well). I prefer keeping it this way
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847759869
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -181,24 +191,43 @@ private void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryRespo
}
}
- private void sendRequest(String rawTableName, AsyncQueryResponse asyncQueryResponse,
- ServerRoutingInstance serverRoutingInstance, byte[] requestBytes)
- throws Exception {
+ void connectWithoutLocking()
+ throws InterruptedException {
if (_channel == null || !_channel.isActive()) {
long startTime = System.currentTimeMillis();
_channel = _bootstrap.connect().sync().channel();
_brokerMetrics.setValueOfGlobalGauge(BrokerGauge.NETTY_CONNECTION_CONNECT_TIME_MS,
System.currentTimeMillis() - startTime);
}
- long sendRequestStartTimeMs = System.currentTimeMillis();
+ }
+
+ void sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse asyncQueryResponse,
+ ServerRoutingInstance serverRoutingInstance, byte[] requestBytes) {
+ long startTimeMs = System.currentTimeMillis();
_channel.writeAndFlush(Unpooled.wrappedBuffer(requestBytes)).addListener(f -> {
- long requestSentLatencyMs = System.currentTimeMillis() - sendRequestStartTimeMs;
+ long requestSentLatencyMs = System.currentTimeMillis() - startTimeMs;
_brokerMetrics.addTimedTableValue(rawTableName, BrokerTimer.NETTY_CONNECTION_SEND_REQUEST_LATENCY,
requestSentLatencyMs, TimeUnit.MILLISECONDS);
asyncQueryResponse.markRequestSent(serverRoutingInstance, requestSentLatencyMs);
});
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_REQUESTS_SENT, 1);
_brokerMetrics.addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_BYTES_SENT, requestBytes.length);
}
+
+ boolean connect() {
+ try {
+ if (_channelLock.tryLock(TRY_CONNECT_CHANNEL_LOCK_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
+ try {
+ connectWithoutLocking();
+ return true;
+ } finally {
+ _channelLock.unlock();
+ }
+ }
+ } catch (Exception e) {
+ // Ignored
Review Comment:
Should probably not swallow this, at least log if at debug
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849964676
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/FailureDetectorFactory.java:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.broker.failuredetector;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class FailureDetectorFactory {
+ private FailureDetectorFactory() {
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FailureDetectorFactory.class);
+
+ public static FailureDetector getFailureDetector(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ String className = config.getProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME);
+ if (StringUtils.isEmpty(className)) {
+ LOGGER.info("Class name is not configured, falling back to NoOpFailureDetector");
+ return new NoOpFailureDetector();
+ } else {
+ LOGGER.info("Initializing failure detector with class: {}", className);
+ try {
+ FailureDetector failureDetector =
+ (FailureDetector) Class.forName(className).getDeclaredConstructor().newInstance();
Review Comment:
Added a config for the failure detector type so that the built-in detectors can be configured using the type
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849815489
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/BrokerRoutingManager.java:
##########
@@ -214,87 +219,178 @@ private void processInstanceConfigChange() {
long startTimeMs = System.currentTimeMillis();
List<ZNRecord> instanceConfigZNRecords =
- _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT,
- CommonConstants.Helix.ZkClient.RETRY_COUNT, CommonConstants.Helix.ZkClient.RETRY_INTERVAL_MS);
+ _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT, Helix.ZkClient.RETRY_COUNT,
+ Helix.ZkClient.RETRY_INTERVAL_MS);
long fetchInstanceConfigsEndTimeMs = System.currentTimeMillis();
- // Calculate new enabled and disabled instances
- Set<String> enabledInstances = new HashSet<>();
- List<String> newEnabledInstances = new ArrayList<>();
+ // Calculate new enabled and disabled servers
+ Set<String> enabledServers = new HashSet<>();
+ List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
- String instance = instanceConfigZNRecord.getId();
- if (isInstanceEnabled(instanceConfigZNRecord)) {
- enabledInstances.add(instance);
+ String instanceId = instanceConfigZNRecord.getId();
+ if (isEnabledServer(instanceConfigZNRecord)) {
+ enabledServers.add(instanceId);
// Always refresh the server instance with the latest instance config in case it changes
ServerInstance serverInstance = new ServerInstance(new InstanceConfig(instanceConfigZNRecord));
- if (_enabledServerInstanceMap.put(instance, serverInstance) == null) {
- newEnabledInstances.add(instance);
+ if (_enabledServerInstanceMap.put(instanceId, serverInstance) == null) {
+ newEnabledServers.add(instanceId);
+
+ // NOTE: Remove new enabled server from excluded servers because the server is likely being restarted
+ if (_excludedServers.remove(instanceId)) {
+ LOGGER.info("Got excluded server: {} re-enabled, including it into the routing", instanceId);
+ }
}
}
}
- List<String> newDisabledInstances = new ArrayList<>();
+ List<String> newDisabledServers = new ArrayList<>();
for (String instance : _enabledServerInstanceMap.keySet()) {
- if (!enabledInstances.contains(instance)) {
- newDisabledInstances.add(instance);
+ if (!enabledServers.contains(instance)) {
+ newDisabledServers.add(instance);
}
}
- List<String> changedInstances = new ArrayList<>(newEnabledInstances.size() + newDisabledInstances.size());
- changedInstances.addAll(newEnabledInstances);
- changedInstances.addAll(newDisabledInstances);
- long calculateChangedInstancesEndTimeMs = System.currentTimeMillis();
-
- // Early terminate if there is no instance changed
- if (changedInstances.isEmpty()) {
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms) without instance change", calculateChangedInstancesEndTimeMs - startTimeMs,
- instanceConfigZNRecords.size(), fetchInstanceConfigsEndTimeMs - startTimeMs,
- calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs);
+
+ // Calculate the routable servers and the changed routable servers
+ List<String> changedServers = new ArrayList<>(newEnabledServers.size() + newDisabledServers.size());
+ if (_excludedServers.isEmpty()) {
+ _routableServers = enabledServers;
+ changedServers.addAll(newEnabledServers);
+ changedServers.addAll(newDisabledServers);
+ } else {
+ enabledServers.removeAll(_excludedServers);
+ _routableServers = enabledServers;
+ // NOTE: All new enabled servers are routable
+ changedServers.addAll(newEnabledServers);
+ for (String newDisabledServer : newDisabledServers) {
+ if (_excludedServers.contains(newDisabledServer)) {
+ changedServers.add(newDisabledServer);
+ }
+ }
+ }
+ long calculateChangedServersEndTimeMs = System.currentTimeMillis();
+
+ // Early terminate if there is no changed servers
+ if (changedServers.isEmpty()) {
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms) without instance change",
+ calculateChangedServersEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
+ fetchInstanceConfigsEndTimeMs - startTimeMs,
+ calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs);
return;
}
// Update routing entry for all tables
for (RoutingEntry routingEntry : _routingEntryMap.values()) {
try {
- routingEntry.onInstancesChange(enabledInstances, changedInstances);
+ routingEntry.onInstancesChange(_routableServers, changedServers);
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry on instances change for table: {}",
routingEntry.getTableNameWithType(), e);
}
}
long updateRoutingEntriesEndTimeMs = System.currentTimeMillis();
- // Remove new disabled instances from _enabledServerInstanceMap after updating all routing entries to ensure it
- // always contains the selected instances
- for (String newDisabledInstance : newDisabledInstances) {
+ // Remove new disabled servers from _enabledServerInstanceMap after updating all routing entries to ensure it
+ // always contains the selected servers
+ for (String newDisabledInstance : newDisabledServers) {
_enabledServerInstanceMap.remove(newDisabledInstance);
}
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms, update {} routing entries: {}ms), new enabled instances: {}, new disabled instances: {}",
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms, update {} routing entries: {}ms), "
+ + "new enabled servers: {}, new disabled servers: {}, excluded servers: {}",
updateRoutingEntriesEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
- fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs,
- _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedInstancesEndTimeMs,
- newEnabledInstances, newDisabledInstances);
+ fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs,
+ _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedServersEndTimeMs, newEnabledServers,
+ newDisabledServers, _excludedServers);
}
- private static boolean isInstanceEnabled(ZNRecord instanceConfigZNRecord) {
+ private static boolean isEnabledServer(ZNRecord instanceConfigZNRecord) {
+ String instanceId = instanceConfigZNRecord.getId();
+ if (!instanceId.startsWith(Helix.PREFIX_OF_SERVER_INSTANCE)) {
+ // NOTE: Some legacy configs might not contain the instance type prefix
+ if (instanceId.startsWith(Helix.PREFIX_OF_CONTROLLER_INSTANCE) || instanceId.startsWith(
+ Helix.PREFIX_OF_BROKER_INSTANCE) || instanceId.startsWith(Helix.PREFIX_OF_MINION_INSTANCE)) {
+ return false;
+ }
+ }
if ("false".equalsIgnoreCase(
instanceConfigZNRecord.getSimpleField(InstanceConfig.InstanceConfigProperty.HELIX_ENABLED.name()))) {
return false;
}
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.IS_SHUTDOWN_IN_PROGRESS))) {
return false;
}
//noinspection RedundantIfStatement
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.QUERIES_DISABLED))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.QUERIES_DISABLED))) {
return false;
}
return true;
}
+ /**
+ * Excludes a server from the routing.
+ */
+ public synchronized void excludeServerFromRouting(String instanceId) {
+ LOGGER.warn("Excluding server: {} from routing", instanceId);
+ if (!_excludedServers.add(instanceId)) {
+ LOGGER.info("Server: {} is already excluded from routing, skipping updating the routing", instanceId);
+ return;
+ }
+ if (!_routableServers.contains(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.remove(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when excluding server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Excluded server: {} from routing in {}ms (updated {} routing entries)", instanceId,
+ System.currentTimeMillis() - startTimeMs, _routingEntryMap.size());
+ }
+
+ /**
+ * Includes a previous excluded server to the routing.
+ */
+ public synchronized void includeServerToRouting(String instanceId) {
+ LOGGER.info("Including server: {} to routing", instanceId);
+ if (!_excludedServers.remove(instanceId)) {
+ LOGGER.info("Server: {} is not previously excluded, skipping updating the routing", instanceId);
+ }
+ if (!_enabledServerInstanceMap.containsKey(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.add(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when including server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Included server: {} to routing in {}ms (updated {} routing entries)", instanceId,
Review Comment:
I see. We do use the same way to record the duration within the same scope (e.g. always use `millis` in the `BrokerRoutingManager`, and always use `nanos` in `BaseExponentialBackoffRetryFailureDetector` for the retry delay), and there is no comparison between them.
For the retry delay, I prefer using the nanos because it guarantees monotonically increasing. In the `BrokerRoutingManager`, since all the existing code already uses millis, I kept it the same
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849837823
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/FailureDetectorFactory.java:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.broker.failuredetector;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class FailureDetectorFactory {
+ private FailureDetectorFactory() {
+ }
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FailureDetectorFactory.class);
+
+ public static FailureDetector getFailureDetector(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ String className = config.getProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME);
+ if (StringUtils.isEmpty(className)) {
+ LOGGER.info("Class name is not configured, falling back to NoOpFailureDetector");
+ return new NoOpFailureDetector();
+ } else {
+ LOGGER.info("Initializing failure detector with class: {}", className);
+ try {
+ FailureDetector failureDetector =
+ (FailureDetector) Class.forName(className).getDeclaredConstructor().newInstance();
Review Comment:
I agree that the plugin manager is worthy of a much bigger discussion, but that's not what's being discussed here.
You're adding a new configuration option associated with the key `Broker.FailureDetector.CONFIG_OF_CLASS_NAME`. That can either be a class name, **which means you can't change that class name ever again**, or it can be a value like "NONE", "DEFAULT" which you then _map_ to a class name to provide to the plugin manger, which allows non-breaking refactoring. By Exposing the class name, your implementation has become your interface.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849800383
##########
pinot-core/src/test/java/org/apache/pinot/core/query/reduce/StreamingReduceServiceTest.java:
##########
@@ -78,19 +74,14 @@ public Void answer(InvocationOnMock invocationOnMock)
}
});
final ExecutorService threadPoolService = Executors.newFixedThreadPool(1);
- final ServerRoutingInstance routingInstance =
- new ServerRoutingInstance("localhost", 9527, TableType.OFFLINE, false);
- //We cannot use TestNG's annotation like @Test(expectedExceptions = { IOException.class }) to verify
+ final ServerRoutingInstance routingInstance = new ServerRoutingInstance("localhost", 9527, TableType.OFFLINE);
+ // We cannot use TestNG's annotation like @Test(expectedExceptions = { IOException.class }) to verify
// because the Exception we hope to verify is nested inside the final exception.
assertTrue(verifyException(() -> {
- StreamingReduceService.processIterativeServerResponse(mock(StreamingReducer.class),
- threadPoolService,
- ImmutableMap.of(routingInstance, mockedResponse),
- 10,
- mock(BaseReduceService.ExecutionStatsAggregator.class));
- return null;
- },
- (cause) -> cause instanceof TimeoutException));
+ StreamingReduceService.processIterativeServerResponse(mock(StreamingReducer.class), threadPoolService,
+ ImmutableMap.of(routingInstance, mockedResponse), 10, mock(BaseReduceService.ExecutionStatsAggregator.class));
+ return null;
+ }, (cause) -> cause instanceof TimeoutException));
Review Comment:
I didn't check the test logic, and this test is actually unrelated to the PR except for the `ServerRoutingInstance` constructor change. Reverted the unnecessary reformat change to reduce the noise for the review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849797221
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerRoutingInstance.java:
##########
@@ -87,17 +100,20 @@ public boolean equals(Object o) {
if (this == o) {
return true;
}
- if (o == null || getClass() != o.getClass()) {
+ if (!(o instanceof ServerRoutingInstance)) {
return false;
}
ServerRoutingInstance that = (ServerRoutingInstance) o;
- return _tlsEnabled == that._tlsEnabled && _port == that._port && _hostname.equals(that._hostname)
- && _tableType == that._tableType;
+ // NOTE: Only check hostname, port and tableType for performance concern because they can identify a routing
+ // instance within the same query
+ return _hostname.equals(that._hostname) && _port == that._port && _tableType == that._tableType;
}
@Override
public int hashCode() {
- return Objects.hash(_tlsEnabled, _hostname, _port, _tableType);
+ // NOTE: Only check hostname, port and tableType for performance concern because they can identify a routing
+ // instance within the same query
+ return Objects.hash(_hostname, _port, _tableType);
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] codecov-commenter commented on pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #8491:
URL: https://github.com/apache/pinot/pull/8491#issuecomment-1092304874
# [Codecov](https://codecov.io/gh/apache/pinot/pull/8491?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#8491](https://codecov.io/gh/apache/pinot/pull/8491?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (aec68ba) into [master](https://codecov.io/gh/apache/pinot/commit/6849441b5b676e1b2685bd4335093f140d423c5b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (6849441) will **decrease** coverage by `1.99%`.
> The diff coverage is `76.87%`.
> :exclamation: Current head aec68ba differs from pull request most recent head 2e82aeb. Consider uploading reports for the commit 2e82aeb to get more accurate results
```diff
@@ Coverage Diff @@
## master #8491 +/- ##
==========================================
- Coverage 29.15% 27.15% -2.00%
==========================================
Files 1660 1665 +5
Lines 87082 87310 +228
Branches 13194 13220 +26
==========================================
- Hits 25386 23707 -1679
- Misses 59367 61384 +2017
+ Partials 2329 2219 -110
```
| Flag | Coverage Δ | |
|---|---|---|
| integration1 | `27.15% <76.87%> (+0.08%)` | :arrow_up: |
| integration2 | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/pinot/pull/8491?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...roker/requesthandler/GrpcBrokerRequestHandler.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcmVxdWVzdGhhbmRsZXIvR3JwY0Jyb2tlclJlcXVlc3RIYW5kbGVyLmphdmE=) | `0.00% <ø> (-75.68%)` | :arrow_down: |
| [...ker/routing/instanceselector/InstanceSelector.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvcm91dGluZy9pbnN0YW5jZXNlbGVjdG9yL0luc3RhbmNlU2VsZWN0b3IuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...rg/apache/pinot/core/transport/ServerResponse.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvU2VydmVyUmVzcG9uc2UuamF2YQ==) | `100.00% <ø> (+7.69%)` | :arrow_up: |
| [...va/org/apache/pinot/spi/utils/CommonConstants.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvdXRpbHMvQ29tbW9uQ29uc3RhbnRzLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...rg/apache/pinot/core/transport/ServerInstance.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvU2VydmVySW5zdGFuY2UuamF2YQ==) | `26.00% <22.22%> (-12.30%)` | :arrow_down: |
| [...a/org/apache/pinot/core/transport/QueryRouter.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvUXVlcnlSb3V0ZXIuamF2YQ==) | `73.41% <50.00%> (+4.58%)` | :arrow_up: |
| [...ot/broker/failuredetector/NoOpFailureDetector.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvZmFpbHVyZWRldGVjdG9yL05vT3BGYWlsdXJlRGV0ZWN0b3IuamF2YQ==) | `60.00% <60.00%> (ø)` | |
| [...he/pinot/core/transport/ServerRoutingInstance.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvU2VydmVyUm91dGluZ0luc3RhbmNlLmphdmE=) | `70.96% <60.00%> (-3.11%)` | :arrow_down: |
| [...rg/apache/pinot/core/transport/ServerChannels.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvU2VydmVyQ2hhbm5lbHMuamF2YQ==) | `64.10% <69.23%> (-12.37%)` | :arrow_down: |
| [...pache/pinot/core/transport/AsyncQueryResponse.java](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS90cmFuc3BvcnQvQXN5bmNRdWVyeVJlc3BvbnNlLmphdmE=) | `84.78% <71.42%> (+11.80%)` | :arrow_up: |
| ... and [194 more](https://codecov.io/gh/apache/pinot/pull/8491/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/pinot/pull/8491?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/pinot/pull/8491?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [6849441...2e82aeb](https://codecov.io/gh/apache/pinot/pull/8491?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847762088
##########
pinot-core/src/test/java/org/apache/pinot/core/query/reduce/StreamingReduceServiceTest.java:
##########
@@ -78,19 +74,14 @@ public Void answer(InvocationOnMock invocationOnMock)
}
});
final ExecutorService threadPoolService = Executors.newFixedThreadPool(1);
- final ServerRoutingInstance routingInstance =
- new ServerRoutingInstance("localhost", 9527, TableType.OFFLINE, false);
- //We cannot use TestNG's annotation like @Test(expectedExceptions = { IOException.class }) to verify
+ final ServerRoutingInstance routingInstance = new ServerRoutingInstance("localhost", 9527, TableType.OFFLINE);
+ // We cannot use TestNG's annotation like @Test(expectedExceptions = { IOException.class }) to verify
// because the Exception we hope to verify is nested inside the final exception.
assertTrue(verifyException(() -> {
- StreamingReduceService.processIterativeServerResponse(mock(StreamingReducer.class),
- threadPoolService,
- ImmutableMap.of(routingInstance, mockedResponse),
- 10,
- mock(BaseReduceService.ExecutionStatsAggregator.class));
- return null;
- },
- (cause) -> cause instanceof TimeoutException));
+ StreamingReduceService.processIterativeServerResponse(mock(StreamingReducer.class), threadPoolService,
+ ImmutableMap.of(routingInstance, mockedResponse), 10, mock(BaseReduceService.ExecutionStatsAggregator.class));
+ return null;
+ }, (cause) -> cause instanceof TimeoutException));
Review Comment:
Does this test actually guarantee the callback executes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849723973
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
Review Comment:
`BaseSegmentFetcher` is also following this pattern, but you are right the majority of the abstract classes use the static logger. IMO it doesn't matter because there is only one single failure detector instance running, so no extra overhead, but I'm okay changing it to a static one for consistency
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849786347
##########
pinot-broker/src/test/java/org/apache/pinot/broker/failuredetector/ConnectionFailureDetectorTest.java:
##########
@@ -0,0 +1,160 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.common.metrics.PinotMetricUtils;
+import org.apache.pinot.core.transport.QueryResponse;
+import org.apache.pinot.core.transport.ServerRoutingInstance;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.apache.pinot.util.TestUtils;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+public class ConnectionFailureDetectorTest {
+ private static final String INSTANCE_ID = "Server_localhost_1234";
+
+ private BrokerMetrics _brokerMetrics;
+ private FailureDetector _failureDetector;
+ private Listener _listener;
+
+ @BeforeClass
+ public void setUp() {
+ PinotConfiguration config = new PinotConfiguration();
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_CLASS_NAME, ConnectionFailureDetector.class.getName());
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS, 100);
+ config.setProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR, 1);
+ _brokerMetrics = new BrokerMetrics(PinotMetricUtils.getPinotMetricsRegistry());
+ _failureDetector = FailureDetectorFactory.getFailureDetector(config, _brokerMetrics);
+ assertTrue(_failureDetector instanceof ConnectionFailureDetector);
+ _listener = new Listener();
+ _failureDetector.register(_listener);
+ _failureDetector.start();
+ }
+
+ @Test
+ public void testConnectionFailure() {
Review Comment:
Makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847748944
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
+ protected final String _name = getClass().getSimpleName();
+ protected final List<Listener> _listeners = new ArrayList<>();
+ protected final ConcurrentHashMap<String, RetryInfo> _unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
+
+ protected BrokerMetrics _brokerMetrics;
+ protected long _retryInitialDelayNs;
+ protected double _retryDelayFactor;
+ protected int _maxRetries;
+ protected Thread _retryThread;
+
+ protected volatile boolean _running;
+
+ @Override
+ public void init(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ _brokerMetrics = brokerMetrics;
+ long retryInitialDelayMs = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
+ Broker.FailureDetector.DEFAULT_RETRY_INITIAL_DELAY_MS);
+ _retryInitialDelayNs = TimeUnit.MILLISECONDS.toNanos(retryInitialDelayMs);
+ _retryDelayFactor = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
+ Broker.FailureDetector.DEFAULT_RETRY_DELAY_FACTOR);
+ _maxRetries =
+ config.getProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES, Broker.FailureDetector.DEFAULT_MAX_RETIRES);
+ _logger.info("Initialized {} with retry initial delay: {}ms, exponential backoff factor: {}, max " + "retries: {}",
+ _name, retryInitialDelayMs, _retryDelayFactor, _maxRetries);
+ }
+
+ @Override
+ public void register(Listener listener) {
+ _listeners.add(listener);
+ }
+
+ @Override
+ public void start() {
+ _logger.info("Starting {}", _name);
+ _running = true;
+
+ _retryThread = new Thread(() -> {
+ while (_running) {
+ try {
+ long earliestRetryTimeNs = System.nanoTime() + _retryInitialDelayNs;
+ for (Map.Entry<String, RetryInfo> entry : _unhealthyServerRetryInfoMap.entrySet()) {
+ RetryInfo retryInfo = entry.getValue();
+ if (System.nanoTime() > retryInfo._retryTimeNs) {
+ String instanceId = entry.getKey();
+ if (retryInfo._numRetires == _maxRetries) {
+ _logger.warn(
+ "Unhealthy server: {} already reaches the max retries: {}, do not retry again and treat it as "
+ + "healthy so that the listeners do not lose track of the server", instanceId, _maxRetries);
+ markServerHealthy(instanceId);
+ continue;
+ }
+ _logger.info("Retry unhealthy server: {}", instanceId);
+ for (Listener listener : _listeners) {
+ listener.retryUnhealthyServer(instanceId, this);
+ }
+ // Update the retry info
+ retryInfo._retryDelayNs = (long) (retryInfo._retryDelayNs * _retryDelayFactor);
+ retryInfo._retryTimeNs = System.nanoTime() + retryInfo._retryDelayNs;
+ retryInfo._numRetires++;
+ } else {
+ earliestRetryTimeNs = Math.min(earliestRetryTimeNs, retryInfo._retryTimeNs);
+ }
+ }
+ //noinspection BusyWait
+ Thread.sleep(TimeUnit.NANOSECONDS.toMillis(earliestRetryTimeNs - System.nanoTime()));
Review Comment:
This would better implemented by putting a task onto a delay queue and waiting on that queue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] kkrugler commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
kkrugler commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847750948
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
Review Comment:
I'm curious why this (and the `_name` variable) aren't static. Is there a convention in the Pinot code base? Just FYI, I found 437 instances of `static final Logger LOGGER`, and 14 of `final Logger _logger` (a few of which were actually static).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r848972208
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
Review Comment:
This is a base class, and we want to log the actual class name that extends this base class, so it cannot be static. E.g. I want the logger to log under `ConnectionFailureDetector` instead of `BaseExponentialBackoffRetryFailureDetector`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r847752986
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/BrokerRoutingManager.java:
##########
@@ -214,87 +219,178 @@ private void processInstanceConfigChange() {
long startTimeMs = System.currentTimeMillis();
List<ZNRecord> instanceConfigZNRecords =
- _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT,
- CommonConstants.Helix.ZkClient.RETRY_COUNT, CommonConstants.Helix.ZkClient.RETRY_INTERVAL_MS);
+ _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT, Helix.ZkClient.RETRY_COUNT,
+ Helix.ZkClient.RETRY_INTERVAL_MS);
long fetchInstanceConfigsEndTimeMs = System.currentTimeMillis();
- // Calculate new enabled and disabled instances
- Set<String> enabledInstances = new HashSet<>();
- List<String> newEnabledInstances = new ArrayList<>();
+ // Calculate new enabled and disabled servers
+ Set<String> enabledServers = new HashSet<>();
+ List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
- String instance = instanceConfigZNRecord.getId();
- if (isInstanceEnabled(instanceConfigZNRecord)) {
- enabledInstances.add(instance);
+ String instanceId = instanceConfigZNRecord.getId();
+ if (isEnabledServer(instanceConfigZNRecord)) {
+ enabledServers.add(instanceId);
// Always refresh the server instance with the latest instance config in case it changes
ServerInstance serverInstance = new ServerInstance(new InstanceConfig(instanceConfigZNRecord));
- if (_enabledServerInstanceMap.put(instance, serverInstance) == null) {
- newEnabledInstances.add(instance);
+ if (_enabledServerInstanceMap.put(instanceId, serverInstance) == null) {
+ newEnabledServers.add(instanceId);
+
+ // NOTE: Remove new enabled server from excluded servers because the server is likely being restarted
+ if (_excludedServers.remove(instanceId)) {
+ LOGGER.info("Got excluded server: {} re-enabled, including it into the routing", instanceId);
+ }
}
}
}
- List<String> newDisabledInstances = new ArrayList<>();
+ List<String> newDisabledServers = new ArrayList<>();
for (String instance : _enabledServerInstanceMap.keySet()) {
- if (!enabledInstances.contains(instance)) {
- newDisabledInstances.add(instance);
+ if (!enabledServers.contains(instance)) {
+ newDisabledServers.add(instance);
}
}
- List<String> changedInstances = new ArrayList<>(newEnabledInstances.size() + newDisabledInstances.size());
- changedInstances.addAll(newEnabledInstances);
- changedInstances.addAll(newDisabledInstances);
- long calculateChangedInstancesEndTimeMs = System.currentTimeMillis();
-
- // Early terminate if there is no instance changed
- if (changedInstances.isEmpty()) {
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms) without instance change", calculateChangedInstancesEndTimeMs - startTimeMs,
- instanceConfigZNRecords.size(), fetchInstanceConfigsEndTimeMs - startTimeMs,
- calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs);
+
+ // Calculate the routable servers and the changed routable servers
+ List<String> changedServers = new ArrayList<>(newEnabledServers.size() + newDisabledServers.size());
+ if (_excludedServers.isEmpty()) {
+ _routableServers = enabledServers;
+ changedServers.addAll(newEnabledServers);
+ changedServers.addAll(newDisabledServers);
+ } else {
+ enabledServers.removeAll(_excludedServers);
+ _routableServers = enabledServers;
+ // NOTE: All new enabled servers are routable
+ changedServers.addAll(newEnabledServers);
+ for (String newDisabledServer : newDisabledServers) {
+ if (_excludedServers.contains(newDisabledServer)) {
+ changedServers.add(newDisabledServer);
+ }
+ }
+ }
+ long calculateChangedServersEndTimeMs = System.currentTimeMillis();
+
+ // Early terminate if there is no changed servers
+ if (changedServers.isEmpty()) {
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms) without instance change",
+ calculateChangedServersEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
+ fetchInstanceConfigsEndTimeMs - startTimeMs,
+ calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs);
return;
}
// Update routing entry for all tables
for (RoutingEntry routingEntry : _routingEntryMap.values()) {
try {
- routingEntry.onInstancesChange(enabledInstances, changedInstances);
+ routingEntry.onInstancesChange(_routableServers, changedServers);
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry on instances change for table: {}",
routingEntry.getTableNameWithType(), e);
}
}
long updateRoutingEntriesEndTimeMs = System.currentTimeMillis();
- // Remove new disabled instances from _enabledServerInstanceMap after updating all routing entries to ensure it
- // always contains the selected instances
- for (String newDisabledInstance : newDisabledInstances) {
+ // Remove new disabled servers from _enabledServerInstanceMap after updating all routing entries to ensure it
+ // always contains the selected servers
+ for (String newDisabledInstance : newDisabledServers) {
_enabledServerInstanceMap.remove(newDisabledInstance);
}
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms, update {} routing entries: {}ms), new enabled instances: {}, new disabled instances: {}",
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms, update {} routing entries: {}ms), "
+ + "new enabled servers: {}, new disabled servers: {}, excluded servers: {}",
updateRoutingEntriesEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
- fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs,
- _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedInstancesEndTimeMs,
- newEnabledInstances, newDisabledInstances);
+ fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs,
+ _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedServersEndTimeMs, newEnabledServers,
+ newDisabledServers, _excludedServers);
}
- private static boolean isInstanceEnabled(ZNRecord instanceConfigZNRecord) {
+ private static boolean isEnabledServer(ZNRecord instanceConfigZNRecord) {
+ String instanceId = instanceConfigZNRecord.getId();
+ if (!instanceId.startsWith(Helix.PREFIX_OF_SERVER_INSTANCE)) {
+ // NOTE: Some legacy configs might not contain the instance type prefix
+ if (instanceId.startsWith(Helix.PREFIX_OF_CONTROLLER_INSTANCE) || instanceId.startsWith(
+ Helix.PREFIX_OF_BROKER_INSTANCE) || instanceId.startsWith(Helix.PREFIX_OF_MINION_INSTANCE)) {
+ return false;
+ }
+ }
if ("false".equalsIgnoreCase(
instanceConfigZNRecord.getSimpleField(InstanceConfig.InstanceConfigProperty.HELIX_ENABLED.name()))) {
return false;
}
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.IS_SHUTDOWN_IN_PROGRESS))) {
return false;
}
//noinspection RedundantIfStatement
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.QUERIES_DISABLED))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.QUERIES_DISABLED))) {
Review Comment:
Not introduced here but prefer Boolean.parseBoolean
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849809680
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
+ protected final String _name = getClass().getSimpleName();
+ protected final List<Listener> _listeners = new ArrayList<>();
+ protected final ConcurrentHashMap<String, RetryInfo> _unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
+
+ protected BrokerMetrics _brokerMetrics;
+ protected long _retryInitialDelayNs;
+ protected double _retryDelayFactor;
+ protected int _maxRetries;
+ protected Thread _retryThread;
+
+ protected volatile boolean _running;
+
+ @Override
+ public void init(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ _brokerMetrics = brokerMetrics;
+ long retryInitialDelayMs = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
+ Broker.FailureDetector.DEFAULT_RETRY_INITIAL_DELAY_MS);
+ _retryInitialDelayNs = TimeUnit.MILLISECONDS.toNanos(retryInitialDelayMs);
+ _retryDelayFactor = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
+ Broker.FailureDetector.DEFAULT_RETRY_DELAY_FACTOR);
+ _maxRetries =
+ config.getProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES, Broker.FailureDetector.DEFAULT_MAX_RETIRES);
+ _logger.info("Initialized {} with retry initial delay: {}ms, exponential backoff factor: {}, max " + "retries: {}",
+ _name, retryInitialDelayMs, _retryDelayFactor, _maxRetries);
+ }
+
+ @Override
+ public void register(Listener listener) {
+ _listeners.add(listener);
+ }
+
+ @Override
+ public void start() {
+ _logger.info("Starting {}", _name);
+ _running = true;
+
+ _retryThread = new Thread(() -> {
Review Comment:
With the `ExecutorService`, I need to use a custom `ThreadFactory` to set the thread name and deamon flag. When shutting down, I need to call `shutdownNow()` and then `awaitTermination()`.
IMO this won't simplify the logic, and developers need to know that `shutdownNow()` will send interrupt to the running thread in order to understand the logic here, which is not as clear as explicitly interrupt.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849783866
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
+ protected final String _name = getClass().getSimpleName();
+ protected final List<Listener> _listeners = new ArrayList<>();
+ protected final ConcurrentHashMap<String, RetryInfo> _unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
+
+ protected BrokerMetrics _brokerMetrics;
+ protected long _retryInitialDelayNs;
+ protected double _retryDelayFactor;
+ protected int _maxRetries;
+ protected Thread _retryThread;
+
+ protected volatile boolean _running;
+
+ @Override
+ public void init(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ _brokerMetrics = brokerMetrics;
+ long retryInitialDelayMs = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
+ Broker.FailureDetector.DEFAULT_RETRY_INITIAL_DELAY_MS);
+ _retryInitialDelayNs = TimeUnit.MILLISECONDS.toNanos(retryInitialDelayMs);
+ _retryDelayFactor = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
+ Broker.FailureDetector.DEFAULT_RETRY_DELAY_FACTOR);
+ _maxRetries =
+ config.getProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES, Broker.FailureDetector.DEFAULT_MAX_RETIRES);
+ _logger.info("Initialized {} with retry initial delay: {}ms, exponential backoff factor: {}, max " + "retries: {}",
+ _name, retryInitialDelayMs, _retryDelayFactor, _maxRetries);
+ }
+
+ @Override
+ public void register(Listener listener) {
+ _listeners.add(listener);
+ }
+
+ @Override
+ public void start() {
+ _logger.info("Starting {}", _name);
+ _running = true;
+
+ _retryThread = new Thread(() -> {
Review Comment:
The code will be simplified, you don’t need to worry about manual interrupts. With an executor all this code is just a call to execute and shutdownNow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849821972
##########
pinot-broker/src/main/java/org/apache/pinot/broker/failuredetector/BaseExponentialBackoffRetryFailureDetector.java:
##########
@@ -0,0 +1,174 @@
+/**
+ * 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.broker.failuredetector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.ThreadSafe;
+import org.apache.pinot.common.metrics.BrokerGauge;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * The {@code BaseExponentialBackoffRetryFailureDetector} is a base failure detector implementation that retries the
+ * unhealthy servers with exponential increasing delays.
+ */
+@ThreadSafe
+public abstract class BaseExponentialBackoffRetryFailureDetector implements FailureDetector {
+ protected final Logger _logger = LoggerFactory.getLogger(getClass());
+ protected final String _name = getClass().getSimpleName();
+ protected final List<Listener> _listeners = new ArrayList<>();
+ protected final ConcurrentHashMap<String, RetryInfo> _unhealthyServerRetryInfoMap = new ConcurrentHashMap<>();
+
+ protected BrokerMetrics _brokerMetrics;
+ protected long _retryInitialDelayNs;
+ protected double _retryDelayFactor;
+ protected int _maxRetries;
+ protected Thread _retryThread;
+
+ protected volatile boolean _running;
+
+ @Override
+ public void init(PinotConfiguration config, BrokerMetrics brokerMetrics) {
+ _brokerMetrics = brokerMetrics;
+ long retryInitialDelayMs = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_INITIAL_DELAY_MS,
+ Broker.FailureDetector.DEFAULT_RETRY_INITIAL_DELAY_MS);
+ _retryInitialDelayNs = TimeUnit.MILLISECONDS.toNanos(retryInitialDelayMs);
+ _retryDelayFactor = config.getProperty(Broker.FailureDetector.CONFIG_OF_RETRY_DELAY_FACTOR,
+ Broker.FailureDetector.DEFAULT_RETRY_DELAY_FACTOR);
+ _maxRetries =
+ config.getProperty(Broker.FailureDetector.CONFIG_OF_MAX_RETRIES, Broker.FailureDetector.DEFAULT_MAX_RETIRES);
+ _logger.info("Initialized {} with retry initial delay: {}ms, exponential backoff factor: {}, max " + "retries: {}",
+ _name, retryInitialDelayMs, _retryDelayFactor, _maxRetries);
+ }
+
+ @Override
+ public void register(Listener listener) {
+ _listeners.add(listener);
+ }
+
+ @Override
+ public void start() {
+ _logger.info("Starting {}", _name);
+ _running = true;
+
+ _retryThread = new Thread(() -> {
Review Comment:
I think we’ll need to agree to disagree here and move on.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] richardstartin commented on a diff in pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
richardstartin commented on code in PR #8491:
URL: https://github.com/apache/pinot/pull/8491#discussion_r849839949
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/BrokerRoutingManager.java:
##########
@@ -214,87 +219,178 @@ private void processInstanceConfigChange() {
long startTimeMs = System.currentTimeMillis();
List<ZNRecord> instanceConfigZNRecords =
- _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT,
- CommonConstants.Helix.ZkClient.RETRY_COUNT, CommonConstants.Helix.ZkClient.RETRY_INTERVAL_MS);
+ _zkDataAccessor.getChildren(_instanceConfigsPath, null, AccessOption.PERSISTENT, Helix.ZkClient.RETRY_COUNT,
+ Helix.ZkClient.RETRY_INTERVAL_MS);
long fetchInstanceConfigsEndTimeMs = System.currentTimeMillis();
- // Calculate new enabled and disabled instances
- Set<String> enabledInstances = new HashSet<>();
- List<String> newEnabledInstances = new ArrayList<>();
+ // Calculate new enabled and disabled servers
+ Set<String> enabledServers = new HashSet<>();
+ List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
- String instance = instanceConfigZNRecord.getId();
- if (isInstanceEnabled(instanceConfigZNRecord)) {
- enabledInstances.add(instance);
+ String instanceId = instanceConfigZNRecord.getId();
+ if (isEnabledServer(instanceConfigZNRecord)) {
+ enabledServers.add(instanceId);
// Always refresh the server instance with the latest instance config in case it changes
ServerInstance serverInstance = new ServerInstance(new InstanceConfig(instanceConfigZNRecord));
- if (_enabledServerInstanceMap.put(instance, serverInstance) == null) {
- newEnabledInstances.add(instance);
+ if (_enabledServerInstanceMap.put(instanceId, serverInstance) == null) {
+ newEnabledServers.add(instanceId);
+
+ // NOTE: Remove new enabled server from excluded servers because the server is likely being restarted
+ if (_excludedServers.remove(instanceId)) {
+ LOGGER.info("Got excluded server: {} re-enabled, including it into the routing", instanceId);
+ }
}
}
}
- List<String> newDisabledInstances = new ArrayList<>();
+ List<String> newDisabledServers = new ArrayList<>();
for (String instance : _enabledServerInstanceMap.keySet()) {
- if (!enabledInstances.contains(instance)) {
- newDisabledInstances.add(instance);
+ if (!enabledServers.contains(instance)) {
+ newDisabledServers.add(instance);
}
}
- List<String> changedInstances = new ArrayList<>(newEnabledInstances.size() + newDisabledInstances.size());
- changedInstances.addAll(newEnabledInstances);
- changedInstances.addAll(newDisabledInstances);
- long calculateChangedInstancesEndTimeMs = System.currentTimeMillis();
-
- // Early terminate if there is no instance changed
- if (changedInstances.isEmpty()) {
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms) without instance change", calculateChangedInstancesEndTimeMs - startTimeMs,
- instanceConfigZNRecords.size(), fetchInstanceConfigsEndTimeMs - startTimeMs,
- calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs);
+
+ // Calculate the routable servers and the changed routable servers
+ List<String> changedServers = new ArrayList<>(newEnabledServers.size() + newDisabledServers.size());
+ if (_excludedServers.isEmpty()) {
+ _routableServers = enabledServers;
+ changedServers.addAll(newEnabledServers);
+ changedServers.addAll(newDisabledServers);
+ } else {
+ enabledServers.removeAll(_excludedServers);
+ _routableServers = enabledServers;
+ // NOTE: All new enabled servers are routable
+ changedServers.addAll(newEnabledServers);
+ for (String newDisabledServer : newDisabledServers) {
+ if (_excludedServers.contains(newDisabledServer)) {
+ changedServers.add(newDisabledServer);
+ }
+ }
+ }
+ long calculateChangedServersEndTimeMs = System.currentTimeMillis();
+
+ // Early terminate if there is no changed servers
+ if (changedServers.isEmpty()) {
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms) without instance change",
+ calculateChangedServersEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
+ fetchInstanceConfigsEndTimeMs - startTimeMs,
+ calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs);
return;
}
// Update routing entry for all tables
for (RoutingEntry routingEntry : _routingEntryMap.values()) {
try {
- routingEntry.onInstancesChange(enabledInstances, changedInstances);
+ routingEntry.onInstancesChange(_routableServers, changedServers);
} catch (Exception e) {
LOGGER.error("Caught unexpected exception while updating routing entry on instances change for table: {}",
routingEntry.getTableNameWithType(), e);
}
}
long updateRoutingEntriesEndTimeMs = System.currentTimeMillis();
- // Remove new disabled instances from _enabledServerInstanceMap after updating all routing entries to ensure it
- // always contains the selected instances
- for (String newDisabledInstance : newDisabledInstances) {
+ // Remove new disabled servers from _enabledServerInstanceMap after updating all routing entries to ensure it
+ // always contains the selected servers
+ for (String newDisabledInstance : newDisabledServers) {
_enabledServerInstanceMap.remove(newDisabledInstance);
}
- LOGGER.info(
- "Processed instance config change in {}ms (fetch {} instance configs: {}ms, calculate changed instances: "
- + "{}ms, update {} routing entries: {}ms), new enabled instances: {}, new disabled instances: {}",
+ LOGGER.info("Processed instance config change in {}ms "
+ + "(fetch {} instance configs: {}ms, calculate changed servers: {}ms, update {} routing entries: {}ms), "
+ + "new enabled servers: {}, new disabled servers: {}, excluded servers: {}",
updateRoutingEntriesEndTimeMs - startTimeMs, instanceConfigZNRecords.size(),
- fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedInstancesEndTimeMs - fetchInstanceConfigsEndTimeMs,
- _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedInstancesEndTimeMs,
- newEnabledInstances, newDisabledInstances);
+ fetchInstanceConfigsEndTimeMs - startTimeMs, calculateChangedServersEndTimeMs - fetchInstanceConfigsEndTimeMs,
+ _routingEntryMap.size(), updateRoutingEntriesEndTimeMs - calculateChangedServersEndTimeMs, newEnabledServers,
+ newDisabledServers, _excludedServers);
}
- private static boolean isInstanceEnabled(ZNRecord instanceConfigZNRecord) {
+ private static boolean isEnabledServer(ZNRecord instanceConfigZNRecord) {
+ String instanceId = instanceConfigZNRecord.getId();
+ if (!instanceId.startsWith(Helix.PREFIX_OF_SERVER_INSTANCE)) {
+ // NOTE: Some legacy configs might not contain the instance type prefix
+ if (instanceId.startsWith(Helix.PREFIX_OF_CONTROLLER_INSTANCE) || instanceId.startsWith(
+ Helix.PREFIX_OF_BROKER_INSTANCE) || instanceId.startsWith(Helix.PREFIX_OF_MINION_INSTANCE)) {
+ return false;
+ }
+ }
if ("false".equalsIgnoreCase(
instanceConfigZNRecord.getSimpleField(InstanceConfig.InstanceConfigProperty.HELIX_ENABLED.name()))) {
return false;
}
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.IS_SHUTDOWN_IN_PROGRESS))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.IS_SHUTDOWN_IN_PROGRESS))) {
return false;
}
//noinspection RedundantIfStatement
- if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(CommonConstants.Helix.QUERIES_DISABLED))) {
+ if ("true".equalsIgnoreCase(instanceConfigZNRecord.getSimpleField(Helix.QUERIES_DISABLED))) {
return false;
}
return true;
}
+ /**
+ * Excludes a server from the routing.
+ */
+ public synchronized void excludeServerFromRouting(String instanceId) {
+ LOGGER.warn("Excluding server: {} from routing", instanceId);
+ if (!_excludedServers.add(instanceId)) {
+ LOGGER.info("Server: {} is already excluded from routing, skipping updating the routing", instanceId);
+ return;
+ }
+ if (!_routableServers.contains(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.remove(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when excluding server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Excluded server: {} from routing in {}ms (updated {} routing entries)", instanceId,
+ System.currentTimeMillis() - startTimeMs, _routingEntryMap.size());
+ }
+
+ /**
+ * Includes a previous excluded server to the routing.
+ */
+ public synchronized void includeServerToRouting(String instanceId) {
+ LOGGER.info("Including server: {} to routing", instanceId);
+ if (!_excludedServers.remove(instanceId)) {
+ LOGGER.info("Server: {} is not previously excluded, skipping updating the routing", instanceId);
+ }
+ if (!_enabledServerInstanceMap.containsKey(instanceId)) {
+ LOGGER.info("Server: {} is not enabled, skipping updating the routing", instanceId);
+ return;
+ }
+
+ // Update routing entry for all tables
+ long startTimeMs = System.currentTimeMillis();
+ Set<String> routableServers = new HashSet<>(_routableServers);
+ routableServers.add(instanceId);
+ _routableServers = routableServers;
+ List<String> changedServers = Collections.singletonList(instanceId);
+ for (RoutingEntry routingEntry : _routingEntryMap.values()) {
+ try {
+ routingEntry.onInstancesChange(_routableServers, changedServers);
+ } catch (Exception e) {
+ LOGGER.error("Caught unexpected exception while updating routing entry when including server: {} for table: {}",
+ instanceId, routingEntry.getTableNameWithType(), e);
+ }
+ }
+ LOGGER.info("Included server: {} to routing in {}ms (updated {} routing entries)", instanceId,
Review Comment:
The durations are tracked for the purpose of observability, and it's confusing that different clocks are used. Let's create a `Timer` SPI so users can choose and get consistency. No need to fix that here and now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] vvivekiyer commented on pull request #8491: Add connection based FailureDetector
Posted by GitBox <gi...@apache.org>.
vvivekiyer commented on PR #8491:
URL: https://github.com/apache/pinot/pull/8491#issuecomment-1100345080
LGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org