azkaban-developers

diff --git a/azkaban-common/src/main/java/azkaban/jobExecutor/AbstractProcessJob.java b/azkaban-common/src/main/java/azkaban/jobExecutor/AbstractProcessJob.java
index 73592f0..831e49b 100644
--- a/azkaban-common/src/main/java/azkaban/jobExecutor/AbstractProcessJob.java
+++ b/azkaban-common/src/main/java/azkaban/jobExecutor/AbstractProcessJob.java
@@ -43,6 +43,9 @@ public abstract class AbstractProcessJob extends AbstractJob {
   public static final String JOB_PROP_ENV = "JOB_PROP_FILE";
   public static final String JOB_NAME_ENV = "JOB_NAME";
   public static final String JOB_OUTPUT_PROP_FILE = "JOB_OUTPUT_PROP_FILE";
+  private static final String SENSITIVE_JOB_PROP_NAME_SUFFIX = "_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;
 
@@ -81,6 +84,31 @@ public abstract class AbstractProcessJob extends AbstractJob {
     jobProps = PropsUtils.resolveProps(jobProps);
   }
 
+  /**
+   * prints the current Job props to the Job log.
+   */
+  protected void logJobProperties() {
+    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_SUFFIX ));
+        for(Map.Entry<String, String> entry : flattenedProps.entrySet()){
+          String key = entry.getKey();
+          String value = key.endsWith(SENSITIVE_JOB_PROP_NAME_SUFFIX)?
+                                      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);
+      }
+    }
+  }
+
   @Override
   public Props getJobGeneratedProperties() {
     return generatedProperties;
@@ -101,7 +129,6 @@ public abstract class AbstractProcessJob extends AbstractJob {
 
     files[1] = createOutputPropsFile(getId(), _cwd);
     jobProps.put(ENV_PREFIX + JOB_OUTPUT_PROP_FILE, files[1].getAbsolutePath());
-
     return files;
   }
 

diff --git a/azkaban-common/src/main/java/azkaban/jobExecutor/LongArgJob.java b/azkaban-common/src/main/java/azkaban/jobExecutor/LongArgJob.java
index 36ba23a..6188e99 100644
--- a/azkaban-common/src/main/java/azkaban/jobExecutor/LongArgJob.java
+++ b/azkaban-common/src/main/java/azkaban/jobExecutor/LongArgJob.java
@@ -69,6 +69,9 @@ public abstract class LongArgJob extends AbstractProcessJob {
 
     File[] propFiles = initPropsFiles();
 
+    // print out the Job properties to the job log.
+    this.logJobProperties();
+
     boolean success = false;
     this.process = builder.build();
     try {

diff --git a/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java b/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
index 18b7ffa..4e268a0 100644
--- a/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
+++ b/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
@@ -93,6 +93,9 @@ public class ProcessJob extends AbstractProcessJob {
       }
       info("Working directory: " + builder.getWorkingDir());
 
+      // print out the Job properties to the job log.
+      this.logJobProperties();
+
       boolean success = false;
       this.process = builder.build();
 
@@ -120,7 +123,7 @@ public class ProcessJob extends AbstractProcessJob {
    * This is used to get the min/max memory size requirement by processes.
    * SystemMemoryInfo can use the info to determine if the memory request
    * can be fulfilled. For Java process, this should be Xms/Xmx setting.
-   *  
+   *
    * @return pair of min/max memory size
    */
   protected Pair<Long, Long> getProcMemoryRequirement() throws Exception {

diff --git a/azkaban-common/src/main/java/azkaban/utils/Props.java b/azkaban-common/src/main/java/azkaban/utils/Props.java
index db2b8fe..68fcbd3 100644
--- a/azkaban-common/src/main/java/azkaban/utils/Props.java
+++ b/azkaban-common/src/main/java/azkaban/utils/Props.java
@@ -34,6 +34,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.util.TreeMap;
 
 import org.apache.log4j.Logger;
 
@@ -811,20 +812,28 @@ public class Props {
   }
 
   /**
+   * Returns a map of all the flattened properties, the item in the returned map is sorted alphabetically
+   * by the key value.
+   *
+   *
+   * @Return
+   */
+  public Map<String,String> getFlattened(){
+    TreeMap<String,String> returnVal = new TreeMap<String,String>(); 
+    returnVal.putAll(getMapByPrefix(""));
+    return returnVal; 
+  }
+
+  /**
    * Get a map of all properties by string prefix
    *
    * @param prefix The string prefix
    */
   public Map<String, String> getMapByPrefix(String prefix) {
-    Map<String, String> values = new HashMap<String, String>();
-
-    if (_parent != null) {
-      for (Map.Entry<String, String> entry : _parent.getMapByPrefix(prefix)
-          .entrySet()) {
-        values.put(entry.getKey(), entry.getValue());
-      }
-    }
+    Map<String, String> values = _parent == null ? new HashMap<String, String>():
+                                                   _parent.getMapByPrefix(prefix);
 
+    // when there is a conflict, value from the child takes the priority.
     for (String key : this.localKeySet()) {
       if (key.startsWith(prefix)) {
         values.put(key.substring(prefix.length()), get(key));

diff --git a/azkaban-common/src/test/java/azkaban/utils/PropsUtilsTest.java b/azkaban-common/src/test/java/azkaban/utils/PropsUtilsTest.java
index 158558e..54464cd 100644
--- a/azkaban-common/src/test/java/azkaban/utils/PropsUtilsTest.java
+++ b/azkaban-common/src/test/java/azkaban/utils/PropsUtilsTest.java
@@ -17,6 +17,7 @@
 package azkaban.utils;
 
 import java.io.IOException;
+import java.util.Map;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -127,6 +128,52 @@ public class PropsUtilsTest {
   }
 
   @Test
+  public void testGetFlattenedProps() throws Exception {
+
+    // for empty props empty flattened map is expected to be returned.
+    Props grandParentProps = new Props();
+    Assert.assertTrue(grandParentProps.getFlattened().isEmpty());
+
+    // single level
+    grandParentProps.put("test1","value1");
+    grandParentProps.put("test2","value2");
+    Map<String,String> set = grandParentProps.getFlattened();
+    Assert.assertEquals(2,set.size());
+    Assert.assertEquals("value1", set.get("test1"));
+    Assert.assertEquals("value2", set.get("test2"));
+
+    // multiple levels .
+    Props parentProps = new Props(grandParentProps);
+    parentProps.put("test3","value3");
+    parentProps.put("test4","value4");
+    set = parentProps.getFlattened();
+    Assert.assertEquals(4,set.size());
+    Assert.assertEquals("value3", set.get("test3"));
+    Assert.assertEquals("value1", set.get("test1"));
+
+    // multiple levels with same keys  .
+    Props props = new Props(parentProps);
+    props.put("test5","value5");
+    props.put("test1","value1.1");
+    set = props.getFlattened();
+    Assert.assertEquals(5,set.size());
+    Assert.assertEquals("value5", set.get("test5"));
+    Assert.assertEquals("value1.1", set.get("test1"));
+
+    // verify when iterating the elements are sorted by the key value.
+    Props props2 = new Props();
+    props2.put("2","2");
+    props2.put("0","0");
+    props2.put("1","1");
+    set = props2.getFlattened();
+    int index = 0 ;
+    for (Map.Entry<String, String> item : set.entrySet())
+    {
+      Assert.assertEquals(item.getKey(),Integer.toString(index++));
+    }
+  }
+
+  @Test
   public void testCyclesResolveProps() throws IOException {
     Props propsGrandParent = new Props();
     Props propsParent = new Props(propsGrandParent);