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

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Set;

import org.apache.oozie.CoordinatorActionBean;
import org.apache.oozie.CoordinatorJobBean;
import org.apache.oozie.ErrorCode;
import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.CoordinatorJob;
import org.apache.oozie.client.CoordinatorJob.Execution;
import org.apache.oozie.client.CoordinatorJob.Timeunit;
import org.apache.oozie.client.Job;
import org.apache.oozie.client.rest.JsonBean;
import org.apache.oozie.command.CommandException;
import org.apache.oozie.executor.jpa.BatchQueryExecutor;
import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
import org.apache.oozie.executor.jpa.CoordJobGetActionByActionNumberJPAExecutor;
import org.apache.oozie.executor.jpa.CoordJobGetJPAExecutor;
import org.apache.oozie.executor.jpa.CoordJobInsertJPAExecutor;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.executor.jpa.SLARegistrationQueryExecutor;
import org.apache.oozie.executor.jpa.SLARegistrationQueryExecutor.SLARegQuery;
import org.apache.oozie.executor.jpa.SLASummaryQueryExecutor;
import org.apache.oozie.executor.jpa.SLASummaryQueryExecutor.SLASummaryQuery;
import org.apache.oozie.service.CallableQueueService;
import org.apache.oozie.service.JPAService;
import org.apache.oozie.service.SchedulerService;
import org.apache.oozie.service.Service;
import org.apache.oozie.service.Services;
import org.apache.oozie.service.StatusTransitService;
import org.apache.oozie.sla.SLARegistrationBean;
import org.apache.oozie.sla.SLASummaryBean;
import org.apache.oozie.store.StoreException;
import org.apache.oozie.test.XDataTestCase;
import org.apache.oozie.util.DateUtils;
import org.apache.oozie.util.XCallable;

public class TestCoordChangeXCommand extends XDataTestCase {
    private Services services;
    public static final int HOURS_IN_MS = 60 * 60 * 1000;

    /**
     * Return the UTC date and time in W3C format down to second
     * (yyyy-MM-ddTHH:mmZ). i.e.: 1997-07-16T19:20Z The input date is a
     * long (Unix Time Stamp)
     *
     * @return the formatted time string.
     */
    public static String convertDateToString(long timeStamp) {
        return DateUtils.formatDateOozieTZ(new Date(timeStamp));
    }

    /**
     * Class is used to change the queueservice, as that one meddles with the actions in the background.
     */
    static class FakeCallableQueueService extends CallableQueueService implements Service {
        @Override
        public void init(Services services){}

        @Override
        public void destroy(){}

        @Override
        public synchronized boolean queueSerial(List<? extends XCallable<?>> callables, long delay){return false;}

        @Override
        public Set<String> getInterruptTypes() {
            return Collections.emptySet();
        }
    }

    @Override
    protected void setUp() throws Exception {
        super.setUp();
        services = new Services();
        services.init();
        services.setService(FakeCallableQueueService.class);
        services.get(SchedulerService.class).destroy();
    }

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

    /**
     * Testing the Coordinator Change Commands with different combination of
     * "Change Value"
     * Changing the end time to be 20 minutes after the current time
     * Changing the pause time to be 10 minutes after the current time
     *
     * @throws Exception
     */

