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 4e268a0..9789799 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;
@@ -41,10 +42,15 @@ public class ProcessJob extends AbstractProcessJob {
private static final String MEMCHECK_ENABLED = "memCheck.enabled";
private static final String MEMCHECK_FREEMEMDECRAMT = "memCheck.freeMemDecrAmt";
public static final String AZKABAN_MEMORY_CHECK = "azkaban.memory.check";
+ 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
@@ -81,6 +87,12 @@ public class ProcessJob extends AbstractProcessJob {
info(commands.size() + " commands to execute.");
File[] propFiles = initPropsFiles();
Map<String, String> envVars = getEnvironmentVariables();
+
+ // change krb5ccname env var
+ String effectiveUser = getEffectiveUser(jobProps);
+ String krb5ccname =
+ String.format("/tmp/krb5cc__%s__%s__%s__%s__%s", jobProps.getString(CommonJobProperties.PROJECT_NAME),jobProps.getString(CommonJobProperties.FLOW_ID), jobProps.getString(CommonJobProperties.JOB_ID), jobProps.getString(CommonJobProperties.EXEC_ID), effectiveUser);
+ envVars.put(KRB5CCNAME, krb5ccname);
for (String command : commands) {
info("Command: " + command);
@@ -119,6 +131,31 @@ public class ProcessJob extends AbstractProcessJob {
generateProperties(propFiles[1]);
}
+ /**
+ * Determines what user id should the process job run as.
+ * Logic should be self explanatory
+ * @param jobProps
+ * @return
+ */
+ 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("There is no user.to.proxy and submit.user in the jobProps. There is something wrong with the system");
+ }
+ 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
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();