You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2016/07/11 01:08:41 UTC
[1/3] storm git commit: STORM-1950 Change response json of "Topology
Lag" REST API to keyed by spoutId, topic, partition
Repository: storm
Updated Branches:
refs/heads/master 122bbefa2 -> d7ba59d8a
STORM-1950 Change response json of "Topology Lag" REST API to keyed by spoutId, topic, partition
* bin/storm-kafka-monitor: guard classpath to only uses storm-kafka-monitor*.jar
* KafkaOffsetLagUtil: when succeed to get, output is keyed by topic and partition
* TopologySpoutLag: modify JSON response keyed by spoutId
* reflect changes on topology page
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/ff595145
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/ff595145
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/ff595145
Branch: refs/heads/master
Commit: ff595145e30c57271eb8f6ab2f1f31805170f672
Parents: b96fef6
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Fri Jul 8 15:07:05 2016 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Fri Jul 8 16:00:03 2016 +0900
----------------------------------------------------------------------
bin/storm-kafka-monitor | 2 +-
.../storm/kafka/monitor/KafkaOffsetLagUtil.java | 23 +++++-
.../kafka/monitor/KafkaPartitionOffsetLag.java | 54 +++++++++++++
.../apache/storm/utils/TopologySpoutLag.java | 32 ++++++--
.../templates/topology-page-template.html | 3 +-
storm-core/src/ui/public/topology.html | 83 +++++++++++---------
6 files changed, 150 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/ff595145/bin/storm-kafka-monitor
----------------------------------------------------------------------
diff --git a/bin/storm-kafka-monitor b/bin/storm-kafka-monitor
index bfe50b7..a47a27d 100755
--- a/bin/storm-kafka-monitor
+++ b/bin/storm-kafka-monitor
@@ -40,4 +40,4 @@ else
JAVA="$JAVA_HOME/bin/java"
fi
-exec "$JAVA" -cp "$STORM_BASE_DIR/toollib/*" org.apache.storm.kafka.monitor.KafkaOffsetLagUtil "$@"
+exec "$JAVA" -cp "$STORM_BASE_DIR/toollib/storm-kafka-monitor*.jar" org.apache.storm.kafka.monitor.KafkaOffsetLagUtil "$@"
http://git-wip-us.apache.org/repos/asf/storm/blob/ff595145/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java
index a4436f9..f63d932 100644
--- a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java
+++ b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java
@@ -134,13 +134,34 @@ public class KafkaOffsetLagUtil {
commandLine.getOptionValue(OPTION_BOOTSTRAP_BROKERS_LONG), commandLine.getOptionValue(OPTION_GROUP_ID_LONG));
results = getOffsetLags(newKafkaSpoutOffsetQuery);
}
- System.out.print(JSONValue.toJSONString(results));
+
+ Map<String, Map<Integer, KafkaPartitionOffsetLag>> keyedResult = keyByTopicAndPartition(results);
+ System.out.print(JSONValue.toJSONString(keyedResult));
} catch (Exception ex) {
System.out.print("Unable to get offset lags for kafka. Reason: ");
ex.printStackTrace(System.out);
}
}
+ private static Map<String, Map<Integer, KafkaPartitionOffsetLag>> keyByTopicAndPartition(
+ List<KafkaOffsetLagResult> results) {
+ Map<String, Map<Integer, KafkaPartitionOffsetLag>> resultKeyedByTopic = new HashMap<>();
+
+ for (KafkaOffsetLagResult result : results) {
+ String topic = result.getTopic();
+ Map<Integer, KafkaPartitionOffsetLag> topicResultKeyedByPartition = resultKeyedByTopic.get(topic);
+ if (topicResultKeyedByPartition == null) {
+ topicResultKeyedByPartition = new HashMap<>();
+ resultKeyedByTopic.put(topic, topicResultKeyedByPartition);
+ }
+
+ topicResultKeyedByPartition.put(result.getPartition(),
+ new KafkaPartitionOffsetLag(result.getConsumerCommittedOffset(), result.getLogHeadOffset()));
+ }
+
+ return resultKeyedByTopic;
+ }
+
private static void printUsageAndExit (Options options, String message) {
System.out.println(message);
HelpFormatter formatter = new HelpFormatter();
http://git-wip-us.apache.org/repos/asf/storm/blob/ff595145/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaPartitionOffsetLag.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaPartitionOffsetLag.java b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaPartitionOffsetLag.java
new file mode 100644
index 0000000..53284ae
--- /dev/null
+++ b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaPartitionOffsetLag.java
@@ -0,0 +1,54 @@
+package org.apache.storm.kafka.monitor;
+
+public class KafkaPartitionOffsetLag {
+ private long consumerCommittedOffset;
+ private long logHeadOffset;
+ private long lag;
+
+ public KafkaPartitionOffsetLag(long consumerCommittedOffset, long logHeadOffset) {
+ this.consumerCommittedOffset = consumerCommittedOffset;
+ this.logHeadOffset = logHeadOffset;
+ this.lag = logHeadOffset - consumerCommittedOffset;
+ }
+
+ public long getConsumerCommittedOffset() {
+ return consumerCommittedOffset;
+ }
+
+ public long getLogHeadOffset() {
+ return logHeadOffset;
+ }
+
+ public long getLag() {
+ return lag;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof KafkaPartitionOffsetLag)) return false;
+
+ KafkaPartitionOffsetLag that = (KafkaPartitionOffsetLag) o;
+
+ if (getConsumerCommittedOffset() != that.getConsumerCommittedOffset()) return false;
+ if (getLogHeadOffset() != that.getLogHeadOffset()) return false;
+ return getLag() == that.getLag();
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = (int) (getConsumerCommittedOffset() ^ (getConsumerCommittedOffset() >>> 32));
+ result = 31 * result + (int) (getLogHeadOffset() ^ (getLogHeadOffset() >>> 32));
+ result = 31 * result + (int) (getLag() ^ (getLag() >>> 32));
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ // JSONAware not working for nested element on Map so write JSON format from here
+ return "{\"consumerCommittedOffset\": " + consumerCommittedOffset + ", " +
+ "\"logHeadOffset\": " + logHeadOffset + ", " +
+ "\"lag\": " + lag + "}";
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/ff595145/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
index 0d3d2f1..6a88f07 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
@@ -24,6 +24,7 @@ import org.apache.storm.generated.ComponentObject;
import org.apache.storm.generated.SpoutSpec;
import org.apache.storm.generated.StormTopology;
import org.json.simple.JSONValue;
+import org.json.simple.parser.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,10 +39,11 @@ public class TopologySpoutLag {
private static final String SPOUT_ID = "spoutId";
private static final String SPOUT_TYPE= "spoutType";
private static final String SPOUT_LAG_RESULT = "spoutLagResult";
+ private static final String ERROR_INFO = "errorInfo";
private final static Logger logger = LoggerFactory.getLogger(TopologySpoutLag.class);
- public static List<Map<String, Object>> lag (StormTopology stormTopology, Map topologyConf) {
- List<Map<String, Object>> result = new ArrayList<>();
+ public static Map<String, Map<String, Object>> lag (StormTopology stormTopology, Map topologyConf) {
+ Map<String, Map<String, Object>> result = new HashMap<>();
Map<String, SpoutSpec> spouts = stormTopology.get_spouts();
String className = null;
for (Map.Entry<String, SpoutSpec> spout: spouts.entrySet()) {
@@ -52,9 +54,9 @@ public class TopologySpoutLag {
className = getClassNameFromComponentObject(componentObject);
logger.debug("spout classname: {}", className);
if (className.endsWith("storm.kafka.spout.KafkaSpout")) {
- result.add(getLagResultForNewKafkaSpout(spout.getKey(), spoutSpec, topologyConf));
+ result.put(spout.getKey(), getLagResultForNewKafkaSpout(spout.getKey(), spoutSpec, topologyConf));
} else if (className.endsWith("storm.kafka.KafkaSpout")) {
- result.add(getLagResultForOldKafkaSpout(spout.getKey(), spoutSpec, topologyConf));
+ result.put(spout.getKey(), getLagResultForOldKafkaSpout(spout.getKey(), spoutSpec, topologyConf));
}
} catch (Exception e) {
logger.warn("Exception thrown while getting lag for spout id: " + spout.getKey() + " and spout class: " + className);
@@ -132,7 +134,8 @@ public class TopologySpoutLag {
private static Map<String, Object> getLagResultForKafka (String spoutId, SpoutSpec spoutSpec, Map topologyConf, boolean old) throws IOException {
ComponentCommon componentCommon = spoutSpec.get_common();
String json = componentCommon.get_json_conf();
- String result = "Offset lags for kafka not supported for older versions. Please update kafka spout to latest version.";
+ Map<String, Object> result = null;
+ String errorMsg = "Offset lags for kafka not supported for older versions. Please update kafka spout to latest version.";
if (json != null && !json.isEmpty()) {
List<String> commands = new ArrayList<>();
String stormHomeDir = System.getenv("STORM_BASE_DIR");
@@ -147,13 +150,28 @@ public class TopologySpoutLag {
// if commands contains one or more null value, spout is compiled with lower version of storm-kafka / storm-kafka-client
if (!commands.contains(null)) {
- result = ShellUtils.execCommand(commands.toArray(new String[0]));
+ String resultFromMonitor = ShellUtils.execCommand(commands.toArray(new String[0]));
+
+ try {
+ result = (Map<String, Object>) JSONValue.parseWithException(resultFromMonitor);
+ } catch (ParseException e) {
+ logger.debug("JSON parsing failed, assuming message as error message: {}", resultFromMonitor);
+ // json parsing fail -> error received
+ errorMsg = resultFromMonitor;
+ }
}
}
+
Map<String, Object> kafkaSpoutLagInfo = new HashMap<>();
kafkaSpoutLagInfo.put(SPOUT_ID, spoutId);
kafkaSpoutLagInfo.put(SPOUT_TYPE, "KAFKA");
- kafkaSpoutLagInfo.put(SPOUT_LAG_RESULT, result);
+
+ if (result != null) {
+ kafkaSpoutLagInfo.put(SPOUT_LAG_RESULT, result);
+ } else {
+ kafkaSpoutLagInfo.put(ERROR_INFO, errorMsg);
+ }
+
return kafkaSpoutLagInfo;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/ff595145/storm-core/src/ui/public/templates/topology-page-template.html
----------------------------------------------------------------------
diff --git a/storm-core/src/ui/public/templates/topology-page-template.html b/storm-core/src/ui/public/templates/topology-page-template.html
index fbdf644..da1796c 100644
--- a/storm-core/src/ui/public/templates/topology-page-template.html
+++ b/storm-core/src/ui/public/templates/topology-page-template.html
@@ -283,7 +283,7 @@
<tr>
<td>{{spoutId}}</td>
<td>{{spoutType}}</td>
- <td>{{spoutLagResult}}</td>
+ <td>{{errorInfo}}</td>
</tr>
{{/spoutsLagErrorResults}}
</tbody>
@@ -603,3 +603,4 @@
</table>
</div>
</script>
+
http://git-wip-us.apache.org/repos/asf/storm/blob/ff595145/storm-core/src/ui/public/topology.html
----------------------------------------------------------------------
diff --git a/storm-core/src/ui/public/topology.html b/storm-core/src/ui/public/topology.html
index 7d3527c..ffdc7c5 100644
--- a/storm-core/src/ui/public/topology.html
+++ b/storm-core/src/ui/public/topology.html
@@ -392,44 +392,52 @@ $(document).ready(function() {
var lagUrl = "/api/v1/topology/"+topologyId+"/lag";
$.getJSON(lagUrl,function(lagResponse,status,jqXHR) {
- if (lagResponse !== null && lagResponse !== undefined && lagResponse instanceof Array && lagResponse.length > 0) {
- var kafkaSpoutsLagTemplate = $(template).filter("#topology-kafka-spouts-lag-template").html();
- var spoutsErrorTemplate = $(template).filter("#topology-spouts-lag-error-template").html();
-
- var kafkaSpoutLags = lagResponse.filter(function(ele) {return ele.spoutType === "KAFKA";});
- var isJson = function (input) {
- try {
- JSON.parse(input);
- } catch (e) {
- return false;
- }
- return true;
- };
- var kafkaSpoutsValidResults = kafkaSpoutLags.filter(function (ele) {return isJson(ele.spoutLagResult);});
- var kafkaSpoutsErrorResults = kafkaSpoutLags.filter(function (ele) {return !isJson(ele.spoutLagResult);});
- var data = {};
- if (kafkaSpoutsValidResults.length > 0) {
- data.kafkaSpoutsLagResults = [];
- kafkaSpoutsValidResults.forEach(function(ele) {
- var spoutLagResult = JSON.parse(ele.spoutLagResult);
- spoutLagResult.forEach(function(ele2) {
- data.kafkaSpoutsLagResults.push({
- id: ele.spoutId,
- topic: ele2.topic,
- partition: ele2.partition,
- logHeadOffset: ele2.logHeadOffset,
- consumerCommittedOffset: ele2.consumerCommittedOffset,
- lag: ele2.lag
- });
- });
- });
- topologySpoutsLag.append(Mustache.render(kafkaSpoutsLagTemplate,data));
+ if (lagResponse !== null && lagResponse !== undefined) {
+ var kafkaSpoutsLagTemplate = $(template).filter("#topology-kafka-spouts-lag-template").html();
+ var spoutsErrorTemplate = $(template).filter("#topology-spouts-lag-error-template").html();
+
+ var data = {};
+ data.kafkaSpoutsLagResults = [];
+ data.spoutsLagErrorResults = [];
+ for (var spoutId in lagResponse) {
+ var spout = lagResponse[spoutId];
+ var spoutType = spout.spoutType;
+ if (spoutType !== "KAFKA") {
+ continue;
+ }
+ var spoutLagResult = spout.spoutLagResult;
+ var errorInfo = spout.errorInfo;
+ if (spoutLagResult !== undefined) {
+ for (var topicName in spoutLagResult) {
+ var topicLagResult = spoutLagResult[topicName];
+ for (var partitionId in topicLagResult) {
+ var partitionLagResult = topicLagResult[partitionId];
+ data.kafkaSpoutsLagResults.push({
+ id: spoutId,
+ topic: topicName,
+ partition: partitionId,
+ logHeadOffset: partitionLagResult.logHeadOffset,
+ consumerCommittedOffset: partitionLagResult.consumerCommittedOffset,
+ lag: partitionLagResult.lag
+ });
+ }
+ }
+ } else if (errorInfo !== undefined) {
+ data.spoutsLagErrorResults.push({
+ spoutId: spoutId,
+ spoutType: spout.spoutType,
+ errorInfo: errorInfo
+ });
+ }
+ }
+
+ if (data.kafkaSpoutsLagResults.length > 0) {
+ topologySpoutsLag.append(Mustache.render(kafkaSpoutsLagTemplate, data));
+ }
+ if (data.spoutsLagErrorResults.length > 0) {
+ topologySpoutsLag.append(Mustache.render(spoutsErrorTemplate, data));
+ }
}
- if (kafkaSpoutsErrorResults.length > 0) {
- data.spoutsLagErrorResults = kafkaSpoutsErrorResults;
- topologySpoutsLag.append(Mustache.render(spoutsErrorTemplate,data));
- }
- }
});
}});
});
@@ -437,3 +445,4 @@ $(document).ready(function() {
});
</script>
</html>
+
[2/3] storm git commit: Merge branch 'STORM-1950' of
https://github.com/HeartSaVioR/storm
Posted by sr...@apache.org.
Merge branch 'STORM-1950' of https://github.com/HeartSaVioR/storm
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/465d58c2
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/465d58c2
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/465d58c2
Branch: refs/heads/master
Commit: 465d58c26438b52161f627eddecf8df5f30ca659
Parents: 122bbef ff59514
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Sun Jul 10 14:41:05 2016 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Sun Jul 10 14:41:05 2016 -0700
----------------------------------------------------------------------
bin/storm-kafka-monitor | 2 +-
.../storm/kafka/monitor/KafkaOffsetLagUtil.java | 23 +++++-
.../kafka/monitor/KafkaPartitionOffsetLag.java | 54 +++++++++++++
.../apache/storm/utils/TopologySpoutLag.java | 32 ++++++--
.../templates/topology-page-template.html | 3 +-
storm-core/src/ui/public/topology.html | 83 +++++++++++---------
6 files changed, 150 insertions(+), 47 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/465d58c2/storm-core/src/ui/public/topology.html
----------------------------------------------------------------------
[3/3] storm git commit: Added STORM-1950 to CHANGELOG.
Posted by sr...@apache.org.
Added STORM-1950 to CHANGELOG.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d7ba59d8
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d7ba59d8
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d7ba59d8
Branch: refs/heads/master
Commit: d7ba59d8ad4e89e21f4e15cea8a457f651fa55a8
Parents: 465d58c
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Sun Jul 10 14:41:43 2016 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Sun Jul 10 14:41:43 2016 -0700
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d7ba59d8/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9aa1f32..cd99415 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -116,6 +116,7 @@
* STORM-1769: Added a test to check local nimbus with notifier plugin
## 1.1.0
+ * STORM-1950: Change response json of "Topology Lag" REST API to keyed by spoutId, topic, partition.
* STORM-1833: Simple equi-join in storm-sql standalone mode
* STORM-1866: Update Resource Aware Scheduler Documentation
* STORM-1930: Kafka New Client API - Support for Topic Wildcards