java - 来自 Java 代码的工作流程暂停

Question

我正在运行一个包含 Java 代码的工作流，该代码又会启动另一个 oozie 工作流。主要工作流程运行良好，但从 Java 代码启动的工作流程始终处于暂停状态。我无法恢复它，因为它的用户是 mapred 而不是我。知道可能是什么问题吗？

这是我的主要工作流程

<java>
        <job-tracker>${jobTracker}</job-tracker>
        <name-node>${nameNode}</name-node>
        <main-class>com.last.play.LaunchJob</main-class>
        <arg>currentUser=${currentUser}</arg>
     </java>

这是java代码:

    Map<String, String> commandArgs = getActionArgs(args);

    Path appPropertyPath = new Path("/user/cmahajan/app.properties");
    Path jobPropertyPath = new Path("/user/cmahajan/job.properties");
    OozieClient wc = new OozieClient("http://host07.com:11000/oozie");

    String userName = commandArgs.get("currentUser");
    System.out.println("User Name recieved ::" + userName);
    Configuration trial = new Configuration();
    FileSystem fs = FileSystem.get(trial);

    Properties conf = wc.createConfiguration();
    Properties jobProperties = new Properties();
    Properties appProperties = new Properties();
    appProperties.load(fs.open(appPropertyPath));
    String version = appProperties.getProperty("version");
    jobProperties.load(fs.open(jobPropertyPath));

    for (Object key : jobProperties.keySet()) {
        String propValue = jobProperties.getProperty((String) key);
        propValue = propValue.replaceAll("\\$\\{user.name\\}", userName);
        conf.setProperty((String) key, propValue);
        System.out.println("Key ::" + key);
        System.out.println("Value ::" + propValue);
        System.out.println(" ===================");
    }
    String appsRoot = "${wfsBasePath}/" + version + "/apps";
    conf.setProperty("appsRoot", appsRoot);
    try {
        String jobId = wc.run(conf);

        System.out.println("Workflow job submitted");

        while (wc.getJobInfo(jobId).getStatus() == WorkflowJob.Status.RUNNING) {
            System.out.println("Workflow job running ...");
            Thread.sleep(10 * 1000);
        }
        System.out.println("Workflow job completed ...");
        System.out.println(wc.getJobInfo(jobId));
    } catch (OozieClientException oozieClientException) {
        oozieClientException.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

Answer 1

您可以使用 UserGroupInformation 来设置用户。

UserGroupInformation ugi = UserGroupInformation.createRemoteUser(username);
ugi.doAs(new PrivilegedExceptionAction<MyMapReduceWrapperClass>() {
public Object run() throws Exception {
    MyMapReduceWrapperClass mr = new MyMapReduceWrapperClass();
    ToolRunner.run(mr, null);
    return mr;
}
});