/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.oozie.event;

import java.io.FileOutputStream;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.io.Writer;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.oozie.AppType;
import org.apache.oozie.CoordinatorActionBean;
import org.apache.oozie.CoordinatorJobBean;
import org.apache.oozie.DagEngine;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
import org.apache.oozie.action.ActionExecutor;
import org.apache.oozie.action.control.ControlNodeActionExecutor;
import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.CoordinatorJob;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.WorkflowAction;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.client.event.Event;
import org.apache.oozie.client.event.JobEvent;
import org.apache.oozie.client.event.JobEvent.EventStatus;
import org.apache.oozie.client.rest.RestConstants;
import org.apache.oozie.command.CommandException;
import org.apache.oozie.command.coord.CoordActionCheckXCommand;
import org.apache.oozie.command.coord.CoordActionInputCheckXCommand;
import org.apache.oozie.command.coord.CoordMaterializeTransitionXCommand;
import org.apache.oozie.command.coord.CoordRerunXCommand;
import org.apache.oozie.command.coord.CoordResumeXCommand;
import org.apache.oozie.command.coord.CoordinatorXCommand;
import org.apache.oozie.command.wf.ActionCheckXCommand;
import org.apache.oozie.command.wf.ActionKillXCommand;
import org.apache.oozie.command.wf.ActionStartXCommand;
import org.apache.oozie.command.wf.ActionXCommand;
import org.apache.oozie.command.wf.KillXCommand;
import org.apache.oozie.command.wf.ResumeXCommand;
import org.apache.oozie.command.wf.SignalXCommand;
import org.apache.oozie.command.wf.StartXCommand;
import org.apache.oozie.command.wf.SuspendXCommand;
import org.apache.oozie.command.wf.WorkflowXCommand;
import org.apache.oozie.coord.CoordELFunctions;
import org.apache.oozie.executor.jpa.CoordActionGetJPAExecutor;
import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
import org.apache.oozie.executor.jpa.CoordJobGetJPAExecutor;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.executor.jpa.WorkflowActionGetJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowActionInsertJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobGetJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobInsertJPAExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery;
import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
import org.apache.oozie.service.ActionService;
import org.apache.oozie.service.EventHandlerService;
import org.apache.oozie.service.JPAService;
import org.apache.oozie.service.LiteWorkflowStoreService;
import org.apache.oozie.service.Services;
import org.apache.oozie.service.UUIDService;
import org.apache.oozie.test.XDataTestCase;
import org.apache.oozie.util.DateUtils;
import org.apache.oozie.util.IOUtils;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.workflow.WorkflowApp;
import org.apache.oozie.workflow.WorkflowInstance;
import org.apache.oozie.workflow.lite.ActionNodeDef;
import org.apache.oozie.workflow.lite.EndNodeDef;
import org.apache.oozie.workflow.lite.LiteWorkflowApp;
import org.apache.oozie.workflow.lite.LiteWorkflowInstance;
import org.apache.oozie.workflow.lite.StartNodeDef;
import org.apache.oozie.workflow.lite.TestLiteWorkflowLib;
import org.apache.oozie.workflow.lite.TestLiteWorkflowLib.TestActionNodeHandler;
import org.apache.oozie.workflow.lite.TestLiteWorkflowLib.TestControlNodeHandler;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

/**
 * Testcase to test that events are correctly generated from corresponding
 * Commands and inserted into events queue
 */
public class TestEventGeneration extends XDataTestCase {

    EventQueue queue;
    Services services;
    EventHandlerService ehs;
    JPAService jpaService;

    @Override
    @Before
    protected void setUp() throws Exception {
        super.setUp();
        services = new Services();
        Configuration conf = services.getConf();
        // The EventHandlerService manipulates the queues in the background, so the actual test results depend on the
        // circumstances (like the speed of the machine, debugging etc).
        conf.setInt("oozie.service.EventHandlerService.worker.threads", 0);
        conf.set(Services.CONF_SERVICE_EXT_CLASSES, "org.apache.oozie.service.EventHandlerService");
        services.init();
        ehs = services.get(EventHandlerService.class);
        queue = ehs.getEventQueue();
        jpaService = services.get(JPAService.class);
    }

