You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by lo...@apache.org on 2016/03/30 09:17:17 UTC
[03/19] storm git commit: update code based on abhishekagarwal87
update code based on abhishekagarwal87
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/399476a9
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/399476a9
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/399476a9
Branch: refs/heads/master
Commit: 399476a97b8ce68579dddc303867d91fda9abdb7
Parents: f22673a
Author: xiaojian.fxj <xi...@alibaba-inc.com>
Authored: Wed Mar 9 20:45:05 2016 +0800
Committer: xiaojian.fxj <xi...@alibaba-inc.com>
Committed: Wed Mar 9 20:45:05 2016 +0800
----------------------------------------------------------------------
.../org/apache/storm/pacemaker/Pacemaker.java | 26 +++++++++-----------
1 file changed, 12 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/399476a9/storm-core/src/jvm/org/apache/storm/pacemaker/Pacemaker.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/pacemaker/Pacemaker.java b/storm-core/src/jvm/org/apache/storm/pacemaker/Pacemaker.java
index ec22a6f..3b15906 100644
--- a/storm-core/src/jvm/org/apache/storm/pacemaker/Pacemaker.java
+++ b/storm-core/src/jvm/org/apache/storm/pacemaker/Pacemaker.java
@@ -34,7 +34,7 @@ public class Pacemaker implements IServerMessageHandler {
private static final Logger LOG = LoggerFactory.getLogger(Pacemaker.class);
- private Map heartbeats;
+ private Map<String, byte[]> heartbeats;
private PacemakerStats pacemakerStats;
private Map conf;
private final long sleepSeconds = 60;
@@ -137,18 +137,17 @@ public class Pacemaker implements IServerMessageHandler {
LOG.debug("List all nodes for path {}", path);
if (authenticated) {
Set<String> pulseIds = new HashSet<>();
- for (Object key : heartbeats.keySet()) {
- String k = (String) key;
- String[] replaceStr = k.replaceFirst(path, "").split("/");
- String trimmmed = null;
+ for (String key : heartbeats.keySet()) {
+ String[] replaceStr = key.replaceFirst(path, "").split("/");
+ String trimmed = null;
for (String str : replaceStr) {
if (!str.equals("")) {
- trimmmed = str;
+ trimmed = str;
break;
}
}
- if (trimmmed != null && k.indexOf(path) == 0) {
- pulseIds.add(trimmmed);
+ if (trimmed != null && key.indexOf(path) == 0) {
+ pulseIds.add(trimmed);
}
}
HBMessageData hbMessageData = HBMessageData.nodes(new HBNodes(new ArrayList(pulseIds)));
@@ -160,7 +159,7 @@ public class Pacemaker implements IServerMessageHandler {
private HBMessage getPulse(String path, boolean authenticated) {
if (authenticated) {
- byte[] details = (byte[]) heartbeats.get(path);
+ byte[] details = heartbeats.get(path);
LOG.debug("Getting Pulse for path [ {} ]...data [ {} ].", path, details);
pacemakerStats.getPulseCount.incrementAndGet();
if (details != null) {
@@ -177,9 +176,9 @@ public class Pacemaker implements IServerMessageHandler {
private HBMessage deletePath(String path) {
String prefix = path.endsWith("/") ? path : (path + "/");
- for (Object key : heartbeats.keySet()) {
- if (((String) key).indexOf(prefix) == 0)
- deletePulseId((String) key);
+ for (String key : heartbeats.keySet()) {
+ if (key.indexOf(prefix) == 0)
+ deletePulseId(key);
}
return new HBMessage(HBServerMessageType.DELETE_PATH_RESPONSE, null);
}
@@ -203,11 +202,10 @@ public class Pacemaker implements IServerMessageHandler {
}
private void updateAverageHbSize(int size) {
- int newValue = size;
while (true) {
int oldValue = pacemakerStats.averageHeartbeatSize.get();
int count = pacemakerStats.sendPulseCount.get();
- newValue = ((count * oldValue) + newValue) / (count + 1);
+ int newValue = ((count * oldValue) + size) / (count + 1);
if (!pacemakerStats.averageHeartbeatSize.compareAndSet(oldValue, newValue))
continue;
break;