    public void testCoordChangeXCommand() throws StoreException, CommandException {
        System.out.println("Running Test");
        String jobId = "0000000-" + new Date().getTime() + "-testCoordChangeXCommand-C";

        try {
            addRecordToJobTable(jobId);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Exception thrown " + ex);
        }

        String pauseTime = convertDateToString(new Date().getTime() + 10 * 60 * 1000);
        String endTime = convertDateToString(new Date().getTime() + 20 * 60 * 1000);

        new CoordChangeXCommand(jobId, "endtime=" + endTime + ";concurrency=200").call();
        try {
            checkCoordJobs(jobId, DateUtils.parseDateOozieTZ(endTime), 200, null, false);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        String changeValue = "endtime=" + endTime + ";concurrency=200;pausetime=" + pauseTime;
        new CoordChangeXCommand(jobId, changeValue).call();
        try {
            checkCoordJobs(jobId, DateUtils.parseDateOozieTZ(endTime), 200, DateUtils.parseDateOozieTZ(pauseTime), true);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        new CoordChangeXCommand(jobId, "endtime=" + endTime + ";concurrency=200;pausetime=").call();
        try {
            checkCoordJobs(jobId, DateUtils.parseDateOozieTZ(endTime), 200, null, true);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        new CoordChangeXCommand(jobId, "endtime=" + endTime + ";pausetime=;concurrency=200").call();
        try {
            checkCoordJobs(jobId, DateUtils.parseDateOozieTZ(endTime), 200, null, true);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        new CoordChangeXCommand(jobId, "endtime=2012-12-20T05:00Z;concurrency=-200").call();
        try {
            checkCoordJobs(jobId, DateUtils.parseDateOozieTZ("2012-12-20T05:00Z"), -200, null, false);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        new CoordChangeXCommand(jobId, "endtime=2012-12-20T05:00Z").call();
        try {
            checkCoordJobs(jobId, DateUtils.parseDateOozieTZ("2012-12-20T05:00Z"), null, null, false);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        new CoordChangeXCommand(jobId, "concurrency=-1").call();
        try {
            checkCoordJobs(jobId, null, -1, null, false);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        try {
            new CoordChangeXCommand(jobId, "a=1;b=-200").call();
            fail("Should not reach here.");
        }
        catch (CommandException ex) {
            if (ex.getErrorCode() != ErrorCode.E1015) {
                fail("Error code should be E1015.");
            }
        }

        try {
            new CoordChangeXCommand(jobId, "endtime=2012-12-20T05:00;concurrency=-200").call();
            fail("Should not reach here.");
        }
        catch (CommandException ex) {
            if (ex.getErrorCode() != ErrorCode.E1015) {
                fail("Error code should be E1015.");
            }
        }

        try {
            new CoordChangeXCommand(jobId, "endtime=2012-12-20T05:00Z;concurrency=2ac").call();
            fail("Should not reach here.");
        }
        catch (CommandException ex) {
            if (ex.getErrorCode() != ErrorCode.E1015) {
                fail("Error code should be E1015.");
            }
        }

        try {
            new CoordChangeXCommand(jobId, "endtime=1900-12-20T05:00Z").call();
        }
        catch (CommandException ex) {
            fail("Should not throw exception");
        }

        try {
            new CoordChangeXCommand(jobId, "pausetime=null").call();
            fail("Should not reach here.");
        }
        catch (CommandException ex) {
            if (ex.getErrorCode() != ErrorCode.E1015) {
                fail("Error code should be E1015.");
            }
        }

        try {
            new CoordChangeXCommand(jobId, "pausetime=" + pauseTime).call();
        }
        catch (CommandException ex) {
            fail("Should not throw exception");
        }
    }

    /**
     * test end time change : pending should mark false if job is running with
     * pending true
     * Changing the end time to be 40 minutes after the current time
     * Changing the pause time to be 10 minutes after the current time
     *
     * @throws Exception
     */
    public void testCoordChangeEndTime() throws Exception {
        Date startTime = new Date();
        Date endTime = new Date(startTime.getTime() + (20 * 60 * 1000));

        final CoordinatorJobBean job = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, startTime, endTime,
                true, true, 0);

        String pauseTime = convertDateToString(startTime.getTime() + 10 * 60 * 1000);
        String newEndTime = convertDateToString(startTime.getTime() + 40 * 60 * 1000);

        new CoordChangeXCommand(job.getId(), "endtime=" + newEndTime + ";pausetime=" + pauseTime).call();
        try {
            checkCoordJobs(job.getId(), DateUtils.parseDateOozieTZ(newEndTime), null, DateUtils.parseDateOozieTZ(
                pauseTime),
                    true);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        JPAService jpaService = Services.get().get(JPAService.class);
        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd);
        assertEquals(Job.Status.RUNNING, coordJob.getStatus());
        assertFalse(coordJob.isDoneMaterialization());
    }

    /**
     * Testcase when changing end-time == nextMaterializedTime
     * reflects correct job status via StatusTransit
     *
     * @throws Exception
     */
    public void testCoordChangeEndTime1() throws Exception {

        JPAService jpaService = Services.get().get(JPAService.class);

        Date startTime = new Date();
        Date endTime = new Date(startTime.getTime() + (50 * 60 * 1000));
        CoordinatorJobBean coordJob = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, startTime, endTime, true, true, 1);
        coordJob.setNextMaterializedTime(new Date(startTime.getTime() + (30 * 60 * 1000)));
        CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob);
        addRecordToCoordActionTable(coordJob.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);

        Runnable runnable = new StatusTransitService.StatusTransitRunnable();
        runnable.run(); // dummy run so we get to the interval check following coord job change
        sleep(1000);

        assertEquals(endTime.getTime(), coordJob.getEndTime().getTime()); // checking before change

        String newEndTime = convertDateToString(startTime.getTime() + 30 * 60 * 1000);

        new CoordChangeXCommand(coordJob.getId(), "endtime=" + newEndTime).call();
        try {
            checkCoordJobs(coordJob.getId(), DateUtils.parseDateOozieTZ(newEndTime), null, null, false);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(coordJob.getId());
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(Job.Status.RUNNING, coordJob.getStatus());
        assertEquals(newEndTime, convertDateToString(coordJob.getEndTime().getTime())); // checking after change
        assertTrue(coordJob.isPending());
        assertTrue(coordJob.isDoneMaterialization());

        runnable.run();
        sleep(1000);
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(Job.Status.SUCCEEDED, coordJob.getStatus());
        assertFalse(coordJob.isPending());
        assertTrue(coordJob.isDoneMaterialization());
    }

    /**
     * Testcase when changing end-time > nextMaterializedTime, but < original end
     * reflects correct job state and values
     *
     * @throws Exception
     */
    public void testCoordChangeEndTime2() throws Exception {

        JPAService jpaService = Services.get().get(JPAService.class);

        Date startTime = new Date();
        Date endTime = new Date(startTime.getTime() + (50 * 60 * 1000));
        CoordinatorJobBean coordJob = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, startTime, endTime, true, true, 1);
        coordJob.setNextMaterializedTime(new Date(startTime.getTime() + (30 * 60 * 1000)));
        CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob);
        addRecordToCoordActionTable(coordJob.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);

        assertTrue(coordJob.isDoneMaterialization()); // checking initial condition before change

        Runnable runnable = new StatusTransitService.StatusTransitRunnable();
        runnable.run(); // dummy run so we get to the interval check following coord job change
        sleep(1000);

        String newEndTime = convertDateToString(startTime.getTime() + 40 * 60 * 1000);

        new CoordChangeXCommand(coordJob.getId(), "endtime=" + newEndTime).call();
        try {
            checkCoordJobs(coordJob.getId(), DateUtils.parseDateOozieTZ(newEndTime), null, null, false);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(coordJob.getId());
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(Job.Status.RUNNING, coordJob.getStatus());
        assertTrue(coordJob.isPending());
        assertFalse(coordJob.isDoneMaterialization()); // <-- changed
        assertEquals(newEndTime, convertDateToString(coordJob.getEndTime().getTime()));

    }

    /**
     * Testcase when changing end-time to after original end-time
     * but before nextMaterializedTime should not cause unnecessary changes
     *
     * @throws Exception
     */
    public void testCoordChangeEndTime3() throws Exception {
        JPAService jpaService = Services.get().get(JPAService.class);
        Date startTime = new Date();
        Date endTime = new Date(startTime.getTime() + (10 * 60 * 1000));
        CoordinatorJobBean coordJob = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, startTime, endTime, true, true, 1);
        coordJob.setNextMaterializedTime(new Date(startTime.getTime() + (40 * 60 * 1000)));
        CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob);
        addRecordToCoordActionTable(coordJob.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);

        Runnable runnable = new StatusTransitService.StatusTransitRunnable();
        runnable.run();

        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(coordJob.getId());
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(Job.Status.SUCCEEDED, coordJob.getStatus());
        assertFalse(coordJob.isPending());
        assertTrue(coordJob.isDoneMaterialization());

        String newEndTime = convertDateToString(startTime.getTime() + 20 * 60 * 1000);
        try{
        new CoordChangeXCommand(coordJob.getId(), "endtime=" + newEndTime).call();
        } catch(Exception e){
            assertTrue(e.getMessage().contains(
                    "Didn't change endtime. Endtime is in between coord end time and next materialization time"));
        }
        coordJob = jpaService.execute(coordGetCmd);
        assertFalse(Job.Status.RUNNING == coordJob.getStatus());
        assertFalse(coordJob.isPending());
        assertTrue(coordJob.isDoneMaterialization());
    }

    /**
     * Testcase when no actions are added to coord action table
     * reflects correct job state and values
     *
     * @throws Exception
     */
    public void testCoordChangeEndTime4() throws Exception {
        JPAService jpaService = Services.get().get(JPAService.class);

        Date startTime = new Date();
        Date endTime = new Date(startTime.getTime() + (50 * 60 * 1000));
        CoordinatorJobBean coordJob = addRecordToCoordJobTable(CoordinatorJob.Status.RUNNING, startTime, endTime, true, true, 1);
        coordJob.setNextMaterializedTime(new Date(startTime.getTime() + (30 * 60 * 1000)));
        CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, coordJob);

        Runnable runnable = new StatusTransitService.StatusTransitRunnable();
        runnable.run(); // dummy run so we get to the interval check following coord job change
        sleep(1000);

        assertEquals(endTime.getTime(), coordJob.getEndTime().getTime()); // checking before change

        String newEndTime = convertDateToString(startTime.getTime() + 30 * 60 * 1000);

        new CoordChangeXCommand(coordJob.getId(), "endtime=" + newEndTime).call();
        try {
            checkCoordJobs(coordJob.getId(), DateUtils.parseDateOozieTZ(newEndTime), null, null, false);
        }
        catch (Exception ex) {
            ex.printStackTrace();
            fail("Invalid date" + ex);
        }

        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(coordJob.getId());
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(Job.Status.RUNNING, coordJob.getStatus());
        assertEquals(newEndTime, convertDateToString(coordJob.getEndTime().getTime())); // checking after change
        assertTrue(coordJob.isPending());
        assertTrue(coordJob.isDoneMaterialization());

        runnable.run();
        sleep(1000);
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(Job.Status.SUCCEEDED, coordJob.getStatus());
        assertFalse(coordJob.isPending());
        assertTrue(coordJob.isDoneMaterialization());
    }

    // Testcase to test deletion of lookahead action in case of end-date change
    public void testCoordChangeEndTimeDeleteAction() throws Exception {
        Date startTime = DateUtils.parseDateOozieTZ("2013-08-01T00:00Z");
        Date endTime = DateUtils.parseDateOozieTZ("2013-08-01T05:00Z");
        Date changeEndTime = DateUtils.parseDateOozieTZ("2013-08-01T02:00Z");
        String endTimeChangeStr = "endtime=" + DateUtils.formatDateOozieTZ(changeEndTime);
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.RUNNING,
                startTime, endTime, endTime, true, false, 4);
        addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-01T00:00Z"));
        addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-01T01:00Z"));
        addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-01T02:00Z"));
        addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-01T03:00Z"));
        JPAService jpaService = Services.get().get(JPAService.class);
        new CoordChangeXCommand(job.getId(), endTimeChangeStr).call();
        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd);
        assertEquals(coordJob.getEndTime(), changeEndTime);
        assertEquals(Job.Status.RUNNING, coordJob.getStatus());
        assertEquals(2, coordJob.getLastActionNumber());
        try {
            jpaService.execute(new CoordJobGetActionByActionNumberJPAExecutor(job.getId(), 2));
        }
        catch (JPAExecutorException jpae) {
            fail(" Action should be there");
        }
        try {
            jpaService.execute(new CoordJobGetActionByActionNumberJPAExecutor(job.getId(), 3));
            fail("Expected to fail as action 3 should have been deleted");
        }
        catch (JPAExecutorException jpae) {
            assertEquals(ErrorCode.E0603, jpae.getErrorCode());
        }
        assertEquals(DateUtils.parseDateOozieTZ("2013-08-01T02:00Z"), coordJob.getNextMaterializedTime());
        assertEquals(DateUtils.parseDateOozieTZ("2013-08-01T02:00Z"), coordJob.getLastActionTime());
    }

    // Testcase to test deletion of lookahead action in case of end-date change
    // Added one more test case to test processLookaheadActions with day frequency and SUSPENDED status.
    public void testProcessLookaheadActions() throws Exception {
        Date startTime = DateUtils.parseDateOozieTZ("2013-08-01T00:00Z");
        Date endTime = DateUtils.parseDateOozieTZ("2013-08-29T00:00Z");

        Date changeEndTime = DateUtils.parseDateOozieTZ("2013-08-05T00:00Z");
        String endTimeChangeStr = "endtime=" + DateUtils.formatDateOozieTZ(changeEndTime);
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.SUSPENDED,
                startTime, endTime, endTime, true, false, 6);
        addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-01T00:00Z"));
        addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-02T00:00Z"));
        addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-03T00:00Z"));
        addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-04T00:00Z"));
        addRecordToCoordActionTable(job.getId(), 5, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-05T00:00Z"));
        addRecordToCoordActionTable(job.getId(), 6, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-06T00:00Z"));

        job.setFrequency("1");
        job.setTimeUnit(Timeunit.DAY);
        CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, job);

        JPAService jpaService = Services.get().get(JPAService.class);
        new CoordChangeXCommand(job.getId(), endTimeChangeStr).call();

        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd);
        assertEquals(coordJob.getEndTime(), changeEndTime);
        assertEquals(Job.Status.SUSPENDED, coordJob.getStatus());
        assertEquals(4, coordJob.getLastActionNumber());
        assertEquals(DateUtils.parseDateOozieTZ("2013-08-05T00:00Z"), coordJob.getNextMaterializedTime());
        assertEquals(DateUtils.parseDateOozieTZ("2013-08-05T00:00Z"), coordJob.getLastActionTime());
        assertEquals(changeEndTime, coordJob.getEndTime());
    }

    // Testcase to check status for coord whose enddate is set before startdate.
    public void testCoordChangeEndTimeBeforeStart() throws Exception {
        Date start = new Date();
        Date end = new Date(start.getTime() + (4 * HOURS_IN_MS)); // 4 hrs
        Date endTime = new Date(start.getTime() - 3000);

        String endTimeChangeStr = "endtime=" + DateUtils.formatDateOozieTZ(endTime);
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.PREP, start, end,
                end, true, false, 0);

        JPAService jpaService = Services.get().get(JPAService.class);
        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd);

        assertEquals(Job.Status.PREP, coordJob.getStatus());
        assertEquals(0, coordJob.getLastActionNumber());
        new CoordChangeXCommand(job.getId(), endTimeChangeStr).call();
        coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(DateUtils.formatDateOozieTZ(coordJob.getEndTime()), DateUtils.formatDateOozieTZ(endTime));
        assertEquals(Job.Status.SUCCEEDED, coordJob.getStatus());

        Date newEndTime = new Date(start.getTime() - 2000);
        endTimeChangeStr = "endtime=" + DateUtils.formatDateOozieTZ(newEndTime);
        new CoordChangeXCommand(job.getId(), endTimeChangeStr).call();
        coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(DateUtils.formatDateOozieTZ(coordJob.getEndTime()), DateUtils.formatDateOozieTZ(newEndTime));
        assertEquals(Job.Status.SUCCEEDED, coordJob.getStatus());

        // setting end date after startdate should make coord in running state
        newEndTime = new Date(start.getTime() + (4 * 60 * 60 * 1000));
        endTimeChangeStr = "endtime=" + DateUtils.formatDateOozieTZ(newEndTime);
        new CoordChangeXCommand(job.getId(), endTimeChangeStr).call();
        coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(DateUtils.formatDateOozieTZ(coordJob.getEndTime()), DateUtils.formatDateOozieTZ(newEndTime));
        assertEquals(Job.Status.RUNNING, coordJob.getStatus());

        // setting end date before startdate should make coord in SUCCEEDED state
        newEndTime = new Date(start.getTime() - 1000);
        endTimeChangeStr = "endtime=" + DateUtils.formatDateOozieTZ(newEndTime);
        new CoordChangeXCommand(job.getId(), endTimeChangeStr).call();
        coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        coordJob = jpaService.execute(coordGetCmd);
        assertEquals(DateUtils.formatDateOozieTZ(coordJob.getEndTime()), DateUtils.formatDateOozieTZ(newEndTime));
        assertEquals(Job.Status.SUCCEEDED, coordJob.getStatus());


    }

    /**
     * Change the pause time and end time of a failed coordinator job. Check whether the status changes
     * to RUNNINGWITHERROR
     * @throws Exception
     */
    public void testCoordChangeStatus() throws Exception {
        Services.get().destroy();
        setSystemProperty(StatusTransitService.CONF_BACKWARD_SUPPORT_FOR_STATES_WITHOUT_ERROR, "false");
        services = new Services();
        services.init();
        services.get(SchedulerService.class).destroy();
        Date startTime = new Date();
        Date endTime = new Date(startTime.getTime() + (20 * 60 * 1000));

        final CoordinatorJobBean job = addRecordToCoordJobTable(CoordinatorJob.Status.FAILED, startTime, endTime,
                true, true, 0);

        String pauseTime = convertDateToString(startTime.getTime() + 10 * 60 * 1000);
        String newEndTime = convertDateToString(startTime.getTime() + 40 * 60 * 1000);

        new CoordChangeXCommand(job.getId(), "endtime=" + newEndTime + ";pausetime=" + pauseTime).call();

        JPAService jpaService = Services.get().get(JPAService.class);
        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd);
        assertEquals(Job.Status.RUNNINGWITHERROR, coordJob.getStatus());
    }

    /**
     * test pause time change : pending should mark false if job is running with
     * pending true. two actions should be removed for pause time changes.
     *
     * @throws Exception
     */
    public void testCoordChangePauseTime() throws Exception {
        Date startTime = DateUtils.parseDateOozieTZ("2013-08-01T00:00Z");
        Date endTime = DateUtils.parseDateOozieTZ("2013-08-01T04:59Z");

        Date pauseTime = new Date(startTime.getTime() + (3 * HOURS_IN_MS));  //2 hrs
        String pauseTimeChangeStr = "pausetime=" + DateUtils.formatDateOozieTZ(pauseTime);
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.RUNNING, startTime,
                endTime, endTime, true, false, 4);
        CoordinatorActionBean ca1 = addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED,
                "coord-action-get.xml", 0);
        CoordinatorActionBean ca2 = addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED,
                "coord-action-get.xml", 0, DateUtils.parseDateOozieTZ("2013-08-01T02:00Z"));
        CoordinatorActionBean ca3 = addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.WAITING,
                "coord-action-get.xml", 0, DateUtils.parseDateOozieTZ("2013-08-01T03:00Z"));
        CoordinatorActionBean ca4 = addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING,
                "coord-action-get.xml", 0, DateUtils.parseDateOozieTZ("2013-08-01T04:00Z"));

        SLARegistrationBean slaRegBean1 = new SLARegistrationBean();
        slaRegBean1.setId(ca1.getId());
        SLARegistrationBean slaRegBean2 = new SLARegistrationBean();
        slaRegBean2.setId(ca2.getId());
        SLARegistrationBean slaRegBean3 = new SLARegistrationBean();
        slaRegBean3.setId(ca3.getId());
        SLARegistrationBean slaRegBean4 = new SLARegistrationBean();
        slaRegBean4.setId(ca4.getId());
        SLASummaryBean slaSummaryBean1 = new SLASummaryBean();
        slaSummaryBean1.setId(ca1.getId());
        SLASummaryBean slaSummaryBean3 = new SLASummaryBean();
        slaSummaryBean3.setId(ca3.getId());
        List<JsonBean> insertList = new ArrayList<JsonBean>();
        insertList.add(slaRegBean1);
        insertList.add(slaRegBean2);
        insertList.add(slaRegBean3);
        insertList.add(slaRegBean4);
        insertList.add(slaSummaryBean1);
        insertList.add(slaSummaryBean3);

        JPAService jpaService = Services.get().get(JPAService.class);
        BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, null, null);

        new CoordChangeXCommand(job.getId(), pauseTimeChangeStr).call();

        CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
        CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd);
        assertEquals(DateUtils.formatDateOozieTZ(coordJob.getPauseTime()), DateUtils.formatDateOozieTZ(pauseTime));
        assertEquals(Job.Status.RUNNING, coordJob.getStatus());
        assertEquals(2, coordJob.getLastActionNumber());
        try {
            jpaService.execute(new CoordJobGetActionByActionNumberJPAExecutor(job.getId(), 3));
            fail("Expected to fail as action 3 should have been deleted");
        }
        catch (JPAExecutorException jpae) {
            assertEquals(ErrorCode.E0603, jpae.getErrorCode());
        }

        try {
            jpaService.execute(new CoordJobGetActionByActionNumberJPAExecutor(job.getId(), 4));
            fail("Expected to fail as action 4 should have been deleted");
        }
        catch (JPAExecutorException jpae) {
            assertEquals(ErrorCode.E0603, jpae.getErrorCode());
        }

        slaRegBean1 = SLARegistrationQueryExecutor.getInstance().get(SLARegQuery.GET_SLA_REG_ALL, slaRegBean1.getId());
        assertNotNull(slaRegBean1);
        slaRegBean2 = SLARegistrationQueryExecutor.getInstance().get(SLARegQuery.GET_SLA_REG_ALL, slaRegBean2.getId());
        assertNotNull(slaRegBean2);
        slaRegBean3 = SLARegistrationQueryExecutor.getInstance().get(SLARegQuery.GET_SLA_REG_ALL, slaRegBean3.getId());
        assertNull(slaRegBean3);
        slaRegBean4 = SLARegistrationQueryExecutor.getInstance().get(SLARegQuery.GET_SLA_REG_ALL, slaRegBean4.getId());
        assertNull(slaRegBean4);
        slaSummaryBean3 = SLASummaryQueryExecutor.getInstance().get(SLASummaryQuery.GET_SLA_SUMMARY, slaSummaryBean3.getId());
        assertNull(slaSummaryBean3);
        slaSummaryBean1 = SLASummaryQueryExecutor.getInstance().get(SLASummaryQuery.GET_SLA_SUMMARY, slaSummaryBean1.getId());
        assertNotNull(slaSummaryBean1);
    }

    public void testCoordChangeConcurrency() throws Exception {
        Date startTime = DateUtils.parseDateOozieTZ("2013-08-01T00:00Z");
        Date endTime = DateUtils.parseDateOozieTZ("2013-08-01T04:59Z");
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.RUNNING,
                startTime, endTime, endTime, true, false, 4);
        CoordinatorActionBean ca1 = addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.RUNNING,
                "coord-action-get.xml", 0);
        CoordinatorActionBean ca2 = addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.RUNNING,
                "coord-action-get.xml", 0);
        CoordinatorActionBean ca3 = addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.READY,
                "coord-action-get.xml", 0);
        CoordinatorActionBean ca4 = addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.READY,
                "coord-action-get.xml", 0);
        new CoordChangeXCommand(job.getId(), "concurrency=4").call();
        Thread.sleep(100);
        ca1 = CoordActionQueryExecutor.getInstance().get(CoordActionQueryExecutor.CoordActionQuery.GET_COORD_ACTION,
                job.getId() + "@1");
        ca2 = CoordActionQueryExecutor.getInstance().get(CoordActionQueryExecutor.CoordActionQuery.GET_COORD_ACTION,
                job.getId() + "@2");
        ca3 = CoordActionQueryExecutor.getInstance().get(CoordActionQueryExecutor.CoordActionQuery.GET_COORD_ACTION,
                job.getId() + "@3");
        ca4 = CoordActionQueryExecutor.getInstance().get(CoordActionQueryExecutor.CoordActionQuery.GET_COORD_ACTION,
                job.getId() + "@4");
        assertEquals(CoordinatorAction.Status.RUNNING.toString(), ca1.getStatusStr());
        assertEquals(CoordinatorAction.Status.RUNNING.toString(), ca2.getStatusStr());
        assertEquals(CoordinatorAction.Status.READY.toString(), ca3.getStatusStr());
        assertEquals(CoordinatorAction.Status.READY.toString(), ca4.getStatusStr());
    }
    // Checks that RUNNING coord action is not deleted
    public void testChangeTimeDeleteRunning() throws Exception {
        Date startTime = DateUtils.parseDateOozieTZ("2013-08-01T00:00Z");
        Date endTime = DateUtils.parseDateOozieTZ("2013-08-01T04:59Z");
        Date pauseTime = new Date(startTime.getTime() + (2 * HOURS_IN_MS)); // 2 hrs
        String pauseTimeChangeStr = "pausetime=" + DateUtils.formatDateOozieTZ(pauseTime);
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.RUNNING,
                startTime, endTime, endTime, true, false, 4);
        addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-01T02:00Z"));
        addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-01T03:00Z"));
        addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0,
                DateUtils.parseDateOozieTZ("2013-08-01T04:00Z"));
        try {
            new CoordChangeXCommand(job.getId(), pauseTimeChangeStr).call();
            fail("Should not reach here.");
        }
        catch (CommandException e) {
            if (e.getErrorCode() != ErrorCode.E1022) {
                fail("Error code should be E1022");
            }
        }
        String endTimeChangeStr = "endtime=" + DateUtils.formatDateOozieTZ(pauseTime);
        try {
            new CoordChangeXCommand(job.getId(), endTimeChangeStr).call();
            fail("Should not reach here.");
        }
        catch (CommandException e) {
            if (e.getErrorCode() != ErrorCode.E1022) {
                fail("Error code should be E1022");
            }
        }
    }

    public void testCoordStatus_Ignored() throws Exception {
        Date start = new Date();
        Date end = new Date(start.getTime() + (5 * HOURS_IN_MS)); // 5 hrs
        String statusToRUNNING = "status=RUNNING";
        String statusToIGNORED = "status=IGNORED";
        final CoordinatorJobBean job1 = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.IGNORED, start,
                end, end, false, false, 4);
        final CoordinatorJobBean job2 = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.KILLED, start,
                end, end, false, false, 4);
        final CoordinatorJobBean job3 = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.RUNNING, start,
                end, end, false, false, 4);
        final CoordinatorJobBean job4 = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.FAILED, start,
                end, end, true, false, 4);

        // Status change from IGNORED to RUNNING
        new CoordChangeXCommand(job1.getId(), statusToRUNNING).call();
        CoordinatorJobBean coordJob = CoordJobQueryExecutor.getInstance()
                .get(CoordJobQuery.GET_COORD_JOB, job1.getId());
        assertEquals(coordJob.getStatus(), Job.Status.RUNNING);

        // Status change from KILLED -> IGNORED
        new CoordChangeXCommand(job2.getId(), statusToIGNORED).call();
        coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, job2.getId());
        assertEquals(coordJob.getStatus(), Job.Status.IGNORED);

        // Status change from RUNNING -> IGNORED
        try {
            new CoordChangeXCommand(job3.getId(), statusToIGNORED).call();
        }
        catch (CommandException ex) {
            assertEquals(ErrorCode.E1015, ex.getErrorCode());
            assertTrue(ex.getMessage().indexOf(
                    "Only FAILED or KILLED non-pending job can be changed to IGNORED") > -1);
        }
        // Status change from FAILED -> IGNORED when coord job is pending
        try {
            new CoordChangeXCommand(job4.getId(), statusToIGNORED).call();
        }
        catch (CommandException ex) {
            assertEquals(ErrorCode.E1015, ex.getErrorCode());
            assertTrue(ex.getMessage().indexOf(
                    "Only FAILED or KILLED non-pending job can be changed to IGNORED") > -1);
        }
    }
    // Status change from failed- successful
    public void testCoordStatus_Failed() throws Exception {
        Date start = new Date();
        Date end = new Date(start.getTime() + (5 * HOURS_IN_MS)); // 5 hrs
        String status = "status=RUNNING";
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.FAILED, start,
                end, end, true, false, 4);
        addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0);

        try {
            new CoordChangeXCommand(job.getId(), status).call();
            JPAService jpaService = Services.get().get(JPAService.class);

            CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
            CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd);

            assertEquals(coordJob.getStatusStr(), "RUNNING");
        }
        catch (CommandException e) {
            e.printStackTrace();
            if (e.getErrorCode() != ErrorCode.E1022) {
                fail("Error code should be E1022");
            }
        }
    }

    //  Status change from Killed- successful
    public void testCoordStatus_Killed() throws Exception {
        Date start = new Date();
        Date end = new Date(start.getTime() + (5 * HOURS_IN_MS)); // 5 hrs
        String status = "status=RUNNING";
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.KILLED, start,
                end, end, true, false, 4);
        addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0);

        try {
            new CoordChangeXCommand(job.getId(), status).call();
            JPAService jpaService = Services.get().get(JPAService.class);

            CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(job.getId());
            CoordinatorJobBean coordJob = jpaService.execute(coordGetCmd);

            assertEquals(coordJob.getStatusStr(), "RUNNING");
        }
        catch (CommandException e) {
            e.printStackTrace();
            if (e.getErrorCode() != ErrorCode.E1022) {
                fail("Error code should be E1022");
            }
        }
    }

    // Check status change from Succeeded-  exception
    public void testCoordStatus_Changefailed() throws Exception {
        Date start = new Date();
        Date end = new Date(start.getTime() + (4 * HOURS_IN_MS)); // 5 hrs
        String status = "status=RUNNING";
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.SUCCEEDED, start,
                end, end, true, false, 4);
        addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.WAITING, "coord-action-get.xml", 0);

        try {
            new CoordChangeXCommand(job.getId(), status).call();
        }
        catch (CommandException e) {
            assertTrue(e.getMessage().contains("Invalid coordinator job change value RUNNING"));
        }
    }

    // Check status change - with multiple option. Pause can't be applied to killed job, old behavior.
    public void testCoord_throwException() throws Exception {
        Date start = new Date();
        Date end = new Date(start.getTime() + (4 * HOURS_IN_MS)); // 4 hrs
        String status = "status=RUNNING;pausetime=" + DateUtils.formatDateOozieTZ(end);
        final CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.KILLED, start,
                end, end, true, false, 4);
        addRecordToCoordActionTable(job.getId(), 1, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 2, CoordinatorAction.Status.SUCCEEDED, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 3, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0);
        addRecordToCoordActionTable(job.getId(), 4, CoordinatorAction.Status.RUNNING, "coord-action-get.xml", 0);

        try {
            new CoordChangeXCommand(job.getId(), status).call();
            fail("should throw Exception");
        }
        catch (CommandException e) {
            assertTrue(e.getMessage().contains("Cannot change a killed coordinator job"));
        }
    }

    // Check status change - when job is killed with no action created
    public void testRunningStatusWithNoAction() throws Exception {
        Date now = new Date();
        Date start = new Date(now.getTime() - (4 * HOURS_IN_MS)); // 4 hrs
        Date end = new Date(now.getTime() + (4 * HOURS_IN_MS)); // 4 hrs

        String status = "status=RUNNING";
        CoordinatorJobBean job = addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status.KILLED, start, end,
                null, true, false, 0);
        job = CoordJobQueryExecutor.getInstance().get(CoordJobQueryExecutor.CoordJobQuery.GET_COORD_JOB, job.getId());
        assertEquals("KILLED", job.getStatusStr());
        assertNull(job.getNextMaterializedTime());
        new CoordChangeXCommand(job.getId(), status).call();
        job = CoordJobQueryExecutor.getInstance().get(CoordJobQueryExecutor.CoordJobQuery.GET_COORD_JOB, job.getId());
        assertEquals("RUNNING", job.getStatusStr());
        // make sure that action is created
        new CoordMaterializeTransitionXCommand(job.getId(), 3600).call();
        job = CoordJobQueryExecutor.getInstance().get(CoordJobQueryExecutor.CoordJobQuery.GET_COORD_JOB, job.getId());
        assertNotNull(job.getNextMaterializedTime());
        assertEquals(1, job.getLastActionNumber());
    }

    protected CoordinatorJobBean addRecordToCoordJobTableForPauseTimeTest(CoordinatorJob.Status status, Date start,
            Date end, Date lastActionTime, boolean pending, boolean doneMatd, int lastActionNum) throws Exception {
        CoordinatorJobBean coordJob = createCoordJob(status, start, end, pending, doneMatd, lastActionNum);
        coordJob.setFrequency("1");
        coordJob.setTimeUnit(Timeunit.HOUR);
        coordJob.setLastActionNumber(lastActionNum);
        coordJob.setLastActionTime(lastActionTime);
        try {
            JPAService jpaService = Services.get().get(JPAService.class);
            assertNotNull(jpaService);
            CoordJobInsertJPAExecutor coordInsertCmd = new CoordJobInsertJPAExecutor(coordJob);
            jpaService.execute(coordInsertCmd);
        }
        catch (JPAExecutorException je) {
            je.printStackTrace();
            fail("Unable to insert the test coord job record to table");
            throw je;
        }

        return coordJob;

    }

    private void addRecordToJobTable(String jobId) throws Exception {
        CoordinatorJobBean coordJob = new CoordinatorJobBean();
        coordJob.setId(jobId);
        coordJob.setAppName("testApp");
        coordJob.setAppPath("testAppPath");
        coordJob.setStatus(CoordinatorJob.Status.SUCCEEDED);
        coordJob.setCreatedTime(new Date());
        coordJob.setLastModifiedTime(DateUtils.parseDateOozieTZ("2009-01-02T23:59Z"));
        coordJob.setTimeZone("UTC");
        coordJob.setTimeUnit(Timeunit.MINUTE);
        coordJob.setUser("testUser");
        coordJob.setGroup("testGroup");

        String confStr = "<configuration></configuration>";
        coordJob.setConf(confStr);
        String appXml = "<coordinator-app xmlns='uri:oozie:coordinator:0.2' name='NAME' frequency=\"5\" start='2009-02-01T01:00Z'"
                + " end='2009-02-01T01:09Z' timezone='UTC' freq_timeunit='MINUTE' end_of_duration='NONE'>";
        appXml += "<controls>";
        appXml += "<timeout>10</timeout>";
        appXml += "<concurrency>2</concurrency>";
        appXml += "<execution>LIFO</execution>";
        appXml += "</controls>";
        appXml += "<input-events>";
        appXml += "<data-in name='A' dataset='a'>";
        appXml += "<dataset name='a' frequency='5' initial-instance='2009-02-01T01:00Z' timezone='UTC' freq_timeunit='MINUTE'"
                + " end_of_duration='NONE'>";
        appXml += "<uri-template>file:///tmp/coord/workflows/${YEAR}/${DAY}</uri-template>";
        appXml += "</dataset>";
        appXml += "<instance>${coord:latest(0)}</instance>";
        appXml += "</data-in>";
        appXml += "</input-events>";
        appXml += "<output-events>";
        appXml += "<data-out name='LOCAL_A' dataset='local_a'>";
        appXml += "<dataset name='local_a' frequency='5' initial-instance='2009-02-01T01:00Z' timezone='UTC' freq_timeunit="
                + "'MINUTE' end_of_duration='NONE'>";
        appXml += "<uri-template>file:///tmp/coord/workflows/${YEAR}/${DAY}</uri-template>";
        appXml += "</dataset>";
        appXml += "<instance>${coord:current(-1)}</instance>";
        appXml += "</data-out>";
        appXml += "</output-events>";
        appXml += "<action>";
        appXml += "<workflow>";
        appXml += "<app-path>hdfs:///tmp/workflows/</app-path>";
        appXml += "<configuration>";
        appXml += "<property>";
        appXml += "<name>inputA</name>";
        appXml += "<value>${coord:dataIn('A')}</value>";
        appXml += "</property>";
        appXml += "<property>";
        appXml += "<name>inputB</name>";
        appXml += "<value>${coord:dataOut('LOCAL_A')}</value>";
        appXml += "</property>";
        appXml += "</configuration>";
        appXml += "</workflow>";
        appXml += "</action>";
        appXml += "</coordinator-app>";
        coordJob.setJobXml(appXml);
        coordJob.setLastActionNumber(0);
        coordJob.setFrequency("5");
        coordJob.setExecutionOrder(Execution.FIFO);
        coordJob.setConcurrency(1);
        try {
            coordJob.setStartTime(DateUtils.parseDateOozieTZ("2009-02-01T01:00Z"));
            coordJob.setEndTime(DateUtils.parseDateOozieTZ("2009-02-01T01:09Z"));
            coordJob.setLastActionTime(DateUtils.parseDateOozieTZ("2009-02-01T01:10Z"));
        }
        catch (Exception e) {
            e.printStackTrace();
            fail("Could not set Date/time");
        }

        try {
            JPAService jpaService = Services.get().get(JPAService.class);
            assertNotNull(jpaService);
            CoordJobInsertJPAExecutor coordInsertCmd = new CoordJobInsertJPAExecutor(coordJob);
            jpaService.execute(coordInsertCmd);
        }
        catch (JPAExecutorException je) {
            je.printStackTrace();
            fail("Unable to insert the test coord job record to table");
            throw je;
        }
    }

    private void checkCoordJobs(String jobId, Date endTime, Integer concurrency, Date pauseTime, boolean checkPauseTime)
            throws StoreException {
        try {
            JPAService jpaService = Services.get().get(JPAService.class);
            assertNotNull(jpaService);
            CoordJobGetJPAExecutor coordGetCmd = new CoordJobGetJPAExecutor(jobId);
            CoordinatorJobBean job = null;

            job = jpaService.execute(coordGetCmd);

            if (endTime != null) {
                Date d = job.getEndTime();
                if (d.compareTo(endTime) != 0) {
                    fail("Endtime is not updated properly job_end_time=" + d + ", expected_end_time=" + endTime);
                }

                CoordinatorJob.Status status = job.getStatus();
                if (status != CoordinatorJob.Status.RUNNING) {
                    fail("Coordinator job's status is not updated properly");
                }
            }

            if (concurrency != null) {
                int c = job.getConcurrency();

                if (c != concurrency) {
                    fail("Concurrency is not updated properly");
                }
            }

            if (checkPauseTime) {
                Date d = job.getPauseTime();
                if (pauseTime == null) {
                    if (d != null) {
                        fail("Pausetime is not updated properly job_pause_time=" + d + ", expected_pause_time="
                                + pauseTime);
                    }
                }
                else {
                    if (d.compareTo(pauseTime) != 0) {
                        fail("Pausetime is not updated properly job_pause_time=" + d + ", expected_pause_time="
                                + pauseTime);
                    }
                }
            }
        }
        catch (JPAExecutorException e) {
            e.printStackTrace();
        }

    }
}