    @Override
    @After
    protected void tearDown() throws Exception {
        Services.get().destroy();
        super.tearDown();
    }

    @Test
    public void testWorkflowJobEvent() throws Exception {
        assertEquals(0, queue.size());
        WorkflowApp app = new LiteWorkflowApp("testApp", "<workflow-app/>", new StartNodeDef(
                LiteWorkflowStoreService.LiteControlNodeHandler.class, "fs-node")).addNode(
                new ActionNodeDef("fs-node", "", TestLiteWorkflowLib.TestActionNodeHandler.class, "end", "end"))
                .addNode(new EndNodeDef("end", LiteWorkflowStoreService.LiteControlNodeHandler.class));
        WorkflowJobBean job = addRecordToWfJobTable(app, WorkflowJob.Status.PREP, WorkflowInstance.Status.PREP);

        // Starting job
        new StartXCommand(job.getId()).call();
        WorkflowJobGetJPAExecutor wfJobGetCmd = new WorkflowJobGetJPAExecutor(job.getId());
        job = jpaService.execute(wfJobGetCmd);
        assertEquals(WorkflowJob.Status.RUNNING, job.getStatus());
        assertEquals(1, queue.size());
        JobEvent event = (JobEvent) queue.poll();
        assertNotNull(event);
        assertEquals(EventStatus.STARTED, event.getEventStatus());
        assertEquals(AppType.WORKFLOW_JOB, event.getAppType());
        assertEquals(job.getId(), event.getId());
        assertEquals(job.getUser(), event.getUser());
        assertEquals(job.getAppName(), event.getAppName());
        assertEquals(job.getStartTime(), event.getStartTime());
        assertEquals(0, queue.size());

        // Suspending job
        new SuspendXCommand(job.getId()).call();
        job = jpaService.execute(wfJobGetCmd);
        assertEquals(WorkflowJob.Status.SUSPENDED, job.getStatus());
        assertEquals(1, queue.size());
        event = (JobEvent) queue.poll();
        assertNotNull(event);
        assertEquals(EventStatus.SUSPEND, event.getEventStatus());
        assertEquals(AppType.WORKFLOW_JOB, event.getAppType());
        assertEquals(job.getId(), event.getId());
        assertEquals(job.getUser(), event.getUser());
        assertEquals(job.getAppName(), event.getAppName());
        assertEquals(0, queue.size());

        // Resuming job
        new ResumeXCommand(job.getId()).call();
        job = jpaService.execute(wfJobGetCmd);
        assertEquals(WorkflowJob.Status.RUNNING, job.getStatus());
        assertEquals(1, queue.size());
        event = (JobEvent) queue.poll();
        assertNotNull(event);
        assertEquals(AppType.WORKFLOW_JOB, event.getAppType());
        assertEquals(job.getId(), event.getId());
        assertEquals(job.getUser(), event.getUser());
        assertEquals(job.getAppName(), event.getAppName());
        assertEquals(job.getStartTime(), event.getStartTime());
        assertEquals(0, queue.size());

        // Killing job
        new KillXCommand(job.getId()).call();
        job = jpaService.execute(wfJobGetCmd);
        assertEquals(WorkflowJob.Status.KILLED, job.getStatus());
        assertEquals(1, queue.size());
        event = (JobEvent) queue.poll();
        assertNotNull(event);
        assertEquals(EventStatus.FAILURE, event.getEventStatus());
        assertEquals(AppType.WORKFLOW_JOB, event.getAppType());
        assertEquals(job.getId(), event.getId());
        assertEquals(job.getUser(), event.getUser());
        assertEquals(job.getAppName(), event.getAppName());
        assertEquals(job.getStartTime(), event.getStartTime());
        assertEquals(job.getEndTime(), event.getEndTime());
        assertEquals(0, queue.size());

        // Successful job (testing SignalX)
        job = _createWorkflowJob();
        LiteWorkflowInstance wfInstance = (LiteWorkflowInstance) job.getWorkflowInstance();
        wfInstance.start();
        job.setWorkflowInstance(wfInstance);
        WorkflowJobQueryExecutor.getInstance().executeUpdate(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MODIFIED, job);
        WorkflowActionBean wfAction = jpaService.execute(new WorkflowActionGetJPAExecutor(job.getId() + "@one"));
        new SignalXCommand(job.getId(), wfAction.getId()).call();
        job = jpaService.execute(new WorkflowJobGetJPAExecutor(job.getId()));
        assertEquals(WorkflowJob.Status.SUCCEEDED, job.getStatus());
        assertEquals(1, queue.size());
        event = (JobEvent) queue.poll();
        assertNotNull(event);
        assertEquals(AppType.WORKFLOW_JOB, event.getAppType());
        assertEquals(job.getId(), event.getId());
        assertEquals(job.getUser(), event.getUser());
        assertEquals(job.getAppName(), event.getAppName());
        assertEquals(job.getStartTime(), event.getStartTime());
        assertEquals(job.getEndTime(), event.getEndTime());

    }

