From b5fe05a3115b8077e0105f70eee89edc3635eb2a Mon Sep 17 00:00:00 2001
From: Satish Duggana <sduggana@hortonworks.com>
Date: Wed, 25 May 2016 11:51:12 +0530
Subject: [PATCH 1/2] STORM-1864 StormSubmitter should throw respective
 exceptions and log respective errors forregistered submitter hook invocation.

---
 .../hive/trident/TridentHiveTopology.java     | 29 ++++++++-----
 .../jvm/org/apache/storm/StormSubmitter.java  | 43 +++++++++++++------
 .../storm/hooks/SubmitterHookException.java   | 41 ++++++++++++++++++
 3 files changed, 90 insertions(+), 23 deletions(-)
 create mode 100644 storm-core/src/jvm/org/apache/storm/hooks/SubmitterHookException.java

diff --git a/external/storm-hive/src/test/java/org/apache/storm/hive/trident/TridentHiveTopology.java b/external/storm-hive/src/test/java/org/apache/storm/hive/trident/TridentHiveTopology.java
index d6c1d65d145..86a35e67047 100644
--- a/external/storm-hive/src/test/java/org/apache/storm/hive/trident/TridentHiveTopology.java
+++ b/external/storm-hive/src/test/java/org/apache/storm/hive/trident/TridentHiveTopology.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
@@ -26,6 +26,7 @@
 import org.apache.storm.LocalCluster;
 import org.apache.storm.StormSubmitter;
 import org.apache.storm.generated.StormTopology;
+import org.apache.storm.hooks.SubmitterHookException;
 import org.apache.storm.tuple.Fields;
 import org.apache.storm.tuple.Values;
 import org.apache.storm.task.TopologyContext;
