azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/executor/ExecutorManager.java b/azkaban-common/src/main/java/azkaban/executor/ExecutorManager.java
index c86648e..7fb61bd 100644
--- a/azkaban-common/src/main/java/azkaban/executor/ExecutorManager.java
+++ b/azkaban-common/src/main/java/azkaban/executor/ExecutorManager.java
@@ -74,7 +74,7 @@ public class ExecutorManager extends EventHandler implements
   private ExecutingManagerUpdaterThread executingManager;
 
   private static final long DEFAULT_EXECUTION_LOGS_RETENTION_MS = 3 * 4 * 7
-      * 24 * 60 * 60 * 1000l;
+      * 24 * 60 * 60 * 1000L;
   private long lastCleanerThreadCheckTime = -1;
 
   private long lastThreadCheckTime = -1;

diff --git a/azkaban-common/src/main/java/azkaban/executor/JdbcExecutorLoader.java b/azkaban-common/src/main/java/azkaban/executor/JdbcExecutorLoader.java
index 8d230ba..25e61c3 100644
--- a/azkaban-common/src/main/java/azkaban/executor/JdbcExecutorLoader.java
+++ b/azkaban-common/src/main/java/azkaban/executor/JdbcExecutorLoader.java
@@ -101,7 +101,7 @@ public class JdbcExecutorLoader extends AbstractJdbcLoader implements
           runner.query(connection, LastInsertID.LAST_INSERT_ID,
               new LastInsertID());
 
