You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by hustfxj <gi...@git.apache.org> on 2016/02/03 14:32:26 UTC
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
GitHub user hustfxj opened a pull request:
https://github.com/apache/storm/pull/1071
[STORM-1273] port backtype.storm.cluster to java
Sorry. Some tests don’t pass, and I will check and fix it. I now check in code, because I want to get some guidance before the Chinese New Year.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/hustfxj/storm cluster
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/1071.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1071
----
commit 682d31c8a5f53a59f47b4df4bd35ff828e7e0aa5
Author: xiaojian.fxj <xi...@alibaba-inc.com>
Date: 2016-02-01T07:17:49Z
port org.apache.storm.cluster.cluster.clj
commit 9a79fb7de0e824e73c294738521e892f1d81fbb0
Author: xiaojian.fxj <xi...@alibaba-inc.com>
Date: 2016-02-03T12:28:05Z
delete zookeeper.clj zookeeper_state_factory.clj cluster.clj, but some tests still can't pass
commit 5916b0b8089f9dd184fdd1ab2f18eb5e5deabc65
Author: xiaojian.fxj <xi...@alibaba-inc.com>
Date: 2016-02-03T13:15:41Z
callback maybe null
commit 2a17c45591692b61df0be922fdec9d10a108de27
Author: xiaojian.fxj <xi...@alibaba-inc.com>
Date: 2016-02-03T13:18:13Z
Merge branch 'master' into cluster
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1071#issuecomment-179294019
@hustfxj there are some major changes that need to be made to the way you setup your class hierarchy and the naming of those classes. I stopped looking at the details of the rest of the patch until those changes can be addressed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51728386
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/Cluster.java ---
@@ -0,0 +1,236 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.ExecutorStats;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Cluster {
+
+ public static final String ZK_SEPERATOR = "/";
+
+ public static final String ASSIGNMENTS_ROOT = "assignments";
+ public static final String CODE_ROOT = "code";
+ public static final String STORMS_ROOT = "storms";
+ public static final String SUPERVISORS_ROOT = "supervisors";
+ public static final String WORKERBEATS_ROOT = "workerbeats";
+ public static final String BACKPRESSURE_ROOT = "backpressure";
+ public static final String ERRORS_ROOT = "errors";
+ public static final String BLOBSTORE_ROOT = "blobstore";
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT = "blobstoremaxkeysequencenumber";
+ public static final String NIMBUSES_ROOT = "nimbuses";
+ public static final String CREDENTIALS_ROOT = "credentials";
+ public static final String LOGCONFIG_ROOT = "logconfigs";
+ public static final String PROFILERCONFIG_ROOT = "profilerconfigs";
+
+ public static final String ASSIGNMENTS_SUBTREE;
+ public static final String STORMS_SUBTREE;
+ public static final String SUPERVISORS_SUBTREE;
+ public static final String WORKERBEATS_SUBTREE;
+ public static final String BACKPRESSURE_SUBTREE;
+ public static final String ERRORS_SUBTREE;
+ public static final String BLOBSTORE_SUBTREE;
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE;
+ public static final String NIMBUSES_SUBTREE;
+ public static final String CREDENTIALS_SUBTREE;
+ public static final String LOGCONFIG_SUBTREE;
+ public static final String PROFILERCONFIG_SUBTREE;
+
+ static {
+ ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
+ STORMS_SUBTREE = ZK_SEPERATOR + STORMS_ROOT;
+ SUPERVISORS_SUBTREE = ZK_SEPERATOR + SUPERVISORS_ROOT;
+ WORKERBEATS_SUBTREE = ZK_SEPERATOR + WORKERBEATS_ROOT;
+ BACKPRESSURE_SUBTREE = ZK_SEPERATOR + BACKPRESSURE_ROOT;
+ ERRORS_SUBTREE = ZK_SEPERATOR + ERRORS_ROOT;
+ BLOBSTORE_SUBTREE = ZK_SEPERATOR + BLOBSTORE_ROOT;
+ BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE = ZK_SEPERATOR + BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT;
+ NIMBUSES_SUBTREE = ZK_SEPERATOR + NIMBUSES_ROOT;
+ CREDENTIALS_SUBTREE = ZK_SEPERATOR + CREDENTIALS_ROOT;
+ LOGCONFIG_SUBTREE = ZK_SEPERATOR + LOGCONFIG_ROOT;
+ PROFILERCONFIG_SUBTREE = ZK_SEPERATOR + PROFILERCONFIG_ROOT;
+ }
+
+ // A singleton instance allows us to mock delegated static methods in our
+ // tests by subclassing.
+ private static final Cluster INSTANCE = new Cluster();
+ private static Cluster _instance = INSTANCE;
+
+ /**
+ * Provide an instance of this class for delegates to use. To mock out
+ * delegated methods, provide an instance of a subclass that overrides the
+ * implementation of the delegated method.
+ *
+ * @param u a Zookeeper instance
+ */
+ public static void setInstance(Cluster u) {
+ _instance = u;
+ }
+
+ /**
+ * Resets the singleton instance to the default. This is helpful to reset
+ * the class to its original functionality when mocking is no longer
+ * desired.
+ */
+ public static void resetInstance() {
+ _instance = INSTANCE;
+ }
+
+ public static List<ACL> mkTopoOnlyAcls(Map topoConf) throws NoSuchAlgorithmException {
+ List<ACL> aclList = null;
+ String payload = (String)topoConf.get(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD);
+ if (Utils.isZkAuthenticationConfiguredStormServer(topoConf)){
+ aclList = new ArrayList<>();
+ ACL acl1 = ZooDefs.Ids.CREATOR_ALL_ACL.get(0);
+ aclList.add(acl1);
+ ACL acl2 = new ACL(ZooDefs.Perms.READ, new Id("digest", DigestAuthenticationProvider.generateDigest(payload)));
+ aclList.add(acl2);
+ }
+ return aclList;
+ }
+
+ public static String supervisorPath(String id) {
+ return SUPERVISORS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String assignmentPath(String id) {
+ return ASSIGNMENTS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String blobstorePath(String key) {
+ return BLOBSTORE_SUBTREE + ZK_SEPERATOR + key;
+ }
+
+ public static String blobstoreMaxKeySequenceNumberPath(String key) {
+ return BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE + ZK_SEPERATOR + key;
+ }
+
+ public static String nimbusPath(String id) {
+ return NIMBUSES_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String stormPath(String id) {
+ return STORMS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String workerbeatStormRoot(String stormId) {
+ return WORKERBEATS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String workerbeatPath(String stormId, String node, Long port) {
+ return workerbeatStormRoot(stormId) + ZK_SEPERATOR + node + "-" + port;
+ }
+
+ public static String backpressureStormRoot(String stormId) {
+ return BACKPRESSURE_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String backpressurePath(String stormId, String node, Long port) {
+ return backpressureStormRoot(stormId) + ZK_SEPERATOR + node + "-" + port;
+ }
+
+ public static String errorStormRoot(String stormId) {
+ return ERRORS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String errorPath(String stormId, String componentId) throws UnsupportedEncodingException {
+ return errorStormRoot(stormId) + ZK_SEPERATOR + URLEncoder.encode(componentId, "UTF-8");
+ }
+
+ public static String lastErrorPath(String stormId, String componentId) throws UnsupportedEncodingException {
+ return errorPath(stormId, componentId) + "-last-error";
+ }
+
+ public static String credentialsPath(String stormId) {
+ return CREDENTIALS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String logConfigPath(String stormId) {
+ return LOGCONFIG_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String profilerConfigPath(String stormId) {
+ return PROFILERCONFIG_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String profilerConfigPath(String stormId, String host, Long port, ProfileAction requestType) {
+ return profilerConfigPath(stormId) + ZK_SEPERATOR + host + "_" + port + "_" + requestType;
+ }
+
+ public static <T> T maybeDeserialize(byte[] serialized, Class<T> clazz){
+ if (serialized != null){
+ return Utils.deserialize(serialized, clazz);
+ }
+ return null;
+ }
+
+ //Ensures that we only return heartbeats for executors assigned to this worker
--- End diff --
Could you make this a javadoc? That is what the doc string represents in clojure.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51728719
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/Cluster.java ---
@@ -0,0 +1,236 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.ExecutorStats;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Cluster {
+
+ public static final String ZK_SEPERATOR = "/";
+
+ public static final String ASSIGNMENTS_ROOT = "assignments";
+ public static final String CODE_ROOT = "code";
+ public static final String STORMS_ROOT = "storms";
+ public static final String SUPERVISORS_ROOT = "supervisors";
+ public static final String WORKERBEATS_ROOT = "workerbeats";
+ public static final String BACKPRESSURE_ROOT = "backpressure";
+ public static final String ERRORS_ROOT = "errors";
+ public static final String BLOBSTORE_ROOT = "blobstore";
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT = "blobstoremaxkeysequencenumber";
+ public static final String NIMBUSES_ROOT = "nimbuses";
+ public static final String CREDENTIALS_ROOT = "credentials";
+ public static final String LOGCONFIG_ROOT = "logconfigs";
+ public static final String PROFILERCONFIG_ROOT = "profilerconfigs";
+
+ public static final String ASSIGNMENTS_SUBTREE;
+ public static final String STORMS_SUBTREE;
+ public static final String SUPERVISORS_SUBTREE;
+ public static final String WORKERBEATS_SUBTREE;
+ public static final String BACKPRESSURE_SUBTREE;
+ public static final String ERRORS_SUBTREE;
+ public static final String BLOBSTORE_SUBTREE;
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE;
+ public static final String NIMBUSES_SUBTREE;
+ public static final String CREDENTIALS_SUBTREE;
+ public static final String LOGCONFIG_SUBTREE;
+ public static final String PROFILERCONFIG_SUBTREE;
+
+ static {
+ ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
+ STORMS_SUBTREE = ZK_SEPERATOR + STORMS_ROOT;
+ SUPERVISORS_SUBTREE = ZK_SEPERATOR + SUPERVISORS_ROOT;
+ WORKERBEATS_SUBTREE = ZK_SEPERATOR + WORKERBEATS_ROOT;
+ BACKPRESSURE_SUBTREE = ZK_SEPERATOR + BACKPRESSURE_ROOT;
+ ERRORS_SUBTREE = ZK_SEPERATOR + ERRORS_ROOT;
+ BLOBSTORE_SUBTREE = ZK_SEPERATOR + BLOBSTORE_ROOT;
+ BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE = ZK_SEPERATOR + BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT;
+ NIMBUSES_SUBTREE = ZK_SEPERATOR + NIMBUSES_ROOT;
+ CREDENTIALS_SUBTREE = ZK_SEPERATOR + CREDENTIALS_ROOT;
+ LOGCONFIG_SUBTREE = ZK_SEPERATOR + LOGCONFIG_ROOT;
+ PROFILERCONFIG_SUBTREE = ZK_SEPERATOR + PROFILERCONFIG_ROOT;
+ }
+
+ // A singleton instance allows us to mock delegated static methods in our
+ // tests by subclassing.
+ private static final Cluster INSTANCE = new Cluster();
+ private static Cluster _instance = INSTANCE;
+
+ /**
+ * Provide an instance of this class for delegates to use. To mock out
+ * delegated methods, provide an instance of a subclass that overrides the
+ * implementation of the delegated method.
+ *
+ * @param u a Zookeeper instance
+ */
+ public static void setInstance(Cluster u) {
+ _instance = u;
+ }
+
+ /**
+ * Resets the singleton instance to the default. This is helpful to reset
+ * the class to its original functionality when mocking is no longer
+ * desired.
+ */
+ public static void resetInstance() {
+ _instance = INSTANCE;
+ }
+
+ public static List<ACL> mkTopoOnlyAcls(Map topoConf) throws NoSuchAlgorithmException {
+ List<ACL> aclList = null;
+ String payload = (String)topoConf.get(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD);
+ if (Utils.isZkAuthenticationConfiguredStormServer(topoConf)){
+ aclList = new ArrayList<>();
+ ACL acl1 = ZooDefs.Ids.CREATOR_ALL_ACL.get(0);
+ aclList.add(acl1);
+ ACL acl2 = new ACL(ZooDefs.Perms.READ, new Id("digest", DigestAuthenticationProvider.generateDigest(payload)));
+ aclList.add(acl2);
+ }
+ return aclList;
+ }
+
+ public static String supervisorPath(String id) {
+ return SUPERVISORS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String assignmentPath(String id) {
+ return ASSIGNMENTS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String blobstorePath(String key) {
+ return BLOBSTORE_SUBTREE + ZK_SEPERATOR + key;
+ }
+
+ public static String blobstoreMaxKeySequenceNumberPath(String key) {
+ return BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE + ZK_SEPERATOR + key;
+ }
+
+ public static String nimbusPath(String id) {
+ return NIMBUSES_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String stormPath(String id) {
+ return STORMS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String workerbeatStormRoot(String stormId) {
+ return WORKERBEATS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String workerbeatPath(String stormId, String node, Long port) {
+ return workerbeatStormRoot(stormId) + ZK_SEPERATOR + node + "-" + port;
+ }
+
+ public static String backpressureStormRoot(String stormId) {
+ return BACKPRESSURE_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String backpressurePath(String stormId, String node, Long port) {
+ return backpressureStormRoot(stormId) + ZK_SEPERATOR + node + "-" + port;
+ }
+
+ public static String errorStormRoot(String stormId) {
+ return ERRORS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String errorPath(String stormId, String componentId) throws UnsupportedEncodingException {
+ return errorStormRoot(stormId) + ZK_SEPERATOR + URLEncoder.encode(componentId, "UTF-8");
+ }
+
+ public static String lastErrorPath(String stormId, String componentId) throws UnsupportedEncodingException {
+ return errorPath(stormId, componentId) + "-last-error";
+ }
+
+ public static String credentialsPath(String stormId) {
+ return CREDENTIALS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String logConfigPath(String stormId) {
+ return LOGCONFIG_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String profilerConfigPath(String stormId) {
+ return PROFILERCONFIG_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String profilerConfigPath(String stormId, String host, Long port, ProfileAction requestType) {
+ return profilerConfigPath(stormId) + ZK_SEPERATOR + host + "_" + port + "_" + requestType;
+ }
+
+ public static <T> T maybeDeserialize(byte[] serialized, Class<T> clazz){
+ if (serialized != null){
+ return Utils.deserialize(serialized, clazz);
+ }
+ return null;
+ }
+
+ //Ensures that we only return heartbeats for executors assigned to this worker
+ public static Map<ExecutorInfo, ClusterWorkerHeartbeat> convertExecutorBeats(List<ExecutorInfo> executors, ClusterWorkerHeartbeat workerHeartbeat){
--- End diff --
Lets make this private. I know it want before but it is only used in this class so it does not need to be public.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51727452
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/Cluster.java ---
@@ -0,0 +1,236 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.ExecutorStats;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Cluster {
+
+ public static final String ZK_SEPERATOR = "/";
+
+ public static final String ASSIGNMENTS_ROOT = "assignments";
+ public static final String CODE_ROOT = "code";
+ public static final String STORMS_ROOT = "storms";
+ public static final String SUPERVISORS_ROOT = "supervisors";
+ public static final String WORKERBEATS_ROOT = "workerbeats";
+ public static final String BACKPRESSURE_ROOT = "backpressure";
+ public static final String ERRORS_ROOT = "errors";
+ public static final String BLOBSTORE_ROOT = "blobstore";
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT = "blobstoremaxkeysequencenumber";
+ public static final String NIMBUSES_ROOT = "nimbuses";
+ public static final String CREDENTIALS_ROOT = "credentials";
+ public static final String LOGCONFIG_ROOT = "logconfigs";
+ public static final String PROFILERCONFIG_ROOT = "profilerconfigs";
+
+ public static final String ASSIGNMENTS_SUBTREE;
+ public static final String STORMS_SUBTREE;
+ public static final String SUPERVISORS_SUBTREE;
+ public static final String WORKERBEATS_SUBTREE;
+ public static final String BACKPRESSURE_SUBTREE;
+ public static final String ERRORS_SUBTREE;
+ public static final String BLOBSTORE_SUBTREE;
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE;
+ public static final String NIMBUSES_SUBTREE;
+ public static final String CREDENTIALS_SUBTREE;
+ public static final String LOGCONFIG_SUBTREE;
+ public static final String PROFILERCONFIG_SUBTREE;
+
+ static {
+ ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
+ STORMS_SUBTREE = ZK_SEPERATOR + STORMS_ROOT;
+ SUPERVISORS_SUBTREE = ZK_SEPERATOR + SUPERVISORS_ROOT;
+ WORKERBEATS_SUBTREE = ZK_SEPERATOR + WORKERBEATS_ROOT;
+ BACKPRESSURE_SUBTREE = ZK_SEPERATOR + BACKPRESSURE_ROOT;
+ ERRORS_SUBTREE = ZK_SEPERATOR + ERRORS_ROOT;
+ BLOBSTORE_SUBTREE = ZK_SEPERATOR + BLOBSTORE_ROOT;
+ BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE = ZK_SEPERATOR + BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT;
+ NIMBUSES_SUBTREE = ZK_SEPERATOR + NIMBUSES_ROOT;
+ CREDENTIALS_SUBTREE = ZK_SEPERATOR + CREDENTIALS_ROOT;
+ LOGCONFIG_SUBTREE = ZK_SEPERATOR + LOGCONFIG_ROOT;
+ PROFILERCONFIG_SUBTREE = ZK_SEPERATOR + PROFILERCONFIG_ROOT;
+ }
+
+ // A singleton instance allows us to mock delegated static methods in our
+ // tests by subclassing.
+ private static final Cluster INSTANCE = new Cluster();
+ private static Cluster _instance = INSTANCE;
+
+ /**
+ * Provide an instance of this class for delegates to use. To mock out
+ * delegated methods, provide an instance of a subclass that overrides the
+ * implementation of the delegated method.
+ *
+ * @param u a Zookeeper instance
+ */
+ public static void setInstance(Cluster u) {
+ _instance = u;
+ }
+
+ /**
+ * Resets the singleton instance to the default. This is helpful to reset
+ * the class to its original functionality when mocking is no longer
+ * desired.
+ */
+ public static void resetInstance() {
+ _instance = INSTANCE;
+ }
+
+ public static List<ACL> mkTopoOnlyAcls(Map topoConf) throws NoSuchAlgorithmException {
+ List<ACL> aclList = null;
+ String payload = (String)topoConf.get(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD);
--- End diff --
This should be STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51744446
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/StormZkClusterState.java ---
@@ -0,0 +1,683 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import clojure.lang.APersistentMap;
+import clojure.lang.IFn;
+import clojure.lang.PersistentArrayMap;
+import clojure.lang.RT;
+import org.apache.commons.lang.StringUtils;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.*;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.storm.callback.Callback;
+import org.apache.storm.generated.*;
+import org.apache.storm.nimbus.NimbusInfo;
+import org.apache.storm.utils.Time;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.zookeeper.Zookeeper;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.UnsupportedEncodingException;
+import java.security.NoSuchAlgorithmException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class StormZkClusterState implements StormClusterState {
+
+ private static Logger LOG = LoggerFactory.getLogger(StormZkClusterState.class);
+
+ private ClusterState clusterState;
+
+ private ConcurrentHashMap<String, IFn> assignmentInfoCallback;
+ private ConcurrentHashMap<String, IFn> assignmentInfoWithVersionCallback;
+ private ConcurrentHashMap<String, IFn> assignmentVersionCallback;
+ private AtomicReference<IFn> supervisorsCallback;
+ // we want to reigister a topo directory getChildren callback for all workers of this dir
+ private ConcurrentHashMap<String, IFn> backPressureCallback;
+ private AtomicReference<IFn> assignmentsCallback;
+ private ConcurrentHashMap<String, IFn> stormBaseCallback;
+ private AtomicReference<IFn> blobstoreCallback;
+ private ConcurrentHashMap<String, IFn> credentialsCallback;
+ private ConcurrentHashMap<String, IFn> logConfigCallback;
+
+ private List<ACL> acls;
+ private String stateId;
+ private boolean solo;
+
+ public StormZkClusterState(Object clusterState, List<ACL> acls, ClusterStateContext context) throws Exception {
+
+ if (clusterState instanceof ClusterState) {
+ solo = false;
+ this.clusterState = (ClusterState) clusterState;
+ } else {
+
+ solo = true;
+ this.clusterState = new DistributedClusterState((Map) clusterState, (Map) clusterState, acls, context);
+ }
+
+ assignmentInfoCallback = new ConcurrentHashMap<>();
+ assignmentInfoWithVersionCallback = new ConcurrentHashMap<>();
+ assignmentVersionCallback = new ConcurrentHashMap<>();
+ supervisorsCallback = new AtomicReference<>();
+ backPressureCallback = new ConcurrentHashMap<>();
+ assignmentsCallback = new AtomicReference<>();
+ stormBaseCallback = new ConcurrentHashMap<>();
+ credentialsCallback = new ConcurrentHashMap<>();
+ logConfigCallback = new ConcurrentHashMap<>();
+ blobstoreCallback = new AtomicReference<>();
+
+ stateId = this.clusterState.register(new Callback() {
+
+ public <T> Object execute(T... args) {
+ if (args == null) {
+ LOG.warn("Input args is null");
+ return null;
+ } else if (args.length < 2) {
+ LOG.warn("Input args is invalid, args length:" + args.length);
+ return null;
+ }
+ String path = (String) args[1];
+
+ List<String> toks = Zookeeper.tokenizePath(path);
+ int size = toks.size();
+ if (size >= 1) {
+ String params = null;
+ String root = toks.get(0);
+ IFn fn = null;
+ if (root.equals(Cluster.ASSIGNMENTS_ROOT)) {
+ if (size == 1) {
+ // set null and get the old value
+ issueCallback(assignmentsCallback);
+ } else {
+ issueMapCallback(assignmentInfoCallback, toks.get(1));
+ issueMapCallback(assignmentVersionCallback, toks.get(1));
+ issueMapCallback(assignmentInfoWithVersionCallback, toks.get(1));
+ }
+
+ } else if (root.equals(Cluster.SUPERVISORS_ROOT)) {
+ issueCallback(supervisorsCallback);
+ } else if (root.equals(Cluster.BLOBSTORE_ROOT)) {
+ issueCallback(blobstoreCallback);
+ } else if (root.equals(Cluster.STORMS_ROOT) && size > 1) {
+ issueMapCallback(stormBaseCallback, toks.get(1));
+ } else if (root.equals(Cluster.CREDENTIALS_ROOT) && size > 1) {
+ issueMapCallback(credentialsCallback, toks.get(1));
+ } else if (root.equals(Cluster.LOGCONFIG_ROOT) && size > 1) {
+ issueMapCallback(logConfigCallback, toks.get(1));
+ } else if (root.equals(Cluster.BACKPRESSURE_ROOT) && size > 1) {
+ issueMapCallback(logConfigCallback, toks.get(1));
+ } else {
+ LOG.error("{} Unknown callback for subtree {}", new RuntimeException("Unknown callback for this path"), path);
+ Runtime.getRuntime().exit(30);
+ }
+
+ }
+
+ return null;
+ }
+
+ });
+
+ String[] pathlist = { Cluster.ASSIGNMENTS_SUBTREE, Cluster.STORMS_SUBTREE, Cluster.SUPERVISORS_SUBTREE, Cluster.WORKERBEATS_SUBTREE,
+ Cluster.ERRORS_SUBTREE, Cluster.BLOBSTORE_SUBTREE, Cluster.NIMBUSES_SUBTREE, Cluster.LOGCONFIG_SUBTREE };
+ for (String path : pathlist) {
+ this.clusterState.mkdirs(path, acls);
+ }
+
+ }
+
+ protected void issueCallback(AtomicReference<IFn> cb) {
+ IFn callback = cb.getAndSet(null);
+ if (callback != null)
+ callback.invoke();
+ }
+
+ protected void issueMapCallback(ConcurrentHashMap<String, IFn> callbackConcurrentHashMap, String key) {
+ IFn callback = callbackConcurrentHashMap.remove(key);
+ if (callback != null)
+ callback.invoke();
+ }
+
+ @Override
+ public List<String> assignments(IFn callback) {
+ if (callback != null) {
+ assignmentsCallback.set(callback);
+ }
+ return clusterState.get_children(Cluster.ASSIGNMENTS_SUBTREE, callback != null);
+ }
+
+ @Override
+ public Assignment assignmentInfo(String stormId, IFn callback) {
+ if (callback != null) {
+ assignmentInfoCallback.put(stormId, callback);
+ }
+ byte[] serialized = clusterState.get_data(Cluster.assignmentPath(stormId), callback != null);
+ return Cluster.maybeDeserialize(serialized, Assignment.class);
+ }
+
+ @Override
+ public APersistentMap assignmentInfoWithVersion(String stormId, IFn callback) {
+ if (callback != null) {
+ assignmentInfoWithVersionCallback.put(stormId, callback);
+ }
+ APersistentMap aPersistentMap = clusterState.get_data_with_version(Cluster.assignmentPath(stormId), callback != null);
+ Assignment assignment = Cluster.maybeDeserialize((byte[]) aPersistentMap.get("data"), Assignment.class);
+ Integer version = (Integer) aPersistentMap.get("version");
+ APersistentMap map = new PersistentArrayMap(new Object[] { RT.keyword(null, "data"), assignment, RT.keyword(null, "version"), version });
+ return map;
+ }
+
+ @Override
+ public Integer assignmentVersion(String stormId, IFn callback) throws Exception {
+ if (callback != null) {
+ assignmentVersionCallback.put(stormId, callback);
+ }
+ return clusterState.get_version(Cluster.assignmentPath(stormId), callback != null);
+ }
+
+ // blobstore state
+ @Override
+ public List<String> blobstoreInfo(String blobKey) {
+ String path = Cluster.blobstorePath(blobKey);
+ clusterState.sync_path(path);
+ return clusterState.get_children(path, false);
+ }
+
+ @Override
+ public List nimbuses() {
+ List<NimbusSummary> nimbusSummaries = new ArrayList<>();
+ List<String> nimbusIds = clusterState.get_children(Cluster.NIMBUSES_SUBTREE, false);
+ for (String nimbusId : nimbusIds) {
+ byte[] serialized = clusterState.get_data(Cluster.nimbusPath(nimbusId), false);
+ NimbusSummary nimbusSummary = Cluster.maybeDeserialize(serialized, NimbusSummary.class);
+ nimbusSummaries.add(nimbusSummary);
+ }
+ return nimbusSummaries;
+ }
+
+ @Override
+ public void addNimbusHost(final String nimbusId, final NimbusSummary nimbusSummary) {
+ // explicit delete for ephmeral node to ensure this session creates the entry.
+ clusterState.delete_node(Cluster.nimbusPath(nimbusId));
+ clusterState.add_listener(new ConnectionStateListener() {
+ @Override
+ public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
+ LOG.info("Connection state listener invoked, zookeeper connection state has changed to {}", connectionState);
+ if (connectionState.equals(ConnectionState.RECONNECTED)) {
+ LOG.info("Connection state has changed to reconnected so setting nimbuses entry one more time");
+ clusterState.set_ephemeral_node(Cluster.nimbusPath(nimbusId), Utils.serialize(nimbusSummary), acls);
+ }
+
+ }
+ });
+
+ clusterState.set_ephemeral_node(Cluster.nimbusPath(nimbusId), Utils.serialize(nimbusSummary), acls);
+ }
+
+ @Override
+ public List<String> activeStorms() {
+ return clusterState.get_children(Cluster.STORMS_SUBTREE, false);
+ }
+
+ @Override
+ public StormBase stormBase(String stormId, IFn callback) {
+ if (callback != null) {
+ stormBaseCallback.put(stormId, callback);
+ }
+ return Cluster.maybeDeserialize(clusterState.get_data(Cluster.stormPath(stormId), callback != null), StormBase.class);
+ }
+
+ @Override
+ public ClusterWorkerHeartbeat getWorkerHeartbeat(String stormId, String node, Long port) {
+ byte[] bytes = clusterState.get_worker_hb(Cluster.workerbeatPath(stormId, node, port), false);
+ if (bytes != null) {
--- End diff --
this if block is not needed
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on the pull request:
https://github.com/apache/storm/pull/1071#issuecomment-179558002
@revans2 @abhishekagarwal87 Thank you for some advice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51726152
--- Diff: storm-core/src/clj/org/apache/storm/pacemaker/pacemaker_state_factory.clj ---
@@ -47,7 +43,7 @@
(reify
ClusterState
;; Let these pass through to the zk-state. We only want to handle heartbeats.
- (register [this callback] (.register zk-state callback))
+ (register [this callback] (.register zk-state callback)) ; need update callback, have questions?? callback is IFn here
--- End diff --
If the callback takes a defined set of parameters that are used consistently lets either create or reuse an interface and then wrap the callback function in a reify call to create an instance of the interface for the callback to talk to.
If it does not take any parameters, it just is a function that is triggered than make it a Runnable. But I think in this case it is the former, not the latter.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51741366
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/DistributedClusterState.java ---
@@ -0,0 +1,268 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import clojure.lang.APersistentMap;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.*;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.storm.Config;
+import org.apache.storm.callback.Callback;
+import org.apache.storm.callback.WatcherCallBack;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.zookeeper.Zookeeper;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class DistributedClusterState implements ClusterState {
+
+ private static Logger LOG = LoggerFactory.getLogger(DistributedClusterState.class);
+
+ private ConcurrentHashMap<String, Callback> callbacks = new ConcurrentHashMap<String, Callback>();
+ private CuratorFramework zkWriter;
+ private CuratorFramework zkReader;
+ private AtomicBoolean active;
+
+ private boolean isNimbus;
+ private Map authConf;
+ private Map<Object, Object> conf;
+
+ public DistributedClusterState(Map<Object, Object> conf, Map authConf, List<ACL> acls, ClusterStateContext context) throws Exception {
+ this.conf = conf;
+ this.authConf = authConf;
+ if (context.getDaemonType().equals(DaemonType.NIMBUS)) this.isNimbus = true;
+
+ // just mkdir STORM_ZOOKEEPER_ROOT dir
+ CuratorFramework zkTemp = mkZk();
+ String rootPath = String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT));
+ Zookeeper.mkdirs(zkTemp, rootPath, acls);
+ zkTemp.close();
+
+ active = new AtomicBoolean(true);
+ zkWriter = mkZk(new WatcherCallBack() {
+ @Override
+ public void execute(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String path) {
+ if (active.get()) {
+ if (!(state.equals(Watcher.Event.KeeperState.SyncConnected))) {
+ LOG.warn("Received event {} : {}: {} with disconnected Zookeeper.", state, type, path);
+ } else {
+ LOG.info("Received event {} : {} : {}", state, type, path);
+ }
+
+ if (!type.equals(Watcher.Event.EventType.None)) {
+ for (Map.Entry<String, Callback> e : callbacks.entrySet()) {
+ Callback fn = e.getValue();
+ fn.execute(type, path);
+ }
+ }
+ }
+ }
+ });
+ if (isNimbus) {
+ zkReader = mkZk(new WatcherCallBack() {
+ @Override
+ public void execute(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String path) {
+ if (active.get()) {
+ if (!(state.equals(Watcher.Event.KeeperState.SyncConnected))) {
+ LOG.warn("Received event {} : {}: {} with disconnected Zookeeper.", state, type, path);
+ } else {
+ LOG.info("Received event {} : {} : {}", state, type, path);
+ }
+
+ if (!type.equals(Watcher.Event.EventType.None)) {
+ for (Map.Entry<String, Callback> e : callbacks.entrySet()) {
+ Callback fn = e.getValue();
+ fn.execute(type, path);
+ }
+ }
+ }
+ }
+ });
+ } else {
+ zkReader = zkWriter;
+ }
+
+ }
+
+ @SuppressWarnings("unchecked")
+ private CuratorFramework mkZk() throws IOException {
+ return Zookeeper.mkClient(conf, (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS), conf.get(Config.STORM_ZOOKEEPER_PORT), "", authConf);
+ }
+
+ @SuppressWarnings("unchecked")
+ private CuratorFramework mkZk(WatcherCallBack watcher) throws NumberFormatException, IOException {
+ return Zookeeper.mkClient(conf, (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS), conf.get(Config.STORM_ZOOKEEPER_PORT),
+ String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT)), watcher, authConf);
+ }
+
+ @Override
+ public void delete_node_blobstore(String path, String nimbusHostPortInfo) {
+
+ }
+
+ @Override
+ public String register( Callback callback) {
+ String id = UUID.randomUUID().toString();
+ this.callbacks.put(id,callback);
+ return id;
+ }
+
+ @Override
+ public void unregister(String id) {
+ this.callbacks.remove(id);
+ }
+
+ @Override
+ public String create_sequential(String path, byte[] data, List<ACL> acls) {
+ return Zookeeper.createNode(zkWriter, path, data, CreateMode.EPHEMERAL_SEQUENTIAL, acls);
+ }
+
+ @Override
+ public void mkdirs(String path, List<ACL> acls) {
+ Zookeeper.mkdirs(zkWriter, path, acls);
+ }
+
+ @Override
+ public void delete_node(String path) {
--- End diff --
should the function names be camelCase instead?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51741222
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/DistributedClusterState.java ---
@@ -0,0 +1,268 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import clojure.lang.APersistentMap;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.*;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.storm.Config;
+import org.apache.storm.callback.Callback;
+import org.apache.storm.callback.WatcherCallBack;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.zookeeper.Zookeeper;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class DistributedClusterState implements ClusterState {
+
+ private static Logger LOG = LoggerFactory.getLogger(DistributedClusterState.class);
+
+ private ConcurrentHashMap<String, Callback> callbacks = new ConcurrentHashMap<String, Callback>();
+ private CuratorFramework zkWriter;
+ private CuratorFramework zkReader;
+ private AtomicBoolean active;
+
+ private boolean isNimbus;
+ private Map authConf;
+ private Map<Object, Object> conf;
+
+ public DistributedClusterState(Map<Object, Object> conf, Map authConf, List<ACL> acls, ClusterStateContext context) throws Exception {
+ this.conf = conf;
+ this.authConf = authConf;
+ if (context.getDaemonType().equals(DaemonType.NIMBUS)) this.isNimbus = true;
+
+ // just mkdir STORM_ZOOKEEPER_ROOT dir
+ CuratorFramework zkTemp = mkZk();
+ String rootPath = String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT));
+ Zookeeper.mkdirs(zkTemp, rootPath, acls);
+ zkTemp.close();
+
+ active = new AtomicBoolean(true);
+ zkWriter = mkZk(new WatcherCallBack() {
+ @Override
+ public void execute(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String path) {
+ if (active.get()) {
+ if (!(state.equals(Watcher.Event.KeeperState.SyncConnected))) {
+ LOG.warn("Received event {} : {}: {} with disconnected Zookeeper.", state, type, path);
+ } else {
+ LOG.info("Received event {} : {} : {}", state, type, path);
+ }
+
+ if (!type.equals(Watcher.Event.EventType.None)) {
+ for (Map.Entry<String, Callback> e : callbacks.entrySet()) {
+ Callback fn = e.getValue();
+ fn.execute(type, path);
+ }
+ }
+ }
+ }
+ });
+ if (isNimbus) {
+ zkReader = mkZk(new WatcherCallBack() {
+ @Override
+ public void execute(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String path) {
+ if (active.get()) {
+ if (!(state.equals(Watcher.Event.KeeperState.SyncConnected))) {
+ LOG.warn("Received event {} : {}: {} with disconnected Zookeeper.", state, type, path);
+ } else {
+ LOG.info("Received event {} : {} : {}", state, type, path);
+ }
+
+ if (!type.equals(Watcher.Event.EventType.None)) {
+ for (Map.Entry<String, Callback> e : callbacks.entrySet()) {
+ Callback fn = e.getValue();
+ fn.execute(type, path);
+ }
+ }
+ }
+ }
+ });
+ } else {
+ zkReader = zkWriter;
+ }
+
+ }
+
+ @SuppressWarnings("unchecked")
+ private CuratorFramework mkZk() throws IOException {
+ return Zookeeper.mkClient(conf, (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS), conf.get(Config.STORM_ZOOKEEPER_PORT), "", authConf);
+ }
+
+ @SuppressWarnings("unchecked")
+ private CuratorFramework mkZk(WatcherCallBack watcher) throws NumberFormatException, IOException {
+ return Zookeeper.mkClient(conf, (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS), conf.get(Config.STORM_ZOOKEEPER_PORT),
+ String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT)), watcher, authConf);
+ }
+
+ @Override
+ public void delete_node_blobstore(String path, String nimbusHostPortInfo) {
+
+ }
+
+ @Override
+ public String register( Callback callback) {
+ String id = UUID.randomUUID().toString();
+ this.callbacks.put(id,callback);
+ return id;
+ }
+
+ @Override
+ public void unregister(String id) {
+ this.callbacks.remove(id);
+ }
+
+ @Override
+ public String create_sequential(String path, byte[] data, List<ACL> acls) {
+ return Zookeeper.createNode(zkWriter, path, data, CreateMode.EPHEMERAL_SEQUENTIAL, acls);
+ }
+
+ @Override
+ public void mkdirs(String path, List<ACL> acls) {
+ Zookeeper.mkdirs(zkWriter, path, acls);
+ }
+
+ @Override
+ public void delete_node(String path) {
+ Zookeeper.deleteNode(zkWriter, path);
+ }
+
+ @Override
+ public void set_ephemeral_node(String path, byte[] data, List<ACL> acls) {
+ Zookeeper.mkdirs(zkWriter, parentPath(path), acls);
+ if (Zookeeper.exists(zkWriter, path, false)) {
+ try {
+ Zookeeper.setData(zkWriter, path, data);
+ } catch (RuntimeException e) {
+ if (Utils.exceptionCauseIsInstanceOf(KeeperException.NodeExistsException.class, e)) {
+ Zookeeper.createNode(zkWriter, path, data, CreateMode.EPHEMERAL, acls);
--- End diff --
you want to recreate if it already exists?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51741578
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/DistributedClusterState.java ---
@@ -0,0 +1,268 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import clojure.lang.APersistentMap;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.*;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.storm.Config;
+import org.apache.storm.callback.Callback;
+import org.apache.storm.callback.WatcherCallBack;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.zookeeper.Zookeeper;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class DistributedClusterState implements ClusterState {
+
+ private static Logger LOG = LoggerFactory.getLogger(DistributedClusterState.class);
+
+ private ConcurrentHashMap<String, Callback> callbacks = new ConcurrentHashMap<String, Callback>();
+ private CuratorFramework zkWriter;
+ private CuratorFramework zkReader;
+ private AtomicBoolean active;
+
+ private boolean isNimbus;
+ private Map authConf;
+ private Map<Object, Object> conf;
+
+ public DistributedClusterState(Map<Object, Object> conf, Map authConf, List<ACL> acls, ClusterStateContext context) throws Exception {
+ this.conf = conf;
+ this.authConf = authConf;
+ if (context.getDaemonType().equals(DaemonType.NIMBUS)) this.isNimbus = true;
+
+ // just mkdir STORM_ZOOKEEPER_ROOT dir
+ CuratorFramework zkTemp = mkZk();
+ String rootPath = String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT));
+ Zookeeper.mkdirs(zkTemp, rootPath, acls);
+ zkTemp.close();
+
+ active = new AtomicBoolean(true);
+ zkWriter = mkZk(new WatcherCallBack() {
+ @Override
+ public void execute(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String path) {
+ if (active.get()) {
+ if (!(state.equals(Watcher.Event.KeeperState.SyncConnected))) {
+ LOG.warn("Received event {} : {}: {} with disconnected Zookeeper.", state, type, path);
+ } else {
+ LOG.info("Received event {} : {} : {}", state, type, path);
+ }
+
+ if (!type.equals(Watcher.Event.EventType.None)) {
+ for (Map.Entry<String, Callback> e : callbacks.entrySet()) {
+ Callback fn = e.getValue();
+ fn.execute(type, path);
+ }
+ }
+ }
+ }
+ });
+ if (isNimbus) {
+ zkReader = mkZk(new WatcherCallBack() {
+ @Override
+ public void execute(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String path) {
+ if (active.get()) {
+ if (!(state.equals(Watcher.Event.KeeperState.SyncConnected))) {
+ LOG.warn("Received event {} : {}: {} with disconnected Zookeeper.", state, type, path);
+ } else {
+ LOG.info("Received event {} : {} : {}", state, type, path);
+ }
+
+ if (!type.equals(Watcher.Event.EventType.None)) {
+ for (Map.Entry<String, Callback> e : callbacks.entrySet()) {
+ Callback fn = e.getValue();
+ fn.execute(type, path);
+ }
+ }
+ }
+ }
+ });
+ } else {
+ zkReader = zkWriter;
+ }
+
+ }
+
+ @SuppressWarnings("unchecked")
+ private CuratorFramework mkZk() throws IOException {
+ return Zookeeper.mkClient(conf, (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS), conf.get(Config.STORM_ZOOKEEPER_PORT), "", authConf);
+ }
+
+ @SuppressWarnings("unchecked")
+ private CuratorFramework mkZk(WatcherCallBack watcher) throws NumberFormatException, IOException {
+ return Zookeeper.mkClient(conf, (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS), conf.get(Config.STORM_ZOOKEEPER_PORT),
+ String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT)), watcher, authConf);
+ }
+
+ @Override
+ public void delete_node_blobstore(String path, String nimbusHostPortInfo) {
+
+ }
+
+ @Override
+ public String register( Callback callback) {
+ String id = UUID.randomUUID().toString();
+ this.callbacks.put(id,callback);
+ return id;
+ }
+
+ @Override
+ public void unregister(String id) {
+ this.callbacks.remove(id);
+ }
+
+ @Override
+ public String create_sequential(String path, byte[] data, List<ACL> acls) {
+ return Zookeeper.createNode(zkWriter, path, data, CreateMode.EPHEMERAL_SEQUENTIAL, acls);
+ }
+
+ @Override
+ public void mkdirs(String path, List<ACL> acls) {
+ Zookeeper.mkdirs(zkWriter, path, acls);
+ }
+
+ @Override
+ public void delete_node(String path) {
+ Zookeeper.deleteNode(zkWriter, path);
+ }
+
+ @Override
+ public void set_ephemeral_node(String path, byte[] data, List<ACL> acls) {
+ Zookeeper.mkdirs(zkWriter, parentPath(path), acls);
+ if (Zookeeper.exists(zkWriter, path, false)) {
+ try {
+ Zookeeper.setData(zkWriter, path, data);
+ } catch (RuntimeException e) {
+ if (Utils.exceptionCauseIsInstanceOf(KeeperException.NodeExistsException.class, e)) {
--- End diff --
It should be NoNodeException.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51735697
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/DistributedClusterState.java ---
@@ -0,0 +1,268 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import clojure.lang.APersistentMap;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.*;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.storm.Config;
+import org.apache.storm.callback.Callback;
+import org.apache.storm.callback.WatcherCallBack;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.zookeeper.Zookeeper;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class DistributedClusterState implements ClusterState {
--- End diff --
I am trying to understand how mk-distributed-cluster-state and mk-storm-cluster-state relate to DistributedClusterState and StormZKClusterState.
mk-distributed-cluster-state just instantiates a class pointed to by Config.STORM_CLUSTER_STATE_STORE with a default of "org.apache.storm.cluster_state.zookeeper_state_factory" and then calls mkInstance on it.
That functionality appears to be completely lost in this patch, and the StormZKStateStore is hardcoded everywhere, we need to fix that.
Now DistributedClusterState appears to be the equivalent of what zookeeper_state_factory was doing without the factory pattern, and without any reference to zookeeper in the name, while StormZKClusterState actually looks like the concrete implementation of ClusterState and does not really have any direct dependency on Zookeeper.
So to clean things up a bit lets do the following.
Rename the ClusterState interface to be StateStorage and ClusterStateFactory to be StateStorageFactory, all of them in the org.apache.storm.cluster package. Along with that please change `:cluster-state` in the worker-data to be `:state-store`. This class, DistributedClusterState, would be renamed to ZKStateStorage and an appropriate ZKStateStorageFactory needs to be created. pacemaker_state_factory.clj needs to be reverted and updated to extend the newly renamed StateStorageFactory, and to return a StateStorage instance.
ClusterState no longer needs to be an interface and implementation split off from one another. It should be a single class that uses the StateStore that is passed into it in the constructor.
mk-distributed-cluster-state should be a static method in mkStateStorage that should probably be a part of ClusterUtils.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51725134
--- Diff: storm-core/src/clj/org/apache/storm/daemon/common.clj ---
@@ -13,14 +13,16 @@
;; 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.
+;TopologyActionOptions TopologyStatus StormBase RebalanceOptions KillOptions
--- End diff --
Not sure what this comment means could you remove it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51728819
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/Cluster.java ---
@@ -0,0 +1,236 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.ExecutorStats;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Cluster {
+
+ public static final String ZK_SEPERATOR = "/";
+
+ public static final String ASSIGNMENTS_ROOT = "assignments";
+ public static final String CODE_ROOT = "code";
+ public static final String STORMS_ROOT = "storms";
+ public static final String SUPERVISORS_ROOT = "supervisors";
+ public static final String WORKERBEATS_ROOT = "workerbeats";
+ public static final String BACKPRESSURE_ROOT = "backpressure";
+ public static final String ERRORS_ROOT = "errors";
+ public static final String BLOBSTORE_ROOT = "blobstore";
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT = "blobstoremaxkeysequencenumber";
+ public static final String NIMBUSES_ROOT = "nimbuses";
+ public static final String CREDENTIALS_ROOT = "credentials";
+ public static final String LOGCONFIG_ROOT = "logconfigs";
+ public static final String PROFILERCONFIG_ROOT = "profilerconfigs";
+
+ public static final String ASSIGNMENTS_SUBTREE;
+ public static final String STORMS_SUBTREE;
+ public static final String SUPERVISORS_SUBTREE;
+ public static final String WORKERBEATS_SUBTREE;
+ public static final String BACKPRESSURE_SUBTREE;
+ public static final String ERRORS_SUBTREE;
+ public static final String BLOBSTORE_SUBTREE;
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE;
+ public static final String NIMBUSES_SUBTREE;
+ public static final String CREDENTIALS_SUBTREE;
+ public static final String LOGCONFIG_SUBTREE;
+ public static final String PROFILERCONFIG_SUBTREE;
+
+ static {
+ ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
+ STORMS_SUBTREE = ZK_SEPERATOR + STORMS_ROOT;
+ SUPERVISORS_SUBTREE = ZK_SEPERATOR + SUPERVISORS_ROOT;
+ WORKERBEATS_SUBTREE = ZK_SEPERATOR + WORKERBEATS_ROOT;
+ BACKPRESSURE_SUBTREE = ZK_SEPERATOR + BACKPRESSURE_ROOT;
+ ERRORS_SUBTREE = ZK_SEPERATOR + ERRORS_ROOT;
+ BLOBSTORE_SUBTREE = ZK_SEPERATOR + BLOBSTORE_ROOT;
+ BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE = ZK_SEPERATOR + BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT;
+ NIMBUSES_SUBTREE = ZK_SEPERATOR + NIMBUSES_ROOT;
+ CREDENTIALS_SUBTREE = ZK_SEPERATOR + CREDENTIALS_ROOT;
+ LOGCONFIG_SUBTREE = ZK_SEPERATOR + LOGCONFIG_ROOT;
+ PROFILERCONFIG_SUBTREE = ZK_SEPERATOR + PROFILERCONFIG_ROOT;
+ }
+
+ // A singleton instance allows us to mock delegated static methods in our
+ // tests by subclassing.
+ private static final Cluster INSTANCE = new Cluster();
+ private static Cluster _instance = INSTANCE;
+
+ /**
+ * Provide an instance of this class for delegates to use. To mock out
+ * delegated methods, provide an instance of a subclass that overrides the
+ * implementation of the delegated method.
+ *
+ * @param u a Zookeeper instance
+ */
+ public static void setInstance(Cluster u) {
+ _instance = u;
+ }
+
+ /**
+ * Resets the singleton instance to the default. This is helpful to reset
+ * the class to its original functionality when mocking is no longer
+ * desired.
+ */
+ public static void resetInstance() {
+ _instance = INSTANCE;
+ }
+
+ public static List<ACL> mkTopoOnlyAcls(Map topoConf) throws NoSuchAlgorithmException {
+ List<ACL> aclList = null;
+ String payload = (String)topoConf.get(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD);
+ if (Utils.isZkAuthenticationConfiguredStormServer(topoConf)){
+ aclList = new ArrayList<>();
+ ACL acl1 = ZooDefs.Ids.CREATOR_ALL_ACL.get(0);
+ aclList.add(acl1);
+ ACL acl2 = new ACL(ZooDefs.Perms.READ, new Id("digest", DigestAuthenticationProvider.generateDigest(payload)));
+ aclList.add(acl2);
+ }
+ return aclList;
+ }
+
+ public static String supervisorPath(String id) {
+ return SUPERVISORS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String assignmentPath(String id) {
+ return ASSIGNMENTS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String blobstorePath(String key) {
+ return BLOBSTORE_SUBTREE + ZK_SEPERATOR + key;
+ }
+
+ public static String blobstoreMaxKeySequenceNumberPath(String key) {
+ return BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE + ZK_SEPERATOR + key;
+ }
+
+ public static String nimbusPath(String id) {
+ return NIMBUSES_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String stormPath(String id) {
+ return STORMS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String workerbeatStormRoot(String stormId) {
+ return WORKERBEATS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String workerbeatPath(String stormId, String node, Long port) {
+ return workerbeatStormRoot(stormId) + ZK_SEPERATOR + node + "-" + port;
+ }
+
+ public static String backpressureStormRoot(String stormId) {
+ return BACKPRESSURE_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String backpressurePath(String stormId, String node, Long port) {
+ return backpressureStormRoot(stormId) + ZK_SEPERATOR + node + "-" + port;
+ }
+
+ public static String errorStormRoot(String stormId) {
+ return ERRORS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String errorPath(String stormId, String componentId) throws UnsupportedEncodingException {
+ return errorStormRoot(stormId) + ZK_SEPERATOR + URLEncoder.encode(componentId, "UTF-8");
+ }
+
+ public static String lastErrorPath(String stormId, String componentId) throws UnsupportedEncodingException {
+ return errorPath(stormId, componentId) + "-last-error";
+ }
+
+ public static String credentialsPath(String stormId) {
+ return CREDENTIALS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String logConfigPath(String stormId) {
+ return LOGCONFIG_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String profilerConfigPath(String stormId) {
+ return PROFILERCONFIG_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String profilerConfigPath(String stormId, String host, Long port, ProfileAction requestType) {
+ return profilerConfigPath(stormId) + ZK_SEPERATOR + host + "_" + port + "_" + requestType;
+ }
+
+ public static <T> T maybeDeserialize(byte[] serialized, Class<T> clazz){
--- End diff --
Lets make this private. a ```(defn- ``` means that the definition is private to the namespace. We should try and follow that where necessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51755050
--- Diff: storm-core/src/clj/org/apache/storm/daemon/nimbus.clj ---
@@ -275,11 +274,11 @@
(defn do-rebalance [nimbus storm-id status storm-base]
(let [rebalance-options (:topology-action-options storm-base)]
- (.update-storm! (:storm-cluster-state nimbus)
+ (.updateStorm (:storm-cluster-state nimbus)
storm-id
- (-> {:topology-action-options nil}
+ (thriftify-storm-base (-> {:topology-action-options nil}
(assoc-non-nil :component->executors (:component->executors rebalance-options))
- (assoc-non-nil :num-workers (:num-workers rebalance-options)))))
+ (assoc-non-nil :num-workers (:num-workers rebalance-options))))))
--- End diff --
Can we put `thriftify-storm-base` at the end of the `->` to follow suit here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51741856
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/DistributedClusterState.java ---
@@ -0,0 +1,268 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import clojure.lang.APersistentMap;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.*;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.storm.Config;
+import org.apache.storm.callback.Callback;
+import org.apache.storm.callback.WatcherCallBack;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.zookeeper.Zookeeper;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class DistributedClusterState implements ClusterState {
+
+ private static Logger LOG = LoggerFactory.getLogger(DistributedClusterState.class);
+
+ private ConcurrentHashMap<String, Callback> callbacks = new ConcurrentHashMap<String, Callback>();
+ private CuratorFramework zkWriter;
+ private CuratorFramework zkReader;
+ private AtomicBoolean active;
+
+ private boolean isNimbus;
+ private Map authConf;
+ private Map<Object, Object> conf;
+
+ public DistributedClusterState(Map<Object, Object> conf, Map authConf, List<ACL> acls, ClusterStateContext context) throws Exception {
+ this.conf = conf;
+ this.authConf = authConf;
+ if (context.getDaemonType().equals(DaemonType.NIMBUS)) this.isNimbus = true;
+
+ // just mkdir STORM_ZOOKEEPER_ROOT dir
+ CuratorFramework zkTemp = mkZk();
+ String rootPath = String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT));
+ Zookeeper.mkdirs(zkTemp, rootPath, acls);
+ zkTemp.close();
+
+ active = new AtomicBoolean(true);
+ zkWriter = mkZk(new WatcherCallBack() {
+ @Override
+ public void execute(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String path) {
+ if (active.get()) {
+ if (!(state.equals(Watcher.Event.KeeperState.SyncConnected))) {
+ LOG.warn("Received event {} : {}: {} with disconnected Zookeeper.", state, type, path);
+ } else {
+ LOG.info("Received event {} : {} : {}", state, type, path);
+ }
+
+ if (!type.equals(Watcher.Event.EventType.None)) {
+ for (Map.Entry<String, Callback> e : callbacks.entrySet()) {
+ Callback fn = e.getValue();
+ fn.execute(type, path);
+ }
+ }
+ }
+ }
+ });
+ if (isNimbus) {
+ zkReader = mkZk(new WatcherCallBack() {
+ @Override
+ public void execute(Watcher.Event.KeeperState state, Watcher.Event.EventType type, String path) {
+ if (active.get()) {
+ if (!(state.equals(Watcher.Event.KeeperState.SyncConnected))) {
+ LOG.warn("Received event {} : {}: {} with disconnected Zookeeper.", state, type, path);
+ } else {
+ LOG.info("Received event {} : {} : {}", state, type, path);
+ }
+
+ if (!type.equals(Watcher.Event.EventType.None)) {
+ for (Map.Entry<String, Callback> e : callbacks.entrySet()) {
+ Callback fn = e.getValue();
+ fn.execute(type, path);
+ }
+ }
+ }
+ }
+ });
+ } else {
+ zkReader = zkWriter;
+ }
+
+ }
+
+ @SuppressWarnings("unchecked")
+ private CuratorFramework mkZk() throws IOException {
+ return Zookeeper.mkClient(conf, (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS), conf.get(Config.STORM_ZOOKEEPER_PORT), "", authConf);
+ }
+
+ @SuppressWarnings("unchecked")
+ private CuratorFramework mkZk(WatcherCallBack watcher) throws NumberFormatException, IOException {
+ return Zookeeper.mkClient(conf, (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS), conf.get(Config.STORM_ZOOKEEPER_PORT),
+ String.valueOf(conf.get(Config.STORM_ZOOKEEPER_ROOT)), watcher, authConf);
+ }
+
+ @Override
+ public void delete_node_blobstore(String path, String nimbusHostPortInfo) {
+
+ }
+
+ @Override
+ public String register( Callback callback) {
+ String id = UUID.randomUUID().toString();
+ this.callbacks.put(id,callback);
+ return id;
+ }
+
+ @Override
+ public void unregister(String id) {
+ this.callbacks.remove(id);
+ }
+
+ @Override
+ public String create_sequential(String path, byte[] data, List<ACL> acls) {
+ return Zookeeper.createNode(zkWriter, path, data, CreateMode.EPHEMERAL_SEQUENTIAL, acls);
+ }
+
+ @Override
+ public void mkdirs(String path, List<ACL> acls) {
+ Zookeeper.mkdirs(zkWriter, path, acls);
+ }
+
+ @Override
+ public void delete_node(String path) {
+ Zookeeper.deleteNode(zkWriter, path);
+ }
+
+ @Override
+ public void set_ephemeral_node(String path, byte[] data, List<ACL> acls) {
+ Zookeeper.mkdirs(zkWriter, parentPath(path), acls);
+ if (Zookeeper.exists(zkWriter, path, false)) {
+ try {
+ Zookeeper.setData(zkWriter, path, data);
+ } catch (RuntimeException e) {
--- End diff --
you can catch Exception here instead of just RuntimeException.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj closed the pull request at:
https://github.com/apache/storm/pull/1071
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by hustfxj <gi...@git.apache.org>.
Github user hustfxj commented on the pull request:
https://github.com/apache/storm/pull/1071#issuecomment-179569559
@revans2 @abhishekagarwal87 I will close this PR to update the codes according to your comments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51735162
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/Cluster.java ---
@@ -0,0 +1,236 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.ExecutorStats;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Cluster {
--- End diff --
Lets rename this ClusterUtils, because it really is a Utils class.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51742410
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/StormZkClusterState.java ---
@@ -0,0 +1,683 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import clojure.lang.APersistentMap;
+import clojure.lang.IFn;
+import clojure.lang.PersistentArrayMap;
+import clojure.lang.RT;
+import org.apache.commons.lang.StringUtils;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.*;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.storm.callback.Callback;
+import org.apache.storm.generated.*;
+import org.apache.storm.nimbus.NimbusInfo;
+import org.apache.storm.utils.Time;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.zookeeper.Zookeeper;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.UnsupportedEncodingException;
+import java.security.NoSuchAlgorithmException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class StormZkClusterState implements StormClusterState {
+
+ private static Logger LOG = LoggerFactory.getLogger(StormZkClusterState.class);
+
+ private ClusterState clusterState;
+
+ private ConcurrentHashMap<String, IFn> assignmentInfoCallback;
+ private ConcurrentHashMap<String, IFn> assignmentInfoWithVersionCallback;
+ private ConcurrentHashMap<String, IFn> assignmentVersionCallback;
+ private AtomicReference<IFn> supervisorsCallback;
+ // we want to reigister a topo directory getChildren callback for all workers of this dir
+ private ConcurrentHashMap<String, IFn> backPressureCallback;
+ private AtomicReference<IFn> assignmentsCallback;
+ private ConcurrentHashMap<String, IFn> stormBaseCallback;
+ private AtomicReference<IFn> blobstoreCallback;
+ private ConcurrentHashMap<String, IFn> credentialsCallback;
+ private ConcurrentHashMap<String, IFn> logConfigCallback;
+
+ private List<ACL> acls;
+ private String stateId;
+ private boolean solo;
+
+ public StormZkClusterState(Object clusterState, List<ACL> acls, ClusterStateContext context) throws Exception {
+
+ if (clusterState instanceof ClusterState) {
+ solo = false;
+ this.clusterState = (ClusterState) clusterState;
+ } else {
+
+ solo = true;
+ this.clusterState = new DistributedClusterState((Map) clusterState, (Map) clusterState, acls, context);
+ }
+
+ assignmentInfoCallback = new ConcurrentHashMap<>();
+ assignmentInfoWithVersionCallback = new ConcurrentHashMap<>();
+ assignmentVersionCallback = new ConcurrentHashMap<>();
+ supervisorsCallback = new AtomicReference<>();
+ backPressureCallback = new ConcurrentHashMap<>();
+ assignmentsCallback = new AtomicReference<>();
+ stormBaseCallback = new ConcurrentHashMap<>();
+ credentialsCallback = new ConcurrentHashMap<>();
+ logConfigCallback = new ConcurrentHashMap<>();
+ blobstoreCallback = new AtomicReference<>();
+
+ stateId = this.clusterState.register(new Callback() {
+
+ public <T> Object execute(T... args) {
+ if (args == null) {
+ LOG.warn("Input args is null");
+ return null;
+ } else if (args.length < 2) {
+ LOG.warn("Input args is invalid, args length:" + args.length);
+ return null;
+ }
+ String path = (String) args[1];
+
+ List<String> toks = Zookeeper.tokenizePath(path);
+ int size = toks.size();
+ if (size >= 1) {
+ String params = null;
+ String root = toks.get(0);
+ IFn fn = null;
+ if (root.equals(Cluster.ASSIGNMENTS_ROOT)) {
+ if (size == 1) {
+ // set null and get the old value
+ issueCallback(assignmentsCallback);
+ } else {
+ issueMapCallback(assignmentInfoCallback, toks.get(1));
+ issueMapCallback(assignmentVersionCallback, toks.get(1));
+ issueMapCallback(assignmentInfoWithVersionCallback, toks.get(1));
+ }
+
+ } else if (root.equals(Cluster.SUPERVISORS_ROOT)) {
+ issueCallback(supervisorsCallback);
+ } else if (root.equals(Cluster.BLOBSTORE_ROOT)) {
+ issueCallback(blobstoreCallback);
+ } else if (root.equals(Cluster.STORMS_ROOT) && size > 1) {
+ issueMapCallback(stormBaseCallback, toks.get(1));
+ } else if (root.equals(Cluster.CREDENTIALS_ROOT) && size > 1) {
+ issueMapCallback(credentialsCallback, toks.get(1));
+ } else if (root.equals(Cluster.LOGCONFIG_ROOT) && size > 1) {
+ issueMapCallback(logConfigCallback, toks.get(1));
+ } else if (root.equals(Cluster.BACKPRESSURE_ROOT) && size > 1) {
+ issueMapCallback(logConfigCallback, toks.get(1));
+ } else {
+ LOG.error("{} Unknown callback for subtree {}", new RuntimeException("Unknown callback for this path"), path);
--- End diff --
can you also log that exit is being called due to the error.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51727157
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/Cluster.java ---
@@ -0,0 +1,236 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.ExecutorStats;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Cluster {
+
+ public static final String ZK_SEPERATOR = "/";
+
+ public static final String ASSIGNMENTS_ROOT = "assignments";
+ public static final String CODE_ROOT = "code";
+ public static final String STORMS_ROOT = "storms";
+ public static final String SUPERVISORS_ROOT = "supervisors";
+ public static final String WORKERBEATS_ROOT = "workerbeats";
+ public static final String BACKPRESSURE_ROOT = "backpressure";
+ public static final String ERRORS_ROOT = "errors";
+ public static final String BLOBSTORE_ROOT = "blobstore";
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT = "blobstoremaxkeysequencenumber";
+ public static final String NIMBUSES_ROOT = "nimbuses";
+ public static final String CREDENTIALS_ROOT = "credentials";
+ public static final String LOGCONFIG_ROOT = "logconfigs";
+ public static final String PROFILERCONFIG_ROOT = "profilerconfigs";
+
+ public static final String ASSIGNMENTS_SUBTREE;
+ public static final String STORMS_SUBTREE;
+ public static final String SUPERVISORS_SUBTREE;
+ public static final String WORKERBEATS_SUBTREE;
+ public static final String BACKPRESSURE_SUBTREE;
+ public static final String ERRORS_SUBTREE;
+ public static final String BLOBSTORE_SUBTREE;
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE;
+ public static final String NIMBUSES_SUBTREE;
+ public static final String CREDENTIALS_SUBTREE;
+ public static final String LOGCONFIG_SUBTREE;
+ public static final String PROFILERCONFIG_SUBTREE;
+
+ static {
+ ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
+ STORMS_SUBTREE = ZK_SEPERATOR + STORMS_ROOT;
+ SUPERVISORS_SUBTREE = ZK_SEPERATOR + SUPERVISORS_ROOT;
+ WORKERBEATS_SUBTREE = ZK_SEPERATOR + WORKERBEATS_ROOT;
+ BACKPRESSURE_SUBTREE = ZK_SEPERATOR + BACKPRESSURE_ROOT;
+ ERRORS_SUBTREE = ZK_SEPERATOR + ERRORS_ROOT;
+ BLOBSTORE_SUBTREE = ZK_SEPERATOR + BLOBSTORE_ROOT;
+ BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE = ZK_SEPERATOR + BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT;
+ NIMBUSES_SUBTREE = ZK_SEPERATOR + NIMBUSES_ROOT;
+ CREDENTIALS_SUBTREE = ZK_SEPERATOR + CREDENTIALS_ROOT;
+ LOGCONFIG_SUBTREE = ZK_SEPERATOR + LOGCONFIG_ROOT;
+ PROFILERCONFIG_SUBTREE = ZK_SEPERATOR + PROFILERCONFIG_ROOT;
+ }
--- End diff --
I would prefer to see these inlined above, instead of in a static block.
for example
```
public static final String ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51727250
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/Cluster.java ---
@@ -0,0 +1,236 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.ExecutorStats;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Cluster {
+
+ public static final String ZK_SEPERATOR = "/";
+
+ public static final String ASSIGNMENTS_ROOT = "assignments";
+ public static final String CODE_ROOT = "code";
+ public static final String STORMS_ROOT = "storms";
+ public static final String SUPERVISORS_ROOT = "supervisors";
+ public static final String WORKERBEATS_ROOT = "workerbeats";
+ public static final String BACKPRESSURE_ROOT = "backpressure";
+ public static final String ERRORS_ROOT = "errors";
+ public static final String BLOBSTORE_ROOT = "blobstore";
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT = "blobstoremaxkeysequencenumber";
+ public static final String NIMBUSES_ROOT = "nimbuses";
+ public static final String CREDENTIALS_ROOT = "credentials";
+ public static final String LOGCONFIG_ROOT = "logconfigs";
+ public static final String PROFILERCONFIG_ROOT = "profilerconfigs";
+
+ public static final String ASSIGNMENTS_SUBTREE;
+ public static final String STORMS_SUBTREE;
+ public static final String SUPERVISORS_SUBTREE;
+ public static final String WORKERBEATS_SUBTREE;
+ public static final String BACKPRESSURE_SUBTREE;
+ public static final String ERRORS_SUBTREE;
+ public static final String BLOBSTORE_SUBTREE;
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE;
+ public static final String NIMBUSES_SUBTREE;
+ public static final String CREDENTIALS_SUBTREE;
+ public static final String LOGCONFIG_SUBTREE;
+ public static final String PROFILERCONFIG_SUBTREE;
+
+ static {
+ ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
+ STORMS_SUBTREE = ZK_SEPERATOR + STORMS_ROOT;
+ SUPERVISORS_SUBTREE = ZK_SEPERATOR + SUPERVISORS_ROOT;
+ WORKERBEATS_SUBTREE = ZK_SEPERATOR + WORKERBEATS_ROOT;
+ BACKPRESSURE_SUBTREE = ZK_SEPERATOR + BACKPRESSURE_ROOT;
+ ERRORS_SUBTREE = ZK_SEPERATOR + ERRORS_ROOT;
+ BLOBSTORE_SUBTREE = ZK_SEPERATOR + BLOBSTORE_ROOT;
+ BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE = ZK_SEPERATOR + BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT;
+ NIMBUSES_SUBTREE = ZK_SEPERATOR + NIMBUSES_ROOT;
+ CREDENTIALS_SUBTREE = ZK_SEPERATOR + CREDENTIALS_ROOT;
+ LOGCONFIG_SUBTREE = ZK_SEPERATOR + LOGCONFIG_ROOT;
+ PROFILERCONFIG_SUBTREE = ZK_SEPERATOR + PROFILERCONFIG_ROOT;
+ }
+
+ // A singleton instance allows us to mock delegated static methods in our
+ // tests by subclassing.
+ private static final Cluster INSTANCE = new Cluster();
+ private static Cluster _instance = INSTANCE;
+
+ /**
+ * Provide an instance of this class for delegates to use. To mock out
+ * delegated methods, provide an instance of a subclass that overrides the
+ * implementation of the delegated method.
+ *
+ * @param u a Zookeeper instance
--- End diff --
Please update the comment to be Cluster not Zookeeper
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51728055
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/Cluster.java ---
@@ -0,0 +1,236 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import org.apache.storm.Config;
+import org.apache.storm.generated.ClusterWorkerHeartbeat;
+import org.apache.storm.generated.ExecutorInfo;
+import org.apache.storm.generated.ExecutorStats;
+import org.apache.storm.generated.ProfileAction;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
+
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class Cluster {
+
+ public static final String ZK_SEPERATOR = "/";
+
+ public static final String ASSIGNMENTS_ROOT = "assignments";
+ public static final String CODE_ROOT = "code";
+ public static final String STORMS_ROOT = "storms";
+ public static final String SUPERVISORS_ROOT = "supervisors";
+ public static final String WORKERBEATS_ROOT = "workerbeats";
+ public static final String BACKPRESSURE_ROOT = "backpressure";
+ public static final String ERRORS_ROOT = "errors";
+ public static final String BLOBSTORE_ROOT = "blobstore";
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT = "blobstoremaxkeysequencenumber";
+ public static final String NIMBUSES_ROOT = "nimbuses";
+ public static final String CREDENTIALS_ROOT = "credentials";
+ public static final String LOGCONFIG_ROOT = "logconfigs";
+ public static final String PROFILERCONFIG_ROOT = "profilerconfigs";
+
+ public static final String ASSIGNMENTS_SUBTREE;
+ public static final String STORMS_SUBTREE;
+ public static final String SUPERVISORS_SUBTREE;
+ public static final String WORKERBEATS_SUBTREE;
+ public static final String BACKPRESSURE_SUBTREE;
+ public static final String ERRORS_SUBTREE;
+ public static final String BLOBSTORE_SUBTREE;
+ public static final String BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE;
+ public static final String NIMBUSES_SUBTREE;
+ public static final String CREDENTIALS_SUBTREE;
+ public static final String LOGCONFIG_SUBTREE;
+ public static final String PROFILERCONFIG_SUBTREE;
+
+ static {
+ ASSIGNMENTS_SUBTREE = ZK_SEPERATOR + ASSIGNMENTS_ROOT;
+ STORMS_SUBTREE = ZK_SEPERATOR + STORMS_ROOT;
+ SUPERVISORS_SUBTREE = ZK_SEPERATOR + SUPERVISORS_ROOT;
+ WORKERBEATS_SUBTREE = ZK_SEPERATOR + WORKERBEATS_ROOT;
+ BACKPRESSURE_SUBTREE = ZK_SEPERATOR + BACKPRESSURE_ROOT;
+ ERRORS_SUBTREE = ZK_SEPERATOR + ERRORS_ROOT;
+ BLOBSTORE_SUBTREE = ZK_SEPERATOR + BLOBSTORE_ROOT;
+ BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE = ZK_SEPERATOR + BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_ROOT;
+ NIMBUSES_SUBTREE = ZK_SEPERATOR + NIMBUSES_ROOT;
+ CREDENTIALS_SUBTREE = ZK_SEPERATOR + CREDENTIALS_ROOT;
+ LOGCONFIG_SUBTREE = ZK_SEPERATOR + LOGCONFIG_ROOT;
+ PROFILERCONFIG_SUBTREE = ZK_SEPERATOR + PROFILERCONFIG_ROOT;
+ }
+
+ // A singleton instance allows us to mock delegated static methods in our
+ // tests by subclassing.
+ private static final Cluster INSTANCE = new Cluster();
+ private static Cluster _instance = INSTANCE;
+
+ /**
+ * Provide an instance of this class for delegates to use. To mock out
+ * delegated methods, provide an instance of a subclass that overrides the
+ * implementation of the delegated method.
+ *
+ * @param u a Zookeeper instance
+ */
+ public static void setInstance(Cluster u) {
+ _instance = u;
+ }
+
+ /**
+ * Resets the singleton instance to the default. This is helpful to reset
+ * the class to its original functionality when mocking is no longer
+ * desired.
+ */
+ public static void resetInstance() {
+ _instance = INSTANCE;
+ }
+
+ public static List<ACL> mkTopoOnlyAcls(Map topoConf) throws NoSuchAlgorithmException {
+ List<ACL> aclList = null;
+ String payload = (String)topoConf.get(Config.STORM_ZOOKEEPER_AUTH_PAYLOAD);
+ if (Utils.isZkAuthenticationConfiguredStormServer(topoConf)){
+ aclList = new ArrayList<>();
+ ACL acl1 = ZooDefs.Ids.CREATOR_ALL_ACL.get(0);
+ aclList.add(acl1);
+ ACL acl2 = new ACL(ZooDefs.Perms.READ, new Id("digest", DigestAuthenticationProvider.generateDigest(payload)));
+ aclList.add(acl2);
+ }
+ return aclList;
+ }
+
+ public static String supervisorPath(String id) {
+ return SUPERVISORS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String assignmentPath(String id) {
+ return ASSIGNMENTS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String blobstorePath(String key) {
+ return BLOBSTORE_SUBTREE + ZK_SEPERATOR + key;
+ }
+
+ public static String blobstoreMaxKeySequenceNumberPath(String key) {
+ return BLOBSTORE_MAX_KEY_SEQUENCE_NUMBER_SUBTREE + ZK_SEPERATOR + key;
+ }
+
+ public static String nimbusPath(String id) {
+ return NIMBUSES_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String stormPath(String id) {
+ return STORMS_SUBTREE + ZK_SEPERATOR + id;
+ }
+
+ public static String workerbeatStormRoot(String stormId) {
+ return WORKERBEATS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String workerbeatPath(String stormId, String node, Long port) {
+ return workerbeatStormRoot(stormId) + ZK_SEPERATOR + node + "-" + port;
+ }
+
+ public static String backpressureStormRoot(String stormId) {
+ return BACKPRESSURE_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String backpressurePath(String stormId, String node, Long port) {
+ return backpressureStormRoot(stormId) + ZK_SEPERATOR + node + "-" + port;
+ }
+
+ public static String errorStormRoot(String stormId) {
+ return ERRORS_SUBTREE + ZK_SEPERATOR + stormId;
+ }
+
+ public static String errorPath(String stormId, String componentId) throws UnsupportedEncodingException {
+ return errorStormRoot(stormId) + ZK_SEPERATOR + URLEncoder.encode(componentId, "UTF-8");
--- End diff --
I know the last time I told you to let the exceptions bubble up, and for the most part I still agree with that, but in this case the UnsupportedEncodingException should never really be thrown. If "UTF-8" is not supported by the JVM, which it should always be supported, we have some really big problems. So I am OK with wrapping this one in a RuntimeException.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51754553
--- Diff: storm-core/src/clj/org/apache/storm/daemon/nimbus.clj ---
@@ -174,11 +173,11 @@
:authorization-handler (mk-authorization-handler (conf NIMBUS-AUTHORIZER) conf)
:impersonation-authorization-handler (mk-authorization-handler (conf NIMBUS-IMPERSONATION-AUTHORIZER) conf)
:submitted-count (atom 0)
- :storm-cluster-state (cluster/mk-storm-cluster-state conf :acls (when
+ :storm-cluster-state (StormZkClusterState. conf (when
(Utils/isZkAuthenticationConfiguredStormServer
conf)
NIMBUS-ZK-ACLS)
--- End diff --
Please correct the indentation on these lines.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by abhishekagarwal87 <gi...@git.apache.org>.
Github user abhishekagarwal87 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51788562
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/StormZkClusterState.java ---
@@ -0,0 +1,683 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.cluster;
+
+import clojure.lang.APersistentMap;
+import clojure.lang.IFn;
+import clojure.lang.PersistentArrayMap;
+import clojure.lang.RT;
+import org.apache.commons.lang.StringUtils;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.*;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.storm.callback.Callback;
+import org.apache.storm.generated.*;
+import org.apache.storm.nimbus.NimbusInfo;
+import org.apache.storm.utils.Time;
+import org.apache.storm.utils.Utils;
+import org.apache.storm.zookeeper.Zookeeper;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.UnsupportedEncodingException;
+import java.security.NoSuchAlgorithmException;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class StormZkClusterState implements StormClusterState {
+
+ private static Logger LOG = LoggerFactory.getLogger(StormZkClusterState.class);
+
+ private ClusterState clusterState;
+
+ private ConcurrentHashMap<String, IFn> assignmentInfoCallback;
+ private ConcurrentHashMap<String, IFn> assignmentInfoWithVersionCallback;
+ private ConcurrentHashMap<String, IFn> assignmentVersionCallback;
+ private AtomicReference<IFn> supervisorsCallback;
+ // we want to reigister a topo directory getChildren callback for all workers of this dir
+ private ConcurrentHashMap<String, IFn> backPressureCallback;
+ private AtomicReference<IFn> assignmentsCallback;
+ private ConcurrentHashMap<String, IFn> stormBaseCallback;
+ private AtomicReference<IFn> blobstoreCallback;
+ private ConcurrentHashMap<String, IFn> credentialsCallback;
+ private ConcurrentHashMap<String, IFn> logConfigCallback;
+
+ private List<ACL> acls;
+ private String stateId;
+ private boolean solo;
+
+ public StormZkClusterState(Object clusterState, List<ACL> acls, ClusterStateContext context) throws Exception {
+
+ if (clusterState instanceof ClusterState) {
+ solo = false;
+ this.clusterState = (ClusterState) clusterState;
+ } else {
+
+ solo = true;
+ this.clusterState = new DistributedClusterState((Map) clusterState, (Map) clusterState, acls, context);
+ }
+
+ assignmentInfoCallback = new ConcurrentHashMap<>();
+ assignmentInfoWithVersionCallback = new ConcurrentHashMap<>();
+ assignmentVersionCallback = new ConcurrentHashMap<>();
+ supervisorsCallback = new AtomicReference<>();
+ backPressureCallback = new ConcurrentHashMap<>();
+ assignmentsCallback = new AtomicReference<>();
+ stormBaseCallback = new ConcurrentHashMap<>();
+ credentialsCallback = new ConcurrentHashMap<>();
+ logConfigCallback = new ConcurrentHashMap<>();
+ blobstoreCallback = new AtomicReference<>();
+
+ stateId = this.clusterState.register(new Callback() {
+
+ public <T> Object execute(T... args) {
+ if (args == null) {
+ LOG.warn("Input args is null");
+ return null;
+ } else if (args.length < 2) {
+ LOG.warn("Input args is invalid, args length:" + args.length);
+ return null;
+ }
+ String path = (String) args[1];
+
+ List<String> toks = Zookeeper.tokenizePath(path);
+ int size = toks.size();
+ if (size >= 1) {
+ String params = null;
+ String root = toks.get(0);
+ IFn fn = null;
+ if (root.equals(Cluster.ASSIGNMENTS_ROOT)) {
+ if (size == 1) {
+ // set null and get the old value
+ issueCallback(assignmentsCallback);
+ } else {
+ issueMapCallback(assignmentInfoCallback, toks.get(1));
+ issueMapCallback(assignmentVersionCallback, toks.get(1));
+ issueMapCallback(assignmentInfoWithVersionCallback, toks.get(1));
+ }
+
+ } else if (root.equals(Cluster.SUPERVISORS_ROOT)) {
+ issueCallback(supervisorsCallback);
+ } else if (root.equals(Cluster.BLOBSTORE_ROOT)) {
+ issueCallback(blobstoreCallback);
+ } else if (root.equals(Cluster.STORMS_ROOT) && size > 1) {
+ issueMapCallback(stormBaseCallback, toks.get(1));
+ } else if (root.equals(Cluster.CREDENTIALS_ROOT) && size > 1) {
+ issueMapCallback(credentialsCallback, toks.get(1));
+ } else if (root.equals(Cluster.LOGCONFIG_ROOT) && size > 1) {
+ issueMapCallback(logConfigCallback, toks.get(1));
+ } else if (root.equals(Cluster.BACKPRESSURE_ROOT) && size > 1) {
+ issueMapCallback(logConfigCallback, toks.get(1));
+ } else {
+ LOG.error("{} Unknown callback for subtree {}", new RuntimeException("Unknown callback for this path"), path);
+ Runtime.getRuntime().exit(30);
+ }
+
+ }
+
+ return null;
+ }
+
+ });
+
+ String[] pathlist = { Cluster.ASSIGNMENTS_SUBTREE, Cluster.STORMS_SUBTREE, Cluster.SUPERVISORS_SUBTREE, Cluster.WORKERBEATS_SUBTREE,
+ Cluster.ERRORS_SUBTREE, Cluster.BLOBSTORE_SUBTREE, Cluster.NIMBUSES_SUBTREE, Cluster.LOGCONFIG_SUBTREE };
+ for (String path : pathlist) {
+ this.clusterState.mkdirs(path, acls);
+ }
+
+ }
+
+ protected void issueCallback(AtomicReference<IFn> cb) {
+ IFn callback = cb.getAndSet(null);
+ if (callback != null)
+ callback.invoke();
+ }
+
+ protected void issueMapCallback(ConcurrentHashMap<String, IFn> callbackConcurrentHashMap, String key) {
+ IFn callback = callbackConcurrentHashMap.remove(key);
+ if (callback != null)
+ callback.invoke();
+ }
+
+ @Override
+ public List<String> assignments(IFn callback) {
+ if (callback != null) {
+ assignmentsCallback.set(callback);
+ }
+ return clusterState.get_children(Cluster.ASSIGNMENTS_SUBTREE, callback != null);
+ }
+
+ @Override
+ public Assignment assignmentInfo(String stormId, IFn callback) {
+ if (callback != null) {
+ assignmentInfoCallback.put(stormId, callback);
+ }
+ byte[] serialized = clusterState.get_data(Cluster.assignmentPath(stormId), callback != null);
+ return Cluster.maybeDeserialize(serialized, Assignment.class);
+ }
+
+ @Override
+ public APersistentMap assignmentInfoWithVersion(String stormId, IFn callback) {
+ if (callback != null) {
+ assignmentInfoWithVersionCallback.put(stormId, callback);
+ }
+ APersistentMap aPersistentMap = clusterState.get_data_with_version(Cluster.assignmentPath(stormId), callback != null);
+ Assignment assignment = Cluster.maybeDeserialize((byte[]) aPersistentMap.get("data"), Assignment.class);
+ Integer version = (Integer) aPersistentMap.get("version");
+ APersistentMap map = new PersistentArrayMap(new Object[] { RT.keyword(null, "data"), assignment, RT.keyword(null, "version"), version });
+ return map;
+ }
+
+ @Override
+ public Integer assignmentVersion(String stormId, IFn callback) throws Exception {
+ if (callback != null) {
+ assignmentVersionCallback.put(stormId, callback);
+ }
+ return clusterState.get_version(Cluster.assignmentPath(stormId), callback != null);
+ }
+
+ // blobstore state
+ @Override
+ public List<String> blobstoreInfo(String blobKey) {
+ String path = Cluster.blobstorePath(blobKey);
+ clusterState.sync_path(path);
+ return clusterState.get_children(path, false);
+ }
+
+ @Override
+ public List nimbuses() {
+ List<NimbusSummary> nimbusSummaries = new ArrayList<>();
+ List<String> nimbusIds = clusterState.get_children(Cluster.NIMBUSES_SUBTREE, false);
+ for (String nimbusId : nimbusIds) {
+ byte[] serialized = clusterState.get_data(Cluster.nimbusPath(nimbusId), false);
+ NimbusSummary nimbusSummary = Cluster.maybeDeserialize(serialized, NimbusSummary.class);
+ nimbusSummaries.add(nimbusSummary);
+ }
+ return nimbusSummaries;
+ }
+
+ @Override
+ public void addNimbusHost(final String nimbusId, final NimbusSummary nimbusSummary) {
+ // explicit delete for ephmeral node to ensure this session creates the entry.
+ clusterState.delete_node(Cluster.nimbusPath(nimbusId));
+ clusterState.add_listener(new ConnectionStateListener() {
+ @Override
+ public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
+ LOG.info("Connection state listener invoked, zookeeper connection state has changed to {}", connectionState);
+ if (connectionState.equals(ConnectionState.RECONNECTED)) {
+ LOG.info("Connection state has changed to reconnected so setting nimbuses entry one more time");
+ clusterState.set_ephemeral_node(Cluster.nimbusPath(nimbusId), Utils.serialize(nimbusSummary), acls);
+ }
+
+ }
+ });
+
+ clusterState.set_ephemeral_node(Cluster.nimbusPath(nimbusId), Utils.serialize(nimbusSummary), acls);
+ }
+
+ @Override
+ public List<String> activeStorms() {
+ return clusterState.get_children(Cluster.STORMS_SUBTREE, false);
+ }
+
+ @Override
+ public StormBase stormBase(String stormId, IFn callback) {
+ if (callback != null) {
+ stormBaseCallback.put(stormId, callback);
+ }
+ return Cluster.maybeDeserialize(clusterState.get_data(Cluster.stormPath(stormId), callback != null), StormBase.class);
+ }
+
+ @Override
+ public ClusterWorkerHeartbeat getWorkerHeartbeat(String stormId, String node, Long port) {
+ byte[] bytes = clusterState.get_worker_hb(Cluster.workerbeatPath(stormId, node, port), false);
+ if (bytes != null) {
+ return Cluster.maybeDeserialize(bytes, ClusterWorkerHeartbeat.class);
+ }
+ return null;
+ }
+
+ @Override
+ public List<ProfileRequest> getWorkerProfileRequests(String stormId, NodeInfo nodeInfo, boolean isThrift) {
+ List<ProfileRequest> requests = new ArrayList<>();
+ List<ProfileRequest> profileRequests = getTopologyProfileRequests(stormId, isThrift);
+ for (ProfileRequest profileRequest : profileRequests) {
+ NodeInfo nodeInfo1 = profileRequest.get_nodeInfo();
+ if (nodeInfo1.equals(nodeInfo))
+ requests.add(profileRequest);
+ }
+ return requests;
+ }
+
+ @Override
+ public List<ProfileRequest> getTopologyProfileRequests(String stormId, boolean isThrift) {
+ List<ProfileRequest> profileRequests = new ArrayList<>();
+ String path = Cluster.profilerConfigPath(stormId);
+ if (clusterState.node_exists(path, false)) {
+ List<String> strs = clusterState.get_children(path, false);
+ for (String str : strs) {
+ String childPath = path + Cluster.ZK_SEPERATOR + str;
+ byte[] raw = clusterState.get_data(childPath, false);
+ ProfileRequest request = Cluster.maybeDeserialize(raw, ProfileRequest.class);
+ if (request != null)
+ profileRequests.add(request);
+ }
+ }
+ return profileRequests;
+ }
+
+ @Override
+ public void setWorkerProfileRequest(String stormId, ProfileRequest profileRequest) {
+ ProfileAction profileAction = profileRequest.get_action();
+ String host = profileRequest.get_nodeInfo().get_node();
+ Long port = profileRequest.get_nodeInfo().get_port_iterator().next();
+ String path = Cluster.profilerConfigPath(stormId, host, port, profileAction);
+ clusterState.set_data(path, Utils.serialize(profileRequest), acls);
+ }
+
+ @Override
+ public void deleteTopologyProfileRequests(String stormId, ProfileRequest profileRequest) {
+ ProfileAction profileAction = profileRequest.get_action();
+ String host = profileRequest.get_nodeInfo().get_node();
+ Long port = profileRequest.get_nodeInfo().get_port_iterator().next();
+ String path = Cluster.profilerConfigPath(stormId, host, port, profileAction);
+ clusterState.delete_node(path);
+ }
+
+ // need to take executor->node+port in explicitly so that we don't run into a situation where a
+ // long dead worker with a skewed clock overrides all the timestamps. By only checking heartbeats
+ // with an assigned node+port, and only reading executors from that heartbeat that are actually assigned,
+ // we avoid situations like that
+ @Override
+ public Map<ExecutorInfo, ClusterWorkerHeartbeat> executorBeats(String stormId, Map<List<Long>, NodeInfo> executorNodePort) {
+ Map<ExecutorInfo, ClusterWorkerHeartbeat> executorWhbs = new HashMap<>();
+
+ LOG.info(executorNodePort.toString());
+ Map<NodeInfo, List<List<Long>>> nodePortExecutors = Cluster.reverseMap(executorNodePort);
+ LOG.info(nodePortExecutors.toString());
+
+ for (Map.Entry<NodeInfo, List<List<Long>>> entry : nodePortExecutors.entrySet()) {
+
+ String node = entry.getKey().get_node();
+ Long port = entry.getKey().get_port_iterator().next();
+ ClusterWorkerHeartbeat whb = getWorkerHeartbeat(stormId, node, port);
+ List<ExecutorInfo> executorInfoList = new ArrayList<>();
+ for (List<Long> list : entry.getValue()) {
+ executorInfoList.add(new ExecutorInfo(list.get(0).intValue(), list.get(list.size() - 1).intValue()));
+ }
+ executorWhbs.putAll(Cluster.convertExecutorBeats(executorInfoList, whb));
+ }
+ return executorWhbs;
+ }
+
+ @Override
+ public List<String> supervisors(IFn callback) {
+ if (callback != null) {
+ supervisorsCallback.set(callback);
+ }
+ return clusterState.get_children(Cluster.SUPERVISORS_SUBTREE, callback != null);
+ }
+
+ @Override
+ public SupervisorInfo supervisorInfo(String supervisorId) {
+ String path = Cluster.supervisorPath(supervisorId);
+ return Cluster.maybeDeserialize(clusterState.get_data(path, false), SupervisorInfo.class);
+ }
+
+ @Override
+ public void setupHeatbeats(String stormId) {
+ clusterState.mkdirs(Cluster.workerbeatStormRoot(stormId), acls);
+ }
+
+ @Override
+ public void teardownHeartbeats(String stormId) {
+ try {
+ clusterState.delete_worker_hb(Cluster.workerbeatStormRoot(stormId));
+ } catch (Exception e) {
+ if (Zookeeper.exceptionCause(KeeperException.class, e)) {
+ // do nothing
+ LOG.warn("Could not teardown heartbeats for {}.", stormId);
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ @Override
+ public void teardownTopologyErrors(String stormId) {
+ try {
+ clusterState.delete_node(Cluster.errorStormRoot(stormId));
+ } catch (Exception e) {
+ if (Zookeeper.exceptionCause(KeeperException.class, e)) {
+ // do nothing
+ LOG.warn("Could not teardown errors for {}.", stormId);
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ @Override
+ public List<String> heartbeatStorms() {
+ return clusterState.get_worker_hb_children(Cluster.WORKERBEATS_SUBTREE, false);
+ }
+
+ @Override
+ public List<String> errorTopologies() {
+ return clusterState.get_children(Cluster.ERRORS_SUBTREE, false);
+ }
+
+ @Override
+ public void setTopologyLogConfig(String stormId, LogConfig logConfig) {
+ clusterState.set_data(Cluster.logConfigPath(stormId), Utils.serialize(logConfig), acls);
+ }
+
+ @Override
+ public LogConfig topologyLogConfig(String stormId, IFn cb) {
+ String path = Cluster.logConfigPath(stormId);
+ return Cluster.maybeDeserialize(clusterState.get_data(path, cb != null), LogConfig.class);
+ }
+
+ @Override
+ public void workerHeartbeat(String stormId, String node, Long port, ClusterWorkerHeartbeat info) {
+ if (info != null) {
+ String path = Cluster.workerbeatPath(stormId, node, port);
+ clusterState.set_worker_hb(path, Utils.serialize(info), acls);
+ }
+ }
+
+ @Override
+ public void removeWorkerHeartbeat(String stormId, String node, Long port) {
+ String path = Cluster.workerbeatPath(stormId, node, port);
+ clusterState.delete_worker_hb(path);
+ }
+
+ @Override
+ public void supervisorHeartbeat(String supervisorId, SupervisorInfo info) {
+ String path = Cluster.supervisorPath(supervisorId);
+ clusterState.set_ephemeral_node(path, Utils.serialize(info), acls);
+ }
+
+ // if znode exists and to be not on?, delete; if exists and on?, do nothing;
+ // if not exists and to be on?, create; if not exists and not on?, do nothing;
+ @Override
+ public void workerBackpressure(String stormId, String node, Long port, boolean on) {
+ String path = Cluster.backpressurePath(stormId, node, port);
+ boolean existed = clusterState.node_exists(path, false);
+ if (existed) {
+ if (on == false)
+ clusterState.delete_node(path);
+
+ } else {
+ if (on == true) {
+ clusterState.set_ephemeral_node(path, null, acls);
+ }
+ }
+ }
+
+ // if the backpresure/storm-id dir is empty, this topology has throttle-on, otherwise not.
+ @Override
+ public boolean topologyBackpressure(String stormId, IFn callback) {
+ if (callback != null) {
+ backPressureCallback.put(stormId, callback);
+ }
+ String path = Cluster.backpressureStormRoot(stormId);
+ List<String> childrens = clusterState.get_children(path, callback != null);
+ return childrens.size() > 0;
+
+ }
+
+ @Override
+ public void setupBackpressure(String stormId) {
+ clusterState.mkdirs(Cluster.backpressureStormRoot(stormId), acls);
+ }
+
+ @Override
+ public void removeWorkerBackpressure(String stormId, String node, Long port) {
+ clusterState.delete_node(Cluster.backpressurePath(stormId, node, port));
+ }
+
+ @Override
+ public void activateStorm(String stormId, StormBase stormBase) {
+ String path = Cluster.stormPath(stormId);
+ clusterState.set_data(path, Utils.serialize(stormBase), acls);
+ }
+
+ // maybe exit some questions for updateStorm
+ @Override
+ public void updateStorm(String stormId, StormBase newElems) {
+
+ StormBase stormBase = stormBase(stormId, null);
+ if (stormBase.get_component_executors() != null) {
+
+ Map<String, Integer> newComponentExecutors = new HashMap<>();
+ Map<String, Integer> componentExecutors = newElems.get_component_executors();
+ //componentExecutors maybe be APersistentMap, which don't support put
+ for (Map.Entry<String, Integer> entry : componentExecutors.entrySet()) {
+ newComponentExecutors.put(entry.getKey(), entry.getValue());
+ }
+ for (Map.Entry<String, Integer> entry : stormBase.get_component_executors().entrySet()) {
+ if (!componentExecutors.containsKey(entry.getKey())) {
+ newComponentExecutors.put(entry.getKey(), entry.getValue());
+ }
+ }
+ if (newComponentExecutors.size() > 0)
+ newElems.set_component_executors(newComponentExecutors);
+ }
+
+ Map<String, DebugOptions> ComponentDebug = new HashMap<>();
+ Map<String, DebugOptions> oldComponentDebug = stormBase.get_component_debug();
+
+ Map<String, DebugOptions> newComponentDebug = newElems.get_component_debug();
+
+ Set<String> debugOptionsKeys = oldComponentDebug.keySet();
+ debugOptionsKeys.addAll(newComponentDebug.keySet());
+ for (String key : debugOptionsKeys) {
+ boolean enable = false;
+ double samplingpct = 0;
+ if (oldComponentDebug.containsKey(key)) {
+ enable = oldComponentDebug.get(key).is_enable();
+ samplingpct = oldComponentDebug.get(key).get_samplingpct();
+ }
+ if (newComponentDebug.containsKey(key)) {
+ enable = newComponentDebug.get(key).is_enable();
+ samplingpct += newComponentDebug.get(key).get_samplingpct();
+ }
+ DebugOptions debugOptions = new DebugOptions();
+ debugOptions.set_enable(enable);
+ debugOptions.set_samplingpct(samplingpct);
+ ComponentDebug.put(key, debugOptions);
+ }
+ if (ComponentDebug.size() > 0) {
+ newElems.set_component_debug(ComponentDebug);
+ }
+
+
+ if (StringUtils.isBlank(newElems.get_name())) {
+ newElems.set_name(stormBase.get_name());
+ }
+ if (newElems.get_status() == null){
+ newElems.set_status(stormBase.get_status());
+ }
+ if (newElems.get_num_workers() == 0){
+ newElems.set_num_workers(stormBase.get_num_workers());
+ }
+ if (newElems.get_launch_time_secs() == 0) {
+ newElems.set_launch_time_secs(stormBase.get_launch_time_secs());
+ }
+ if (StringUtils.isBlank(newElems.get_owner())) {
+ newElems.set_owner(stormBase.get_owner());
+ }
+ if (newElems.get_topology_action_options() == null) {
+ newElems.set_topology_action_options(stormBase.get_topology_action_options());
+ }
+ if (newElems.get_status() == null) {
+ newElems.set_status(stormBase.get_status());
+ }
+ clusterState.set_data(Cluster.stormPath(stormId), Utils.serialize(newElems), acls);
+ }
+
+ @Override
+ public void removeStormBase(String stormId) {
+ clusterState.delete_node(Cluster.stormPath(stormId));
+ }
+
+ @Override
+ public void setAssignment(String stormId, Assignment info) {
+ clusterState.set_data(Cluster.assignmentPath(stormId), Utils.serialize(info), acls);
+ }
+
+ @Override
+ public void setupBlobstore(String key, NimbusInfo nimbusInfo, Integer versionInfo) {
+ String path = Cluster.blobstorePath(key) + Cluster.ZK_SEPERATOR + nimbusInfo.toHostPortString() + "-" + versionInfo;
+ LOG.info("set-path: {}", path);
+ clusterState.mkdirs(Cluster.blobstorePath(key), acls);
+ clusterState.delete_node_blobstore(Cluster.blobstorePath(key), nimbusInfo.toHostPortString());
+ clusterState.set_ephemeral_node(path, null, acls);
+ }
+
+ @Override
+ public List<String> activeKeys() {
+ return clusterState.get_children(Cluster.BLOBSTORE_SUBTREE, false);
+ }
+
+ // blobstore state
+ @Override
+ public List<String> blobstore(IFn callback) {
+ if (callback != null) {
+ blobstoreCallback.set(callback);
+ }
+ clusterState.sync_path(Cluster.BLOBSTORE_SUBTREE);
+ return clusterState.get_children(Cluster.BLOBSTORE_SUBTREE, callback != null);
+
+ }
+
+ @Override
+ public void removeStorm(String stormId) {
+ clusterState.delete_node(Cluster.assignmentPath(stormId));
+ clusterState.delete_node(Cluster.credentialsPath(stormId));
+ clusterState.delete_node(Cluster.logConfigPath(stormId));
+ clusterState.delete_node(Cluster.profilerConfigPath(stormId));
+ removeStormBase(stormId);
+ }
+
+ @Override
+ public void removeBlobstoreKey(String blobKey) {
+ LOG.debug("remove key {}", blobKey);
+ clusterState.delete_node(Cluster.blobstorePath(blobKey));
+ }
+
+ @Override
+ public void removeKeyVersion(String blobKey) {
+ clusterState.delete_node(Cluster.blobstoreMaxKeySequenceNumberPath(blobKey));
+ }
+
+ @Override
+ public void reportError(String stormId, String componentId, String node, Integer port, String error) {
+
+ try {
+ String path = Cluster.errorPath(stormId, componentId);
+ String lastErrorPath = Cluster.lastErrorPath(stormId, componentId);
+ ErrorInfo errorInfo = new ErrorInfo(error, Time.currentTimeSecs());
+ errorInfo.set_host(node);
+ errorInfo.set_port(port.intValue());
+ byte[] serData = Utils.serialize(errorInfo);
+ clusterState.mkdirs(path, acls);
+ clusterState.create_sequential(path + Cluster.ZK_SEPERATOR + "e", serData, acls);
+ clusterState.set_data(lastErrorPath, serData, acls);
+ List<String> childrens = clusterState.get_children(path, false);
+
+ Collections.sort(childrens);
--- End diff --
this will sort e1, e2, e12 to e1, e12, e2 while you want it to be e1, e2, e12.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the pull request:
https://github.com/apache/storm/pull/1071#issuecomment-179258530
The travis CI tests failed with
```
13881 [timer] ERROR o.a.s.d.nimbus - Error when processing event
java.lang.NullPointerException
at org.apache.storm.cluster.Cluster.convertExecutorBeats(Cluster.java:202) ~[classes/:?]
at org.apache.storm.cluster.StormZkClusterState.executorBeats(StormZkClusterState.java:327) ~[classes/:?]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.7.0_76]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) ~[?:1.7.0_76]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.7.0_76]
at java.lang.reflect.Method.invoke(Method.java:606) ~[?:1.7.0_76]
at clojure.lang.Reflector.invokeMatchingMethod(Reflector.java:93) ~[clojure-1.7.0.jar:?]
at clojure.lang.Reflector.invokeInstanceMethod(Reflector.java:28) ~[clojure-1.7.0.jar:?]
at org.apache.storm.daemon.nimbus$update_heartbeats_BANG_.invoke(nimbus.clj:589) ~[classes/:?]
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request: [STORM-1273] port backtype.storm.cluster to ja...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1071#discussion_r51726849
--- Diff: storm-core/src/jvm/org/apache/storm/callback/Callback.java ---
@@ -18,6 +18,9 @@
package org.apache.storm.callback;
+import clojure.lang.IFn;
+
+// To remove IFn after porting all callbacks to java
public interface Callback {
public <T> Object execute(T... args);
--- End diff --
I don't like this interface. The generic <T> for the parameters is just confusing because it implies that all of the parameters should be of the same type, when in reality they are not. I would much rather see something like
```
public interface ZKStateChangedCallback {
public void changed(Watcher.Event.EventType type, String path);
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---