@Test
public void oozieShouldStart() throws Exception {

    LOGGER.info("OOZIE: Test Submit Workflow Start");
    FileSystem hdfsFs = ((HdfsBootstrap) HadoopBootstrap.INSTANCE.getService("HDFS")).getHdfsFileSystemHandle();
    OozieClient oozieClient = ((OozieBootstrap) HadoopBootstrap.INSTANCE.getService("OOZIE")).getOozieClient();
    Path appPath = new Path(hdfsFs.getHomeDirectory(), "testApp");
    hdfsFs.mkdirs(new Path(appPath, "lib"));
    Path workflow = new Path(appPath, "workflow.xml");
    String testInputFile = "test_input.txt";
    String testInputDir = "/tmp/test_input_dir";
    String testOutputDir = "/tmp/test_output_dir";
    // Setup input directory and file
    hdfsFs.mkdirs(new Path(testInputDir));
    hdfsFs.copyFromLocalFile(
            new Path(getClass().getClassLoader().getResource(testInputFile).toURI()), new Path(testInputDir));
    //write workflow.xml
    String wfApp = "<workflow-app name=\"sugar-option-decision\" xmlns=\"uri:oozie:workflow:0.5\">\n" +
            "  <global>\n" +
            "    <job-tracker>${jobTracker}</job-tracker>\n" +
            "    <name-node>${nameNode}</name-node>\n" +
            "    <configuration>\n" +
            "      <property>\n" +
            "        <name>mapreduce.output.fileoutputformat.outputdir</name>\n" +
            "        <value>" + testOutputDir + "</value>\n" +
            "      </property>\n" +
            "      <property>\n" +
            "        <name>mapreduce.input.fileinputformat.inputdir</name>\n" +
            "        <value>" + testInputDir + "</value>\n" +
            "      </property>\n" +
            "    </configuration>\n" +
            "  </global>\n" +
            "  <start to=\"first\"/>\n" +
            "  <action name=\"first\">\n" +
            "    <map-reduce> <prepare><delete path=\"" + testOutputDir + "\"/></prepare></map-reduce>\n" +
            "    <ok to=\"decision-second-option\"/>\n" +
            "    <error to=\"kill\"/>\n" +
            "  </action>\n" +
            "  <decision name=\"decision-second-option\">\n" +
            "    <switch>\n" +
            "      <case to=\"option\">${doOption}</case>\n" +
            "      <default to=\"second\"/>\n" +
            "    </switch>\n" +
            "  </decision>\n" +
            "  <action name=\"option\">\n" +
            "    <map-reduce> <prepare><delete path=\"" + testOutputDir + "\"/></prepare></map-reduce>\n" +
            "    <ok to=\"second\"/>\n" +
            "    <error to=\"kill\"/>\n" +
            "  </action>\n" +
            "  <action name=\"second\">\n" +
            "    <map-reduce> <prepare><delete path=\"" + testOutputDir + "\"/></prepare></map-reduce>\n" +
            "    <ok to=\"end\"/>\n" +
            "    <error to=\"kill\"/>\n" +
            "  </action>\n" +
            "  <kill name=\"kill\">\n" +
            "    <message>\n" +
            "      Failed to workflow, error message[${wf: errorMessage (wf: lastErrorNode ())}]\n" +
            "    </message>\n" +
            "  </kill>\n" +
            "  <end name=\"end\"/>\n" +
            "</workflow-app>";
    Writer writer = new OutputStreamWriter(hdfsFs.create(workflow));
    writer.write(wfApp);
    writer.close();
    //write job.properties
    Properties oozieConf = oozieClient.createConfiguration();
    oozieConf.setProperty(OozieClient.APP_PATH, workflow.toString());
    oozieConf.setProperty(OozieClient.USER_NAME, UserGroupInformation.getCurrentUser().getUserName());
    oozieConf.setProperty("jobTracker", "localhost:37001");
    oozieConf.setProperty("nameNode", "hdfs://localhost:20112");
    oozieConf.setProperty("user.name", System.getProperty("user.name"));
    oozieConf.setProperty("doOption", "true");
    //submit and check
    final String jobId = oozieClient.run(oozieConf);
    WorkflowJob wf = oozieClient.getJobInfo(jobId);
    assertNotNull(wf);
    LOGGER.info("OOZIE: Workflow: {}", wf.toString());
    while (oozieClient.getJobInfo(jobId).getStatus() != WorkflowJob.Status.RUNNING) {
        System.out.println("========== workflow job status " + oozieClient.getJobInfo(jobId).getStatus());
        Thread.sleep(1000);
    }
    while (oozieClient.getJobInfo(jobId).getStatus() == WorkflowJob.Status.RUNNING) {
        System.out.println("========== workflow job status " + oozieClient.getJobInfo(jobId).getStatus());
        System.out.println("========== job is running");
        Thread.sleep(1000);
    }
    System.out.println("=============== OOZIE: Final Workflow status" + oozieClient.getJobInfo(jobId).getStatus());
    wf = oozieClient.getJobInfo(jobId);
    System.out.println("=============== OOZIE: Workflow: {}" + wf.toString());
    assertEquals(WorkflowJob.Status.SUCCEEDED, wf.getStatus());
    hdfsFs.close();
}