azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/jobExecutor/AbstractProcessJob.java b/azkaban-common/src/main/java/azkaban/jobExecutor/AbstractProcessJob.java
index c3ff1bf..b1c7c8d 100644
--- a/azkaban-common/src/main/java/azkaban/jobExecutor/AbstractProcessJob.java
+++ b/azkaban-common/src/main/java/azkaban/jobExecutor/AbstractProcessJob.java
@@ -45,6 +45,7 @@ public abstract class AbstractProcessJob extends AbstractJob {
   public static final String JOB_OUTPUT_PROP_FILE = "JOB_OUTPUT_PROP_FILE";
   private static final String SENSITIVE_JOB_PROP_NAME_SURFFIX = "_X";
   private static final String SENSITIVE_JOB_PROP_VALUE_PLACEHOLDER = "[MASKED]";
+  private static final String JOB_DUMP_PROPERTIES_IN_LOG = "job.dump.properties";
 
   protected final String _jobPath;
 
@@ -87,18 +88,25 @@ public abstract class AbstractProcessJob extends AbstractJob {
    * prints the current Job props to the Job log.
    */
   protected void logJobProperties() {
-    Map<String,String> flattenedProps = this.jobProps.getFlattened();
-    this.info("******   Job properties   ******");
-    this.info(String.format("- Note : value is masked if property name ends with '%s'.",
-                SENSITIVE_JOB_PROP_NAME_SURFFIX ));
-    for(Map.Entry<String, String> entry : flattenedProps.entrySet()){
-      String key = entry.getKey();
-      String value = key.endsWith(SENSITIVE_JOB_PROP_NAME_SURFFIX)?
-                                  SENSITIVE_JOB_PROP_VALUE_PLACEHOLDER :
-                                  entry.getValue();
-      this.info(String.format("%s=%s",key,value));
+    if (this.jobProps != null && 
+        this.jobProps.getBoolean(JOB_DUMP_PROPERTIES_IN_LOG, false)){
+      try {
+        Map<String,String> flattenedProps = this.jobProps.getFlattened();
+        this.info("******   Job properties   ******");
+        this.info(String.format("- Note : value is masked if property name ends with '%s'.",
+            SENSITIVE_JOB_PROP_NAME_SURFFIX ));
+        for(Map.Entry<String, String> entry : flattenedProps.entrySet()){
+          String key = entry.getKey();
+          String value = key.endsWith(SENSITIVE_JOB_PROP_NAME_SURFFIX)?
+                                      SENSITIVE_JOB_PROP_VALUE_PLACEHOLDER :
+                                      entry.getValue();
+          this.info(String.format("%s=%s",key,value));
+        }
+        this.info("****** End Job properties  ******");
+      } catch (Exception ex){
+        log.error("failed to log job properties ", ex);
+      }
     }
-    this.info("****** End Job properties  ******");
   }
 
   @Override