You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by jl...@apache.org on 2019/04/30 17:19:58 UTC
[incubator-pinot] branch master updated: Reduce ZK access (#4173)
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 971c9ad Reduce ZK access (#4173)
971c9ad is described below
commit 971c9adf091905d938cce0f97db63d806f0a1e84
Author: Jialiang Li <jl...@linkedin.com>
AuthorDate: Tue Apr 30 10:19:52 2019 -0700
Reduce ZK access (#4173)
---
.../helix/core/SegmentDeletionManager.java | 10 ++++------
.../controller/helix/core/TableRebalancer.java | 22 +++++++++++-----------
.../apache/pinot/tools/PinotSegmentRebalancer.java | 6 +++---
3 files changed, 18 insertions(+), 20 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
index 19ad746..46c84c0 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
@@ -100,9 +100,10 @@ public class SegmentDeletionManager {
protected synchronized void deleteSegmentFromPropertyStoreAndLocal(String tableName, Collection<String> segmentIds,
long deletionDelay) {
- // Check if segment got removed from ExternalView and IdealStates
- if (_helixAdmin.getResourceExternalView(_helixClusterName, tableName) == null
- || _helixAdmin.getResourceIdealState(_helixClusterName, tableName) == null) {
+ // Check if segment got removed from ExternalView or IdealState
+ ExternalView externalView = _helixAdmin.getResourceExternalView(_helixClusterName, tableName);
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, tableName);
+ if (externalView == null || idealState == null) {
LOGGER.warn("Resource: {} is not set up in idealState or ExternalView, won't do anything", tableName);
return;
}
@@ -111,9 +112,6 @@ public class SegmentDeletionManager {
Set<String> segmentsToRetryLater = new HashSet<>(segmentIds.size()); // List of segments that we need to retry
try {
- ExternalView externalView = _helixAdmin.getResourceExternalView(_helixClusterName, tableName);
- IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, tableName);
-
for (String segmentId : segmentIds) {
Map<String, String> segmentToInstancesMapFromExternalView = externalView.getStateMap(segmentId);
Map<String, String> segmentToInstancesMapFromIdealStates = idealState.getInstanceStateMap(segmentId);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/TableRebalancer.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/TableRebalancer.java
index 7d52ba9..fabbffe 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/TableRebalancer.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/TableRebalancer.java
@@ -96,11 +96,11 @@ public class TableRebalancer {
RebalanceResult result = new RebalanceResult();
- String tableName = tableConfig.getTableName();
+ String tableNameWithType = tableConfig.getTableName();
HelixDataAccessor dataAccessor = _helixManager.getHelixDataAccessor();
ZkBaseDataAccessor zkBaseDataAccessor = (ZkBaseDataAccessor) dataAccessor.getBaseDataAccessor();
- PropertyKey idealStateKey = dataAccessor.keyBuilder().idealStates(tableName);
+ PropertyKey idealStateKey = dataAccessor.keyBuilder().idealStates(tableNameWithType);
IdealState previousIdealState = dataAccessor.getProperty(idealStateKey);
if (rebalanceConfig.getBoolean(RebalanceUserConfigConstants.DRYRUN, RebalanceUserConfigConstants.DEFAULT_DRY_RUN)) {
@@ -123,7 +123,7 @@ public class TableRebalancer {
IdealState currentIdealState = dataAccessor.getProperty(idealStateKey);
if (targetIdealState == null || !EqualityUtils.isEqual(previousIdealState, currentIdealState)) {
- LOGGER.info("Computing new rebalanced state for table {}", tableName);
+ LOGGER.info("Computing new rebalanced state for table {}", tableNameWithType);
// we need to recompute target state
@@ -139,9 +139,9 @@ public class TableRebalancer {
}
if (EqualityUtils.isEqual(targetIdealState, currentIdealState)) {
- LOGGER.info("Table {} is rebalanced.", tableName);
+ LOGGER.info("Table {} is rebalanced.", tableNameWithType);
- LOGGER.info("Finished rebalancing table {} in {} ms.", tableName,
+ LOGGER.info("Finished rebalancing table {} in {} ms.", tableNameWithType,
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime));
result.setIdealStateMapping(targetIdealState.getRecord().getMapFields());
result.setPartitionAssignment(targetPartitionAssignment);
@@ -158,33 +158,33 @@ public class TableRebalancer {
// Check version and set ideal state
try {
- LOGGER.info("Updating IdealState for table {}", tableName);
+ LOGGER.info("Updating IdealState for table {}", tableNameWithType);
if (zkBaseDataAccessor
.set(idealStateKey.getPath(), nextIdealState.getRecord(), currentIdealState.getRecord().getVersion(),
AccessOption.PERSISTENT)) {
// if we succeeded, wait for the change to stabilize
- waitForStable(tableName);
+ waitForStable(tableNameWithType);
// clear retries as it tracks failures with each idealstate update attempt
retries = 0;
continue;
}
// in case of any error, we retry a bounded number of types
} catch (ZkBadVersionException e) {
- LOGGER.warn("Version changed while updating ideal state for resource: {}", tableName);
+ LOGGER.warn("Version changed while updating ideal state for resource: {}", tableNameWithType);
} catch (Exception e) {
- LOGGER.warn("Caught exception while updating ideal state for resource: {}", tableName, e);
+ LOGGER.warn("Caught exception while updating ideal state for resource: {}", tableNameWithType, e);
}
previousIdealState = currentIdealState;
if (retries++ > MAX_RETRIES) {
- LOGGER.error("Unable to rebalance table {} in {} attempts. Giving up", tableName, MAX_RETRIES);
+ LOGGER.error("Unable to rebalance table {} in {} attempts. Giving up", tableNameWithType, MAX_RETRIES);
return result;
}
// wait before retrying
try {
Thread.sleep(retryDelayMs);
} catch (InterruptedException e) {
- LOGGER.error("Got interrupted while rebalancing table {}", tableName);
+ LOGGER.error("Got interrupted while rebalancing table {}", tableNameWithType);
Thread.currentThread().interrupt();
return result;
}
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/PinotSegmentRebalancer.java b/pinot-tools/src/main/java/org/apache/pinot/tools/PinotSegmentRebalancer.java
index e5ca5ff..e94cf8b 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/PinotSegmentRebalancer.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/PinotSegmentRebalancer.java
@@ -59,9 +59,9 @@ public class PinotSegmentRebalancer extends PinotZKChanger {
* return true if IdealState = ExternalView
* @return
*/
- public int isStable(String tableName) {
- IdealState idealState = helixAdmin.getResourceIdealState(clusterName, tableName);
- ExternalView externalView = helixAdmin.getResourceExternalView(clusterName, tableName);
+ public int isStable(String tableNameWithType) {
+ IdealState idealState = helixAdmin.getResourceIdealState(clusterName, tableNameWithType);
+ ExternalView externalView = helixAdmin.getResourceExternalView(clusterName, tableNameWithType);
Map<String, Map<String, String>> mapFieldsIS = idealState.getRecord().getMapFields();
Map<String, Map<String, String>> mapFieldsEV = externalView.getRecord().getMapFields();
int numDiff = 0;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org