    @Test
    public void testCoordinatorActionEvent() throws Exception {
        // avoid noise from other apptype events by setting it to only
        // coord action
        ehs.setAppTypes(new HashSet<String>(Arrays.asList("coordinator_action")));
        assertEquals(queue.size(), 0);
        Date startTime = DateUtils.parseDateOozieTZ("2013-01-01T10:00Z");
        Date endTime = DateUtils.parseDateOozieTZ("2013-01-01T10:01Z");
        CoordinatorJobBean coord = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, startTime, endTime, false,
                false, 0);
        modifyCoordForRunning(coord);

        // Action WAITING on materialization
        new CoordMaterializeTransitionXCommand(coord.getId(), 3600).call();
        final CoordActionGetJPAExecutor coordGetCmd = new CoordActionGetJPAExecutor(coord.getId() + "@1");
        CoordinatorActionBean action = jpaService.execute(coordGetCmd);
        assertEquals(CoordinatorAction.Status.WAITING, action.getStatus());
        assertEquals(1, queue.size());
        JobEvent event = (JobEvent) queue.poll();
        assertNotNull(event);
        assertEquals(EventStatus.WAITING, event.getEventStatus());
        assertEquals(AppType.COORDINATOR_ACTION, event.getAppType());
        assertEquals(action.getId(), event.getId());
        assertEquals(action.getJobId(), event.getParentId());
        assertEquals(action.getNominalTime(), ((CoordinatorActionEvent) event).getNominalTime());
        assertNull(event.getStartTime());
        assertEquals(coord.getUser(), event.getUser());
        assertEquals(coord.getAppName(), event.getAppName());
        assertEquals(0, queue.size());

        // Make Action ready
        // In this case it will proceed to Running since n(ready_actions) < concurrency
        new CoordActionInputCheckXCommand(action.getId(), coord.getId()).call();
        action = jpaService.execute(coordGetCmd);
        assertEquals(CoordinatorAction.Status.RUNNING, action.getStatus());

        event = (JobEvent) queue.poll();
        assertEquals(EventStatus.STARTED, event.getEventStatus());
        assertEquals(AppType.COORDINATOR_ACTION, event.getAppType());
        assertEquals(action.getId(), event.getId());
        assertEquals(action.getJobId(), event.getParentId());
        assertEquals(action.getNominalTime(), ((CoordinatorActionEvent) event).getNominalTime());
        WorkflowJobBean wfJob = jpaService.execute(new WorkflowJobGetJPAExecutor(action.getExternalId()));
        assertEquals(wfJob.getStartTime(), event.getStartTime());
        assertEquals(coord.getUser(), event.getUser());
        assertEquals(coord.getAppName(), event.getAppName());

        sleep(2000);

