You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@helix.apache.org by GitBox <gi...@apache.org> on 2020/02/27 19:45:52 UTC
[GitHub] [helix] zhangmeng916 opened a new pull request #827: Add
intermediate storage for customized state
zhangmeng916 opened a new pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827
### Issues
- [X] My PR addresses the following Helix issues and references them in the PR description:
(#826 )
### Description
- [X] Here are some details about my PR, including screenshots of any UI changes:
To support the storing and computation over customized state, Helix needs to implement some intermediate storage for better performance and easy operation. This PR adds a cache for store customized state and refresh it when necessary. It also adds a new customized state output data structure to store the data that will be used for computation.
### Tests
- [X] The following is the result of the "mvn test" command on the appropriate module:
### Commits
- [X] My commits all reference appropriate Apache Helix GitHub issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
1. Subject is limited to 50 characters (not including Jira issue reference)
1. Subject does not end with a period
1. Subject uses the imperative mood ("add", not "adding")
1. Body wraps at 72 characters
1. Body explains "what" and "why", not "how"
### Code Quality
- [X] My diff has been formatted using helix-style.xml
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386740708
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String keyLevel1 = params[1];
Review comment:
Although it is an abstract class, level one for the instancename should be fixed. We can keep it here for clearer code.
As for level 2 and 3, we might want to comment it with an example for the other code readers.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387864883
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,183 @@
+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.List;
+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 = Collections.emptyMap();
+ }
+
+ /**
+ * 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,
+ Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ 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, Boolean snapshotEnabled,
+ Set<String> restrictedKeys) {
+
+ long start = System.currentTimeMillis();
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
+ PopulateParticipantKeys(accessor, participantStateKeys, liveInstanceMap, restrictedKeys);
+
+ // 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));
+
+ if (snapshotEnabled) {
+ 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());
+ }
+ }
+
+ public void PopulateParticipantKeys(HelixDataAccessor accessor,
+ Set<PropertyKey> participantStateKeys, Map<String, LiveInstance> liveInstanceMap,
+ Set<String> restrictedKeys) {
+ }
+
+ protected void refreshSnapshot(Map<PropertyKey, T> newStateCache,
Review comment:
This one as well. Try to use more abstract methods if the implementation is not possible in this abstract 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385449973
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,186 @@
+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.List;
+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.CustomizedState;
+import org.apache.helix.model.LiveInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomizedStateCache extends AbstractDataCache<CustomizedState> {
Review comment:
As we discussed offline, we should either leverage propertyCache class or try to combine the common part of this cache and the current state cache.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386738894
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
Review comment:
Add a simple comment for this class, please.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386739955
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
Review comment:
As an abstract class, usually, we only need one constructor for the children.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r388112109
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,177 @@
+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.List;
+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,
+ Boolean snapshotEnabled) {
Review comment:
Should this boolean be fixed once the class is coded? Overall, I don't think we need it. Just call refreshSnapshot() anyway. If there is no such implementation, the child class won't implement it anyway.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387862692
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CurrentStateCache.java
##########
@@ -55,149 +46,33 @@ public CurrentStateCache(String clusterName) {
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()));
+ protected void refreshSnapshot(Map<PropertyKey, CurrentState> newStateCache,
+ Map<PropertyKey, CurrentState> participantStateCache, Set<PropertyKey> reloadedKeys) {
+ if (_initialized) {
+ _snapshot = new CurrentStateSnapshot(newStateCache, participantStateCache, reloadedKeys);
+ } else {
+ _snapshot = new CurrentStateSnapshot(newStateCache);
+ _initialized = true;
}
- return true;
}
- // reload current states that has been changed from zk to local cache.
- private void refreshCurrentStatesCache(HelixDataAccessor accessor,
- Map<String, LiveInstance> liveInstanceMap) {
-
- long start = System.currentTimeMillis();
+ public void PopulateParticipantKeys(HelixDataAccessor accessor,
Review comment:
If the input participantStateKeys is always empty and is a return value, why not just let this method return a set for the keys?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #827:
Add intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387283782
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String keyLevel1 = params[1];
+ String keyLevel2 = params[2];
+ String keyLevel3 = params[3];
+ Map<String, Map<String, T>> stateMapLevel1 =
+ allParticipantStateMap.get(keyLevel1);
+ if (stateMapLevel1 == null) {
+ stateMapLevel1 = Maps.newHashMap();
+ allParticipantStateMap.put(keyLevel1, stateMapLevel1);
+ }
+ Map<String, T> stateMapLevel2 = stateMapLevel1.get(keyLevel2);
+ if (stateMapLevel2 == null) {
+ stateMapLevel2 = Maps.newHashMap();
+ stateMapLevel1.put(keyLevel2, stateMapLevel2);
+ }
+ stateMapLevel2.put(keyLevel3, participantState);
Review comment:
I tried a bit, but it cannot really be simplified by using putIfAbsent, as the next level map still need to use the previous map.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang merged pull request #827: Add intermediate
storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang merged pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386745840
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String keyLevel1 = params[1];
+ String keyLevel2 = params[2];
+ String keyLevel3 = params[3];
+ Map<String, Map<String, T>> stateMapLevel1 =
+ allParticipantStateMap.get(keyLevel1);
+ if (stateMapLevel1 == null) {
+ stateMapLevel1 = Maps.newHashMap();
+ allParticipantStateMap.put(keyLevel1, stateMapLevel1);
+ }
+ Map<String, T> stateMapLevel2 = stateMapLevel1.get(keyLevel2);
+ if (stateMapLevel2 == null) {
+ stateMapLevel2 = Maps.newHashMap();
+ stateMapLevel1.put(keyLevel2, stateMapLevel2);
+ }
+ stateMapLevel2.put(keyLevel3, participantState);
+ }
+ }
+
+ for (String instance : allParticipantStateMap.keySet()) {
+ allParticipantStateMap.put(instance, Collections.unmodifiableMap(allParticipantStateMap.get(instance)));
+ }
+ _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, Boolean snapshotEnabled,
+ Set<String> restrictedKeys) {
+
+ long start = System.currentTimeMillis();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
+ for (String instanceName : liveInstanceMap.keySet()) {
+ LiveInstance liveInstance = liveInstanceMap.get(instanceName);
+
+ if (restrictedKeys.isEmpty()) {
+ String sessionId = liveInstance.getEphemeralOwner();
+ List<String> currentStateNames =
+ accessor.getChildNames(keyBuilder.currentStates(instanceName, sessionId));
+
+ for (String currentStateName : currentStateNames) {
+ participantStateKeys
+ .add(keyBuilder.currentState(instanceName, sessionId, currentStateName));
+ }
+ } else {
+ List<String> resourceNames = new ArrayList<>();
+ for (String customizedStateType : restrictedKeys) {
+ resourceNames = accessor
+ .getChildNames(keyBuilder.customizedStates(instanceName, customizedStateType));
+ for (String resourceName : resourceNames) {
+ participantStateKeys
+ .add(keyBuilder.customizedState(instanceName, customizedStateType, resourceName));
+ }
+ }
+ }
+ }
+ // 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));
+
+ if (snapshotEnabled) {
+ 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 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 level1 key.
+ * @param keyLevel1
+ * @return
+ */
+ public Map<String, Map<String, T>> getParticipantStates(String keyLevel1) {
+ if (!_participantStateMap.containsKey(keyLevel1)) {
+ return Collections.emptyMap();
+ }
+ return Collections.unmodifiableMap(_participantStateMap.get(keyLevel1));
+ }
+
+ /**
+ * Provides the participant state map for a certain level1 key and level2 key
+ * @param keyLevel1
+ * @param keyLevel2
+ * @return
+ */
+ public Map<String, T> getParticipantState(String keyLevel1,
Review comment:
Optional, both getParticipantStates() methods could be protected and let the children class wrap them and implement with more meaningful public methods.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r388114768
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,126 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
Review comment:
How this is used? I would suggest adding this only when we have the usage in the PR.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385967597
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,37 @@
+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 org.apache.helix.common.controllers.ControlContextProvider;
+import org.apache.helix.model.CustomizedState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomizedStateCache extends AbstractDataCache<CustomizedState> {
+ private static final Logger LOG = LoggerFactory.getLogger(CurrentStateCache.class.getName());
Review comment:
nit, I think you can directly use CurrentStateCache.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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385971667
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,135 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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, resource, partition, instance), returns customized state
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @param instanceName
+ * @return
+ */
+ public String getCustomizedState(String stateType, String resourceName, Partition partition,
+ String instanceName) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map != null) {
+ Map<Partition, Map<String, String>> resourceMap = map.get(resourceName);
+ if (resourceMap != null) {
+ Map<String, String> instanceStateMap = resourceMap.get(partition);
+ if (instanceStateMap != null) {
+ return instanceStateMap.get(instanceName);
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Given stateType, returns resource customized state map (resource -> parition -> instance ->
+ * customizedState)
+ * @param stateType
+ * @return
+ */
+ public Map<String, Map<Partition, Map<String, String>>> getResourceCustomizedStateMap(
+ String stateType) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ return _customizedStateMap.get(stateType);
Review comment:
duplicate and return is safer.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385428420
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,186 @@
+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.List;
+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.CustomizedState;
+import org.apache.helix.model.LiveInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomizedStateCache extends AbstractDataCache<CustomizedState> {
+ private static final Logger LOG = LoggerFactory.getLogger(CurrentStateCache.class.getName());
+
+ // instance -> customizedStateType -> resource -> CustomizedState
+ private Map<String, Map<String, Map<String, CustomizedState>>> _customizedStateMap;
+ private Map<PropertyKey, CustomizedState> _customizedStateCache = Maps.newHashMap();
+
+ public CustomizedStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+ public CustomizedStateCache(ControlContextProvider contextProvider) {
+ super(contextProvider);
+ _customizedStateMap = Collections.emptyMap();
+ }
+
+ /**
+ * This refreshes the CustomizedStates 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,
+ Set<String> aggregationEnabledTypes) {
+ long startTime = System.currentTimeMillis();
+
+ refreshCustomizedStatesCache(accessor, liveInstanceMap, aggregationEnabledTypes);
+
+ Map<String, Map<String, Map<String, CustomizedState>>> allCustomizedStateMap = new HashMap<>();
+ for (PropertyKey key : _customizedStateCache.keySet()) {
+ CustomizedState customizedState = _customizedStateCache.get(key);
+ String[] params = key.getParams();
+ if (customizedState != null && params.length >= 4) {
+ String instanceName = params[1];
+ String customizedStateType = params[2];
+ String resourceName = params[3];
+ Map<String, Map<String, CustomizedState>> instanceCustomizedStateMap =
+ allCustomizedStateMap.get(instanceName);
+ if (instanceCustomizedStateMap == null) {
+ instanceCustomizedStateMap = Maps.newHashMap();
+ allCustomizedStateMap.put(instanceName, instanceCustomizedStateMap);
+ }
+ Map<String, CustomizedState> customizedStateNameCustomizedStateMap =
+ instanceCustomizedStateMap.get(customizedStateType);
+ if (customizedStateNameCustomizedStateMap == null) {
+ customizedStateNameCustomizedStateMap = Maps.newHashMap();
+ instanceCustomizedStateMap.put(customizedStateType,
+ customizedStateNameCustomizedStateMap);
+ }
+ customizedStateNameCustomizedStateMap.put(resourceName, customizedState);
+ }
+ }
+
+ for (String instance : allCustomizedStateMap.keySet()) {
+ allCustomizedStateMap.put(instance,
+ Collections.unmodifiableMap(allCustomizedStateMap.get(instance)));
+ }
+ _customizedStateMap = Collections.unmodifiableMap(allCustomizedStateMap);
+
+ long endTime = System.currentTimeMillis();
+ LogUtil.logInfo(LOG, genEventInfo(),
+ "END: CustomizedStateCache.refresh() for cluster "
+ + _controlContextProvider.getClusterName() + ", started at : " + startTime + ", took "
+ + (endTime - startTime) + " ms");
+ if (LOG.isDebugEnabled()) {
+ LogUtil.logDebug(LOG, genEventInfo(),
+ String.format("Customized State refreshed : %s", _customizedStateMap.toString()));
+ }
+ return true;
+ }
+
+ // reload customized states that has been changed from zk to local cache.
+ private void refreshCustomizedStatesCache(HelixDataAccessor accessor,
+ Map<String, LiveInstance> liveInstanceMap, Set<String> aggregationEnabledTypes) {
+ long start = System.currentTimeMillis();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ Set<PropertyKey> customizedStateKeys = new HashSet<>();
+ List<String> resourceNames = new ArrayList<>();
+ for (String instanceName : liveInstanceMap.keySet()) {
+ for (String customizedStateName : aggregationEnabledTypes) {
Review comment:
And lamda shall fit here very well.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385971886
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,135 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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, resource, partition, instance), returns customized state
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @param instanceName
+ * @return
+ */
+ public String getCustomizedState(String stateType, String resourceName, Partition partition,
+ String instanceName) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map != null) {
+ Map<Partition, Map<String, String>> resourceMap = map.get(resourceName);
+ if (resourceMap != null) {
+ Map<String, String> instanceStateMap = resourceMap.get(partition);
+ if (instanceStateMap != null) {
+ return instanceStateMap.get(instanceName);
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Given stateType, returns resource customized state map (resource -> parition -> instance ->
+ * customizedState)
+ * @param stateType
+ * @return
+ */
+ public Map<String, Map<Partition, Map<String, String>>> getResourceCustomizedStateMap(
+ String stateType) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ return _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>> getCustomizedStateMap(String stateType,
Review comment:
This method should call getResourceCustomizedStateMap() to avoid duplicate code.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r388114359
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,126 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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>>> getResourceCustomizedStateMap(
+ 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>> getPartitionCustomizedStateMap(String stateType,
+ String resourceName) {
+ if (getResourceCustomizedStateMap(stateType).containsKey(resourceName)) {
+ return Collections
+ .unmodifiableMap(getResourceCustomizedStateMap(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> getInstanceCustomizedStateMap(String stateType, String resourceName,
Review comment:
Intuitively, this one should be called getPartitionCustomizedStateMap()
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r388114552
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,126 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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>>> getResourceCustomizedStateMap(
+ 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>> getPartitionCustomizedStateMap(String stateType,
+ String resourceName) {
+ if (getResourceCustomizedStateMap(stateType).containsKey(resourceName)) {
+ return Collections
+ .unmodifiableMap(getResourceCustomizedStateMap(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> getInstanceCustomizedStateMap(String stateType, String resourceName,
+ Partition partition) {
+ if (getResourceCustomizedStateMap(stateType).containsKey(resourceName)
+ && getPartitionCustomizedStateMap(stateType, resourceName).containsKey(partition)) {
+ return Collections
+ .unmodifiableMap(getPartitionCustomizedStateMap(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 getCustomizedState(String stateType, String resourceName, Partition partition,
Review comment:
This one can be called getReplicaCustomizedState()
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385966866
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/AbstractDataCache.java
##########
@@ -104,6 +113,157 @@ public AbstractDataCache(ControlContextProvider controlContextProvider) {
return refreshedPropertyMap;
}
+
+ /**
+ * 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 refreshParticipantStatesCache(HelixDataAccessor accessor,
Review comment:
You can keep the original name refresh() so it fits the style.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #827:
Add intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387288169
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,37 @@
+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 org.apache.helix.common.controllers.ControlContextProvider;
+import org.apache.helix.model.CustomizedState;
+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);
+ }
Review comment:
Yeah, actually they're both used.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386743747
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String keyLevel1 = params[1];
+ String keyLevel2 = params[2];
+ String keyLevel3 = params[3];
+ Map<String, Map<String, T>> stateMapLevel1 =
+ allParticipantStateMap.get(keyLevel1);
+ if (stateMapLevel1 == null) {
+ stateMapLevel1 = Maps.newHashMap();
+ allParticipantStateMap.put(keyLevel1, stateMapLevel1);
+ }
+ Map<String, T> stateMapLevel2 = stateMapLevel1.get(keyLevel2);
+ if (stateMapLevel2 == null) {
+ stateMapLevel2 = Maps.newHashMap();
+ stateMapLevel1.put(keyLevel2, stateMapLevel2);
+ }
+ stateMapLevel2.put(keyLevel3, participantState);
Review comment:
nit, can we simplify this block a little bit. For example, putIfAbsent, etc.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #827:
Add intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385967157
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,135 @@
+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.CustomizedState;
+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;
Review comment:
CurrentStateOutput has more maps that CustomizedStateOutput. And those maps are with different types of keys. I feel it's actually easier to know what the map has and what exactly each level of keys are by looking at the map. If we have another class of hierarchical map, we will still need to pass the types for the key, and how many levels there are, and if we only manipulate map in that class, it is harder to users to understand the map logic.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385428853
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,186 @@
+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.List;
+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.CustomizedState;
+import org.apache.helix.model.LiveInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomizedStateCache extends AbstractDataCache<CustomizedState> {
+ private static final Logger LOG = LoggerFactory.getLogger(CurrentStateCache.class.getName());
+
+ // instance -> customizedStateType -> resource -> CustomizedState
+ private Map<String, Map<String, Map<String, CustomizedState>>> _customizedStateMap;
+ private Map<PropertyKey, CustomizedState> _customizedStateCache = Maps.newHashMap();
+
+ public CustomizedStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+ public CustomizedStateCache(ControlContextProvider contextProvider) {
+ super(contextProvider);
+ _customizedStateMap = Collections.emptyMap();
+ }
+
+ /**
+ * This refreshes the CustomizedStates 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,
+ Set<String> aggregationEnabledTypes) {
+ long startTime = System.currentTimeMillis();
+
+ refreshCustomizedStatesCache(accessor, liveInstanceMap, aggregationEnabledTypes);
+
+ Map<String, Map<String, Map<String, CustomizedState>>> allCustomizedStateMap = new HashMap<>();
+ for (PropertyKey key : _customizedStateCache.keySet()) {
+ CustomizedState customizedState = _customizedStateCache.get(key);
+ String[] params = key.getParams();
+ if (customizedState != null && params.length >= 4) {
+ String instanceName = params[1];
+ String customizedStateType = params[2];
+ String resourceName = params[3];
+ Map<String, Map<String, CustomizedState>> instanceCustomizedStateMap =
+ allCustomizedStateMap.get(instanceName);
+ if (instanceCustomizedStateMap == null) {
+ instanceCustomizedStateMap = Maps.newHashMap();
+ allCustomizedStateMap.put(instanceName, instanceCustomizedStateMap);
+ }
+ Map<String, CustomizedState> customizedStateNameCustomizedStateMap =
+ instanceCustomizedStateMap.get(customizedStateType);
+ if (customizedStateNameCustomizedStateMap == null) {
+ customizedStateNameCustomizedStateMap = Maps.newHashMap();
+ instanceCustomizedStateMap.put(customizedStateType,
+ customizedStateNameCustomizedStateMap);
+ }
+ customizedStateNameCustomizedStateMap.put(resourceName, customizedState);
+ }
+ }
+
+ for (String instance : allCustomizedStateMap.keySet()) {
+ allCustomizedStateMap.put(instance,
+ Collections.unmodifiableMap(allCustomizedStateMap.get(instance)));
+ }
+ _customizedStateMap = Collections.unmodifiableMap(allCustomizedStateMap);
+
+ long endTime = System.currentTimeMillis();
+ LogUtil.logInfo(LOG, genEventInfo(),
+ "END: CustomizedStateCache.refresh() for cluster "
+ + _controlContextProvider.getClusterName() + ", started at : " + startTime + ", took "
+ + (endTime - startTime) + " ms");
+ if (LOG.isDebugEnabled()) {
+ LogUtil.logDebug(LOG, genEventInfo(),
+ String.format("Customized State refreshed : %s", _customizedStateMap.toString()));
+ }
+ return true;
+ }
+
+ // reload customized states that has been changed from zk to local cache.
+ private void refreshCustomizedStatesCache(HelixDataAccessor accessor,
+ Map<String, LiveInstance> liveInstanceMap, Set<String> aggregationEnabledTypes) {
+ long start = System.currentTimeMillis();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ Set<PropertyKey> customizedStateKeys = new HashSet<>();
+ List<String> resourceNames = new ArrayList<>();
+ for (String instanceName : liveInstanceMap.keySet()) {
+ for (String customizedStateName : aggregationEnabledTypes) {
+ resourceNames =
+ accessor.getChildNames(keyBuilder.customizedStates(instanceName, customizedStateName));
+ for (String resourceName : resourceNames) {
+ customizedStateKeys
Review comment:
nit, Just call it keys or stateKeys? For shortening the code a little bit.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385971758
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,135 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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, resource, partition, instance), returns customized state
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @param instanceName
+ * @return
+ */
+ public String getCustomizedState(String stateType, String resourceName, Partition partition,
+ String instanceName) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map != null) {
+ Map<Partition, Map<String, String>> resourceMap = map.get(resourceName);
+ if (resourceMap != null) {
+ Map<String, String> instanceStateMap = resourceMap.get(partition);
+ if (instanceStateMap != null) {
+ return instanceStateMap.get(instanceName);
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Given stateType, returns resource customized state map (resource -> parition -> instance ->
+ * customizedState)
+ * @param stateType
+ * @return
+ */
+ public Map<String, Map<Partition, Map<String, String>>> getResourceCustomizedStateMap(
+ String stateType) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ return _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>> getCustomizedStateMap(String stateType,
+ String resourceName) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map.containsKey(resourceName)) {
+ return map.get(resourceName);
Review comment:
Same here. Please copy the map and then return.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r388112511
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,177 @@
+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.List;
+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,
+ Boolean snapshotEnabled) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled);
+ 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, Boolean snapshotEnabled) {
+
+ 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));
+
+ if (snapshotEnabled) {
+ 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);
+
+ protected abstract void refreshSnapshot(Map<PropertyKey, T> newStateCache, Map<PropertyKey, T> participantStateCache,
Review comment:
Since the customized state cache won't support snapshot for now, just comment on this method as OPTIONAL to be implemented. If the child class does not support snapshot, just do nothing. And we don't need the boolean to differentiate.
We might want to support it later though.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386743446
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
Review comment:
log error (or warning) if any mismatch keys.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387864354
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,183 @@
+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.List;
+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 = Collections.emptyMap();
+ }
+
+ /**
+ * 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,
+ Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ 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, Boolean snapshotEnabled,
+ Set<String> restrictedKeys) {
+
+ long start = System.currentTimeMillis();
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
+ PopulateParticipantKeys(accessor, participantStateKeys, liveInstanceMap, restrictedKeys);
+
+ // 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));
+
+ if (snapshotEnabled) {
+ 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());
+ }
+ }
+
+ public void PopulateParticipantKeys(HelixDataAccessor accessor,
Review comment:
Usually, this class should be protected only. Since it is only called in side this 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387870303
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,183 @@
+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.List;
+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 = Collections.emptyMap();
+ }
+
+ /**
+ * 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,
+ Boolean snapshotEnabled, Set<String> restrictedKeys) {
Review comment:
As we talked offline, let's try to hide restrictedKeys as an internal field and don't require the caller to pass on every refresh call.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r388114206
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,126 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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>>> getResourceCustomizedStateMap(
Review comment:
Intuitively, this one should be called getCustomizedStateMap()
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #827:
Add intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387266136
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String keyLevel1 = params[1];
+ String keyLevel2 = params[2];
+ String keyLevel3 = params[3];
+ Map<String, Map<String, T>> stateMapLevel1 =
+ allParticipantStateMap.get(keyLevel1);
+ if (stateMapLevel1 == null) {
+ stateMapLevel1 = Maps.newHashMap();
+ allParticipantStateMap.put(keyLevel1, stateMapLevel1);
+ }
+ Map<String, T> stateMapLevel2 = stateMapLevel1.get(keyLevel2);
+ if (stateMapLevel2 == null) {
+ stateMapLevel2 = Maps.newHashMap();
+ stateMapLevel1.put(keyLevel2, stateMapLevel2);
+ }
+ stateMapLevel2.put(keyLevel3, participantState);
+ }
+ }
+
+ for (String instance : allParticipantStateMap.keySet()) {
+ allParticipantStateMap.put(instance, Collections.unmodifiableMap(allParticipantStateMap.get(instance)));
+ }
+ _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, Boolean snapshotEnabled,
+ Set<String> restrictedKeys) {
+
+ long start = System.currentTimeMillis();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
+ for (String instanceName : liveInstanceMap.keySet()) {
+ LiveInstance liveInstance = liveInstanceMap.get(instanceName);
+
+ if (restrictedKeys.isEmpty()) {
+ String sessionId = liveInstance.getEphemeralOwner();
+ List<String> currentStateNames =
+ accessor.getChildNames(keyBuilder.currentStates(instanceName, sessionId));
+
+ for (String currentStateName : currentStateNames) {
+ participantStateKeys
+ .add(keyBuilder.currentState(instanceName, sessionId, currentStateName));
+ }
+ } else {
+ List<String> resourceNames = new ArrayList<>();
+ for (String customizedStateType : restrictedKeys) {
+ resourceNames = accessor
+ .getChildNames(keyBuilder.customizedStates(instanceName, customizedStateType));
+ for (String resourceName : resourceNames) {
+ participantStateKeys
+ .add(keyBuilder.customizedState(instanceName, customizedStateType, resourceName));
+ }
+ }
+ }
+ }
+ // 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));
+
+ if (snapshotEnabled) {
+ 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 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 level1 key.
+ * @param keyLevel1
+ * @return
+ */
+ public Map<String, Map<String, T>> getParticipantStates(String keyLevel1) {
+ if (!_participantStateMap.containsKey(keyLevel1)) {
+ return Collections.emptyMap();
+ }
+ return Collections.unmodifiableMap(_participantStateMap.get(keyLevel1));
+ }
+
+ /**
+ * Provides the participant state map for a certain level1 key and level2 key
+ * @param keyLevel1
+ * @param keyLevel2
+ * @return
+ */
+ public Map<String, T> getParticipantState(String keyLevel1,
Review comment:
I changed the key names here. I feel it's ok to leave it here as it only returns some mapping.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #827:
Add intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387899648
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CurrentStateCache.java
##########
@@ -55,149 +46,33 @@ public CurrentStateCache(String clusterName) {
public CurrentStateCache(ControlContextProvider contextProvider) {
super(contextProvider);
- _currentStateMap = Collections.emptyMap();
- _snapshot = new CurrentStateSnapshot(_currentStateCache);
+ _snapshot = new CurrentStateSnapshot(_participantStateCache);
Review comment:
I changed the map in ParticipantCache.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387861358
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CurrentStateCache.java
##########
@@ -55,149 +46,33 @@ public CurrentStateCache(String clusterName) {
public CurrentStateCache(ControlContextProvider contextProvider) {
super(contextProvider);
- _currentStateMap = Collections.emptyMap();
- _snapshot = new CurrentStateSnapshot(_currentStateCache);
+ _snapshot = new CurrentStateSnapshot(_participantStateCache);
Review comment:
The original logic is actually confusing here. Passing an empty new map here is much easier for people to understand.
Since it won't change the logic, I suggest you just change it to make it clean : )
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385428088
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,186 @@
+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.List;
+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.CustomizedState;
+import org.apache.helix.model.LiveInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomizedStateCache extends AbstractDataCache<CustomizedState> {
+ private static final Logger LOG = LoggerFactory.getLogger(CurrentStateCache.class.getName());
+
+ // instance -> customizedStateType -> resource -> CustomizedState
+ private Map<String, Map<String, Map<String, CustomizedState>>> _customizedStateMap;
+ private Map<PropertyKey, CustomizedState> _customizedStateCache = Maps.newHashMap();
+
+ public CustomizedStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+ public CustomizedStateCache(ControlContextProvider contextProvider) {
+ super(contextProvider);
+ _customizedStateMap = Collections.emptyMap();
+ }
+
+ /**
+ * This refreshes the CustomizedStates 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,
+ Set<String> aggregationEnabledTypes) {
+ long startTime = System.currentTimeMillis();
+
+ refreshCustomizedStatesCache(accessor, liveInstanceMap, aggregationEnabledTypes);
+
+ Map<String, Map<String, Map<String, CustomizedState>>> allCustomizedStateMap = new HashMap<>();
+ for (PropertyKey key : _customizedStateCache.keySet()) {
+ CustomizedState customizedState = _customizedStateCache.get(key);
+ String[] params = key.getParams();
+ if (customizedState != null && params.length >= 4) {
+ String instanceName = params[1];
+ String customizedStateType = params[2];
+ String resourceName = params[3];
+ Map<String, Map<String, CustomizedState>> instanceCustomizedStateMap =
+ allCustomizedStateMap.get(instanceName);
+ if (instanceCustomizedStateMap == null) {
+ instanceCustomizedStateMap = Maps.newHashMap();
+ allCustomizedStateMap.put(instanceName, instanceCustomizedStateMap);
+ }
+ Map<String, CustomizedState> customizedStateNameCustomizedStateMap =
+ instanceCustomizedStateMap.get(customizedStateType);
+ if (customizedStateNameCustomizedStateMap == null) {
+ customizedStateNameCustomizedStateMap = Maps.newHashMap();
+ instanceCustomizedStateMap.put(customizedStateType,
+ customizedStateNameCustomizedStateMap);
+ }
+ customizedStateNameCustomizedStateMap.put(resourceName, customizedState);
+ }
+ }
+
+ for (String instance : allCustomizedStateMap.keySet()) {
+ allCustomizedStateMap.put(instance,
+ Collections.unmodifiableMap(allCustomizedStateMap.get(instance)));
+ }
+ _customizedStateMap = Collections.unmodifiableMap(allCustomizedStateMap);
+
+ long endTime = System.currentTimeMillis();
+ LogUtil.logInfo(LOG, genEventInfo(),
+ "END: CustomizedStateCache.refresh() for cluster "
+ + _controlContextProvider.getClusterName() + ", started at : " + startTime + ", took "
+ + (endTime - startTime) + " ms");
+ if (LOG.isDebugEnabled()) {
+ LogUtil.logDebug(LOG, genEventInfo(),
+ String.format("Customized State refreshed : %s", _customizedStateMap.toString()));
+ }
+ return true;
+ }
+
+ // reload customized states that has been changed from zk to local cache.
+ private void refreshCustomizedStatesCache(HelixDataAccessor accessor,
+ Map<String, LiveInstance> liveInstanceMap, Set<String> aggregationEnabledTypes) {
+ long start = System.currentTimeMillis();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ Set<PropertyKey> customizedStateKeys = new HashSet<>();
+ List<String> resourceNames = new ArrayList<>();
+ for (String instanceName : liveInstanceMap.keySet()) {
+ for (String customizedStateName : aggregationEnabledTypes) {
Review comment:
nit, customizedStateName -> customizedStateType or 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r388114307
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,126 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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>>> getResourceCustomizedStateMap(
+ 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>> getPartitionCustomizedStateMap(String stateType,
Review comment:
Intuitively, this one should be called getResourceCustomizedStateMap()
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386739065
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
Review comment:
getName() is not preferred. Please just pass the class as the parameter.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #827:
Add intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385977246
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,37 @@
+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 org.apache.helix.common.controllers.ControlContextProvider;
+import org.apache.helix.model.CustomizedState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomizedStateCache extends AbstractDataCache<CustomizedState> {
+ private static final Logger LOG = LoggerFactory.getLogger(CurrentStateCache.class.getName());
Review comment:
Let's keep it separated, so that in the computation logic, there's no confusion.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r387864201
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,183 @@
+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.List;
+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 = Collections.emptyMap();
+ }
+
+ /**
+ * 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,
+ Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ 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, Boolean snapshotEnabled,
+ Set<String> restrictedKeys) {
+
+ long start = System.currentTimeMillis();
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
+ PopulateParticipantKeys(accessor, participantStateKeys, liveInstanceMap, restrictedKeys);
+
+ // 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));
+
+ if (snapshotEnabled) {
+ 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());
+ }
+ }
+
+ public void PopulateParticipantKeys(HelixDataAccessor accessor,
Review comment:
This should be an abstract method with no implementation. So the child class is required to implement it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on a change in pull request #827:
Add intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386665912
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,186 @@
+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.List;
+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.CustomizedState;
+import org.apache.helix.model.LiveInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CustomizedStateCache extends AbstractDataCache<CustomizedState> {
+ private static final Logger LOG = LoggerFactory.getLogger(CurrentStateCache.class.getName());
+
+ // instance -> customizedStateType -> resource -> CustomizedState
+ private Map<String, Map<String, Map<String, CustomizedState>>> _customizedStateMap;
+ private Map<PropertyKey, CustomizedState> _customizedStateCache = Maps.newHashMap();
+
+ public CustomizedStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+ public CustomizedStateCache(ControlContextProvider contextProvider) {
+ super(contextProvider);
+ _customizedStateMap = Collections.emptyMap();
+ }
+
+ /**
+ * This refreshes the CustomizedStates 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,
+ Set<String> aggregationEnabledTypes) {
+ long startTime = System.currentTimeMillis();
+
+ refreshCustomizedStatesCache(accessor, liveInstanceMap, aggregationEnabledTypes);
+
+ Map<String, Map<String, Map<String, CustomizedState>>> allCustomizedStateMap = new HashMap<>();
+ for (PropertyKey key : _customizedStateCache.keySet()) {
+ CustomizedState customizedState = _customizedStateCache.get(key);
+ String[] params = key.getParams();
+ if (customizedState != null && params.length >= 4) {
+ String instanceName = params[1];
+ String customizedStateType = params[2];
+ String resourceName = params[3];
+ Map<String, Map<String, CustomizedState>> instanceCustomizedStateMap =
+ allCustomizedStateMap.get(instanceName);
+ if (instanceCustomizedStateMap == null) {
+ instanceCustomizedStateMap = Maps.newHashMap();
+ allCustomizedStateMap.put(instanceName, instanceCustomizedStateMap);
+ }
+ Map<String, CustomizedState> customizedStateNameCustomizedStateMap =
+ instanceCustomizedStateMap.get(customizedStateType);
+ if (customizedStateNameCustomizedStateMap == null) {
+ customizedStateNameCustomizedStateMap = Maps.newHashMap();
+ instanceCustomizedStateMap.put(customizedStateType,
+ customizedStateNameCustomizedStateMap);
+ }
+ customizedStateNameCustomizedStateMap.put(resourceName, customizedState);
+ }
+ }
+
+ for (String instance : allCustomizedStateMap.keySet()) {
+ allCustomizedStateMap.put(instance,
+ Collections.unmodifiableMap(allCustomizedStateMap.get(instance)));
+ }
+ _customizedStateMap = Collections.unmodifiableMap(allCustomizedStateMap);
+
+ long endTime = System.currentTimeMillis();
+ LogUtil.logInfo(LOG, genEventInfo(),
+ "END: CustomizedStateCache.refresh() for cluster "
+ + _controlContextProvider.getClusterName() + ", started at : " + startTime + ", took "
+ + (endTime - startTime) + " ms");
+ if (LOG.isDebugEnabled()) {
+ LogUtil.logDebug(LOG, genEventInfo(),
+ String.format("Customized State refreshed : %s", _customizedStateMap.toString()));
+ }
+ return true;
+ }
+
+ // reload customized states that has been changed from zk to local cache.
+ private void refreshCustomizedStatesCache(HelixDataAccessor accessor,
+ Map<String, LiveInstance> liveInstanceMap, Set<String> aggregationEnabledTypes) {
+ long start = System.currentTimeMillis();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+ Set<PropertyKey> customizedStateKeys = new HashSet<>();
+ List<String> resourceNames = new ArrayList<>();
+ for (String instanceName : liveInstanceMap.keySet()) {
+ for (String customizedStateName : aggregationEnabledTypes) {
Review comment:
It's moved to participantStateCache logic and updated.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385966377
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/AbstractDataCache.java
##########
@@ -43,8 +47,13 @@
protected ControlContextProvider _controlContextProvider;
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+
public AbstractDataCache(ControlContextProvider controlContextProvider) {
Review comment:
Thanks for the change. Looks great.
One suggestion is that this change will reduce the genericity of AbstractDataCache. Can we add a middle class called "AbstractParticipantStateCache" as a child class of the AbstractDataCache?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] zhangmeng916 commented on issue #827: Add intermediate
storage for customized state
Posted by GitBox <gi...@apache.org>.
zhangmeng916 commented on issue #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#issuecomment-595375695
This PR is ready to be merged, approved by @jiajunwang
Final commit message:
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.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386744471
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String keyLevel1 = params[1];
+ String keyLevel2 = params[2];
+ String keyLevel3 = params[3];
+ Map<String, Map<String, T>> stateMapLevel1 =
+ allParticipantStateMap.get(keyLevel1);
+ if (stateMapLevel1 == null) {
+ stateMapLevel1 = Maps.newHashMap();
+ allParticipantStateMap.put(keyLevel1, stateMapLevel1);
+ }
+ Map<String, T> stateMapLevel2 = stateMapLevel1.get(keyLevel2);
+ if (stateMapLevel2 == null) {
+ stateMapLevel2 = Maps.newHashMap();
+ stateMapLevel1.put(keyLevel2, stateMapLevel2);
+ }
+ stateMapLevel2.put(keyLevel3, participantState);
+ }
+ }
+
+ for (String instance : allParticipantStateMap.keySet()) {
Review comment:
Since you are creating a new map every time already, is it still necessary to make it unmodifiable?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386745397
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String keyLevel1 = params[1];
+ String keyLevel2 = params[2];
+ String keyLevel3 = params[3];
+ Map<String, Map<String, T>> stateMapLevel1 =
+ allParticipantStateMap.get(keyLevel1);
+ if (stateMapLevel1 == null) {
+ stateMapLevel1 = Maps.newHashMap();
+ allParticipantStateMap.put(keyLevel1, stateMapLevel1);
+ }
+ Map<String, T> stateMapLevel2 = stateMapLevel1.get(keyLevel2);
+ if (stateMapLevel2 == null) {
+ stateMapLevel2 = Maps.newHashMap();
+ stateMapLevel1.put(keyLevel2, stateMapLevel2);
+ }
+ stateMapLevel2.put(keyLevel3, participantState);
+ }
+ }
+
+ for (String instance : allParticipantStateMap.keySet()) {
+ allParticipantStateMap.put(instance, Collections.unmodifiableMap(allParticipantStateMap.get(instance)));
+ }
+ _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, Boolean snapshotEnabled,
+ Set<String> restrictedKeys) {
+
+ long start = System.currentTimeMillis();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
+ for (String instanceName : liveInstanceMap.keySet()) {
+ LiveInstance liveInstance = liveInstanceMap.get(instanceName);
+
+ if (restrictedKeys.isEmpty()) {
Review comment:
Basically you have an abstract method getParticipantStateKeys and called here. The children will implement this method separately with the different logic.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385972007
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,135 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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, resource, partition, instance), returns customized state
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @param instanceName
+ * @return
+ */
+ public String getCustomizedState(String stateType, String resourceName, Partition partition,
+ String instanceName) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map != null) {
+ Map<Partition, Map<String, String>> resourceMap = map.get(resourceName);
+ if (resourceMap != null) {
+ Map<String, String> instanceStateMap = resourceMap.get(partition);
+ if (instanceStateMap != null) {
+ return instanceStateMap.get(instanceName);
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Given stateType, returns resource customized state map (resource -> parition -> instance ->
+ * customizedState)
+ * @param stateType
+ * @return
+ */
+ public Map<String, Map<Partition, Map<String, String>>> getResourceCustomizedStateMap(
+ String stateType) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ return _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>> getCustomizedStateMap(String stateType,
+ String resourceName) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map.containsKey(resourceName)) {
+ return map.get(resourceName);
+ }
+ }
+ return Collections.emptyMap();
+ }
+
+ /**
+ * given (stateType, resource, partition), returns (instance-> customizedState) map
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @return
+ */
+ public Map<String, String> getCustomizedStateMap(String stateType, String resourceName,
Review comment:
This method should call the other getCustomizedStateMap() to avoid duplicate code.
BTW, can we have a better name to differentiate?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385423566
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,135 @@
+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.CustomizedState;
+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;
Review comment:
This map is long and potentially introduces duplicate code since we have a similar code for the current state.
Can we abstract a type for the stateMap that covers (resourceName -> (Partition -> (instanceName -> customizedState))) section? Then this map class can be used in both CustomizedStateOutput and CurrentStateOutput.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386739839
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/CustomizedStateCache.java
##########
@@ -0,0 +1,37 @@
+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 org.apache.helix.common.controllers.ControlContextProvider;
+import org.apache.helix.model.CustomizedState;
+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);
+ }
Review comment:
Do we really need these 2 constructors?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r386745111
##########
File path: helix-core/src/main/java/org/apache/helix/common/caches/ParticipantStateCache.java
##########
@@ -0,0 +1,203 @@
+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.List;
+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;
+
+
+public abstract class ParticipantStateCache<T> extends AbstractDataCache {
+ private static Logger LOG = LoggerFactory.getLogger(ParticipantStateCache.class.getName());
+ protected Map<String, Map<String, Map<String, T>>> _participantStateMap;
+
+ protected Map<PropertyKey, T> _participantStateCache = Maps.newHashMap();
+ public ParticipantStateCache(ControlContextProvider controlContextProvider) {
+ super(controlContextProvider);
+ _participantStateMap = Collections.emptyMap();
+ }
+
+ public ParticipantStateCache(String clusterName) {
+ this(createDefaultControlContextProvider(clusterName));
+ }
+
+
+ /**
+ * 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, Boolean snapshotEnabled, Set<String> restrictedKeys) {
+ long startTime = System.currentTimeMillis();
+
+ refreshParticipantStatesCacheFromZk(accessor, liveInstanceMap, snapshotEnabled, restrictedKeys);
+ Map<String, Map<String, Map<String, T>>> allParticipantStateMap = new HashMap<>();
+ for (PropertyKey key : _participantStateCache.keySet()) {
+ T participantState = _participantStateCache.get(key);
+ String[] params = key.getParams();
+ if (participantState != null && params.length >= 4) {
+ String keyLevel1 = params[1];
+ String keyLevel2 = params[2];
+ String keyLevel3 = params[3];
+ Map<String, Map<String, T>> stateMapLevel1 =
+ allParticipantStateMap.get(keyLevel1);
+ if (stateMapLevel1 == null) {
+ stateMapLevel1 = Maps.newHashMap();
+ allParticipantStateMap.put(keyLevel1, stateMapLevel1);
+ }
+ Map<String, T> stateMapLevel2 = stateMapLevel1.get(keyLevel2);
+ if (stateMapLevel2 == null) {
+ stateMapLevel2 = Maps.newHashMap();
+ stateMapLevel1.put(keyLevel2, stateMapLevel2);
+ }
+ stateMapLevel2.put(keyLevel3, participantState);
+ }
+ }
+
+ for (String instance : allParticipantStateMap.keySet()) {
+ allParticipantStateMap.put(instance, Collections.unmodifiableMap(allParticipantStateMap.get(instance)));
+ }
+ _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, Boolean snapshotEnabled,
+ Set<String> restrictedKeys) {
+
+ long start = System.currentTimeMillis();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
+
+ Set<PropertyKey> participantStateKeys = new HashSet<>();
+ for (String instanceName : liveInstanceMap.keySet()) {
+ LiveInstance liveInstance = liveInstanceMap.get(instanceName);
+
+ if (restrictedKeys.isEmpty()) {
Review comment:
So for the first condition, the liveInstance is kind of the restrictedKeys, right? Can we put this logic to a protected method and leave it to the child class? This will help to cleanup the abstract 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org
[GitHub] [helix] jiajunwang commented on a change in pull request #827: Add
intermediate storage for customized state
Posted by GitBox <gi...@apache.org>.
jiajunwang commented on a change in pull request #827: Add intermediate storage for customized state
URL: https://github.com/apache/helix/pull/827#discussion_r385972092
##########
File path: helix-core/src/main/java/org/apache/helix/controller/stages/CustomizedStateOutput.java
##########
@@ -0,0 +1,135 @@
+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.CustomizedState;
+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;
+
+ private final Map<String, CustomizedState> _customizedStateMetaMap;
+
+ public CustomizedStateOutput() {
+ _customizedStateMap = new HashMap<>();
+ _customizedStateMetaMap = 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, resource, partition, instance), returns customized state
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @param instanceName
+ * @return
+ */
+ public String getCustomizedState(String stateType, String resourceName, Partition partition,
+ String instanceName) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map != null) {
+ Map<Partition, Map<String, String>> resourceMap = map.get(resourceName);
+ if (resourceMap != null) {
+ Map<String, String> instanceStateMap = resourceMap.get(partition);
+ if (instanceStateMap != null) {
+ return instanceStateMap.get(instanceName);
+ }
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Given stateType, returns resource customized state map (resource -> parition -> instance ->
+ * customizedState)
+ * @param stateType
+ * @return
+ */
+ public Map<String, Map<Partition, Map<String, String>>> getResourceCustomizedStateMap(
+ String stateType) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ return _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>> getCustomizedStateMap(String stateType,
+ String resourceName) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map.containsKey(resourceName)) {
+ return map.get(resourceName);
+ }
+ }
+ return Collections.emptyMap();
+ }
+
+ /**
+ * given (stateType, resource, partition), returns (instance-> customizedState) map
+ * @param stateType
+ * @param resourceName
+ * @param partition
+ * @return
+ */
+ public Map<String, String> getCustomizedStateMap(String stateType, String resourceName,
+ Partition partition) {
+ if (_customizedStateMap.containsKey(stateType)) {
+ Map<String, Map<Partition, Map<String, String>>> map = _customizedStateMap.get(stateType);
+ if (map.containsKey(resourceName)) {
+ Map<Partition, Map<String, String>> partitionMap = map.get(resourceName);
+ if (partitionMap != null) {
+ return partitionMap.get(partition);
+ }
+ }
+ }
+ return Collections.emptyMap();
+ }
+
+ public Set<String> getAllStateTypes() {
+ return _customizedStateMap.keySet();
Review comment:
Always copy and then return.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@helix.apache.org
For additional commands, e-mail: reviews-help@helix.apache.org