@@ -41,9 +42,13 @@
 import org.apache.storm.trident.TridentState;
 import org.apache.storm.trident.TridentTopology;
 import org.apache.storm.trident.state.StateFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 
 public class TridentHiveTopology {
+    private static final Logger LOG = LoggerFactory.getLogger(TridentHiveTopology.class);
+
     public static StormTopology buildTopology(String metaStoreURI, String dbName, String tblName, Object keytab, Object principal) {
         int batchSize = 100;
         FixedBatchSpout spout = new FixedBatchSpout(batchSize);
@@ -93,28 +98,32 @@ public static void main(String[] args) {
         if(args.length == 3) {
             LocalCluster cluster = new LocalCluster();
             cluster.submitTopology("tridentHiveTopology", conf, buildTopology(metaStoreURI, dbName, tblName,null,null));
-            System.out.println("waiting for 60 seconds");
+            LOG.info("waiting for 60 seconds");
             waitForSeconds(60);
-            System.out.println("killing topology");
+            LOG.info("killing topology");
             cluster.killTopology("tridenHiveTopology");
-            System.out.println("cluster shutdown");
+            LOG.info("cluster shutdown");
             cluster.shutdown();
-            System.out.println("cluster shutdown");
+            LOG.info("cluster shutdown");
             System.exit(0);
         } else if(args.length == 4) {
             try {
                 StormSubmitter.submitTopology(args[3], conf, buildTopology(metaStoreURI, dbName, tblName,null,null));
-            } catch(Exception e) {
-                System.out.println("Failed to submit topology "+e);
+            } catch(SubmitterHookException e) {
+                LOG.warn("Topology is submitted but invoking ISubmitterHook failed", e);
+            } catch (Exception e) {
+                LOG.warn("Failed to submit topology ", e);
             }
         } else if (args.length == 6) {
             try {
                 StormSubmitter.submitTopology(args[3], conf, buildTopology(metaStoreURI, dbName, tblName,args[4],args[5]));
-            } catch(Exception e) {
-                System.out.println("Failed to submit topology "+e);
+            } catch(SubmitterHookException e) {
+                LOG.warn("Topology is submitted but invoking ISubmitterHook failed", e);
+            } catch (Exception e) {
+                LOG.warn("Failed to submit topology ", e);
             }
         } else {
-            System.out.println("Usage: TridentHiveTopology metastoreURI dbName tableName [topologyNamey]");
+            LOG.info("Usage: TridentHiveTopology metastoreURI dbName tableName [topologyNamey]");
         }
     }
 
diff --git a/storm-core/src/jvm/org/apache/storm/StormSubmitter.java b/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
index c199f97944a..077c5779416 100644
--- a/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
+++ b/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
@@ -24,6 +24,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import org.apache.storm.hooks.SubmitterHookException;
 import org.apache.storm.scheduler.resource.ResourceUtils;
 import org.apache.storm.validation.ConfigValidation;
 import org.apache.commons.lang.StringUtils;
@@ -149,6 +150,7 @@ public static void pushCredentials(String name, Map stormConf, Map<String, Strin
      * @throws AlreadyAliveException if a topology with this name is already running
      * @throws InvalidTopologyException if an invalid topology was submitted
      * @throws AuthorizationException if authorization is failed
+     * @thorws SubmitterHookException if any Exception occurs during initialization or invocation of registered {@link ISubmitterHook}
      */
     public static void submitTopology(String name, Map stormConf, StormTopology topology)
             throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
@@ -166,6 +168,7 @@ public static void submitTopology(String name, Map stormConf, StormTopology topo
      * @throws AlreadyAliveException if a topology with this name is already running
      * @throws InvalidTopologyException if an invalid topology was submitted
      * @throws AuthorizationException if authorization is failed
+     * @thorws SubmitterHookException if any Exception occurs during initialization or invocation of registered {@link ISubmitterHook}
      */
     public static void submitTopology(String name, Map stormConf, StormTopology topology, SubmitOptions opts)
             throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
@@ -186,6 +189,7 @@ public static void submitTopology(String name, Map stormConf, StormTopology topo
      * @throws InvalidTopologyException
      * @throws AuthorizationException
      * @throws IllegalArgumentException thrown if configs will yield an unschedulable topology. validateConfs validates confs
+     * @thorws SubmitterHookException if any Exception occurs during initialization or invocation of registered {@link ISubmitterHook}
      */
     public static void submitTopologyAs(String name, Map stormConf, StormTopology topology, SubmitOptions opts, ProgressListener progressListener, String asUser)
             throws AlreadyAliveException, InvalidTopologyException, AuthorizationException, IllegalArgumentException {
@@ -223,32 +227,30 @@ public static void submitTopologyAs(String name, Map stormConf, StormTopology to
                     // this is for backwards compatibility
                     localNimbus.submitTopology(name, stormConf, topology);
                 }
+                LOG.info("Finished submitting topology: " +  name);
             } else {
                 String serConf = JSONValue.toJSONString(stormConf);
-                NimbusClient client = NimbusClient.getConfiguredClientAs(conf, asUser);
                 if(topologyNameExists(conf, name, asUser)) {
                     throw new RuntimeException("Topology with name `" + name + "` already exists on cluster");
                 }
                 String jar = submitJarAs(conf, System.getProperty("storm.jar"), progressListener, asUser);
-                try {
-                    LOG.info("Submitting topology " +  name + " in distributed mode with conf " + serConf);
-                    if(opts!=null) {
+                try (NimbusClient client = NimbusClient.getConfiguredClientAs(conf, asUser)) {
+                    LOG.info("Submitting topology " + name + " in distributed mode with conf " + serConf);
+                    if (opts != null) {
                         client.getClient().submitTopologyWithOpts(name, jar, serConf, topology, opts);
                     } else {
                         // this is for backwards compatibility
                         client.getClient().submitTopology(name, jar, serConf, topology);
                     }
-                } catch(InvalidTopologyException e) {
-                    LOG.warn("Topology submission exception: "+e.get_msg());
+                    LOG.info("Finished submitting topology: " + name);
+                } catch (InvalidTopologyException e) {
+                    LOG.warn("Topology submission exception: " + e.get_msg());
                     throw e;
-                } catch(AlreadyAliveException e) {
+                } catch (AlreadyAliveException e) {
                     LOG.warn("Topology already alive exception", e);
                     throw e;
-                } finally {
-                    client.close();
                 }
             }
-            LOG.info("Finished submitting topology: " +  name);
         } catch(TException e) {
             throw new RuntimeException(e);
         }
@@ -256,15 +258,29 @@ public static void submitTopologyAs(String name, Map stormConf, StormTopology to
 
     }
 
+    /**
+     *
+     * @param name
+     * @param asUser
+     * @param stormConf
+     * @param topology
+     *
+     * @thorws SubmitterHookException This is thrown when any Exception occurs during initialization or invocation of registered {@link ISubmitterHook}
+     */
     private static void invokeSubmitterHook(String name, String asUser, Map stormConf, StormTopology topology) {
+        String submissionNotifierClassName = null;
         try {
             if (stormConf.containsKey(Config.STORM_TOPOLOGY_SUBMISSION_NOTIFIER_PLUGIN)) {
-                ISubmitterHook submitterHook = (ISubmitterHook) Class.forName(stormConf.get(Config.STORM_TOPOLOGY_SUBMISSION_NOTIFIER_PLUGIN).toString()).newInstance();
+                submissionNotifierClassName = stormConf.get(Config.STORM_TOPOLOGY_SUBMISSION_NOTIFIER_PLUGIN).toString();
+                LOG.info("Initializing the registered ISubmitterHook [{}]", submissionNotifierClassName);
+                ISubmitterHook submitterHook = (ISubmitterHook) Class.forName(submissionNotifierClassName).newInstance();
                 TopologyInfo topologyInfo = Utils.getTopologyInfo(name, asUser, stormConf);
+                LOG.info("Invoking the registered ISubmitterHook [{}]", submissionNotifierClassName);
                 submitterHook.notify(topologyInfo, stormConf, topology);
             }
         } catch (Exception e) {
-            throw new RuntimeException(e);
+            LOG.warn("Error occurred in invoking submitter hook: "+submissionNotifierClassName, e);
+            throw new SubmitterHookException(e);
         }
     }
 
@@ -281,6 +297,7 @@ private static void invokeSubmitterHook(String name, String asUser, Map stormCon
      * @throws AlreadyAliveException if a topology with this name is already running
      * @throws InvalidTopologyException if an invalid topology was submitted
      * @throws AuthorizationException if authorization is failed
+     * @thorws SubmitterHookException if any Exception occurs during initialization or invocation of registered {@link ISubmitterHook}
      */
     @SuppressWarnings("unchecked")
     public static void submitTopology(String name, Map stormConf, StormTopology topology, SubmitOptions opts,
@@ -317,8 +334,8 @@ public static void submitTopologyWithProgressBar(String name, Map stormConf, Sto
      * @throws AlreadyAliveException if a topology with this name is already running
      * @throws InvalidTopologyException if an invalid topology was submitted
      * @throws AuthorizationException if authorization is failed
+     * @thorws SubmitterHookException if any Exception occurs during initialization or invocation of registered {@link ISubmitterHook}
      */
-
     public static void submitTopologyWithProgressBar(String name, Map stormConf, StormTopology topology, SubmitOptions opts) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {
         // show a progress bar so we know we're not stuck (especially on slow connections)
         submitTopology(name, stormConf, topology, opts, new StormSubmitter.ProgressListener() {
diff --git a/storm-core/src/jvm/org/apache/storm/hooks/SubmitterHookException.java b/storm-core/src/jvm/org/apache/storm/hooks/SubmitterHookException.java
new file mode 100644
index 00000000000..b7679a71510
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/hooks/SubmitterHookException.java
@@ -0,0 +1,41 @@
+/*
+ * 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.hooks;
+
+/**
+ * This Exception is thrown when registered {@link org.apache.storm.ISubmitterHook} could not be initialized or invoked.
+ */
+public class SubmitterHookException extends RuntimeException {
+
+    public SubmitterHookException() {
+    }
+
+    public SubmitterHookException(String message) {
+        super(message);
+    }
+
+    public SubmitterHookException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public SubmitterHookException(Throwable cause) {
+        super(cause);
+    }
+
+}

From fbcd0ab780b9c3047de2f522e4286ec5321cba09 Mon Sep 17 00:00:00 2001
From: Satish Duggana <sduggana@hortonworks.com>
Date: Wed, 25 May 2016 12:35:26 +0530
Subject: [PATCH 2/2] STORM-1864 log messages with format

---
 storm-core/src/jvm/org/apache/storm/StormSubmitter.java | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/storm-core/src/jvm/org/apache/storm/StormSubmitter.java b/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
index 077c5779416..97667409e18 100644
--- a/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
+++ b/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
@@ -273,13 +273,18 @@ private static void invokeSubmitterHook(String name, String asUser, Map stormCon
             if (stormConf.containsKey(Config.STORM_TOPOLOGY_SUBMISSION_NOTIFIER_PLUGIN)) {
                 submissionNotifierClassName = stormConf.get(Config.STORM_TOPOLOGY_SUBMISSION_NOTIFIER_PLUGIN).toString();
                 LOG.info("Initializing the registered ISubmitterHook [{}]", submissionNotifierClassName);
+
+                if(submissionNotifierClassName == null || submissionNotifierClassName.isEmpty()) {
+                    throw new IllegalArgumentException(Config.STORM_TOPOLOGY_SUBMISSION_NOTIFIER_PLUGIN + " property must be a non empty string.");
+                }
+
                 ISubmitterHook submitterHook = (ISubmitterHook) Class.forName(submissionNotifierClassName).newInstance();
                 TopologyInfo topologyInfo = Utils.getTopologyInfo(name, asUser, stormConf);
                 LOG.info("Invoking the registered ISubmitterHook [{}]", submissionNotifierClassName);
                 submitterHook.notify(topologyInfo, stormConf, topology);
             }
         } catch (Exception e) {
-            LOG.warn("Error occurred in invoking submitter hook: "+submissionNotifierClassName, e);
+            LOG.warn("Error occurred in invoking submitter hook:[{}] ",submissionNotifierClassName, e);
             throw new SubmitterHookException(e);
         }
     }