You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by ji...@apache.org on 2020/04/23 19:27:55 UTC
[helix] 09/23: Add intermediate storage for customized state (#827)
This is an automated email from the ASF dual-hosted git repository.
jiajunwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 55f3b60e55a9a450f3dba11856d392802cafc349
Author: Meng Zhang <mn...@linkedin.com>
AuthorDate: Thu Mar 5 10:24:17 2020 -0800
Add intermediate storage for customized state (#827)
1. Implement a participant state cache for generalizing functions in both current state cache and customized state cache.
2. Implement an intermediate data structure to store the result of customized state computation and prepare the data to be the format that can be used by customized view computation later.
---
.../helix/common/caches/CurrentStateCache.java | 149 ++---------------
.../helix/common/caches/CustomizedStateCache.java | 66 ++++++++
.../helix/common/caches/ParticipantStateCache.java | 184 +++++++++++++++++++++
.../dataproviders/BaseControllerDataProvider.java | 4 +-
.../controller/stages/CustomizedStateOutput.java | 120 ++++++++++++++
.../model/CustomizedStateAggregationConfig.java | 6 +-
.../apache/helix/spectator/RoutingDataCache.java | 2 +-
7 files changed, 390 insertions(+), 141 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/common/caches/CurrentStateCache.java b/helix-core/src/main/java/org/apache/helix/common/caches/CurrentStateCache.java
index d266196..fc7fa78 100644
--- a/helix-core/src/main/java/org/apache/helix/common/caches/CurrentStateCache.java
+++ b/helix-core/src/main/java/org/apache/helix/common/caches/CurrentStateCache.java
@@ -19,33 +19,26 @@ package org.apache.helix.common.caches;
* under the License.
*/
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import com.google.common.collect.Maps;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.PropertyKey;
import org.apache.helix.common.controllers.ControlContextProvider;
-import org.apache.helix.controller.LogUtil;
import org.apache.helix.model.CurrentState;
import org.apache.helix.model.LiveInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
/**
* Cache to hold all CurrentStates of a cluster.
*/
-public class CurrentStateCache extends AbstractDataCache<CurrentState> {
+public class CurrentStateCache extends ParticipantStateCache<CurrentState> {
private static final Logger LOG = LoggerFactory.getLogger(CurrentStateCache.class.getName());
-
- private Map<String, Map<String, Map<String, CurrentState>>> _currentStateMap;
- private Map<PropertyKey, CurrentState> _currentStateCache = Maps.newHashMap();
- // If the cache is already refreshed with current state data.
+ // If the snapshot is already refreshed with current state data.
private boolean _initialized = false;
private CurrentStateSnapshot _snapshot;
@@ -55,149 +48,35 @@ public class CurrentStateCache extends AbstractDataCache<CurrentState> {
public CurrentStateCache(ControlContextProvider contextProvider) {
super(contextProvider);
- _currentStateMap = Collections.emptyMap();
- _snapshot = new CurrentStateSnapshot(_currentStateCache);
+ _snapshot = new CurrentStateSnapshot(_participantStateCache);
}
- /**
- * This refreshes the CurrentStates data by re-fetching the data from zookeeper in an efficient
- * way
- *
- * @param accessor
- * @param liveInstanceMap map of all liveInstances in cluster
- *
- * @return
- */
- public boolean refresh(HelixDataAccessor accessor,
- Map<String, LiveInstance> liveInstanceMap) {
- long startTime = System.currentTimeMillis();
-
- refreshCurrentStatesCache(accessor, liveInstanceMap);
- Map<String, Map<String, Map<String, CurrentState>>> allCurStateMap = new HashMap<>();
- for (PropertyKey key : _currentStateCache.keySet()) {
- CurrentState currentState = _currentStateCache.get(key);
- String[] params = key.getParams();
- if (currentState != null && params.length >= 4) {
- String instanceName = params[1];
- String sessionId = params[2];
- String stateName = params[3];
- Map<String, Map<String, CurrentState>> instanceCurStateMap =
- allCurStateMap.get(instanceName);
- if (instanceCurStateMap == null) {
- instanceCurStateMap = Maps.newHashMap();
- allCurStateMap.put(instanceName, instanceCurStateMap);
- }
- Map<String, CurrentState> sessionCurStateMap = instanceCurStateMap.get(sessionId);
- if (sessionCurStateMap == null) {
- sessionCurStateMap = Maps.newHashMap();
- instanceCurStateMap.put(sessionId, sessionCurStateMap);
- }
- sessionCurStateMap.put(stateName, currentState);
- }
- }
-
- for (String instance : allCurStateMap.keySet()) {
- allCurStateMap.put(instance, Collections.unmodifiableMap(allCurStateMap.get(instance)));
- }
- _currentStateMap = Collections.unmodifiableMap(allCurStateMap);
-
- long endTime = System.currentTimeMillis();
- LogUtil.logInfo(LOG, genEventInfo(),
- "END: CurrentStateCache.refresh() for cluster " + _controlContextProvider.getClusterName()
- + ", started at : " + startTime + ", took " + (endTime - startTime) + " ms");
- if (LOG.isDebugEnabled()) {
- LogUtil.logDebug(LOG, genEventInfo(),
- String.format("Current State refreshed : %s", _currentStateMap.toString()));
- }
- return true;
- }
-
- // reload current states that has been changed from zk to local cache.
- private void refreshCurrentStatesCache(HelixDataAccessor accessor,
+ @Override
+ protected Set<PropertyKey> PopulateParticipantKeys(HelixDataAccessor accessor,
Map<String, LiveInstance> liveInstanceMap) {
-
- long start = System.currentTimeMillis();
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
PropertyKey.Builder keyBuilder = accessor.keyBuilder();
-
- Set<PropertyKey> currentStateKeys = new HashSet<>();
for (String instanceName : liveInstanceMap.keySet()) {
LiveInstance liveInstance = liveInstanceMap.get(instanceName);
String sessionId = liveInstance.getEphemeralOwner();
List<String> currentStateNames =
accessor.getChildNames(keyBuilder.currentStates(instanceName, sessionId));
for (String currentStateName : currentStateNames) {
- currentStateKeys.add(keyBuilder.currentState(instanceName, sessionId, currentStateName));
+ participantStateKeys
+ .add(keyBuilder.currentState(instanceName, sessionId, currentStateName));
}
}
- // All new entries from zk not cached locally yet should be read from ZK.
- Set<PropertyKey> reloadKeys = new HashSet<>(currentStateKeys);
- reloadKeys.removeAll(_currentStateCache.keySet());
-
- Set<PropertyKey> cachedKeys = new HashSet<>(_currentStateCache.keySet());
- cachedKeys.retainAll(currentStateKeys);
-
- Set<PropertyKey> reloadedKeys = new HashSet<>();
- Map<PropertyKey, CurrentState> newStateCache = Collections.unmodifiableMap(
- refreshProperties(accessor, reloadKeys, new ArrayList<>(cachedKeys),
- _currentStateCache, reloadedKeys));
+ return participantStateKeys;
+ }
- // if the cache was not initialized, the previous state should not be included in the snapshot
+ protected void refreshSnapshot(Map<PropertyKey, CurrentState> newStateCache,
+ Map<PropertyKey, CurrentState> participantStateCache, Set<PropertyKey> reloadedKeys) {
if (_initialized) {
- _snapshot = new CurrentStateSnapshot(newStateCache, _currentStateCache, reloadedKeys);
+ _snapshot = new CurrentStateSnapshot(newStateCache, participantStateCache, reloadedKeys);
} else {
_snapshot = new CurrentStateSnapshot(newStateCache);
_initialized = true;
}
-
- _currentStateCache = newStateCache;
-
- if (LOG.isDebugEnabled()) {
- LogUtil.logDebug(LOG, genEventInfo(),
- "# of CurrentStates reload: " + reloadKeys.size() + ", skipped:" + (
- currentStateKeys.size() - reloadKeys.size()) + ". took " + (System.currentTimeMillis()
- - start) + " ms to reload new current states for cluster: " + _controlContextProvider
- .getClusterName());
- }
- }
-
- /**
- * Return CurrentStates map for all instances.
- *
- * @return
- */
- public Map<String, Map<String, Map<String, CurrentState>>> getCurrentStatesMap() {
- return Collections.unmodifiableMap(_currentStateMap);
- }
-
- /**
- * Return all CurrentState on the given instance.
- *
- * @param instance
- *
- * @return
- */
- public Map<String, Map<String, CurrentState>> getCurrentStates(String instance) {
- if (!_currentStateMap.containsKey(instance)) {
- return Collections.emptyMap();
- }
- return Collections.unmodifiableMap(_currentStateMap.get(instance));
- }
-
- /**
- * Provides the current state of the node for a given session id, the sessionid can be got from
- * LiveInstance
- *
- * @param instance
- * @param clientSessionId
- *
- * @return
- */
- public Map<String, CurrentState> getCurrentState(String instance, String clientSessionId) {
- if (!_currentStateMap.containsKey(instance) || !_currentStateMap.get(instance)
- .containsKey(clientSessionId)) {
- return Collections.emptyMap();
- }
- return Collections.unmodifiableMap(_currentStateMap.get(instance).get(clientSessionId));
}
@Override
diff --git a/helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java b/helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
new file mode 100644
index 0000000..1f33948
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
@@ -0,0 +1,66 @@
+package org.apache.helix.common.caches;
+
+/*
+ * 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.
+ */
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.common.controllers.ControlContextProvider;
+import org.apache.helix.model.CustomizedState;
+import org.apache.helix.model.CustomizedStateAggregationConfig;
+import org.apache.helix.model.LiveInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class CustomizedStateCache extends ParticipantStateCache<CustomizedState> {
+ private static final Logger LOG = LoggerFactory.getLogger(CurrentStateCache.class.getName());
+
+ public CustomizedStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+ public CustomizedStateCache(ControlContextProvider contextProvider) {
+ super(contextProvider);
+ }
+
+ @Override
+ protected Set<PropertyKey> PopulateParticipantKeys(HelixDataAccessor accessor,
+ Map<String, LiveInstance> liveInstanceMap) {
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ Set<String> restrictedKeys = new HashSet<>(
+ accessor.getProperty(accessor.keyBuilder().customizedStateAggregationConfig()).getRecord()
+ .getListFields().get(
+ CustomizedStateAggregationConfig.CustomizedStateAggregationProperty.AGGREGATION_ENABLED_TYPES
+ .name()));
+ for (String instanceName : liveInstanceMap.keySet()) {
+ for (String customizedStateType : restrictedKeys) {
+ accessor.getChildNames(keyBuilder.customizedStates(instanceName, customizedStateType))
+ .stream().forEach(resourceName -> participantStateKeys
+ .add(keyBuilder.customizedState(instanceName, customizedStateType, resourceName)));
+ }
+ }
+ return participantStateKeys;
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java b/helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
new file mode 100644
index 0000000..61556b1
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
@@ -0,0 +1,184 @@
+package org.apache.helix.common.caches;
+
+/*
+ * 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.
+ */
+
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.common.controllers.ControlContextProvider;
+import org.apache.helix.controller.LogUtil;
+import org.apache.helix.model.LiveInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Represent a cache that holds a certain participant side state of for the whole cluster.
+ */
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class);
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = new HashMap<>();
+ }
+
+ /**
+ * This refreshes the participant state cache data by re-fetching the data from zookeeper in an
+ * efficient way
+ * @param accessor
+ * @param liveInstanceMap map of all liveInstances in cluster
+ * @return
+ */
+ public boolean refresh(HelixDataAccessor accessor, Map<String, LiveInstance> liveInstanceMap) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ // There should be 4 levels of keys. The first one is the cluster name, the second one is the
+ // instance name, the third one is a customized key (could be session Id or customized state
+ // type), the fourth one is the resourceName
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String instanceName = params[1];
+ String customizedName = params[2];
+ String resourceName = params[3];
+ Map<String, Map<String, T>> instanceMap = allParticipantStateMap.get(instanceName);
+ if (instanceMap == null) {
+ instanceMap = Maps.newHashMap();
+ allParticipantStateMap.put(instanceName, instanceMap);
+ }
+ Map<String, T> customizedMap = instanceMap.get(customizedName);
+ if (customizedMap == null) {
+ customizedMap = Maps.newHashMap();
+ instanceMap.put(customizedName, customizedMap);
+ }
+ customizedMap.put(resourceName, participantState);
+ } else {
+ LogUtil.logError(LOG, genEventInfo(),
+ "Invalid key found in the participant state cache" + key);
+ }
+ }
+
+ _participantStateMap = Collections.unmodifiableMap(allParticipantStateMap);
+
+ long endTime = System.currentTimeMillis();
+ LogUtil.logInfo(LOG, genEventInfo(),
+ "END: participantStateCache.refresh() for cluster " + _controlContextProvider
+ .getClusterName() + ", started at : " + startTime + ", took " + (endTime - startTime)
+ + " ms");
+ if (LOG.isDebugEnabled()) {
+ LogUtil.logDebug(LOG, genEventInfo(),
+ String.format("Participant State refreshed : %s", _participantStateMap.toString()));
+ }
+ return true;
+ }
+
+ // reload participant states that has been changed from zk to local cache.
+ private void refreshParticipantStatesCacheFromZk(HelixDataAccessor accessor,
+ Map<String, LiveInstance> liveInstanceMap) {
+
+ long start = System.currentTimeMillis();
+ Set<PropertyKey> participantStateKeys = PopulateParticipantKeys(accessor, liveInstanceMap);
+
+ // All new entries from zk not cached locally yet should be read from ZK.
+ Set<PropertyKey> reloadKeys = new HashSet<>(participantStateKeys);
+ reloadKeys.removeAll(_participantStateCache.keySet());
+
+ Set<PropertyKey> cachedKeys = new HashSet<>(_participantStateCache.keySet());
+ cachedKeys.retainAll(participantStateKeys);
+
+ Set<PropertyKey> reloadedKeys = new HashSet<>();
+ Map<PropertyKey, T> newStateCache = Collections.unmodifiableMap(
+ refreshProperties(accessor, reloadKeys, new ArrayList<>(cachedKeys), _participantStateCache,
+ reloadedKeys));
+
+ refreshSnapshot(newStateCache, _participantStateCache, reloadedKeys);
+
+ _participantStateCache = newStateCache;
+
+ if (LOG.isDebugEnabled()) {
+ LogUtil.logDebug(LOG, genEventInfo(),
+ "# of participant state reload: " + reloadKeys.size() + ", skipped:" + (
+ participantStateKeys.size() - reloadKeys.size()) + ". took " + (
+ System.currentTimeMillis() - start)
+ + " ms to reload new participant states for cluster: " + _controlContextProvider
+ .getClusterName() + "and state: " + this.getClass().getName());
+ }
+ }
+
+ protected abstract Set<PropertyKey> PopulateParticipantKeys(HelixDataAccessor accessor,
+ Map<String, LiveInstance> liveInstanceMap);
+
+ /**
+ * Refresh the snapshot of the cache. This method is optional for child class to extend. If the
+ * child class does not need to refresh snapshot, it just does nothing.
+ * @return
+ */
+ protected void refreshSnapshot(Map<PropertyKey, T> newStateCache,
+ Map<PropertyKey, T> participantStateCache, Set<PropertyKey> reloadedKeys) {
+ }
+
+ /**
+ * Return the whole participant state map.
+ * @return
+ */
+ public Map<String, Map<String, Map<String, T>>> getParticipantStatesMap() {
+ return Collections.unmodifiableMap(_participantStateMap);
+ }
+
+ /**
+ * Return all participant states for a certain instance.
+ * @param instanceName
+ * @return
+ */
+ public Map<String, Map<String, T>> getParticipantStates(String instanceName) {
+ if (!_participantStateMap.containsKey(instanceName)) {
+ return Collections.emptyMap();
+ }
+ return Collections.unmodifiableMap(_participantStateMap.get(instanceName));
+ }
+
+ /**
+ * Provides the participant state map for a certain instance and a customized key
+ * @param instanceName
+ * @param customizedKey
+ * @return
+ */
+ public Map<String, T> getParticipantState(String instanceName, String customizedKey) {
+ if (!_participantStateMap.containsKey(instanceName) || !_participantStateMap.get(instanceName)
+ .containsKey(customizedKey)) {
+ return Collections.emptyMap();
+ }
+ return Collections.unmodifiableMap(_participantStateMap.get(instanceName).get(customizedKey));
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
index 405af47..0cd9355 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/dataproviders/BaseControllerDataProvider.java
@@ -337,7 +337,7 @@ public class BaseControllerDataProvider implements ControlContextProvider {
// current state must be refreshed before refreshing relay messages
// because we need to use current state to validate all relay messages.
_instanceMessagesCache.updateRelayMessages(_liveInstanceCache.getPropertyMap(),
- _currentStateCache.getCurrentStatesMap());
+ _currentStateCache.getParticipantStatesMap());
updateIdealRuleMap();
updateDisabledInstances();
@@ -526,7 +526,7 @@ public class BaseControllerDataProvider implements ControlContextProvider {
* @return
*/
public Map<String, CurrentState> getCurrentState(String instanceName, String clientSessionId) {
- return _currentStateCache.getCurrentState(instanceName, clientSessionId);
+ return _currentStateCache.getParticipantState(instanceName, clientSessionId);
}
/**
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java b/helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
new file mode 100644
index 0000000..3863743
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
@@ -0,0 +1,120 @@
+package org.apache.helix.controller.stages;
+
+/*
+ * 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.
+ */
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.helix.model.Partition;
+
+
+public class CustomizedStateOutput {
+ // stateType -> (resourceName -> (Partition -> (instanceName -> customizedState)))
+ private final Map<String, Map<String, Map<Partition, Map<String, String>>>> _customizedStateMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ }
+
+ public void setCustomizedState(String stateType, String resourceName, Partition partition,
+ String instanceName, String state) {
+ if (!_customizedStateMap.containsKey(stateType)) {
+ _customizedStateMap
+ .put(stateType, new HashMap<String, Map<Partition, Map<String, String>>>());
+ }
+ if (!_customizedStateMap.get(stateType).containsKey(resourceName)) {
+ _customizedStateMap.get(stateType)
+ .put(resourceName, new HashMap<Partition, Map<String, String>>());
+ }
+ if (!_customizedStateMap.get(stateType).get(resourceName).containsKey(partition)) {
+ _customizedStateMap.get(stateType).get(resourceName)
+ .put(partition, new HashMap<String, String>());
+ }
+ _customizedStateMap.get(stateType).get(resourceName).get(partition).put(instanceName, state);
+ }
+
+ /**
+ * Given stateType, returns resource customized state map (resource -> parition -> instance ->
+ * customizedState)
+ * @param stateType
+ * @return
+ */
+ public Map<String, Map<Partition, Map<String, String>>> getCustomizedStateMap(String stateType) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ return Collections.unmodifiableMap(_customizedStateMap.get(stateType));
+ }
+ return Collections.emptyMap();
+ }
+
+ /**
+ * given (stateType, resource), returns (partition -> instance-> customizedState) map
+ * @param stateType
+ * @param resourceName
+ * @return
+ */
+ public Map<Partition, Map<String, String>> getResourceCustomizedStateMap(String stateType,
+ String resourceName) {
+ if (getCustomizedStateMap(stateType).containsKey(resourceName)) {
+ return Collections.unmodifiableMap(getCustomizedStateMap(stateType).get(resourceName));
+ }
+ return Collections.emptyMap();
+ }
+
+ /**
+ * given (stateType, resource, partition), returns (instance-> customizedState) map
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @return
+ */
+ public Map<String, String> getPartitionCustomizedStateMap(String stateType, String resourceName,
+ Partition partition) {
+ if (getCustomizedStateMap(stateType).containsKey(resourceName) && getResourceCustomizedStateMap(
+ stateType, resourceName).containsKey(partition)) {
+ return Collections
+ .unmodifiableMap(getResourceCustomizedStateMap(stateType, resourceName).get(partition));
+ }
+ return Collections.emptyMap();
+ }
+
+ /**
+ * given (stateType, resource, partition, instance), returns customized state
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @param instanceName
+ * @return
+ */
+ public String getPartitionCustomizedState(String stateType, String resourceName,
+ Partition partition, String instanceName) {
+ if (getCustomizedStateMap(stateType).containsKey(resourceName) && getResourceCustomizedStateMap(
+ stateType, resourceName).containsKey(partition) && getPartitionCustomizedStateMap(stateType,
+ resourceName, partition).containsKey(instanceName)) {
+ return getPartitionCustomizedStateMap(stateType, resourceName, partition).get(instanceName);
+ }
+ return null;
+ }
+
+ public Set<String> getAllStateTypes() {
+ return _customizedStateMap.keySet();
+ }
+}
diff --git a/helix-core/src/main/java/org/apache/helix/model/CustomizedStateAggregationConfig.java b/helix-core/src/main/java/org/apache/helix/model/CustomizedStateAggregationConfig.java
index 8d13fda..4acd0e6 100644
--- a/helix-core/src/main/java/org/apache/helix/model/CustomizedStateAggregationConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/model/CustomizedStateAggregationConfig.java
@@ -63,7 +63,7 @@ public class CustomizedStateAggregationConfig extends HelixProperty {
}
/**
- * Set the AGGREGATION_ENABLED_STATES field.
+ * Set the AGGREGATION_ENABLED_TYPES field.
* @param aggregationEnabledTypes
*/
public void setAggregationEnabledTypes(List<String> aggregationEnabledTypes) {
@@ -72,8 +72,8 @@ public class CustomizedStateAggregationConfig extends HelixProperty {
}
/**
- * Get the AGGREGATION_ENABLED_STATES field.
- * @return AGGREGATION_ENABLED_STATES field.
+ * Get the AGGREGATION_ENABLED_TYPES field.
+ * @return AGGREGATION_ENABLED_TYPES field.
*/
public List<String> getAggregationEnabledTypes() {
return _record
diff --git a/helix-core/src/main/java/org/apache/helix/spectator/RoutingDataCache.java b/helix-core/src/main/java/org/apache/helix/spectator/RoutingDataCache.java
index 7af0d28..4896455 100644
--- a/helix-core/src/main/java/org/apache/helix/spectator/RoutingDataCache.java
+++ b/helix-core/src/main/java/org/apache/helix/spectator/RoutingDataCache.java
@@ -121,7 +121,7 @@ class RoutingDataCache extends BasicClusterDataCache {
* @return
*/
public Map<String, Map<String, Map<String, CurrentState>>> getCurrentStatesMap() {
- return _currentStateCache.getCurrentStatesMap();
+ return _currentStateCache.getParticipantStatesMap();
}
public CurrentStateSnapshot getCurrentStateSnapshot() {