/**
 * 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.service;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.AppType;
import org.apache.oozie.CoordinatorActionBean;
import org.apache.oozie.CoordinatorJobBean;
import org.apache.oozie.WorkflowJobBean;
import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.client.event.JobEvent.EventStatus;
import org.apache.oozie.client.event.SLAEvent;
import org.apache.oozie.client.event.SLAEvent.SLAStatus;
import org.apache.oozie.client.rest.JsonBean;
import org.apache.oozie.client.rest.RestConstants;
import org.apache.oozie.event.EventQueue;
import org.apache.oozie.executor.jpa.BatchQueryExecutor;
import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.executor.jpa.SLASummaryQueryExecutor;
import org.apache.oozie.executor.jpa.SLASummaryQueryExecutor.SLASummaryQuery;
import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
import org.apache.oozie.sla.SLACalcStatus;
import org.apache.oozie.sla.SLACalculatorMemory;
import org.apache.oozie.sla.SLARegistrationBean;
import org.apache.oozie.sla.SLASummaryBean;
import org.apache.oozie.sla.TestSLAService;
import org.apache.oozie.sla.listener.SLAEventListener;
import org.apache.oozie.sla.service.SLAService;
import org.apache.oozie.test.ZKXTestCase;
import org.apache.oozie.util.JobUtils;
import org.apache.oozie.util.Pair;
import org.apache.oozie.workflow.WorkflowInstance;

public class TestHASLAService extends ZKXTestCase {

    private static StringBuilder output = new StringBuilder();

    protected void setUp() throws Exception {
        Configuration conf = new Configuration(false);
        conf.set(Services.CONF_SERVICE_EXT_CLASSES, "org.apache.oozie.service.EventHandlerService,"
                + "org.apache.oozie.sla.service.SLAService");
        conf.setClass(EventHandlerService.CONF_LISTENERS, DummySLAEventListener.class, SLAEventListener.class);
        conf.setLong(SLAService.CONF_JOB_EVENT_LATENCY, 0);
        // manually do check in this test
        conf.setInt(SLAService.CONF_SLA_CHECK_INITIAL_DELAY, 100000);
        conf.setInt(SLAService.CONF_SLA_CHECK_INTERVAL, 100000);
        conf.setInt(EventHandlerService.CONF_WORKER_THREADS, 0);
        super.setUp(conf);
        Services.get().setService(ZKJobsConcurrencyService.class);
    }

    @Override
    protected void tearDown() throws Exception {
        super.tearDown();
    }

    public void testSLAFailOverWithHA() throws Exception {

        SLAService slas = Services.get().get(SLAService.class);
        SLACalculatorMemory slaCalcMem = (SLACalculatorMemory) slas.getSLACalculator();
        EventHandlerService ehs = Services.get().get(EventHandlerService.class);

        // start another dummy oozie instance (dummy sla and eventhandler
        // services)
        DummyZKOozie dummyOozie_1 = null;
        try {
            dummyOozie_1 = new DummyZKOozie("a", "http://blah");
            DummySLACalculatorMemory dummyCalc = new DummySLACalculatorMemory();
            EventHandlerService dummyEhs = new EventHandlerService();
            dummyCalc.setEventHandlerService(dummyEhs);
            dummyEhs.init(Services.get());
            dummyCalc.init(Services.get().getConf());

            // Case 1 workflow job submitted to dummy server,
            // but before start running, the dummy server is down
            WorkflowJobBean wfJob1 = createWorkflow("job-1-W");
            SLARegistrationBean sla1 = TestSLAService._createSLARegistration("job-1-W", AppType.WORKFLOW_JOB);
            sla1.setExpectedStart(new Date(System.currentTimeMillis() - 2 * 3600 * 1000)); // 2 hr before
            sla1.setExpectedEnd(new Date(System.currentTimeMillis() - 1 * 3600 * 1000)); // 1 hr before
            sla1.setExpectedDuration(10 * 60 * 1000); // 10 mins
            dummyCalc.addRegistration(sla1.getId(), sla1);
            dummyCalc.updateAllSlaStatus();

            // Case 2. workflow job submitted to dummy server, start running,
            // then the dummy server is down
            WorkflowJobBean wfJob2 = createWorkflow("job-2-W");
            SLARegistrationBean sla2 = TestSLAService._createSLARegistration("job-2-W", AppType.WORKFLOW_JOB);
            sla2.setExpectedStart(new Date(System.currentTimeMillis() - 2 * 3600 * 1000)); // 2hr before
            sla2.setExpectedEnd(new Date(System.currentTimeMillis() + 1 * 3600 * 1000)); // 1hr ahead
            sla2.setExpectedDuration(10 * 60 * 1000); // 10 mins
            dummyCalc.addRegistration(sla2.getId(), sla2);
            dummyCalc.addJobStatus(sla2.getId(), WorkflowJob.Status.RUNNING.name(), EventStatus.STARTED, new Date(),
                    new Date());

            dummyCalc.updateAllSlaStatus();
            dummyEhs.new EventWorker().run();
            assertTrue(output.toString().contains(sla2.getId() + " Sla START - MISS!!!"));

            // suppose dummy Server is down
            dummyCalc.clear();
            dummyCalc = null;
            dummyOozie_1.teardown();

            slaCalcMem.updateAllSlaStatus();

            // Job 1 started running on the living server --> start miss
            slaCalcMem.addJobStatus(sla1.getId(), WorkflowJob.Status.RUNNING.name(), EventStatus.STARTED, new Date(),
                    new Date());
            // job 1 is added to slamap of living oozie server
            assertNotNull(slaCalcMem.get(sla1.getId()));
            ehs.new EventWorker().run();
            assertTrue(output.toString().contains(sla1.getId() + " Sla START - MISS!!!"));

            wfJob1.setStatus(WorkflowJob.Status.SUCCEEDED);
            wfJob1.setEndTime(new Date());
            wfJob1.setStartTime(new Date());
            WorkflowJobQueryExecutor.getInstance().executeUpdate(
                    WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MOD_START_END, wfJob1);

            // Job 1 succeeded on the living server --> duration met and end miss
            slaCalcMem.addJobStatus(sla1.getId(), WorkflowJob.Status.SUCCEEDED.name(), EventStatus.SUCCESS, new Date(),
                    new Date());
            ehs.new EventWorker().run();
            assertTrue(output.toString().contains(sla1.getId() + " Sla DURATION - MET!!!"));
            assertTrue(output.toString().contains(sla1.getId() + " Sla END - MISS!!!"));

            wfJob2.setStatus(WorkflowJob.Status.SUCCEEDED);
            wfJob2.setEndTime(new Date());
            wfJob2.setStartTime(new Date());
            WorkflowJobQueryExecutor.getInstance().executeUpdate(
                    WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MOD_START_END, wfJob2);

            // Job 2 succeeded on the living server --> duration met and end met
            slaCalcMem.addJobStatus(sla2.getId(), WorkflowJob.Status.SUCCEEDED.name(), EventStatus.SUCCESS, new Date(),
                    new Date());
            // eventProc >= 7(already processed duration/end met), should be removed from slaMap
            assertNull(slaCalcMem.get(sla2.getId()));
            ehs.new EventWorker().run();
            assertTrue(output.toString().contains(sla2.getId() + " Sla DURATION - MET!!!"));
            assertTrue(output.toString().contains(sla2.getId() + " Sla END - MET!!!"));
        }
        finally {
            if (dummyOozie_1 != null) {
                dummyOozie_1.teardown();
            }
        }
    }

    public CoordinatorActionBean updateCoordAction(String id, String status) throws JPAExecutorException {
        CoordinatorActionBean action = new CoordinatorActionBean();
        action.setId(id);
        action.setStatusStr(status);
        action.setLastModifiedTime(new Date());
        CoordActionQueryExecutor.getInstance().executeUpdate(CoordActionQuery.UPDATE_COORD_ACTION_STATUS_PENDING_TIME,
                action);
        return action;
    }

    public void testSLAUpdateWithHA() throws Exception {

        String id1 = "0000001-130521183438837-oozie-test-C@1";
        String id2 = "0000002-130521183438837-oozie-test-C@1";
        String id3 = "0000003-130521183438837-oozie-test-C@1";
        String id4 = "0000004-130521183438837-oozie-test-C@1";
        String id5 = "0000005-130521183438837-oozie-test-C@1";
        String id6 = "0000006-130521183438837-oozie-test-C@1";
        Date expectedStartTS = new Date(System.currentTimeMillis() - 2 * 3600 * 1000); // 2 hrs passed
        Date expectedEndTS1 = new Date(System.currentTimeMillis() + 1 * 3600 * 1000); // 1 hour ahead
        Date expectedEndTS2 = new Date(System.currentTimeMillis() - 1 * 3600 * 1000); // 1 hour passed
        // Coord Action of jobs 1-4 not started yet
        createDBEntry(id1, expectedStartTS, expectedEndTS1);
        createDBEntry(id2, expectedStartTS, expectedEndTS1);
        createDBEntry(id3, expectedStartTS, expectedEndTS1);
        createDBEntry(id4, expectedStartTS, expectedEndTS1);
        // Coord Action of jobs 5-6 already started and currently running (to test history set)
        createDBEntryForStarted(id5, expectedStartTS, expectedEndTS2, 1);
        createDBEntryForStarted(id6, expectedStartTS, expectedEndTS2, 1);

        SLAService slas = Services.get().get(SLAService.class);
        SLACalculatorMemory slaCalcMem = (SLACalculatorMemory) slas.getSLACalculator();
        slaCalcMem.init(Services.get().getConf());
        slaCalcMem.updateAllSlaStatus();
        List<String> slaMapKeys = new ArrayList<String>();
        Iterator<String> itr = slaCalcMem.iterator();
        while (itr.hasNext()) {
            slaMapKeys.add(itr.next());
        }
        //4 jobs expected end is not yet reached
        //2 jobs has end miss, waiting for job to complete
        assertEquals(4, slaMapKeys.size());
        assertEquals(2, slaCalcMem.getHistorySet().size());

        DummyZKOozie dummyOozie_1 = null;
        try {
            // start another dummy oozie instance (dummy sla and event handler services)
            dummyOozie_1 = new DummyZKOozie("a", "http://blah");
            DummySLACalculatorMemory dummySlaCalcMem = new DummySLACalculatorMemory();
            EventHandlerService dummyEhs = new EventHandlerService();
            dummySlaCalcMem.setEventHandlerService(dummyEhs);
            dummyEhs.init(Services.get());
            dummySlaCalcMem.init(Services.get().getConf());
            dummySlaCalcMem.updateAllSlaStatus();
            slaMapKeys = new ArrayList<String>();
            itr = dummySlaCalcMem.iterator();
            while (itr.hasNext()) {
                slaMapKeys.add(itr.next());
            }
            assertEquals(4, slaMapKeys.size());
            assertEquals(2, dummySlaCalcMem.getHistorySet().size());
            // Coord Action 1,3 run and update status on *non-dummy* server
            updateCoordAction(id1, "RUNNING");
            slaCalcMem
                    .addJobStatus(id1, CoordinatorAction.Status.RUNNING.name(), EventStatus.STARTED, new Date(), null);
            updateCoordAction(id3, "FAILED");
            slaCalcMem.addJobStatus(id3, CoordinatorAction.Status.FAILED.name(), EventStatus.FAILURE, null, new Date());

            // Coord Action 2,4 run and update status on *dummy* server
            updateCoordAction(id2, "RUNNING");
            dummySlaCalcMem.addJobStatus(id2, CoordinatorAction.Status.RUNNING.name(), EventStatus.STARTED, new Date(),
                    null);
            updateCoordAction(id4, "FAILED");
            dummySlaCalcMem.addJobStatus(id4, CoordinatorAction.Status.FAILED.name(), EventStatus.FAILURE, null,
                    new Date());

            // Both servers iterate SlaMap (updateAllSlaStatus)
            slaCalcMem.updateAllSlaStatus();
            dummySlaCalcMem.updateAllSlaStatus();

            // SlaMap on both Servers synced
            SLACalcStatus sla1_nodummy = slaCalcMem.get(id1);
            SLACalcStatus sla1_dummy = dummySlaCalcMem.get(id1);
            SLACalcStatus sla2_nodummy = slaCalcMem.get(id2);
            SLACalcStatus sla2_dummy = dummySlaCalcMem.get(id2);
            assertEquals(1, sla1_nodummy.getEventProcessed());
            assertEquals(1, sla1_dummy.getEventProcessed());
            assertEquals(1, sla2_dummy.getEventProcessed());
            assertEquals(1, sla2_nodummy.getEventProcessed());
            assertFalse(slaCalcMem.isJobIdInSLAMap(id3));
            assertFalse(dummySlaCalcMem.isJobIdInSLAMap(id3));
            assertFalse(slaCalcMem.isJobIdInSLAMap(id4));
            assertFalse(dummySlaCalcMem.isJobIdInSLAMap(id4));

            Byte eventProc = (Byte) SLASummaryQueryExecutor.getInstance().getSingleValue(
                    SLASummaryQuery.GET_SLA_SUMMARY_EVENTPROCESSED, id3);
            assertEquals(8, eventProc.byteValue());
            eventProc = (Byte) SLASummaryQueryExecutor.getInstance().getSingleValue(
                    SLASummaryQuery.GET_SLA_SUMMARY_EVENTPROCESSED, id4);
            assertEquals(8, eventProc.byteValue());

            // Action 5 was processed as END_MISS in updateAllSlaStatus, put into history set
            assertTrue(slaCalcMem.isJobIdInHistorySet(id5));
            assertTrue(dummySlaCalcMem.isJobIdInHistorySet(id6));
            // Action 6 was processed as END_MISS in updateAllSlaStatus, put into history set
            assertTrue(slaCalcMem.isJobIdInHistorySet(id5));
            assertTrue(dummySlaCalcMem.isJobIdInHistorySet(id6));

            eventProc = (Byte) SLASummaryQueryExecutor.getInstance().getSingleValue(
                    SLASummaryQuery.GET_SLA_SUMMARY_EVENTPROCESSED, id5);
            assertEquals(7, eventProc.byteValue());
            eventProc = (Byte) SLASummaryQueryExecutor.getInstance().getSingleValue(
                    SLASummaryQuery.GET_SLA_SUMMARY_EVENTPROCESSED, id6);
            assertEquals(7, eventProc.byteValue());

            // Action 1 Succeeded on non-dummy server
            updateCoordAction(id1, "SUCCEEDED");
            slaCalcMem.addJobStatus(id1, CoordinatorAction.Status.SUCCEEDED.name(), EventStatus.SUCCESS, new Date(
                    System.currentTimeMillis() - 1800 * 1000), new Date());

            // Action 2 Succeeded on dummy server
            updateCoordAction(id2, "SUCCEEDED");
            dummySlaCalcMem.addJobStatus(id2, CoordinatorAction.Status.SUCCEEDED.name(), EventStatus.SUCCESS, new Date(
                    System.currentTimeMillis() - 1800 * 1000), new Date());

            // Both servers iterate SlaMap (updateAllSlaStatus)
            slaCalcMem.updateAllSlaStatus();
            dummySlaCalcMem.updateAllSlaStatus();

            // Action 1, 2 are removed from both servers
            assertNull(slaCalcMem.get(id1));
            assertNull(dummySlaCalcMem.get(id1));
            assertNull(slaCalcMem.get(id2));
            assertNull(dummySlaCalcMem.get(id2));
            eventProc = (Byte) SLASummaryQueryExecutor.getInstance().getSingleValue(
                    SLASummaryQuery.GET_SLA_SUMMARY_EVENTPROCESSED, id1);
            assertEquals(8, eventProc.byteValue());
            eventProc = (Byte) SLASummaryQueryExecutor.getInstance().getSingleValue(
                    SLASummaryQuery.GET_SLA_SUMMARY_EVENTPROCESSED, id2);
            assertEquals(8, eventProc.byteValue());

            // Test HistoryPurgeWorker purges Action 5,6 from history set
            updateCoordAction(id5, "SUCCEEDED");
            slaCalcMem.new HistoryPurgeWorker().run();
            assertFalse(slaCalcMem.isJobIdInHistorySet(id5));
            updateCoordAction(id6, "SUCCEEDED");
            dummySlaCalcMem.new HistoryPurgeWorker().run();
            assertFalse(dummySlaCalcMem.isJobIdInHistorySet(id6));

        }
        finally {
            if (dummyOozie_1 != null) {
                dummyOozie_1.teardown();
            }
        }
    }

    public void testNoDuplicateEventsInHA() throws Exception {
        String id1 = "0000001-130521183438837-oozie-test-C@1";

        SLAService slas = Services.get().get(SLAService.class);
        SLACalculatorMemory slaCalcMem = (SLACalculatorMemory) slas.getSLACalculator();
        slaCalcMem.init(Services.get().getConf()); // loads the job in sla map


        EventHandlerService ehs = Services.get().get(EventHandlerService.class);
        EventQueue ehs_q = ehs.getEventQueue();

        DummyZKOozie dummyOozie_1 = null;
        try {
            // start another dummy oozie instance (dummy sla and event handler services)
            dummyOozie_1 = new DummyZKOozie("a", "http://blah");
            DummySLACalculatorMemory dummySlaCalcMem = new DummySLACalculatorMemory();
            dummySlaCalcMem.init(Services.get().getConf());
            EventHandlerService dummyEhs = new EventHandlerService();
            dummySlaCalcMem.setEventHandlerService(dummyEhs);
            dummyEhs.init(Services.get());
            EventQueue dummyEhs_q = dummyEhs.getEventQueue();

            Date expectedStartTS = new Date(System.currentTimeMillis() + 2 * 3600 * 1000); // get MISS
            Date expectedEndTS = new Date(System.currentTimeMillis() + 1 * 3600 * 1000); // get MISS
            SLASummaryBean sla = createDBEntryForStarted(id1, expectedStartTS, expectedEndTS, 0);
            sla.setExpectedDuration(-1);
            sla.setLastModifiedTime(new Date());
            SLASummaryQueryExecutor.getInstance().executeUpdate(SLASummaryQuery.UPDATE_SLA_SUMMARY_FOR_EXPECTED_TIMES,
                    sla);

            // Action started on Server 1
            updateCoordAction(id1, "RUNNING");

            slaCalcMem
                    .addJobStatus(id1, CoordinatorAction.Status.RUNNING.name(), EventStatus.STARTED, new Date(), null);
            assertEquals(1, ehs_q.size());
            SLACalcStatus s1 = (SLACalcStatus) ehs_q.poll();

            assertEquals(SLAStatus.IN_PROCESS, s1.getSLAStatus());
            // Action ended on Server 2
            updateCoordAction(id1, "FAILED");
            dummySlaCalcMem.addJobStatus(id1, CoordinatorAction.Status.FAILED.name(), EventStatus.FAILURE, new Date(
                    System.currentTimeMillis() - 1800 * 1000),
                    new Date());
            SLACalcStatus s2 = (SLACalcStatus) dummyEhs_q.poll();

            assertEquals(SLAStatus.MISS, s2.getSLAStatus());

            slaCalcMem.updateAllSlaStatus();
            dummySlaCalcMem.updateAllSlaStatus();
            assertEquals(0, ehs_q.size()); // no dupe event should be created again by Server 1
        }
        finally {
            if (dummyOozie_1 != null) {
                dummyOozie_1.teardown();
            }
        }
    }

    public void testSLAAlertCommandWithHA() throws Exception {

        //Test SLA ALERT commands in HA mode.
        //slaCalcMem1 is for server 1 and slaCalcMem2 is for server2

        String id = "0000001-130521183438837-oozie-test-C@1";
        Date expectedStartTS = new Date(System.currentTimeMillis() - 2 * 3600 * 1000); // 2 hrs passed
        Date expectedEndTS1 = new Date(System.currentTimeMillis() + 1 * 3600 * 1000); // 1 hour ahead
        // Coord Action of jobs 1-4 not started yet
        createDBEntry(id, expectedStartTS, expectedEndTS1);

        SLAService slas = Services.get().get(SLAService.class);
        SLACalculatorMemory slaCalcMem1 = (SLACalculatorMemory) slas.getSLACalculator();
        slaCalcMem1.init(Services.get().get(ConfigurationService.class).getConf());
        slaCalcMem1.updateAllSlaStatus();
        List<String> idList = new ArrayList<String>();
        idList.add(id);
        slaCalcMem1.disableAlert(idList);
        assertTrue(slaCalcMem1.get(id).getSLAConfigMap().containsKey(OozieClient.SLA_DISABLE_ALERT));

        DummyZKOozie dummyOozie_1 = null;
        try {
            // start another dummy oozie instance (dummy sla and event handler services)
            dummyOozie_1 = new DummyZKOozie("a", "http://blah");
            DummySLACalculatorMemory slaCalcMem2 = new DummySLACalculatorMemory();
            EventHandlerService dummyEhs = new EventHandlerService();
            slaCalcMem2.setEventHandlerService(dummyEhs);

            // So that job sla updated doesn't run automatically
            Services.get().get(ConfigurationService.class).getConf().setInt(SLAService.CONF_SLA_CHECK_INTERVAL, 100000);
            Services.get().get(ConfigurationService.class).getConf().setInt(SLAService.CONF_SLA_CHECK_INITIAL_DELAY, 100000);
            dummyEhs.init(Services.get());
            slaCalcMem2.init(Services.get().get(ConfigurationService.class).getConf());

            slaCalcMem2.updateAllSlaStatus();
            assertTrue(slaCalcMem2.get(id).getSLAConfigMap().containsKey(OozieClient.SLA_DISABLE_ALERT));

            String newParams = RestConstants.SLA_MAX_DURATION + "=5";
            List<Pair<String, Map<String, String>>> jobIdsSLAPair = new ArrayList<Pair<String, Map<String, String>>>();
            jobIdsSLAPair.add(new Pair<String, Map<String, String>>(id, JobUtils.parseChangeValue(newParams)));
            slaCalcMem1.changeDefinition(jobIdsSLAPair);
            assertEquals(slaCalcMem1.get(id).getExpectedDuration(), 5 * 60 * 1000);

            //Before update, default is 10.
            assertEquals(slaCalcMem2.get(id).getExpectedDuration(), 10 * 60 * 1000);

            slaCalcMem2.updateAllSlaStatus();
            assertEquals(slaCalcMem2.get(id).getExpectedDuration(), 5 * 60 * 1000);

            newParams = RestConstants.SLA_MAX_DURATION + "=15";
            jobIdsSLAPair.clear();
            jobIdsSLAPair.add(new Pair<String, Map<String, String>>(id, JobUtils.parseChangeValue(newParams)));
            slaCalcMem1.changeDefinition(jobIdsSLAPair);

            // Before update
            assertEquals(slaCalcMem2.get(id).getExpectedDuration(), 5 * 60 * 1000);
            slaCalcMem2.updateAllSlaStatus();
            assertEquals(slaCalcMem2.get(id).getExpectedDuration(), 15 * 60 * 1000);

        }
        finally {
            if (dummyOozie_1 != null) {
                dummyOozie_1.teardown();
            }
        }
    }

    private void createDBEntry(String actionId, Date expectedStartTS, Date expectedEndTS) throws Exception {
        ArrayList<JsonBean> insertList = new ArrayList<JsonBean>();
        CoordinatorActionBean coordAction = new CoordinatorActionBean();
        Date modTime = new Date(System.currentTimeMillis() - 1000 * 3600 * 2);
        coordAction.setId(actionId);
        coordAction.setJobId(actionId.split("@", -1)[0]);
        coordAction.setStatusStr("WAITING");
        coordAction.setLastModifiedTime(modTime);

        CoordinatorJobBean coordJob = new CoordinatorJobBean();
        coordJob.setId(actionId.split("@", -1)[0]);
        coordJob.setUser("dummy");
        coordJob.setAppName("dummy");
        coordJob.setStatusStr("RUNNING");
        coordJob.setAppNamespace("dummy");

        SLASummaryBean sla = new SLASummaryBean();
        sla.setId(actionId);
        sla.setAppType(AppType.COORDINATOR_ACTION);
        sla.setJobStatus("WAITING");
        sla.setSLAStatus(SLAStatus.NOT_STARTED);
        sla.setEventProcessed(0);
        sla.setLastModifiedTime(modTime);
        sla.setExpectedStart(expectedStartTS);
        sla.setExpectedEnd(expectedEndTS);
        sla.setExpectedDuration(10 * 60 * 1000);
        SLARegistrationBean reg = new SLARegistrationBean();
        reg.setId(actionId);
        insertList.add(coordAction);
        insertList.add(coordJob);
        insertList.add(sla);
        insertList.add(reg);
        BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, null, null);
    }

    private SLASummaryBean createDBEntryForStarted(String actionId, Date expectedStartTS, Date expectedEndTS,
            int eventProcessed) throws Exception {
        ArrayList<JsonBean> insertList = new ArrayList<JsonBean>();
        Date modTime = new Date();
        WorkflowJobBean wf = addRecordToWfJobTable(WorkflowJob.Status.RUNNING, WorkflowInstance.Status.RUNNING,
                actionId);
        wf.setStatus(wf.getStatus());
        wf.setStartTime(expectedStartTS);
        wf.setLastModifiedTime(modTime);
        WorkflowJobQueryExecutor.getInstance().executeUpdate(
                WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MOD_START_END, wf);

        CoordinatorActionBean coordAction = new CoordinatorActionBean();
        coordAction.setId(actionId);
        coordAction.setJobId(actionId.split("@", -1)[0]);
        coordAction.setStatusStr("RUNNING");
        coordAction.setLastModifiedTime(modTime);
        coordAction.setExternalId(wf.getId());

        CoordinatorJobBean coordJob = new CoordinatorJobBean();
        coordJob.setId(actionId.split("@", -1)[0]);
        coordJob.setUser("dummy");
        coordJob.setAppName("dummy");
        coordJob.setStatusStr("RUNNING");
        coordJob.setAppNamespace("dummy");

        SLASummaryBean sla = new SLASummaryBean();
        sla.setId(actionId);
        sla.setAppType(AppType.COORDINATOR_ACTION);
        sla.setJobStatus("RUNNING");
        sla.setSLAStatus(SLAStatus.IN_PROCESS);
        sla.setEventProcessed(eventProcessed);
        sla.setLastModifiedTime(modTime);
        sla.setExpectedStart(expectedStartTS);
        sla.setActualStart(expectedStartTS);
        sla.setExpectedEnd(expectedEndTS);
        sla.setExpectedDuration(10 * 60 * 1000);
        SLARegistrationBean reg = new SLARegistrationBean();
        reg.setId(actionId);
        insertList.add(coordAction);

        insertList.add(coordJob);
        insertList.add(sla);
        insertList.add(reg);
        BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, null, null);
        return sla;
    }

    private WorkflowJobBean createWorkflow(String id) throws Exception {
        List<JsonBean> insertList = new ArrayList<JsonBean>();
        WorkflowJobBean workflow = new WorkflowJobBean();
        workflow.setId(id);
        workflow.setStatusStr("PREP");
        workflow.setStartTime(new Date());
        workflow.setSlaXml("<sla></sla>");
        insertList.add(workflow);
        BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, null, null);
        return workflow;
    }

    public static class DummySLAEventListener extends SLAEventListener {

        @Override
        public void onStartMet(SLAEvent sla) {
            output.append(sla.getId() + " Sla START - MET!!!");
        }

        @Override
        public void onStartMiss(SLAEvent sla) {
            output.append(sla.getId() + " Sla START - MISS!!!");
        }

        @Override
        public void onEndMet(SLAEvent sla) {
            output.append(sla.getId() + " Sla END - MET!!!");
        }

        @Override
        public void onEndMiss(SLAEvent sla) {
            output.append(sla.getId() + " Sla END - MISS!!!");
        }

        @Override
        public void onDurationMet(SLAEvent sla) {
            output.append(sla.getId() + " Sla DURATION - MET!!!");
        }

        @Override
        public void onDurationMiss(SLAEvent sla) {
            output.append(sla.getId() + " Sla DURATION - MISS!!!");
        }

        @Override
        public void init(Configuration conf) {
        }

        @Override
        public void destroy() {
        }
    }
}

class DummySLACalculatorMemory extends SLACalculatorMemory {
    public void setEventHandlerService(EventHandlerService ehs) {
        this.eventHandler = ehs;
    }
}