        // Action Success
        wfJob.setStatus(WorkflowJob.Status.SUCCEEDED);
        WorkflowJobQueryExecutor.getInstance().executeUpdate(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_MODTIME, wfJob);
        action.setStatus(CoordinatorAction.Status.RUNNING);
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_STATUS_PENDING_TIME, action);
        new CoordActionCheckXCommand(action.getId(), 0).call();
        action = jpaService.execute(coordGetCmd);
        assertEquals(CoordinatorAction.Status.SUCCEEDED, action.getStatus());
        List<Event> list =  queue.pollBatch();
        event = (JobEvent)list.get(list.size()-1);
        assertEquals(EventStatus.SUCCESS, event.getEventStatus());
        assertEquals(AppType.COORDINATOR_ACTION, event.getAppType());
        assertEquals(action.getId(), event.getId());
        assertEquals(action.getJobId(), event.getParentId());
        assertEquals(action.getNominalTime(), ((CoordinatorActionEvent) event).getNominalTime());
        assertEquals(wfJob.getStartTime(), event.getStartTime());
        assertEquals(coord.getUser(), event.getUser());
        assertEquals(coord.getAppName(), event.getAppName());

        // Action Failure
        wfJob.setStatus(WorkflowJob.Status.FAILED);
        action.setStatus(CoordinatorAction.Status.RUNNING);
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_STATUS_PENDING_TIME, action);
        WorkflowJobQueryExecutor.getInstance().executeUpdate(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_MODTIME, wfJob);
        new CoordActionCheckXCommand(action.getId(), 0).call();
        action = jpaService.execute(coordGetCmd);
        assertEquals(CoordinatorAction.Status.FAILED, action.getStatus());
        event = (JobEvent) queue.poll();
        assertEquals(EventStatus.FAILURE, event.getEventStatus());
        assertEquals(AppType.COORDINATOR_ACTION, event.getAppType());
        assertEquals(action.getId(), event.getId());
        assertEquals(action.getJobId(), event.getParentId());
        assertEquals(action.getNominalTime(), ((CoordinatorActionEvent) event).getNominalTime());
        assertEquals(wfJob.getStartTime(), event.getStartTime());
        assertEquals(coord.getUser(), event.getUser());
        assertEquals(coord.getAppName(), event.getAppName());

        // Action Suspended
        wfJob.setStatus(WorkflowJob.Status.SUSPENDED);
        action.setStatus(CoordinatorAction.Status.RUNNING);
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_STATUS_PENDING_TIME,
                action);
        WorkflowJobQueryExecutor.getInstance().executeUpdate(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_MODTIME, wfJob);
        new CoordActionCheckXCommand(action.getId(), 0).call();
        action = jpaService.execute(coordGetCmd);
        assertEquals(CoordinatorAction.Status.SUSPENDED, action.getStatus());
        event = (JobEvent) queue.poll();
        assertEquals(EventStatus.SUSPEND, event.getEventStatus());
        assertEquals(AppType.COORDINATOR_ACTION, event.getAppType());
        assertEquals(action.getId(), event.getId());
        assertEquals(action.getJobId(), event.getParentId());
        assertEquals(action.getNominalTime(), ((CoordinatorActionEvent) event).getNominalTime());
        assertEquals(wfJob.getStartTime(), event.getStartTime());
        assertEquals(coord.getUser(), event.getUser());
        assertEquals(coord.getAppName(), event.getAppName());

        // Action start on Coord Resume
        coord.setStatus(CoordinatorJobBean.Status.SUSPENDED);
        CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB_STATUS, coord);
        action.setStatus(CoordinatorAction.Status.SUSPENDED);
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_STATUS_PENDING_TIME, action);
        wfJob.setStatus(WorkflowJob.Status.SUSPENDED);
        WorkflowInstance wfInstance = wfJob.getWorkflowInstance();
        ((LiteWorkflowInstance) wfInstance).setStatus(WorkflowInstance.Status.SUSPENDED);
        wfJob.setWorkflowInstance(wfInstance);
        WorkflowJobQueryExecutor.getInstance().executeUpdate(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MODIFIED, wfJob);
        queue.clear();
        new CoordResumeXCommand(coord.getId()).call();
        waitForEventGeneration(1000);
        CoordinatorActionEvent cevent = (CoordinatorActionEvent) queue.poll();
        assertEquals(EventStatus.STARTED, cevent.getEventStatus());
        assertEquals(AppType.COORDINATOR_ACTION, cevent.getAppType());
        assertEquals(action.getId(), cevent.getId());
        assertEquals(action.getJobId(), cevent.getParentId());
        assertEquals(action.getNominalTime(), cevent.getNominalTime());
        coord = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, coord.getId());
        assertEquals(coord.getLastModifiedTime(), cevent.getStartTime());

        // Action going to WAITING on Coord Rerun
        action.setStatus(CoordinatorAction.Status.KILLED);
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_STATUS_PENDING_TIME, action);
        queue.clear();
        new CoordRerunXCommand(coord.getId(), RestConstants.JOB_COORD_SCOPE_ACTION, "1", false, true, false,
                null).call();
        waitFor(3 * 100, new Predicate() {
            @Override
            public boolean evaluate() throws Exception {
                return jpaService.execute(coordGetCmd).getStatus() == CoordinatorAction.Status.WAITING;
            }
        });
        cevent = (CoordinatorActionEvent) queue.poll();
        assertEquals(EventStatus.WAITING, cevent.getEventStatus());
        assertEquals(AppType.COORDINATOR_ACTION, cevent.getAppType());
        assertEquals(action.getId(), cevent.getId());
        assertEquals(action.getJobId(), cevent.getParentId());
        assertEquals(action.getNominalTime(), cevent.getNominalTime());
        assertEquals(wfJob.getStartTime(), event.getStartTime());
        assertNotNull(cevent.getMissingDeps());

    }

    @Test
    public void testWorkflowActionEvent() throws Exception {
        assertEquals(queue.size(), 0);
        // avoid noise from other apptype events by setting it to only
        // workflow action
        ehs.setAppTypes(new HashSet<String>(Arrays.asList("workflow_action")));
        WorkflowJobBean job = this.addRecordToWfJobTable(WorkflowJob.Status.RUNNING, WorkflowInstance.Status.RUNNING);
        WorkflowActionBean action = this.addRecordToWfActionTable(job.getId(), "1", WorkflowAction.Status.PREP, true);
        // adding record sets externalChildID to dummy workflow-id so resetting it
        action.setExternalChildIDs(null);
        WorkflowActionQueryExecutor.getInstance().executeUpdate(WorkflowActionQuery.UPDATE_ACTION_START, action);

        // Starting job
        new ActionStartXCommand(action.getId(), "map-reduce").call();
        WorkflowActionGetJPAExecutor wfActionGetCmd = new WorkflowActionGetJPAExecutor(action.getId());
        action = jpaService.execute(wfActionGetCmd);
        assertEquals(WorkflowAction.Status.RUNNING, action.getStatus());
        assertEquals(1, queue.size());
        WorkflowActionEvent event = (WorkflowActionEvent) queue.poll();
        assertNotNull(event);
        assertEquals(EventStatus.STARTED, event.getEventStatus());
        assertEquals(AppType.WORKFLOW_ACTION, event.getAppType());
        assertEquals(action.getId(), event.getId());
        assertEquals(job.getUser(), event.getUser());
        assertEquals(action.getName(), event.getAppName());
        assertEquals(action.getStartTime(), event.getStartTime());
        assertEquals(0, queue.size());

        // Suspending job
        ActionExecutor.Context context = new ActionXCommand.ActionExecutorContext(job, action, false, false);
        ActionExecutor executor = Services.get().get(ActionService.class).getExecutor(action.getType());
        ActionCheckXCommandForTest dac = new ActionCheckXCommandForTest(context, executor, action.getId());
        dac.execute();
        action = dac.getAction();
        assertEquals(WorkflowAction.Status.START_MANUAL, action.getStatus());
        assertEquals(1, queue.size());
        event = (WorkflowActionEvent) queue.poll();
        assertNotNull(event);
        assertEquals(EventStatus.SUSPEND, event.getEventStatus());
        assertEquals(AppType.WORKFLOW_ACTION, event.getAppType());
        assertEquals(action.getId(), event.getId());
        assertEquals(job.getUser(), event.getUser());
        assertEquals(action.getName(), event.getAppName());
        assertEquals(0, queue.size());

        // Killing job
        action.setStatus(WorkflowAction.Status.KILLED);
        action.setPendingOnly();
        action.setEndTime(null); //its already set by XTestCase add action record method above
        WorkflowActionQueryExecutor.getInstance().executeUpdate(WorkflowActionQuery.UPDATE_ACTION_END, action);
        new ActionKillXCommand(action.getId()).call();
        action = jpaService.execute(wfActionGetCmd);
        assertEquals(WorkflowAction.Status.KILLED, action.getStatus());
        assertEquals(1, queue.size());
        event = (WorkflowActionEvent) queue.poll();
        assertNotNull(event);
        assertEquals(EventStatus.FAILURE, event.getEventStatus());
        assertEquals(AppType.WORKFLOW_ACTION, event.getAppType());
        assertEquals(action.getId(), event.getId());
        assertEquals(job.getUser(), event.getUser());
        assertEquals(action.getName(), event.getAppName());
        assertEquals(action.getStartTime(), event.getStartTime());
        assertNotNull(action.getEndTime());
        assertNotNull(event.getEndTime());
        assertEquals(action.getEndTime(), event.getEndTime());
        assertEquals(0, queue.size());

    }

    @Test
    public void testWorkflowJobEventError() throws Exception {
        final WorkflowJobBean job = addRecordToWfJobTable(WorkflowJob.Status.FAILED, WorkflowInstance.Status.FAILED);
        // create event with error code and message
        WorkflowXCommand<Void> myCmd = new KillXCommand(job.getId()) {
            @Override
            protected Void execute() {
                WorkflowXCommand.generateEvent(job, "errorCode", "errorMsg");
                return null;
            }
        };
        myCmd.call();
        WorkflowJobEvent event = (WorkflowJobEvent) queue.poll();
        assertNotNull(event);
        assertEquals("errorCode", event.getErrorCode());
        assertEquals("errorMsg", event.getErrorMessage());
        assertEquals(EventStatus.FAILURE, event.getEventStatus());

    }

    @Test
    public void testCoordinatorActionEventDependencies() throws Exception {
        final CoordinatorJobBean coord = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, false, false);
        final CoordinatorActionBean action = addRecordToCoordActionTable(coord.getId(), 1,
                CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0);
        WorkflowJobBean wjb = new WorkflowJobBean();
        wjb.setId(action.getExternalId());
        wjb.setLastModifiedTime(new Date());
        WorkflowJobQueryExecutor.getInstance().insert(wjb);

        CoordinatorXCommand<Void> myCmd = new CoordActionCheckXCommand(action.getId(), 0) {
            @Override
            protected Void execute() {
                CoordinatorXCommand.generateEvent(action, coord.getUser(), coord.getAppName(), null);
                return null;
            }
        };

        // CASE 1: Only pull missing deps
        action.setMissingDependencies("pull");
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_DEPENDENCIES, action);
        myCmd.call();
        CoordinatorActionEvent event = (CoordinatorActionEvent) queue.poll();
        assertNotNull(event);
        assertEquals("pull", event.getMissingDeps());

        // CASE 2: Only hcat (push) missing deps
        action.setMissingDependencies(null);
        action.setPushMissingDependencies("push");
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_DEPENDENCIES, action);
        myCmd.call();
        event = (CoordinatorActionEvent) queue.poll();
        assertNotNull(event);
        assertEquals("push", event.getMissingDeps());

        // CASE 3: Both types
        action.setMissingDependencies("pull");
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_DEPENDENCIES, action);
        myCmd.call();
        event = (CoordinatorActionEvent) queue.poll();
        assertNotNull(event);
        assertEquals("pull" + CoordELFunctions.INSTANCE_SEPARATOR + "push", event.getMissingDeps());

    }

    @Test
    public void testForNoDuplicatesWorkflowEvents() throws Exception {
        // test workflow job events
        Reader reader = IOUtils.getResourceAsReader("wf-no-op.xml", -1);
        Writer writer = new OutputStreamWriter(new FileOutputStream(getTestCaseDir() + "/workflow.xml"),
                StandardCharsets.UTF_8);
        IOUtils.copyCharStream(reader, writer);

        final DagEngine engine = new DagEngine(getTestUser());
        Configuration conf = new XConfiguration();
        conf.set(OozieClient.APP_PATH, getTestCaseFileUri("workflow.xml"));
        conf.set(OozieClient.USER_NAME, getTestUser());

        final String jobId1 = engine.submitJob(conf, true);
        final WorkflowJobGetJPAExecutor readCmd = new WorkflowJobGetJPAExecutor(jobId1);

        waitFor(1 * 100, new Predicate() {
            @Override
            public boolean evaluate() throws Exception {
                return jpaService.execute(readCmd).getStatus() == WorkflowJob.Status.SUCCEEDED;
            }
        });
        assertEquals(2, queue.size());
        assertEquals(EventStatus.STARTED, ((JobEvent)queue.poll()).getEventStatus());
        assertEquals(EventStatus.SUCCESS, ((JobEvent)queue.poll()).getEventStatus());
        queue.clear();
    }

    @Test
    public void testForNoDuplicatesCoordinatorActionEvents() throws Exception {
        // test coordinator action events (failure case)
        Date startTime = DateUtils.parseDateOozieTZ("2009-02-01T23:59Z");
        Date endTime = DateUtils.parseDateOozieTZ("2009-02-02T23:59Z");
        CoordinatorJobBean coord = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, startTime, endTime, false,
                false, 0);
        _modifyCoordForFailureAction(coord, "wf-invalid-fork.xml");
        new CoordMaterializeTransitionXCommand(coord.getId(), 3600).call();
        final CoordJobGetJPAExecutor readCmd1 = new CoordJobGetJPAExecutor(coord.getId());
        waitFor(1 * 100, new Predicate() {
            @Override
            public boolean evaluate() throws Exception {
                CoordinatorJobBean bean = jpaService.execute(readCmd1);
                return bean.getStatus() == CoordinatorJob.Status.SUCCEEDED
                        || bean.getStatus() == CoordinatorJob.Status.KILLED;
            }
        });
        assertEquals(2, queue.size());
        assertEquals(EventStatus.WAITING, ((JobEvent)queue.poll()).getEventStatus());
        assertEquals(EventStatus.FAILURE, ((JobEvent)queue.poll()).getEventStatus());
        queue.clear();
    }

    @Test
    public void testInvalidXMLCoordinatorFailsForNoDuplicates() throws Exception {
        Date startTime = DateUtils.parseDateOozieTZ("2009-02-01T23:59Z");
        Date endTime = DateUtils.parseDateOozieTZ("2009-02-02T23:59Z");

        // test coordinator action events (failure from ActionStartX)
        ehs.getAppTypes().add("workflow_action");
        CoordinatorJobBean coord = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, startTime, endTime, false, false, 0);
        CoordinatorActionBean action = addRecordToCoordActionTable(coord.getId(), 1, CoordinatorAction.Status.RUNNING,
                "coord-action-sla1.xml", 0);
        WorkflowJobBean wf = addRecordToWfJobTable(WorkflowJob.Status.RUNNING, WorkflowInstance.Status.RUNNING,
                action.getId());
        action.setExternalId(wf.getId());
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION, action);

        String waId = _createWorkflowAction(wf.getId(), "wf-action");
        new ActionStartXCommand(waId, action.getType()).call();

        final CoordJobGetJPAExecutor readCmd2 = new CoordJobGetJPAExecutor(coord.getId());
        waitFor(1 * 100, new Predicate() {
            @Override
            public boolean evaluate() throws Exception {
                return jpaService.execute(readCmd2).getStatus() == CoordinatorJob.Status.KILLED;
            }
        });

        assertEquals(3, queue.size());

        HashMap<AppType,JobEvent> eventsMap = new HashMap<AppType,JobEvent>();
        while (queue.size() > 0){
            JobEvent event = (JobEvent) queue.poll();
            eventsMap.put(event.getAppType(), event);
        }

        assertEquals(3, eventsMap.size());

        //Check the WF action
        {
            JobEvent wfActionEvent = eventsMap.get(AppType.WORKFLOW_ACTION);
            assertNotNull("There should be a WF action", wfActionEvent);
            assertEquals(EventStatus.FAILURE, wfActionEvent.getEventStatus());
            assertEquals(waId, wfActionEvent.getId());
            assertEquals(AppType.WORKFLOW_ACTION, wfActionEvent.getAppType());
        }

        //Check the WF job
        {
            JobEvent wfJobEvent = eventsMap.get(AppType.WORKFLOW_JOB);
            assertNotNull("There should be a WF job", wfJobEvent);
            assertEquals(EventStatus.FAILURE, wfJobEvent.getEventStatus());
            assertEquals(wf.getId(), wfJobEvent.getId());
            assertEquals(AppType.WORKFLOW_JOB, wfJobEvent.getAppType());
        }

        //Check the Coordinator action
        {
            JobEvent coordActionEvent = eventsMap.get(AppType.COORDINATOR_ACTION);
            assertNotNull("There should be a Coordinator action", coordActionEvent);
            assertEquals(EventStatus.FAILURE, coordActionEvent.getEventStatus());
            assertEquals(action.getId(), coordActionEvent.getId());
            assertEquals(AppType.COORDINATOR_ACTION, coordActionEvent.getAppType());
        }
        queue.clear();
    }

    private class ActionCheckXCommandForTest extends ActionCheckXCommand {

        ActionExecutor.Context context;
        ActionExecutor executor;
        WorkflowActionBean action;
        JPAService jpa;

        public ActionCheckXCommandForTest(ActionExecutor.Context context, ActionExecutor executor, String actionId)
                throws JPAExecutorException {
            super(actionId);
            this.context = context;
            this.executor = executor;
            jpa = Services.get().get(JPAService.class);
            this.action = jpa.execute(new WorkflowActionGetJPAExecutor(actionId));
        }

        @Override
        public Void execute() throws CommandException {
            handleNonTransient(context, executor, WorkflowAction.Status.START_MANUAL);
            action = (WorkflowActionBean) ((ActionExecutorContext) context).getAction();
            if (!(executor instanceof ControlNodeActionExecutor) && EventHandlerService.isEnabled()) {
                generateEvent(action, getTestUser());
            }
            return null;
        }

        public WorkflowActionBean getAction() {
            return action;
        }

    }

    private WorkflowJobBean _createWorkflowJob() throws Exception {
        LiteWorkflowApp app = new LiteWorkflowApp("my-app", "<workflow-app/>",
                new StartNodeDef(TestControlNodeHandler.class, "one"))
                .addNode(new ActionNodeDef("one", "<java></java>", TestActionNodeHandler.class, "end", "end"))
                .addNode(new EndNodeDef("end", TestControlNodeHandler.class));
        Configuration conf = new Configuration();
        Path appUri = new Path(getAppPath(), "workflow.xml");
        conf.set(OozieClient.APP_PATH, appUri.toString());
        conf.set(OozieClient.LOG_TOKEN, "testToken");
        conf.set(OozieClient.USER_NAME, getTestUser());
        WorkflowJobBean workflow = createWorkflow(app, conf, WorkflowJob.Status.PREP,
                WorkflowInstance.Status.PREP);
        String executionPath = "/";

        assertNotNull(jpaService);
        WorkflowJobInsertJPAExecutor wfInsertCmd = new WorkflowJobInsertJPAExecutor(workflow);
        jpaService.execute(wfInsertCmd);
        WorkflowActionBean wfAction = addRecordToWfActionTable(workflow.getId(), "one", WorkflowAction.Status.OK,
                executionPath, true);
        wfAction.setPending();
        wfAction.setSignalValue(WorkflowAction.Status.OK.name());
        WorkflowActionQueryExecutor.getInstance().executeUpdate(WorkflowActionQuery.UPDATE_ACTION, wfAction);

        return workflow;
    }

    private void _modifyCoordForFailureAction(CoordinatorJobBean coord, String resourceXml) throws Exception {
        String wfXml = IOUtils.getResourceAsString(resourceXml, -1);
        writeToFile(wfXml, getFsTestCaseDir(), "workflow.xml");
        String coordXml = coord.getJobXml();
        coord.setJobXml(coordXml.replace("hdfs:///tmp/workflows/", getFsTestCaseDir() + "/workflow.xml"));
        CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coord);
    }

    private String _createWorkflowAction(String wfId, String actionName) throws JPAExecutorException {
        WorkflowActionBean action = new WorkflowActionBean();
        action.setName(actionName);
        action.setId(Services.get().get(UUIDService.class).generateChildId(wfId, actionName));
        action.setJobId(wfId);
        action.setType("java");
        action.setTransition("transition");
        action.setStatus(WorkflowAction.Status.PREP);
        action.setStartTime(new Date());
        action.setEndTime(new Date());
        action.setLastCheckTime(new Date());
        action.setPendingOnly();

        String actionXml = "<java>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>"
                + getNameNodeUri() + "</name-node>" + "<main-class>" + "${dummy}" + "</java>";
        action.setConf(actionXml);
        jpaService.execute(new WorkflowActionInsertJPAExecutor(action));
        return action.getId();
    }

    private void waitForEventGeneration(int wait) {
        waitFor(wait, new Predicate() {
            @Override
            public boolean evaluate() throws Exception {
                return ehs.getEventQueue().peek() != null;
            }
        });
    }

}