You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by knusbaum <gi...@git.apache.org> on 2016/09/23 07:36:01 UTC
[GitHub] storm pull request #1710: STORM-1546: Adding Read and Write Aggregations for...
GitHub user knusbaum opened a pull request:
https://github.com/apache/storm/pull/1710
STORM-1546: Adding Read and Write Aggregations for Pacemaker to make it HA compatible
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/knusbaum/incubator-storm Pacemaker-HA
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/1710.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 #1710
----
commit c71314fbd2101ff081bee7aef3ecc434c3541044
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-06T21:04:07Z
Inspecting travis's maven cache
commit efa546862f7717ee1db222895ea14e6432666ab3
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-08T20:07:49Z
Working on Pacemaker HA
commit f46cfc15fa2069cabaa742ccfd2b5689fe5a3777
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-19T19:22:46Z
cleanup.
commit 0c002928e49f63070915e7a41f97f70b40336f23
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-19T19:47:04Z
Merge branch 'master' of http://git-wip-us.apache.org/repos/asf/storm
commit 6cd9661bde2ce4c93c5628aae2565e167b89aeb3
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T06:48:33Z
Fix ups.
commit 6a8be9f8c19bacb646c119c162c6cdba69cc66ff
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T07:03:46Z
Fix ups.
commit 523317b06abe87d7e745885c32c8a9c5dfbbeb0d
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T07:29:59Z
Cleaning up tests.
commit 410ef3a816fc6a961771abfbcac2d83512192040
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T07:31:54Z
Cleaning up.
----
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80719082
--- Diff: storm-core/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java ---
@@ -0,0 +1,113 @@
+/**
+ * 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.pacemaker;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import org.apache.storm.Config;
+import org.apache.storm.generated.HBMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PacemakerClientPool {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PacemakerClientPool.class);
+
+ private ConcurrentHashMap<String, PacemakerClient> clientForServer = new ConcurrentHashMap<>();
+ private ConcurrentLinkedQueue<String> servers;
+ private Map config;
+
+ public PacemakerClientPool(Map config) {
+ this.config = config;
+ List<String> serverList = (List<String>)config.get(Config.PACEMAKER_SERVERS);
+ if(serverList == null) {
+ serverList = new ArrayList<String>();
+ }
+ else {
+ serverList = new ArrayList<String>(serverList);
+ }
+ Collections.shuffle(serverList);
+ if(serverList != null) {
+ servers = new ConcurrentLinkedQueue<String>(serverList);
+ }
+ else {
+ servers = new ConcurrentLinkedQueue<String>();
+ }
+ }
+
+ public HBMessage send(HBMessage m) throws PacemakerConnectionException {
+ try {
+ return getWriteClient().send(m);
+ } catch (Exception e) {
+ rotateClients();
+ throw e;
+ }
+ }
+
+ public List<HBMessage> sendAll(HBMessage m) throws PacemakerConnectionException {
+ List<HBMessage> responses = new ArrayList<HBMessage>();
+ LOG.info("Using servers: {}", servers);
--- End diff --
Perhaps this should be debug (especially if we are worried about filling the logs)
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/1710
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80820718
--- Diff: storm-core/src/jvm/org/apache/storm/Config.java ---
@@ -926,10 +926,10 @@
public static final String UI_HTTPS_NEED_CLIENT_AUTH = "ui.https.need.client.auth";
/**
- * The host that Pacemaker is running on.
+ * The list of servers that Pacemaker is running on.
*/
- @isString
- public static final String PACEMAKER_HOST = "pacemaker.host";
+ @isStringList
+ public static final String PACEMAKER_SERVERS = "pacemaker.servers";
--- End diff --
I'm in favor of backward compatibility change for this, like nimbus.host and nimbus.seeds. For 2.0 we don't need to keep nimbus.host and pacemaker.host but 1.x still need to have 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 issue #1710: STORM-1546: Adding Read and Write Aggregations for Pacema...
Posted by d2r <gi...@git.apache.org>.
Github user d2r commented on the issue:
https://github.com/apache/storm/pull/1710
Thanks for doc update. +1
---
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 issue #1710: STORM-1546: Adding Read and Write Aggregations for Pacema...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on the issue:
https://github.com/apache/storm/pull/1710
Documentation updated.
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80791012
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/PaceMakerStateStorage.java ---
@@ -159,10 +162,6 @@ public void set_worker_hb(String path, byte[] data, List<ACL> acls) {
ret = details;
}
}
- if(ret == null) {
- throw new HBExecutionException("Failed to get a response.");
- }
- LOG.debug("Successful get_worker_hb");
return ret;
--- End diff --
Do we need the ret == null check before this? Is it possible to get more then one response and all of them are bogus?
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80508806
--- Diff: storm-core/src/jvm/org/apache/storm/pacemaker/PacemakerClientHandler.java ---
@@ -69,7 +69,7 @@ else if(evm instanceof HBMessage) {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent event) {
- LOG.error("Connection to pacemaker failed", event.getCause());
+ LOG.error("Connection to pacemaker failed. Trying to reconnect {}", event.getCause().getMessage());
--- End diff --
I don't think we want the stack. It ends up absolutely filling Nimbus logs with relatively useless messages when a Pacemaker node goes down. I'm open to ideas, though, because a trace for an NPE or something would be nice.
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80718833
--- Diff: storm-core/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java ---
@@ -0,0 +1,113 @@
+/**
+ * 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.pacemaker;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import org.apache.storm.Config;
+import org.apache.storm.generated.HBMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PacemakerClientPool {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PacemakerClientPool.class);
+
+ private ConcurrentHashMap<String, PacemakerClient> clientForServer = new ConcurrentHashMap<>();
+ private ConcurrentLinkedQueue<String> servers;
+ private Map config;
+
+ public PacemakerClientPool(Map config) {
+ this.config = config;
+ List<String> serverList = (List<String>)config.get(Config.PACEMAKER_SERVERS);
+ if(serverList == null) {
+ serverList = new ArrayList<String>();
--- End diff --
nit: I don't think `<String>` is needed just `<>`
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by d2r <gi...@git.apache.org>.
Github user d2r commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80319044
--- Diff: storm-core/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java ---
@@ -0,0 +1,114 @@
+/**
+ * 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.pacemaker;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import org.apache.storm.Config;
+import org.apache.storm.generated.HBMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PacemakerClientPool {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PacemakerClientPool.class);
--- End diff --
check indentation
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by d2r <gi...@git.apache.org>.
Github user d2r commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80324279
--- Diff: storm-core/test/jvm/org/apache/storm/PaceMakerStateStorageFactoryTest.java ---
@@ -20,13 +20,22 @@
import org.apache.storm.cluster.PaceMakerStateStorage;
import org.apache.storm.generated.*;
import org.apache.storm.pacemaker.PacemakerClient;
+import org.apache.storm.pacemaker.PacemakerClientPool;
import org.apache.storm.utils.Utils;
import org.junit.Assert;
import org.junit.Test;
-public class PaceMakerStateStorageFactoryTest {
+import java.util.HashMap;
+import java.util.List;
+import java.util.ArrayList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
- private PacemakerClient clientProxy;
+public class PaceMakerStateStorageFactoryTest {
+ private static Logger LOG = LoggerFactory.getLogger(PaceMakerStateStorageFactoryTest.class);
--- End diff --
We don't seem to be logging anything in this test. Do we need the Logger?
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80718914
--- Diff: storm-core/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java ---
@@ -0,0 +1,113 @@
+/**
+ * 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.pacemaker;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import org.apache.storm.Config;
+import org.apache.storm.generated.HBMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PacemakerClientPool {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PacemakerClientPool.class);
+
+ private ConcurrentHashMap<String, PacemakerClient> clientForServer = new ConcurrentHashMap<>();
+ private ConcurrentLinkedQueue<String> servers;
+ private Map config;
+
+ public PacemakerClientPool(Map config) {
+ this.config = config;
+ List<String> serverList = (List<String>)config.get(Config.PACEMAKER_SERVERS);
+ if(serverList == null) {
+ serverList = new ArrayList<String>();
--- End diff --
and in the following few lines too.
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by d2r <gi...@git.apache.org>.
Github user d2r commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80323247
--- Diff: storm-core/src/jvm/org/apache/storm/pacemaker/PacemakerClientHandler.java ---
@@ -69,7 +69,7 @@ else if(evm instanceof HBMessage) {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent event) {
- LOG.error("Connection to pacemaker failed", event.getCause());
+ LOG.error("Connection to pacemaker failed. Trying to reconnect {}", event.getCause().getMessage());
--- End diff --
We don't want to log the stack here, only the message?
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r81219539
--- Diff: storm-core/src/jvm/org/apache/storm/Config.java ---
@@ -926,10 +926,10 @@
public static final String UI_HTTPS_NEED_CLIENT_AUTH = "ui.https.need.client.auth";
/**
- * The host that Pacemaker is running on.
+ * The list of servers that Pacemaker is running on.
*/
- @isString
- public static final String PACEMAKER_HOST = "pacemaker.host";
+ @isStringList
+ public static final String PACEMAKER_SERVERS = "pacemaker.servers";
--- End diff --
@HeartSaVioR Yep, the 1.x-branch PR will leave pacemaker.host in, but have it deprecated.
---
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 issue #1710: STORM-1546: Adding Read and Write Aggregations for Pacema...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on the issue:
https://github.com/apache/storm/pull/1710
@d2r @revans2 Any additional 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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by d2r <gi...@git.apache.org>.
Github user d2r commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80321800
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/PaceMakerStateStorage.java ---
@@ -160,13 +179,23 @@ public void set_worker_hb(String path, byte[] data, List<ACL> acls) {
int retry = maxRetries;
while (true) {
try {
+ HashSet<String> retSet = new HashSet<>();
+ int latest_time_secs = 0;
--- End diff --
`latest_time_sec` not used?
---
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 issue #1710: STORM-1546: Adding Read and Write Aggregations for Pacema...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on the issue:
https://github.com/apache/storm/pull/1710
+1
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum closed the pull request at:
https://github.com/apache/storm/pull/1710
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80717469
--- Diff: storm-core/src/jvm/org/apache/storm/Config.java ---
@@ -926,10 +926,10 @@
public static final String UI_HTTPS_NEED_CLIENT_AUTH = "ui.https.need.client.auth";
/**
- * The host that Pacemaker is running on.
+ * The list of servers that Pacemaker is running on.
*/
- @isString
- public static final String PACEMAKER_HOST = "pacemaker.host";
+ @isStringList
+ public static final String PACEMAKER_SERVERS = "pacemaker.servers";
--- End diff --
This is a breaking change that would be good to go back to 1.x as well. I am OK with this, because I don't think anyone is really using pacemaker, but I would like to hear others opinions on 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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80718589
--- Diff: storm-core/src/jvm/org/apache/storm/pacemaker/PacemakerClientHandler.java ---
@@ -69,7 +69,7 @@ else if(evm instanceof HBMessage) {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent event) {
- LOG.error("Connection to pacemaker failed", event.getCause());
+ LOG.error("Connection to pacemaker failed. Trying to reconnect {}", event.getCause().getMessage());
--- End diff --
If we know which exceptions we want to ignore, perhaps just IOExceptions, then we can explicitly blacklist them, and even log them under warn, instead of 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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by revans2 <gi...@git.apache.org>.
Github user revans2 commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80801551
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/PaceMakerStateStorage.java ---
@@ -159,10 +162,6 @@ public void set_worker_hb(String path, byte[] data, List<ACL> acls) {
ret = details;
}
}
- if(ret == null) {
- throw new HBExecutionException("Failed to get a response.");
- }
- LOG.debug("Successful get_worker_hb");
return ret;
--- End diff --
I think that is best, if all of them return something invalid or we get nothing back at all then we throw.
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by d2r <gi...@git.apache.org>.
Github user d2r commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80319682
--- Diff: storm-core/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java ---
@@ -0,0 +1,114 @@
+/**
+ * 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.pacemaker;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import org.apache.storm.Config;
+import org.apache.storm.generated.HBMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PacemakerClientPool {
+
+ private static final Logger LOG = LoggerFactory.getLogger(PacemakerClientPool.class);
+
+ private ConcurrentHashMap<String, PacemakerClient> clientForServer = new ConcurrentHashMap<>();
+ private ConcurrentLinkedQueue<String> servers;
+ private Map config;
+
+ public PacemakerClientPool(Map config) {
+ this.config = config;
+ List<String> serverList = (List<String>)config.get(Config.PACEMAKER_SERVERS);
+ if(serverList == null) {
+ serverList = new ArrayList<String>();
+ }
+ else {
+ serverList = new ArrayList<String>(serverList);
+ }
+ Collections.shuffle(serverList);
+ if(serverList != null) {
+ servers = new ConcurrentLinkedQueue<String>(serverList);
+ }
+ else {
+ servers = new ConcurrentLinkedQueue<String>();
+ }
+ }
+
+ public HBMessage send(HBMessage m) throws PacemakerConnectionException {
+ try {
+ return getWriteClient().send(m);
+ } catch (Exception e) {
+ rotateClients();
+ throw e;
+ }
+ }
+
+ public List<HBMessage> sendAll(HBMessage m) throws PacemakerConnectionException {
+ List<HBMessage> responses = new ArrayList<HBMessage>();
+ LOG.info("Using servers: {}", servers);
+ for(String s : servers) {
+ try {
+ HBMessage response = getClientForServer(s).send(m);
+ responses.add(response);
+ } catch (PacemakerConnectionException e) {
+ LOG.error("Failed to send message to Pacemaker " + s);
+ }
+ }
+ if(responses.size() == 0) {
+ throw new PacemakerConnectionException("Failed to connect to any Pacemaker.");
+ }
+ return responses;
+ }
+
+ public void close() {
+ for(PacemakerClient client : clientForServer.values()) {
+ client.shutdown();
+ client.close();
+ }
+ }
+
+ private void rotateClients() {
+ PacemakerClient c = getWriteClient();
+ String server = servers.peek();
+ // Servers should be rotated **BEFORE** the old client is removed from clientForServer
+ // or a race with getWriteClient() could cause it to be put back in the map.
+ servers.add(servers.remove());
+ clientForServer.remove(server);
+ c.shutdown();
+ c.close();
+ }
+
+ private PacemakerClient getWriteClient() {
+ return getClientForServer(servers.peek());
+ }
+
+ private PacemakerClient getClientForServer(String server) {
+ PacemakerClient client = clientForServer.get(server);
+ if(client == null) {
+ client = new PacemakerClient(config, server);
+ clientForServer.put(server, client);
+ }
+ return client;
+ }
+
--- End diff --
minor: extra line
---
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 issue #1710: STORM-1546: Adding Read and Write Aggregations for Pacema...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on the issue:
https://github.com/apache/storm/pull/1710
Addressed your comments. Not sure why they have not disappeared.
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by knusbaum <gi...@git.apache.org>.
Github user knusbaum commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80798282
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/PaceMakerStateStorage.java ---
@@ -159,10 +162,6 @@ public void set_worker_hb(String path, byte[] data, List<ACL> acls) {
ret = details;
}
}
- if(ret == null) {
- throw new HBExecutionException("Failed to get a response.");
- }
- LOG.debug("Successful get_worker_hb");
return ret;
--- End diff --
Yes, I wasn't quite sure how we wanted to handle this, NULL responses are valid, and right now the code treats mismatched responses as if they were NULL. We can distinguish between them if we want the behavior to differ whether the pacemakers returned NULL details or returned an invalid message. Throwing on null is not what we wanted, though, since it will cause Nimbus to crash when nimbus tries to read a topology's heartbeats before it has started sending them.
Maybe we want to add some logic so that if ALL pacemakers return an invalid response, then we throw?
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by d2r <gi...@git.apache.org>.
Github user d2r commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80314321
--- Diff: conf/defaults.yaml ---
@@ -276,6 +276,7 @@ resource.aware.scheduler.priority.strategy: "org.apache.storm.scheduler.resource
dev.zookeeper.path: "/tmp/dev-storm-zookeeper"
pacemaker.host: "localhost"
+pacemaker.servers: []
--- End diff --
Let's remove `pacemaker.host` if it is no longer used.
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by knusbaum <gi...@git.apache.org>.
GitHub user knusbaum reopened a pull request:
https://github.com/apache/storm/pull/1710
STORM-1546: Adding Read and Write Aggregations for Pacemaker to make it HA compatible
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/knusbaum/incubator-storm Pacemaker-HA
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/1710.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 #1710
----
commit c71314fbd2101ff081bee7aef3ecc434c3541044
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-06T21:04:07Z
Inspecting travis's maven cache
commit efa546862f7717ee1db222895ea14e6432666ab3
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-08T20:07:49Z
Working on Pacemaker HA
commit f46cfc15fa2069cabaa742ccfd2b5689fe5a3777
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-19T19:22:46Z
cleanup.
commit 0c002928e49f63070915e7a41f97f70b40336f23
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-19T19:47:04Z
Merge branch 'master' of http://git-wip-us.apache.org/repos/asf/storm
commit 6cd9661bde2ce4c93c5628aae2565e167b89aeb3
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T06:48:33Z
Fix ups.
commit 6a8be9f8c19bacb646c119c162c6cdba69cc66ff
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T07:03:46Z
Fix ups.
commit 523317b06abe87d7e745885c32c8a9c5dfbbeb0d
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T07:29:59Z
Cleaning up tests.
commit 410ef3a816fc6a961771abfbcac2d83512192040
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T07:31:54Z
Cleaning up.
commit 158228b7ad434287e7d1ec92aca9c5c08f144869
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T16:03:32Z
Kick travis
commit 6a657f16301306b5b0d8d4ab91f916c6e7757ffe
Author: Kyle Nusbaum <ky...@gmail.com>
Date: 2016-09-23T16:28:39Z
Fixing travis, adding license to new code.
----
---
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 #1710: STORM-1546: Adding Read and Write Aggregations for...
Posted by d2r <gi...@git.apache.org>.
Github user d2r commented on a diff in the pull request:
https://github.com/apache/storm/pull/1710#discussion_r80321976
--- Diff: storm-core/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java ---
@@ -18,48 +18,22 @@
package org.apache.storm.cluster;
import org.apache.storm.pacemaker.PacemakerClient;
+import org.apache.storm.pacemaker.PacemakerClientPool;
--- End diff --
Remove PacemakerClient import, as it's not used?
---
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.
---