storm-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bo...@apache.org
Subject storm git commit: Merge branch 'STORM-3201-1.x' of STORM-3201-1.x
Date Thu, 23 Aug 2018 02:28:39 GMT
Repository: storm
Updated Branches:
  refs/heads/1.1.x-branch ed79d96dd -> 242ad03c9


Merge branch 'STORM-3201-1.x' of STORM-3201-1.x

STORM-3201: Cleanup Lag


Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/242ad03c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/242ad03c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/242ad03c

Branch: refs/heads/1.1.x-branch
Commit: 242ad03c981974a33b1eabbf671d7bf41a0d151d
Parents: ed79d96
Author: Robert Evans <evans@yahoo-inc.com>
Authored: Wed Aug 22 12:48:00 2018 -0500
Committer: Robert Evans <evans@yahoo-inc.com>
Committed: Wed Aug 22 17:46:04 2018 -0500

----------------------------------------------------------------------
 .../apache/storm/utils/TopologySpoutLag.java    | 101 ++++++++++---------
 1 file changed, 52 insertions(+), 49 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/storm/blob/242ad03c/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 bb327ee..1217577 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java
@@ -1,4 +1,4 @@
-/**
+/*
  * 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
@@ -18,9 +18,14 @@
 
 package org.apache.storm.utils;
 
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import org.apache.storm.Config;
 import org.apache.storm.generated.ComponentCommon;
-import org.apache.storm.generated.ComponentObject;
 import org.apache.storm.generated.SpoutSpec;
 import org.apache.storm.generated.StormTopology;
 import org.json.simple.JSONValue;
@@ -28,70 +33,45 @@ import org.json.simple.parser.ParseException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
 public class TopologySpoutLag {
     private static final String SPOUT_ID = "spoutId";
-    private static final String SPOUT_TYPE= "spoutType";
+    private static final String SPOUT_TYPE = "spoutType";
     private static final String SPOUT_LAG_RESULT = "spoutLagResult";
     private static final String ERROR_INFO = "errorInfo";
+    private static final String CONFIG_KEY_PREFIX = "config.";
+    private static final String TOPICS_CONFIG = CONFIG_KEY_PREFIX + "topics";
+    private static final String GROUPID_CONFIG = CONFIG_KEY_PREFIX + "groupid";
+    private static final String BOOTSTRAP_CONFIG = CONFIG_KEY_PREFIX + "bootstrap.servers";
+    private static final String LEADERS_CONFIG = CONFIG_KEY_PREFIX + "leaders";
+    private static final String ZKROOT_CONFIG = CONFIG_KEY_PREFIX + "zkRoot";
     private final static Logger logger = LoggerFactory.getLogger(TopologySpoutLag.class);
 
-    public static Map<String, Map<String, Object>> lag (StormTopology stormTopology,
Map topologyConf) {
+    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()) {
             try {
                 SpoutSpec spoutSpec = spout.getValue();
-                ComponentObject componentObject = spoutSpec.get_spout_object();
-                // FIXME: yes it's a trick so we might be better to find alternative way...
-                className = getClassNameFromComponentObject(componentObject);
-                logger.debug("spout classname: {}", className);
-                if (className.endsWith("storm.kafka.spout.KafkaSpout")) {
-                    result.put(spout.getKey(), getLagResultForNewKafkaSpout(spout.getKey(),
spoutSpec, topologyConf));
-                } else if (className.endsWith("storm.kafka.KafkaSpout")) {
-                    result.put(spout.getKey(), getLagResultForOldKafkaSpout(spout.getKey(),
spoutSpec, topologyConf));
-                }
+                addLagResultForKafkaSpout(result, spout.getKey(), spoutSpec, topologyConf);
             } catch (Exception e) {
-                logger.warn("Exception thrown while getting lag for spout id: " + spout.getKey()
+ " and spout class: " + className);
+                logger.warn("Exception thrown while getting lag for spout id: " + spout.getKey());
                 logger.warn("Exception message:" + e.getMessage(), e);
             }
         }
         return result;
     }
 
-    private static String getClassNameFromComponentObject(ComponentObject componentObject)
{
-        try {
-            Object object = Utils.getSetComponentObject(componentObject);
-            return object.getClass().getCanonicalName();
-        } catch (RuntimeException e) {
-
-            if (e.getCause() instanceof ClassNotFoundException) {
-                return e.getCause().getMessage().trim();
-            }
-
-            throw e;
-        }
-    }
-
     private static List<String> getCommandLineOptionsForNewKafkaSpout (Map<String,
Object> jsonConf) {
         logger.debug("json configuration: {}", jsonConf);
 
         List<String> commands = new ArrayList<>();
-        String configKeyPrefix = "config.";
         commands.add("-t");
-        commands.add((String)jsonConf.get(configKeyPrefix + "topics"));
+        commands.add((String) jsonConf.get(TOPICS_CONFIG));
         commands.add("-g");
-        commands.add((String)jsonConf.get(configKeyPrefix + "groupid"));
+        commands.add((String) jsonConf.get(GROUPID_CONFIG));
         commands.add("-b");
-        commands.add((String)jsonConf.get(configKeyPrefix + "bootstrap.servers"));
-        String securityProtocol = (String)jsonConf.get(configKeyPrefix + "security.protocol");
+        commands.add((String) jsonConf.get(BOOTSTRAP_CONFIG));
+        String securityProtocol = (String) jsonConf.get(CONFIG_KEY_PREFIX + "security.protocol");
         if (securityProtocol != null && !securityProtocol.isEmpty()) {
             commands.add("-s");
             commands.add(securityProtocol);
@@ -103,13 +83,12 @@ public class TopologySpoutLag {
         logger.debug("json configuration: {}", jsonConf);
 
         List<String> commands = new ArrayList<>();
-        String configKeyPrefix = "config.";
         commands.add("-o");
         commands.add("-t");
-        commands.add((String)jsonConf.get(configKeyPrefix + "topics"));
+        commands.add((String)jsonConf.get(TOPICS_CONFIG));
         commands.add("-n");
-        commands.add((String)jsonConf.get(configKeyPrefix + "zkRoot"));
-        String zkServers = (String)jsonConf.get(configKeyPrefix + "zkServers");
+        commands.add((String)jsonConf.get(ZKROOT_CONFIG));
+        String zkServers = (String)jsonConf.get(CONFIG_KEY_PREFIX + "zkServers");
         if (zkServers == null || zkServers.isEmpty()) {
             StringBuilder zkServersBuilder = new StringBuilder();
             Integer zkPort = ((Number) topologyConf.get(Config.STORM_ZOOKEEPER_PORT)).intValue();
@@ -120,14 +99,14 @@ public class TopologySpoutLag {
         }
         commands.add("-z");
         commands.add(zkServers);
-        if (jsonConf.get(configKeyPrefix + "leaders") != null) {
+        if (jsonConf.get(LEADERS_CONFIG) != null) {
             commands.add("-p");
-            commands.add((String)jsonConf.get(configKeyPrefix + "partitions"));
+            commands.add((String)jsonConf.get(CONFIG_KEY_PREFIX + "partitions"));
             commands.add("-l");
-            commands.add((String)jsonConf.get(configKeyPrefix + "leaders"));
+            commands.add((String)jsonConf.get(LEADERS_CONFIG));
         } else {
             commands.add("-r");
-            commands.add((String)jsonConf.get(configKeyPrefix + "zkNodeBrokers"));
+            commands.add((String)jsonConf.get(CONFIG_KEY_PREFIX + "zkNodeBrokers"));
             Boolean isWildCard = (Boolean) topologyConf.get("kafka.topic.wildcard.match");
             if (isWildCard != null && isWildCard.booleanValue()) {
                 commands.add("-w");
@@ -136,6 +115,30 @@ public class TopologySpoutLag {
         return commands;
     }
 
+    private static void addLagResultForKafkaSpout(Map<String, Map<String, Object>>
finalResult, String spoutId, SpoutSpec spoutSpec,
+                                                  Map topologyConf) throws IOException {
+        ComponentCommon componentCommon = spoutSpec.get_common();
+        String json = componentCommon.get_json_conf();
+        if (json != null && !json.isEmpty()) {
+            Map<String, Object> jsonMap = null;
+            try {
+                jsonMap = (Map<String, Object>) JSONValue.parseWithException(json);
+            } catch (ParseException e) {
+                throw new IOException(e);
+            }
+
+            if (jsonMap.containsKey(TOPICS_CONFIG)
+                && jsonMap.containsKey(GROUPID_CONFIG)
+                && jsonMap.containsKey(BOOTSTRAP_CONFIG)) {
+                finalResult.put(spoutId, getLagResultForNewKafkaSpout(spoutId, spoutSpec,
topologyConf));
+            } else if (jsonMap.containsKey(TOPICS_CONFIG)
+                && jsonMap.containsKey(ZKROOT_CONFIG)) {
+                //Probably the old spout
+                finalResult.put(spoutId, getLagResultForOldKafkaSpout(spoutId, spoutSpec,
topologyConf));
+            }
+        }
+    }
+
     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();


Mime
View raw message