-      if (id == -1l) {
+      if (id == -1L) {
         throw new ExecutorManagerException(
             "Execution id is not properly created.");
       }
@@ -779,7 +779,7 @@ public class JdbcExecutorLoader extends AbstractJdbcLoader implements
     @Override
     public Long handle(ResultSet rs) throws SQLException {
       if (!rs.next()) {
-        return -1l;
+        return -1L;
       }
       long id = rs.getLong(1);
       return id;

diff --git a/azkaban-common/src/main/java/azkaban/flow/CommonJobProperties.java b/azkaban-common/src/main/java/azkaban/flow/CommonJobProperties.java
index 0c98ecb..8c208a5 100644
--- a/azkaban-common/src/main/java/azkaban/flow/CommonJobProperties.java
+++ b/azkaban-common/src/main/java/azkaban/flow/CommonJobProperties.java
@@ -133,6 +133,11 @@ public class CommonJobProperties {
    * hotspot occurs.
    */
   public static final String PROJECT_VERSION = "azkaban.flow.projectversion";
+  
+  /**
+   * Find out who is the submit user, in addition to the user.to.proxy (they may be different)
+   */
+  public static final String SUBMIT_USER = "azkaban.flow.submituser";
 
   /**
    * A uuid assigned to every execution

diff --git a/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java b/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
index 75cc0dd..7636b21 100644
--- a/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
+++ b/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
@@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.log4j.Logger;
 
+import azkaban.flow.CommonJobProperties;
 import azkaban.jobExecutor.utils.process.AzkabanProcess;
 import azkaban.jobExecutor.utils.process.AzkabanProcessBuilder;
 import azkaban.utils.Pair;
@@ -36,17 +37,30 @@ import azkaban.utils.SystemMemoryInfo;
 public class ProcessJob extends AbstractProcessJob {
 
   public static final String COMMAND = "command";
+
   private static final long KILL_TIME_MS = 5000;
+
   private volatile AzkabanProcess process;
+
   private static final String MEMCHECK_ENABLED = "memCheck.enabled";
-  private static final String MEMCHECK_FREEMEMDECRAMT = "memCheck.freeMemDecrAmt";
+
+  private static final String MEMCHECK_FREEMEMDECRAMT =
+      "memCheck.freeMemDecrAmt";
+
   public static final String AZKABAN_MEMORY_CHECK = "azkaban.memory.check";
+  
   public static final String NATIVE_LIB_FOLDER = "azkaban.native.lib";
   public static final String EXECUTE_AS_USER = "execute.as.user";
-  
+  public static final String USER_TO_PROXY = "user.to.proxy";
+  public static final String KRB5CCNAME = "KRB5CCNAME";
+
   public ProcessJob(final String jobId, final Props sysProps,
       final Props jobProps, final Logger log) {
     super(jobId, sysProps, jobProps, log);
+
+    // this is in line with what other job types (hadoopJava, spark, pig, hive)
+    // is doing
+    jobProps.put(CommonJobProperties.JOB_ID, jobId);
   }
 
   @Override
@@ -57,13 +71,19 @@ public class ProcessJob extends AbstractProcessJob {
       handleError("Bad property definition! " + e.getMessage(), e);
     }
 
-    if (sysProps.getBoolean(MEMCHECK_ENABLED, true) && jobProps.getBoolean(AZKABAN_MEMORY_CHECK, true)) {
+    if (sysProps.getBoolean(MEMCHECK_ENABLED, true)
+        && jobProps.getBoolean(AZKABAN_MEMORY_CHECK, true)) {
       long freeMemDecrAmt = sysProps.getLong(MEMCHECK_FREEMEMDECRAMT, 0);
       Pair<Long, Long> memPair = getProcMemoryRequirement();
-      boolean isMemGranted = SystemMemoryInfo.canSystemGrantMemory(memPair.getFirst(), memPair.getSecond(), freeMemDecrAmt);
+      boolean isMemGranted =
+          SystemMemoryInfo.canSystemGrantMemory(memPair.getFirst(),
+              memPair.getSecond(), freeMemDecrAmt);
       if (!isMemGranted) {
-        throw new Exception(String.format("Cannot request memory (Xms %d kb, Xmx %d kb) from system for job %s",
-                memPair.getFirst(), memPair.getSecond(), getId()));
+        throw new Exception(
+            String
+                .format(
+                    "Cannot request memory (Xms %d kb, Xmx %d kb) from system for job %s",
+                    memPair.getFirst(), memPair.getSecond(), getId()));
       }
     }
     
@@ -82,7 +102,11 @@ public class ProcessJob extends AbstractProcessJob {
 
     info(commands.size() + " commands to execute.");
     File[] propFiles = initPropsFiles();
+    
+    // change krb5ccname env var so that each job execution gets its own cache
     Map<String, String> envVars = getEnvironmentVariables();
+    envVars.put(KRB5CCNAME, getKrb5ccname(jobProps));
+
   
     String nativeLibFolder = null;
     String executeAsUserBinary = null;
@@ -137,9 +161,61 @@ public class ProcessJob extends AbstractProcessJob {
   }
 
   /**
+   * <pre>
+   * This method extracts the kerberos ticket cache file name from the jobprops.
+   * This method will ensure that each job execution will have its own kerberos ticket cache file
+   * Given that the code only sets an environmental variable, the number of files created corresponds
+   * to the number of processes that are doing kinit in their flow, which should not be an inordinately 
+   * high number.
+   * </pre>
+   * 
+   * @return file name: the kerberos ticket cache file to use
+   */
+  private String getKrb5ccname(Props jobProps) {
+    String effectiveUser = getEffectiveUser(jobProps);
+    String projectName =
+        jobProps.getString(CommonJobProperties.PROJECT_NAME).replace(" ", "_");
+    String flowId =
+        jobProps.getString(CommonJobProperties.FLOW_ID).replace(" ", "_");
+    String jobId =
+        jobProps.getString(CommonJobProperties.JOB_ID).replace(" ", "_");
+    // execId should be an int and should not have space in it, ever
+    String execId = jobProps.getString(CommonJobProperties.EXEC_ID);
+    String krb5ccname =
+        String.format("/tmp/krb5cc__%s__%s__%s__%s__%s", projectName, flowId,
+            jobId, execId, effectiveUser);
+
+    return krb5ccname;
+  }
+
+  /**
+   * <pre>
+   * Determines what user id should the process job run as, in the following order of precedence:
+   * 1. USER_TO_PROXY
+   * 2. SUBMIT_USER
+   * </pre>
+   * 
+   * @param jobProps
+   * @return the user that Azkaban is going to execute as
+   */
+  private String getEffectiveUser(Props jobProps) {
+    String effectiveUser = null;
+    if (jobProps.containsKey(USER_TO_PROXY)) {
+      effectiveUser = jobProps.getString(USER_TO_PROXY);
+    } else if (jobProps.containsKey(CommonJobProperties.SUBMIT_USER)) {
+      effectiveUser = jobProps.getString(CommonJobProperties.SUBMIT_USER);
+    } else {
+      throw new RuntimeException(
+          "Internal Error: No user.to.proxy or submit.user in the jobProps");
+    }
+    info("effective user is: " + effectiveUser);
+    return effectiveUser;
+  }
+
+  /**
    * 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.
+   * 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
    */

diff --git a/azkaban-common/src/main/java/azkaban/project/ProjectManager.java b/azkaban-common/src/main/java/azkaban/project/ProjectManager.java
index 0f09b9c..1e5e522 100644
--- a/azkaban-common/src/main/java/azkaban/project/ProjectManager.java
+++ b/azkaban-common/src/main/java/azkaban/project/ProjectManager.java
@@ -269,7 +269,7 @@ public class ProjectManager {
           "Project names must start with a letter, followed by any number of letters, digits, '-' or '_'.");
     }
 
-    if (projectsByName.contains(projectName)) {
+    if (projectsByName.containsKey(projectName)) {
       throw new ProjectManagerException("Project already exists.");
     }
 

diff --git a/azkaban-common/src/main/java/azkaban/server/session/SessionCache.java b/azkaban-common/src/main/java/azkaban/server/session/SessionCache.java
index fc4d64f..1237678 100644
--- a/azkaban-common/src/main/java/azkaban/server/session/SessionCache.java
+++ b/azkaban-common/src/main/java/azkaban/server/session/SessionCache.java
@@ -58,7 +58,7 @@ public class SessionCache {
    * @return
    */
   public Session getSession(String sessionId) {
-    Session elem = cache.<Session> get(sessionId);
+    Session elem = cache.get(Session.class, sessionId);
 
     return elem;
   }

diff --git a/azkaban-common/src/main/java/azkaban/trigger/JdbcTriggerLoader.java b/azkaban-common/src/main/java/azkaban/trigger/JdbcTriggerLoader.java
index 99b32f9..007d872 100644
--- a/azkaban-common/src/main/java/azkaban/trigger/JdbcTriggerLoader.java
+++ b/azkaban-common/src/main/java/azkaban/trigger/JdbcTriggerLoader.java
@@ -178,7 +178,7 @@ public class JdbcTriggerLoader extends AbstractJdbcLoader implements
           runner.query(connection, LastInsertID.LAST_INSERT_ID,
               new LastInsertID());
 
-      if (id == -1l) {
+      if (id == -1L) {
         logger.error("trigger id is not properly created.");
         throw new TriggerLoaderException("trigger id is not properly created.");
       }
@@ -257,7 +257,7 @@ public class JdbcTriggerLoader extends AbstractJdbcLoader implements
     @Override
     public Long handle(ResultSet rs) throws SQLException {
       if (!rs.next()) {
-        return -1l;
+        return -1L;
       }
 
       long id = rs.getLong(1);

diff --git a/azkaban-common/src/main/java/azkaban/utils/cache/Cache.java b/azkaban-common/src/main/java/azkaban/utils/cache/Cache.java
index a2ddc60..231d419 100644
--- a/azkaban-common/src/main/java/azkaban/utils/cache/Cache.java
+++ b/azkaban-common/src/main/java/azkaban/utils/cache/Cache.java
@@ -38,17 +38,16 @@ public class Cache {
     LRU, FIFO
   }
 
-  /* package */Cache(CacheManager manager) {
+  /* package */ Cache(CacheManager manager) {
     this.manager = manager;
   }
 
-  @SuppressWarnings("unchecked")
-  public <T> T get(Object key) {
+  public <T> T get(Class<T> clazz, Object key) {
     Element<?> element = elementMap.get(key);
     if (element == null) {
       return null;
     }
-    return (T) element.getElement();
+    return clazz.cast(element.getElement());
   }
 
   public <T> void put(Object key, T item) {

diff --git a/azkaban-common/src/main/java/azkaban/utils/PropsUtils.java b/azkaban-common/src/main/java/azkaban/utils/PropsUtils.java
index 596e8b3..3412425 100644
--- a/azkaban-common/src/main/java/azkaban/utils/PropsUtils.java
+++ b/azkaban-common/src/main/java/azkaban/utils/PropsUtils.java
@@ -292,6 +292,7 @@ public class PropsUtils {
     props.put(CommonJobProperties.FLOW_UUID, UUID.randomUUID().toString());
     props.put(CommonJobProperties.PROJECT_LAST_CHANGED_BY, flow.getLastModifiedByUser());
     props.put(CommonJobProperties.PROJECT_LAST_CHANGED_DATE, flow.getLastModifiedTimestamp());
+    props.put(CommonJobProperties.SUBMIT_USER, flow.getExecutableFlow().getSubmitUser());  
 
     DateTime loadTime = new DateTime();
 

diff --git a/azkaban-common/src/main/java/azkaban/utils/TypedMapWrapper.java b/azkaban-common/src/main/java/azkaban/utils/TypedMapWrapper.java
index 7118e8c..62208f3 100644
--- a/azkaban-common/src/main/java/azkaban/utils/TypedMapWrapper.java
+++ b/azkaban-common/src/main/java/azkaban/utils/TypedMapWrapper.java
@@ -71,7 +71,7 @@ public class TypedMapWrapper<K, V> {
   }
 
   public Long getLong(K key) {
-    return getLong(key, -1l);
+    return getLong(key, -1L);
   }
 
   public Long getLong(K key, Long defaultVal) {

diff --git a/azkaban-common/src/test/java/azkaban/executor/JavaJobRunnerMain.java b/azkaban-common/src/test/java/azkaban/executor/JavaJobRunnerMain.java
index 67d0284..cf46637 100644
--- a/azkaban-common/src/test/java/azkaban/executor/JavaJobRunnerMain.java
+++ b/azkaban-common/src/test/java/azkaban/executor/JavaJobRunnerMain.java
@@ -196,7 +196,7 @@ public class JavaJobRunnerMain {
       }
       writer.write("}".getBytes());
     } catch (Exception e) {
-      new RuntimeException("Unable to store output properties to: "
+      throw new RuntimeException("Unable to store output properties to: "
           + outputFileStr);
     } finally {
       try {

diff --git a/azkaban-common/src/test/java/azkaban/jobExecutor/JavaProcessJobTest.java b/azkaban-common/src/test/java/azkaban/jobExecutor/JavaProcessJobTest.java
index 05a0a28..068935f 100644
--- a/azkaban-common/src/test/java/azkaban/jobExecutor/JavaProcessJobTest.java
+++ b/azkaban-common/src/test/java/azkaban/jobExecutor/JavaProcessJobTest.java
@@ -22,7 +22,6 @@ import java.util.Date;
 import java.util.Properties;
 
 import org.apache.log4j.Logger;
-
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Assert;
@@ -33,6 +32,7 @@ import org.junit.Test;
 import org.junit.Rule;
 import org.junit.rules.TemporaryFolder;
 
+import azkaban.flow.CommonJobProperties;
 import azkaban.utils.Props;
 
 public class JavaProcessJobTest {
@@ -108,6 +108,14 @@ public class JavaProcessJobTest {
     props.put(AbstractProcessJob.WORKING_DIR, workingDir.getCanonicalPath());
     props.put("type", "java");
     props.put("fullPath", ".");
+    
+    props.put(CommonJobProperties.PROJECT_NAME, "test_project");
+    props.put(CommonJobProperties.FLOW_ID, "test_flow");
+    props.put(CommonJobProperties.JOB_ID, "test_job");
+    props.put(CommonJobProperties.EXEC_ID, "123");
+    props.put(CommonJobProperties.SUBMIT_USER, "test_user");
+    
+
     job = new JavaProcessJob("testJavaProcess", props, props, log);
   }
 

diff --git a/azkaban-common/src/test/java/azkaban/jobExecutor/ProcessJobTest.java b/azkaban-common/src/test/java/azkaban/jobExecutor/ProcessJobTest.java
index 974895b..57f4505 100644
--- a/azkaban-common/src/test/java/azkaban/jobExecutor/ProcessJobTest.java
+++ b/azkaban-common/src/test/java/azkaban/jobExecutor/ProcessJobTest.java
@@ -20,7 +20,6 @@ import java.io.File;
 import java.io.IOException;
 
 import org.apache.log4j.Logger;
-
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -28,6 +27,7 @@ import org.junit.Test;
 import org.junit.Rule;
 import org.junit.rules.TemporaryFolder;
 
+import azkaban.flow.CommonJobProperties;
 import azkaban.utils.Props;
 
 public class ProcessJobTest {
@@ -46,6 +46,12 @@ public class ProcessJobTest {
     props.put(AbstractProcessJob.WORKING_DIR, workingDir.getCanonicalPath());
     props.put("type", "command");
     props.put("fullPath", ".");
+    
+    props.put(CommonJobProperties.PROJECT_NAME, "test_project");
+    props.put(CommonJobProperties.FLOW_ID, "test_flow");
+    props.put(CommonJobProperties.JOB_ID, "test_job");
+    props.put(CommonJobProperties.EXEC_ID, "123");
+    props.put(CommonJobProperties.SUBMIT_USER, "test_user");
 
     job = new ProcessJob("TestProcess", props, props, log);
   }
@@ -62,6 +68,37 @@ public class ProcessJobTest {
     job.run();
 
   }
+  
+  /**
+   * this job should run fine if the props contain user.to.proxy
+   * @throws Exception
+   */
+  @Test
+  public void testOneUnixCommandWithProxyUserInsteadOfSubmitUser() throws Exception {
+    
+    // Initialize the Props
+    props.removeLocal(CommonJobProperties.SUBMIT_USER);
+    props.put("user.to.proxy", "test_user");
+    props.put(ProcessJob.COMMAND, "ls -al");
+    
+    job.run();
+
+  }
+  
+  /**
+   * this job should fail because there is no user.to.proxy and no CommonJobProperties.SUBMIT_USER
+   * @throws Exception
+   */
+  @Test (expected=RuntimeException.class)
+  public void testOneUnixCommandWithNoUser() throws Exception {
+    
+    // Initialize the Props
+    props.removeLocal(CommonJobProperties.SUBMIT_USER);    
+    props.put(ProcessJob.COMMAND, "ls -al");
+    
+    job.run();
+
+  }
 
   @Test
   public void testFailedUnixCommand() throws Exception {

diff --git a/azkaban-common/src/test/java/azkaban/jobExecutor/PythonJobTest.java b/azkaban-common/src/test/java/azkaban/jobExecutor/PythonJobTest.java
index ee767de..7320c15 100644
--- a/azkaban-common/src/test/java/azkaban/jobExecutor/PythonJobTest.java
+++ b/azkaban-common/src/test/java/azkaban/jobExecutor/PythonJobTest.java
@@ -24,6 +24,7 @@ import org.apache.log4j.Logger;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import azkaban.utils.Props;
@@ -82,6 +83,7 @@ public class PythonJobTest {
     Utils.removeFile(scriptFile);
   }
 
+  @Ignore("Test appears to hang.")
   @Test
   public void testPythonJob() {
 

diff --git a/azkaban-common/src/test/java/azkaban/project/ProjectTest.java b/azkaban-common/src/test/java/azkaban/project/ProjectTest.java
index da36d46..63c19f0 100644
--- a/azkaban-common/src/test/java/azkaban/project/ProjectTest.java
+++ b/azkaban-common/src/test/java/azkaban/project/ProjectTest.java
@@ -28,8 +28,8 @@ public class ProjectTest {
   @Test
   public void testToAndFromObject() throws Exception {
     Project project = new Project(1, "tesTing");
-    project.setCreateTimestamp(1l);
-    project.setLastModifiedTimestamp(2l);
+    project.setCreateTimestamp(1L);
+    project.setLastModifiedTimestamp(2L);
     project.setDescription("I am a test");
     project.setUserPermission("user1", new Permission(new Type[] { Type.ADMIN,
         Type.EXECUTE }));

diff --git a/azkaban-common/src/test/java/azkaban/utils/cache/CacheTest.java b/azkaban-common/src/test/java/azkaban/utils/cache/CacheTest.java
index 905cf5e..c01d3f8 100644
--- a/azkaban-common/src/test/java/azkaban/utils/cache/CacheTest.java
+++ b/azkaban-common/src/test/java/azkaban/utils/cache/CacheTest.java
@@ -34,19 +34,19 @@ public class CacheTest {
     cache.insertElement("key3", "val3");
     cache.insertElement("key4", "val4");
 
-    Assert.assertEquals(cache.get("key2"), "val2");
-    Assert.assertEquals(cache.get("key3"), "val3");
-    Assert.assertEquals(cache.get("key4"), "val4");
-    Assert.assertEquals(cache.get("key1"), "val1");
+    Assert.assertEquals(cache.get(String.class, "key2"), "val2");
+    Assert.assertEquals(cache.get(String.class, "key3"), "val3");
+    Assert.assertEquals(cache.get(String.class, "key4"), "val4");
+    Assert.assertEquals(cache.get(String.class, "key1"), "val1");
     Assert.assertEquals(4, cache.getSize());
 
     cache.insertElement("key5", "val5");
     Assert.assertEquals(4, cache.getSize());
-    Assert.assertEquals(cache.get("key3"), "val3");
-    Assert.assertEquals(cache.get("key4"), "val4");
-    Assert.assertEquals(cache.get("key1"), "val1");
-    Assert.assertEquals(cache.get("key5"), "val5");
-    Assert.assertNull(cache.get("key2"));
+    Assert.assertEquals(cache.get(String.class, "key3"), "val3");
+    Assert.assertEquals(cache.get(String.class, "key4"), "val4");
+    Assert.assertEquals(cache.get(String.class, "key1"), "val1");
+    Assert.assertEquals(cache.get(String.class, "key5"), "val5");
+    Assert.assertNull(cache.get(String.class, "key2"));
   }
 
   @Test
@@ -67,19 +67,19 @@ public class CacheTest {
     cache.insertElement("key3", "val3");
     cache.insertElement("key4", "val4");
 
-    Assert.assertEquals(cache.get("key2"), "val2");
-    Assert.assertEquals(cache.get("key3"), "val3");
-    Assert.assertEquals(cache.get("key4"), "val4");
-    Assert.assertEquals(cache.get("key1"), "val1");
+    Assert.assertEquals(cache.get(String.class, "key2"), "val2");
+    Assert.assertEquals(cache.get(String.class, "key3"), "val3");
+    Assert.assertEquals(cache.get(String.class, "key4"), "val4");
+    Assert.assertEquals(cache.get(String.class, "key1"), "val1");
     Assert.assertEquals(4, cache.getSize());
 
     cache.insertElement("key5", "val5");
     Assert.assertEquals(4, cache.getSize());
-    Assert.assertEquals(cache.get("key3"), "val3");
-    Assert.assertEquals(cache.get("key4"), "val4");
-    Assert.assertEquals(cache.get("key2"), "val2");
-    Assert.assertEquals(cache.get("key5"), "val5");
-    Assert.assertNull(cache.get("key1"));
+    Assert.assertEquals(cache.get(String.class, "key3"), "val3");
+    Assert.assertEquals(cache.get(String.class, "key4"), "val4");
+    Assert.assertEquals(cache.get(String.class, "key2"), "val2");
+    Assert.assertEquals(cache.get(String.class, "key5"), "val5");
+    Assert.assertNull(cache.get(String.class, "key1"));
   }
 
   @Test
@@ -99,7 +99,7 @@ public class CacheTest {
       } catch (InterruptedException e) {
       }
     }
-    Assert.assertEquals(cache.get("key1"), "val1");
+    Assert.assertEquals(cache.get(String.class, "key1"), "val1");
     cache.insertElement("key2", "val2");
     synchronized (this) {
       try {
@@ -107,8 +107,8 @@ public class CacheTest {
       } catch (InterruptedException e) {
       }
     }
-    Assert.assertNull(cache.get("key1"));
-    Assert.assertEquals("val2", cache.get("key2"));
+    Assert.assertNull(cache.get(String.class, "key1"));
+    Assert.assertEquals("val2", cache.get(String.class, "key2"));
 
     synchronized (this) {
       try {
@@ -117,7 +117,7 @@ public class CacheTest {
       }
     }
 
-    Assert.assertNull(cache.get("key2"));
+    Assert.assertNull(cache.get(String.class, "key2"));
   }
 
   @Test
@@ -137,7 +137,7 @@ public class CacheTest {
       } catch (InterruptedException e) {
       }
     }
-    Assert.assertEquals(cache.get("key1"), "val1");
+    Assert.assertEquals(cache.get(String.class, "key1"), "val1");
     cache.insertElement("key2", "val2");
     synchronized (this) {
       try {
@@ -145,8 +145,8 @@ public class CacheTest {
       } catch (InterruptedException e) {
       }
     }
-    Assert.assertEquals("val1", cache.get("key1"));
-    Assert.assertNull(cache.get("key3"));
+    Assert.assertEquals("val1", cache.get(String.class, "key1"));
+    Assert.assertNull(cache.get(String.class, "key3"));
     synchronized (this) {
       try {
         wait(1000);
@@ -154,6 +154,6 @@ public class CacheTest {
       }
     }
 
-    Assert.assertNull(cache.get("key2"));
+    Assert.assertNull(cache.get(String.class, "key2"));
   }
 }

diff --git a/azkaban-execserver/src/package/bin/start-exec.sh b/azkaban-execserver/src/package/bin/start-exec.sh
index fbb7124..d8f0f73 100755
--- a/azkaban-execserver/src/package/bin/start-exec.sh
+++ b/azkaban-execserver/src/package/bin/start-exec.sh
@@ -1,6 +1,5 @@
 #!/bin/bash
 
-base_dir=$(dirname $0)/..
-
-bin/azkaban-executor-start.sh $base_dir 2>&1>logs/executorServerLog__`date +%F+%T`.out &
+# pass along command line arguments to azkaban-executor-start.sh script
+bin/azkaban-executor-start.sh "$@" 2>&1>logs/executorServerLog__`date +%F+%T`.out &
 

diff --git a/build.gradle b/build.gradle
index d8afcca..7a5db1e 100644
--- a/build.gradle
+++ b/build.gradle
@@ -1,10 +1,14 @@
 buildscript {
   repositories {
     mavenCentral()
+    maven {
+      url 'https://plugins.gradle.org/m2/'
+    }
   }
   dependencies {
     classpath 'com.linkedin:gradle-dustjs-plugin:1.0.0'
     classpath 'de.obqo.gradle:gradle-lesscss-plugin:1.0-1.3.3'
+    classpath 'net.ltgt.gradle:gradle-errorprone-plugin:0.0.8'
   }
 }
 
@@ -32,6 +36,11 @@ def cmdCaller = { commandln ->
 subprojects {
   apply plugin: 'java'
   apply plugin: 'eclipse'
+  apply plugin: 'net.ltgt.errorprone'
+
+  configurations.errorprone {
+    resolutionStrategy.force 'com.google.errorprone:error_prone_core:2.0.5'
+  }
 
   /**
    * Gets the version name from the latest Git tag
@@ -78,6 +87,9 @@ project(':azkaban-common') {
     all {
       transitive = false
     }
+    errorprone {
+      transitive = true
+    }
   }
 
   dependencies {
@@ -114,11 +126,11 @@ project(':azkaban-common') {
     testCompile('junit:junit:4.11')
     testCompile('org.hamcrest:hamcrest-all:1.3')
   }
-  
+
   tasks.withType(JavaCompile) {
     options.encoding = "UTF-8"
   }
-  
+
 }
 
 project(':azkaban-migration') {
@@ -126,6 +138,9 @@ project(':azkaban-migration') {
     all {
       transitive = false
     }
+    errorprone {
+      transitive = true
+    }
   }
 
   dependencies {
@@ -174,6 +189,9 @@ project(':azkaban-webserver') {
     generateRestli {
       transitive = true
     }
+    errorprone {
+      transitive = true
+    }
   }
 
   dependencies {
@@ -297,6 +315,9 @@ project(':azkaban-execserver') {
     all {
       transitive = false
     }
+    errorprone {
+      transitive = true
+    }
   }
 
   dependencies {
@@ -312,7 +333,6 @@ project(':azkaban-execserver') {
     compile('org.mortbay.jetty:jetty-util:6.1.26')
     compile('org.codehaus.jackson:jackson-core-asl:1.9.5')
     compile('org.codehaus.jackson:jackson-mapper-asl:1.9.5')
-    
 
     testCompile('junit:junit:4.11')
     testCompile('org.hamcrest:hamcrest-all:1.3')
@@ -478,11 +498,11 @@ project(':azkaban-test') {
     }
   }
 
-  distZip.dependsOn build, animalDistZip, embeddedDistZip, embedded2DistZip,
+  distZip.dependsOn animalDistZip, embeddedDistZip, embedded2DistZip,
       embedded3DistZip, embeddedBadDistZip, execpropstestDistZip,
       exectest1DistZip, exectest2DistZip, logtestDistZip
 
-  distTar.dependsOn build, animalDistTar, embeddedDistTar, embedded2DistTar,
+  distTar.dependsOn animalDistTar, embeddedDistTar, embedded2DistTar,
       embedded3DistTar, embeddedBadDistTar, execpropstestDistTar,
       exectest1DistTar, exectest2DistTar, logtestDistTar
 }
@@ -570,5 +590,5 @@ distZip.dependsOn migrationDistZip, webserverDistZip, execserverDistZip, soloser
  * Gradle wrapper task.
  */
 task wrapper(type: Wrapper) {
-  gradleVersion = '1.12'
+  gradleVersion = '2.7'
 }

diff --git a/gradle.properties b/gradle.properties
index 6f7f687..ef3a7de 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -1,3 +1,3 @@
 org.gradle.daemon=true
 group=com.linkedin
-version=2.6.4
+version=2.7.0

diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar
index 0087cd3..e8c6bf7 100644
Binary files a/gradle/wrapper/gradle-wrapper.jar and b/gradle/wrapper/gradle-wrapper.jar differ

diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index 506745b..bbc82a1 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,6 @@
-#Wed Jun 11 01:55:01 PDT 2014
+#Sat Sep 26 15:48:43 PDT 2015
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-1.12-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-2.7-bin.zip

diff --git a/gradlew b/gradlew
index 91a7e26..97fac78 100755
--- a/gradlew
+++ b/gradlew
@@ -42,11 +42,6 @@ case "`uname`" in
     ;;
 esac
 
-# For Cygwin, ensure paths are in UNIX format before anything is touched.
-if $cygwin ; then
-    [ -n "$JAVA_HOME" ] && JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
-fi
-
 # Attempt to set APP_HOME
 # Resolve links: $0 may be a link
 PRG="$0"
@@ -114,6 +109,7 @@ fi
 if $cygwin ; then
     APP_HOME=`cygpath --path --mixed "$APP_HOME"`
     CLASSPATH=`cygpath --path --mixed "$CLASSPATH"`
+    JAVACMD=`cygpath --unix "$JAVACMD"`
 
     # We build the pattern for arguments to be converted via cygpath
     ROOTDIRSRAW=`find -L / -maxdepth 1 -mindepth 1 -type d 2>/dev/null`