azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/jobtype/JobTypeManager.java b/azkaban-common/src/main/java/azkaban/jobtype/JobTypeManager.java
index e5d0dae..2a5a376 100644
--- a/azkaban-common/src/main/java/azkaban/jobtype/JobTypeManager.java
+++ b/azkaban-common/src/main/java/azkaban/jobtype/JobTypeManager.java
@@ -78,7 +78,7 @@ public class JobTypeManager {
         try {
           loadPluginJobTypes(plugins);
         } catch (Exception e) {
-          logger.info("Plugin jobtypes failed to load. " + e.getCause());
+          logger.info("Plugin jobtypes failed to load. " + e.getCause(), e);
           throw new JobTypeManagerException(e);
         }
       }
@@ -163,8 +163,8 @@ public class JobTypeManager {
         try {
           loadJobTypes(dir, plugins);
         } catch (Exception e) {
-          logger.error("Failed to load jobtype " + dir.getName()
-              + e.getMessage());
+          logger.error(
+              "Failed to load jobtype " + dir.getName() + e.getMessage(), e);
           throw new JobTypeManagerException(e);
         }
       }
@@ -202,8 +202,10 @@ public class JobTypeManager {
       pluginLoadProps = new Props(commonPluginLoadProps, pluginLoadPropsFile);
       pluginLoadProps = PropsUtils.resolveProps(pluginLoadProps);
     } catch (Exception e) {
+      logger
+          .error("pluginLoadProps to help with debugging: " + pluginLoadProps);
       throw new JobTypeManagerException("Failed to get jobtype properties"
-          + e.getMessage());
+          + e.getMessage(), e);
     }
     // Add properties into the plugin set
     pluginLoadProps.put("plugin.dir", pluginDir.getAbsolutePath());
@@ -232,9 +234,6 @@ public class JobTypeManager {
       Job job =
           (Job) Utils.callConstructor(clazz, "dummy", fakeSysProps,
               fakeJobProps, logger);
-    } catch (Exception e) {
-      logger.info("Jobtype " + jobTypeName + " failed test!", e);
-      throw new JobExecutionException(e);
     } catch (Throwable t) {
       logger.info("Jobtype " + jobTypeName + " failed test!", t);
       throw new JobExecutionException(t);
@@ -245,7 +244,7 @@ public class JobTypeManager {
 
   /**
    * Creates and loads all plugin resources (jars) into a ClassLoader
-   *
+   * 
    * @param pluginDir
    * @param jobTypeName
    * @param plugins

diff --git a/azkaban-execserver/src/main/java/azkaban/execapp/JobRunner.java b/azkaban-execserver/src/main/java/azkaban/execapp/JobRunner.java
index 63e139c..a58a5a2 100644
--- a/azkaban-execserver/src/main/java/azkaban/execapp/JobRunner.java
+++ b/azkaban-execserver/src/main/java/azkaban/execapp/JobRunner.java
@@ -43,6 +43,7 @@ import azkaban.executor.ExecutorManagerException;
 import azkaban.executor.Status;
 import azkaban.flow.CommonJobProperties;
 import azkaban.jobExecutor.AbstractProcessJob;
+import azkaban.jobExecutor.JavaProcessJob;
 import azkaban.jobExecutor.Job;
 import azkaban.jobtype.JobTypeManager;
 import azkaban.jobtype.JobTypeManagerException;
@@ -50,7 +51,7 @@ import azkaban.utils.Props;
 
 public class JobRunner extends EventHandler implements Runnable {
   private final Layout DEFAULT_LAYOUT = new EnhancedPatternLayout(
-      "%d{dd-MM-yyyy HH:mm:ss z} %c{1} %p - %m\n");
+      "%d{dd-MM-yyyy HH:mm:ss z} %c %p - %m\n");
 
   private ExecutorLoader loader;
   private Props props;
@@ -488,6 +489,7 @@ public class JobRunner extends EventHandler implements Runnable {
       }
 
       insertLinks();
+      insertJVMAargs();
 
       props.put(CommonJobProperties.JOB_ATTEMPT, node.getAttempt());
       props.put(CommonJobProperties.JOB_METADATA_FILE,
@@ -521,6 +523,29 @@ public class JobRunner extends EventHandler implements Runnable {
   }
 
   /**
+   * Add useful JVM arguments so it is easier to map a running Java process to a
+   * flow, execution id and job
+   */
+  private void insertJVMAargs() {
+    String flowName = node.getParentFlow().getFlowId();
+    String jobId = node.getId();
+
+    String jobJVMArgs =
+        String.format(
+            "-Dazkaban.flowid=%s -Dazkaban.execid=%s -Dazkaban.jobid=%s",
+            flowName, executionId, jobId);
+    String previousJVMArgs = props.get(JavaProcessJob.JVM_PARAMS);
+    if (previousJVMArgs == null) {
+      previousJVMArgs = jobJVMArgs;
+    } else {
+      previousJVMArgs += " " + jobJVMArgs;
+    }
+
+    logger.info("job JVM args: " + previousJVMArgs);
+    props.put(JavaProcessJob.JVM_PARAMS, previousJVMArgs);
+  }
+
+  /**
    * Add relevant links to the job properties so that downstream consumers may
    * know what executions initiated their execution.
    */