/**
 * 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.falcon.regression.prism;

import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.Frequency;
import org.apache.falcon.entity.v0.Frequency.TimeUnit;
import org.apache.falcon.entity.v0.feed.ClusterType;
import org.apache.falcon.entity.v0.process.ExecutionType;
import org.apache.falcon.regression.Entities.ProcessMerlin;
import org.apache.falcon.regression.core.bundle.Bundle;
import org.apache.falcon.regression.core.helpers.ColoHelper;
import org.apache.falcon.regression.core.response.APIResult;
import org.apache.falcon.regression.core.response.ServiceResponse;
import org.apache.falcon.regression.core.supportClasses.HadoopFileEditor;
import org.apache.falcon.regression.core.util.AssertUtil;
import org.apache.falcon.regression.core.util.BundleUtil;
import org.apache.falcon.regression.core.util.HadoopUtil;
import org.apache.falcon.regression.core.util.InstanceUtil;
import org.apache.falcon.regression.core.util.OSUtil;
import org.apache.falcon.regression.core.util.OozieUtil;
import org.apache.falcon.regression.core.util.TimeUtil;
import org.apache.falcon.regression.core.util.Util;
import org.apache.falcon.regression.core.util.XmlUtil;
import org.apache.falcon.regression.testHelper.BaseTestClass;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.apache.log4j.Logger;
import org.apache.oozie.client.BundleJob;
import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.CoordinatorJob;
import org.apache.oozie.client.Job;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.client.OozieClientException;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Minutes;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import javax.xml.bind.JAXBException;
import java.io.IOException;
import java.lang.reflect.Method;
import java.net.URISyntaxException;
import java.util.Date;
import java.util.List;
import java.util.Random;

/**
 * test for process update.
 */
@Test(groups = "distributed")
public class NewPrismProcessUpdateTest extends BaseTestClass {

    private String baseTestDir = baseHDFSDir + "/NewPrismProcessUpdateTest";
    private String inputFeedPath = baseTestDir + MINUTE_DATE_PATTERN;
    private String workflowPath = baseTestDir + "/falcon-oozie-wf";
    private String workflowPath2 = baseTestDir + "/falcon-oozie-wf2";
    private String aggregatorPath = baseTestDir + "/aggregator";
    private String aggregator1Path = baseTestDir + "/aggregator1";
    private ColoHelper cluster1 = servers.get(0);
    private ColoHelper cluster2 = servers.get(1);
    private ColoHelper cluster3 = servers.get(2);
    private FileSystem cluster1FS = serverFS.get(0);
    private OozieClient cluster2OC = serverOC.get(1);
    private OozieClient cluster3OC = serverOC.get(2);
    private static final Logger LOGGER = Logger.getLogger(NewPrismProcessUpdateTest.class);

    @BeforeMethod(alwaysRun = true)
    public void testSetup(Method method) throws Exception {
        LOGGER.info("test name: " + method.getName());
        Bundle b = BundleUtil.readUpdateBundle();
        bundles[0] = new Bundle(b, cluster1);
        bundles[0].generateUniqueBundle();
        bundles[1] = new Bundle(b, cluster2);
        bundles[1].generateUniqueBundle();
        bundles[2] = new Bundle(b, cluster3);
        bundles[2].generateUniqueBundle();
        setBundleWFPath(bundles[0], bundles[1], bundles[2]);
        bundles[1].addClusterToBundle(bundles[2].getClusters().get(0),
                ClusterType.TARGET, null, null);
        usualGrind(bundles[1]);
        Util.restartService(cluster3.getClusterHelper());
    }

    @BeforeClass(alwaysRun = true)
    public void setup() throws Exception {
        for (String wfPath : new String[] { workflowPath, workflowPath2, aggregatorPath, aggregator1Path }) {
            uploadDirToClusters(wfPath, OSUtil.RESOURCES_OOZIE);
        }
        Util.restartService(cluster3.getClusterHelper());
        Util.restartService(cluster1.getClusterHelper());
        Util.restartService(cluster2.getClusterHelper());
    }

    @AfterMethod(alwaysRun = true)
    public void tearDown() {
        removeBundles();
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessFrequencyInEachColoWithOneProcessRunningMonthly()
        throws Exception {
        final String startTIme = TimeUtil.getTimeWrtSystemTime(-20);
        String endTime = TimeUtil.getTimeWrtSystemTime(4000 * 60);
        bundles[1].setProcessPeriodicity(1, TimeUnit.months);
        bundles[1].setOutputFeedPeriodicity(1, TimeUnit.months);
        bundles[1].setProcessValidity(startTIme, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);
        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        String updatedProcess = InstanceUtil
                .setProcessFrequency(bundles[1].getProcessData(),
                        new Frequency("" + 5, TimeUnit.minutes));

        LOGGER.info("updated process: " + Util.prettyPrintXml(updatedProcess));

        //now to update
        while (Util
                .parseResponse(prism.getProcessHelper()
                        .update((bundles[1].getProcessData()), updatedProcess))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            LOGGER.info("update didn't SUCCEED in last attempt");
            TimeUtil.sleepSeconds(10);
        }

        String prismString = getResponse(prism, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(prismString).getFrequency(),
                Util.getProcessObject(updatedProcess).getFrequency());
        TimeUtil.sleepSeconds(60);
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, false);
        waitingForBundleFinish(cluster3, oldBundleId, 5);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 1, 10);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);

    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    //failing due to falcon bug : https://issues.apache.org/jira/browse/FALCON-458
    public void updateProcessRollStartTimeForwardInEachColoWithOneProcessRunning()
        throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(3);
        String endTime = TimeUtil.getTimeWrtSystemTime(7);
        bundles[1].setProcessValidity(startTime, endTime);
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        List<String> oldNominalTimes =
                OozieUtil.getActionsNominalTime(cluster3, oldBundleId, EntityType.PROCESS);

        String newStartTime = TimeUtil.addMinsToTime(TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getStart()
        ), 20);
        String newEndTime = TimeUtil.addMinsToTime(TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getStart()
        ), 25);

        bundles[1].setProcessValidity(newStartTime, newEndTime);
        bundles[1].setProcessConcurrency(10);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        LOGGER.info("updated process: " + Util.prettyPrintXml(bundles[1].getProcessData()));
        while (Util.parseResponse(
                prism.getProcessHelper()
                        .update(bundles[1].getProcessData(), Util.prettyPrintXml(bundles[1]
                                .getProcessData())))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            LOGGER.info("update didn't SUCCEED in last attempt");
            TimeUtil.sleepSeconds(10);
        }

        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, false);

        dualComparison(prism, cluster3, bundles[1].getProcessData());

        waitingForBundleFinish(cluster3, oldBundleId, 15);
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        int finalNumberOfInstances =
                InstanceUtil.getProcessInstanceListFromAllBundles(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS).size();
        Assert.assertEquals(finalNumberOfInstances,
                getExpectedNumberOfWorkflowInstances(TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                                .getValidity().getStart()),
                        TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters()
                                                .get(0).getValidity()
                                                .getEnd())));
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
        int expectedNumberOfWorkflows =
                getExpectedNumberOfWorkflowInstances(newStartTime, TimeUtil
                        .dateToOozieDate(
                                bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                        .getValidity().getEnd()));
        Assert.assertEquals(OozieUtil.getNumberOfWorkflowInstances(cluster3, oldBundleId),
                expectedNumberOfWorkflows);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1800000)
    public void updateProcessConcurrencyWorkflowExecutionInEachColoWithOneColoDown()
        throws Exception {
        //bundles[1].generateUniqueBundle();
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        TimeUtil.sleepSeconds(25);

        int initialConcurrency = bundles[1].getProcessObject().getParallel();

        bundles[1].setProcessConcurrency(bundles[1].getProcessObject().getParallel() + 3);
        bundles[1].setProcessWorkflow(workflowPath2);
        bundles[1].getProcessObject().setOrder(getRandomExecutionType(bundles[1]));

        //stop cluster 3 where process is scheduled
        Util.shutDownService(cluster3.getProcessHelper());

        //now to update
        AssertUtil.assertPartial(
                prism.getProcessHelper()
                        .update(bundles[1].getProcessData(), bundles[1].getProcessData()));
        String prismString = getResponse(prism, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(prismString).getParallel(),
                initialConcurrency);
        Assert.assertEquals(Util.getProcessObject(prismString).getWorkflow().getPath(),
                workflowPath);
        Assert.assertEquals(Util.getProcessObject(prismString).getOrder(),
                bundles[1].getProcessObject().getOrder());

        String coloString = getResponse(cluster2, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(coloString).getWorkflow().getPath(),
                workflowPath2);

        Util.startService(cluster3.getProcessHelper());
        dualComparisonFailure(prism, cluster2, bundles[1].getProcessData());
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        AssertUtil
                .checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        while (Util.parseResponse(
                prism.getProcessHelper()
                        .update(bundles[1].getProcessData(), bundles[1].getProcessData()))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            LOGGER.info("WARNING: update did not succeed, retrying ");
            TimeUtil.sleepSeconds(20);
        }
        prismString = getResponse(prism, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(prismString).getParallel(),
                initialConcurrency + 3);
        Assert.assertEquals(Util.getProcessObject(prismString).getWorkflow().getPath(),
                workflowPath2);
        Assert.assertEquals(Util.getProcessObject(prismString).getOrder(),
                bundles[1].getProcessObject().getOrder());
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        AssertUtil
                .checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
        waitingForBundleFinish(cluster3, oldBundleId);
        int finalNumberOfInstances =
                InstanceUtil.getProcessInstanceListFromAllBundles(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS).size();

        int expectedInstances =
                getExpectedNumberOfWorkflowInstances(TimeUtil.dateToOozieDate(
                                bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                        .getValidity().getStart()),
                        TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters()
                                                .get(0).getValidity()
                                                .getEnd()));
        Assert.assertEquals(finalNumberOfInstances, expectedInstances,
                "number of instances doesnt match :(");

    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessFrequencyInEachColoWithOneProcessRunning() throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(-2);
        String endTime = TimeUtil.getTimeWrtSystemTime(20);
        bundles[1].setProcessValidity(startTime, endTime);
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);
        LOGGER.info("original process: " + Util.prettyPrintXml(bundles[1].getProcessData()));

        String updatedProcess = InstanceUtil
                .setProcessFrequency(bundles[1].getProcessData(),
                        new Frequency("" + 7, TimeUnit.minutes));

        LOGGER.info("updated process: " + updatedProcess);

        //now to update

        ServiceResponse response =
                prism.getProcessHelper().update(updatedProcess, updatedProcess);
        AssertUtil.assertSucceeded(response);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, false);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 1, 10);

        String prismString = getResponse(prism, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(prismString).getFrequency(),
                Util.getProcessObject(updatedProcess).getFrequency());
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessNameInEachColoWithOneProcessRunning() throws Exception {
        //bundles[1].generateUniqueBundle();
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        String originalProcessData = bundles[1].getProcessData();
        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        TimeUtil.sleepSeconds(20);
        List<String> oldNominalTimes =
                OozieUtil.getActionsNominalTime(cluster3, oldBundleId, EntityType.PROCESS);
        bundles[1].setProcessName("myNewProcessName");

        //now to update
        ServiceResponse response =
                prism.getProcessHelper()
                        .update((bundles[1].getProcessData()), bundles[1].getProcessData());
        AssertUtil.assertFailed(response);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                originalProcessData, false, false);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessConcurrencyInEachColoWithOneProcessRunning()
        throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(-2);
        String endTime = TimeUtil.getTimeWrtSystemTime(10);
        bundles[1].setProcessValidity(startTime, endTime);

        //bundles[1].generateUniqueBundle();
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        //now to update
        DateTime updateTime = new DateTime(DateTimeZone.UTC);
        TimeUtil.sleepSeconds(60);
        List<String> oldNominalTimes =
                OozieUtil.getActionsNominalTime(cluster3, oldBundleId, EntityType.PROCESS);
        LOGGER.info("updating at " + updateTime);
        while (Util
                .parseResponse(updateProcessConcurrency(bundles[1],
                        bundles[1].getProcessObject().getParallel() + 3))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            LOGGER.info("WARNING: update did not scceed, retyring ");
            TimeUtil.sleepSeconds(20);
        }

        String prismString = getResponse(prism, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(prismString).getParallel(),
                bundles[1].getProcessObject().getParallel() + 3);
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated
        // correctly.
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(),
                false, true);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);

        // future : should be verified using cord xml
        Job.Status status = OozieUtil.getOozieJobStatus(cluster3.getFeedHelper().getOozieClient(),
                Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        boolean doesExist = false;
        OozieUtil.createMissingDependencies(cluster3, EntityType.PROCESS, bundles[1].getProcessName(), 0);
        while (status != Job.Status.SUCCEEDED && status != Job.Status.FAILED
                &&
                status != Job.Status.DONEWITHERROR) {
            int statusCount = InstanceUtil
                    .getInstanceCountWithStatus(cluster3,
                            Util.readEntityName(bundles[1].getProcessData()),
                            org.apache.oozie.client.CoordinatorAction.Status.RUNNING,
                            EntityType.PROCESS);
            if (statusCount == bundles[1].getProcessObject().getParallel() + 3) {
                doesExist = true;
                break;
            }
            status = OozieUtil.getOozieJobStatus(cluster3.getFeedHelper().getOozieClient(),
                    Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
            Assert.assertNotNull(status,
                    "status must not be null!");
            TimeUtil.sleepSeconds(30);
        }

        Assert.assertTrue(doesExist, "Er! The desired concurrency levels are never reached!!!");
        int expectedNumberOfInstances =
                getExpectedNumberOfWorkflowInstances(TimeUtil.dateToOozieDate(
                                bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                        .getValidity().getStart()),
                        TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters()
                                                .get(0).getValidity()
                                                .getEnd()));
        Assert.assertEquals(OozieUtil.getNumberOfWorkflowInstances(cluster3, oldBundleId),
                expectedNumberOfInstances);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessIncreaseValidityInEachColoWithOneProcessRunning() throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(3);
        String endTime = TimeUtil.getTimeWrtSystemTime(8);
        bundles[1].setProcessValidity(startTime, endTime);
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        String newEndTime = TimeUtil.addMinsToTime(TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getEnd()
        ), 4);
        bundles[1].setProcessValidity(TimeUtil.dateToOozieDate(
                        bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                                .getStart()),
                newEndTime);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        ServiceResponse response = prism.getProcessHelper()
                .update(bundles[1].getProcessData(), bundles[1].getProcessData());
        for (int i = 0; i < 10
                &&
                Util.parseResponse(response).getStatus() != APIResult.Status.SUCCEEDED; ++i) {
            response = prism.getProcessHelper()
                    .update(bundles[1].getProcessData(), bundles[1].getProcessData());
            TimeUtil.sleepSeconds(6);
        }
        Assert.assertEquals(Util.parseResponse(response).getStatus(),
                APIResult.Status.SUCCEEDED, "Process update did not succeed.");

        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), false, true);

        int i = 0;

        while (OozieUtil.getNumberOfWorkflowInstances(cluster3, oldBundleId)
                != getExpectedNumberOfWorkflowInstances(TimeUtil.dateToOozieDate(
                        bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                                .getStart()
                ), TimeUtil.dateToOozieDate(
                        bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                .getValidity()
                                .getEnd()))
                && i < 10) {
            TimeUtil.sleepSeconds(1);
            i++;
        }

        bundles[1].verifyDependencyListing(cluster2);

        dualComparison(prism, cluster3, bundles[1].getProcessData());
        waitingForBundleFinish(cluster3, oldBundleId);
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated
        // correctly.
        int finalNumberOfInstances = InstanceUtil
                .getProcessInstanceList(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS)
                .size();
        Assert.assertEquals(finalNumberOfInstances,
                getExpectedNumberOfWorkflowInstances(TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                                .getValidity().getStart()),
                        TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters()
                                                .get(0).getValidity()
                                                .getEnd())));
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessConcurrencyInEachColoWithOneProcessSuspended()
        throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(3);
        String endTime = TimeUtil.getTimeWrtSystemTime(7);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().suspend(bundles[1].getProcessData()));
        //now to update
        while (Util
                .parseResponse(updateProcessConcurrency(bundles[1],
                        bundles[1].getProcessObject().getParallel() + 3))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            LOGGER.info("WARNING: update did not scceed, retyring ");
            TimeUtil.sleepSeconds(20);
        }

        String prismString = getResponse(prism, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(prismString).getParallel(),
                bundles[1].getProcessObject().getParallel() + 3);
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), false, true);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
        AssertUtil.assertSucceeded(cluster3.getProcessHelper().resume(bundles[1].getProcessData()));
        AssertUtil.checkStatus(cluster3OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);

        Job.Status status = OozieUtil.getOozieJobStatus(cluster3.getFeedHelper().getOozieClient(),
                Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        boolean doesExist = false;
        OozieUtil.createMissingDependencies(cluster3, EntityType.PROCESS, bundles[1].getProcessName(), 0);
        while (status != Job.Status.SUCCEEDED && status != Job.Status.FAILED
                &&
                status != Job.Status.DONEWITHERROR) {
            if (InstanceUtil
                    .getInstanceCountWithStatus(cluster3,
                            Util.readEntityName(bundles[1].getProcessData()),
                            org.apache.oozie.client.CoordinatorAction.Status.RUNNING,
                            EntityType.PROCESS)
                    ==
                    bundles[1].getProcessObject().getParallel()) {
                doesExist = true;
                break;
            }
            status = OozieUtil.getOozieJobStatus(cluster3.getFeedHelper().getOozieClient(),
                    Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        }

        Assert.assertTrue(doesExist, "Er! The desired concurrency levels are never reached!!!");
        waitingForBundleFinish(cluster3, oldBundleId);

        int finalNumberOfInstances =
                InstanceUtil.getProcessInstanceListFromAllBundles(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS).size();

        int expectedInstances =
                getExpectedNumberOfWorkflowInstances(TimeUtil.dateToOozieDate(
                                bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                        .getValidity().getStart()),
                        TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters()
                                                .get(0).getValidity()
                                                .getEnd()));

        Assert.assertEquals(finalNumberOfInstances, expectedInstances,
                "number of instances doesnt match :(");
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessConcurrencyInEachColoWithOneColoDown() throws Exception {

        String startTime = TimeUtil.getTimeWrtSystemTime(-1);
        String endTime = TimeUtil.getTimeWrtSystemTime(5);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another

        LOGGER.info("process to be scheduled: " + Util.prettyPrintXml(bundles[1].getProcessData()));

        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        //now to update
        Util.shutDownService(cluster3.getClusterHelper());

        ServiceResponse response =
                updateProcessConcurrency(bundles[1],
                        bundles[1].getProcessObject().getParallel() + 3);
        AssertUtil.assertPartial(response);

        Util.startService(cluster3.getClusterHelper());

        String prismString = getResponse(prism, bundles[1].getProcessData(), true);
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        Assert.assertEquals(Util.getProcessObject(prismString).getParallel(),
                bundles[1].getProcessObject().getParallel());

        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1],
                Job.Status.RUNNING);

        while (Util
                .parseResponse(updateProcessConcurrency(bundles[1],
                        bundles[1].getProcessObject().getParallel() + 3))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            LOGGER.info("WARNING: update did not scceed, retyring ");
            TimeUtil.sleepSeconds(20);
        }
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        dualComparison(prism, cluster2, bundles[1].getProcessData());

        Job.Status status =
                OozieUtil.getOozieJobStatus(cluster3.getFeedHelper().getOozieClient(),
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        boolean doesExist = false;
        OozieUtil.createMissingDependencies(cluster3, EntityType.PROCESS, bundles[1].getProcessName(), 0);
        while (status != Job.Status.SUCCEEDED && status != Job.Status.FAILED
                &&
                status != Job.Status.DONEWITHERROR) {
            if (InstanceUtil
                    .getInstanceCountWithStatus(cluster3,
                            Util.readEntityName(bundles[1].getProcessData()),
                            org.apache.oozie.client.CoordinatorAction.Status.RUNNING,
                            EntityType.PROCESS)
                    ==
                    bundles[1].getProcessObject().getParallel() + 3) {
                doesExist = true;
                break;
            }
            status = OozieUtil.getOozieJobStatus(cluster3.getFeedHelper().getOozieClient(),
                    Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
            TimeUtil.sleepSeconds(30);
        }
        Assert.assertTrue(doesExist, "Er! The desired concurrency levels are never reached!!!");
        OozieUtil.verifyNewBundleCreation(cluster3, InstanceUtil
                        .getLatestBundleID(cluster3,
                                Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS),
                oldNominalTimes, bundles[1].getProcessData(), false,
                true
        );

        waitingForBundleFinish(cluster3, oldBundleId);

        int finalNumberOfInstances =
                InstanceUtil.getProcessInstanceListFromAllBundles(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS).size();

        int expectedInstances =
                getExpectedNumberOfWorkflowInstances(TimeUtil.dateToOozieDate(
                                bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                        .getValidity().getStart()),
                        TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters()
                                                .get(0).getValidity()
                                                .getEnd()));
        Assert.assertEquals(finalNumberOfInstances, expectedInstances,
                "number of instances doesnt match :(");
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessConcurrencyExecutionWorkflowInEachColoWithOneProcessRunning()
        throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(-1);
        String endTime = TimeUtil.getTimeWrtSystemTime(7);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        int initialConcurrency = bundles[1].getProcessObject().getParallel();

        bundles[1].setProcessConcurrency(bundles[1].getProcessObject().getParallel() + 3);
        bundles[1].setProcessWorkflow(aggregator1Path);
        bundles[1].getProcessObject().setOrder(getRandomExecutionType(bundles[1]));

        //now to update

        String updateTime = new DateTime(DateTimeZone.UTC).plusMinutes(2).toString();

        LOGGER.info("updating @ " + updateTime);

        while (Util.parseResponse(
                prism.getProcessHelper().update((bundles[1].getProcessData()), bundles[1]
                        .getProcessData())).getStatus() != APIResult.Status.SUCCEEDED) {
            TimeUtil.sleepSeconds(10);
        }
        String prismString = getResponse(prism, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(prismString).getParallel(),
                initialConcurrency + 3);
        Assert.assertEquals(Util.getProcessObject(prismString).getWorkflow().getPath(),
                aggregator1Path);
        Assert.assertEquals(Util.getProcessObject(prismString).getOrder(),
                bundles[1].getProcessObject().getOrder());
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        waitingForBundleFinish(cluster3, oldBundleId);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
        int finalNumberOfInstances =
                InstanceUtil.getProcessInstanceListFromAllBundles(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS).size();
        int expectedInstances =
                getExpectedNumberOfWorkflowInstances(TimeUtil.dateToOozieDate(
                                bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                        .getValidity().getStart()),
                        TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters()
                                                .get(0).getValidity()
                                                .getEnd()));
        Assert.assertEquals(finalNumberOfInstances, expectedInstances,
                "number of instances doesnt match :(");
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessConcurrencyExecutionWorkflowInEachColoWithOneProcessSuspended()
        throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(2);
        String endTime = TimeUtil.getTimeWrtSystemTime(6);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        int initialConcurrency = bundles[1].getProcessObject().getParallel();

        bundles[1].setProcessConcurrency(bundles[1].getProcessObject().getParallel() + 3);
        bundles[1].setProcessWorkflow(aggregator1Path);
        bundles[1].getProcessObject().setOrder(getRandomExecutionType(bundles[1]));
        //suspend
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().suspend(bundles[1].getProcessData()));

        //now to update
        String updateTime = new DateTime(DateTimeZone.UTC).plusMinutes(2).toString();
        LOGGER.info("updating @ " + updateTime);
        while (Util.parseResponse(
                prism.getProcessHelper()
                        .update((bundles[1].getProcessData()), bundles[1].getProcessData()))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            TimeUtil.sleepSeconds(10);
        }

        AssertUtil.assertSucceeded(cluster3.getProcessHelper().resume(bundles[1].getProcessData()));

        String prismString = getResponse(prism, bundles[1].getProcessData(), true);
        Assert.assertEquals(Util.getProcessObject(prismString).getParallel(),
                initialConcurrency + 3);
        Assert.assertEquals(Util.getProcessObject(prismString).getWorkflow().getPath(),
                aggregator1Path);
        Assert.assertEquals(Util.getProcessObject(prismString).getOrder(),
                bundles[1].getProcessObject().getOrder());
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        waitingForBundleFinish(cluster3, oldBundleId);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);
        AssertUtil.checkNotStatus(cluster3OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
        int finalNumberOfInstances =
                InstanceUtil.getProcessInstanceListFromAllBundles(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS).size();

        int expectedInstances =
                getExpectedNumberOfWorkflowInstances(TimeUtil.dateToOozieDate(
                                bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                        .getValidity().getStart()),
                        TimeUtil
                                .dateToOozieDate(
                                        bundles[1].getProcessObject().getClusters().getClusters()
                                                .get(0).getValidity()
                                                .getEnd()));
        Assert.assertEquals(finalNumberOfInstances, expectedInstances,
                "number of instances doesnt match :(");
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessAddNewInputInEachColoWithOneProcessRunning() throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(-1);
        String endTime = TimeUtil.getTimeWrtSystemTime(6);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        TimeUtil.sleepSeconds(20);
        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        String newFeedName = bundles[1].getInputFeedNameFromBundle() + "2";
        String inputFeed = bundles[1].getInputFeedFromBundle();

        bundles[1].addProcessInput(newFeedName, "inputData2");
        inputFeed = Util.setFeedName(inputFeed, newFeedName);

        LOGGER.info(inputFeed);
        AssertUtil.assertSucceeded(prism.getFeedHelper().submitEntity(inputFeed));

        while (Util.parseResponse(
                prism.getProcessHelper()
                        .update((bundles[1].getProcessData()), bundles[1].getProcessData()))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            TimeUtil.sleepSeconds(20);
        }
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, false);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 1, 10);

        bundles[1].verifyDependencyListing(cluster2);

        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        waitingForBundleFinish(cluster3, oldBundleId);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessAddNewInputInEachColoWithOneProcessSuspended() throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(1);
        String endTime = TimeUtil.getTimeWrtSystemTime(6);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        String newFeedName = bundles[1].getInputFeedNameFromBundle() + "2";
        String inputFeed = bundles[1].getInputFeedFromBundle();

        bundles[1].addProcessInput(newFeedName, "inputData2");
        inputFeed = Util.setFeedName(inputFeed, newFeedName);

        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().suspend(bundles[1].getProcessData()));
        AssertUtil.assertSucceeded(prism.getFeedHelper().submitEntity(inputFeed));

        while (Util.parseResponse(
                prism.getProcessHelper()
                        .update((bundles[1].getProcessData()), bundles[1].getProcessData()))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            TimeUtil.sleepSeconds(10);
        }

        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, false);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 1, 10);

        AssertUtil.assertSucceeded(cluster3.getProcessHelper().resume(bundles[1].getProcessData()));

        bundles[1].verifyDependencyListing(cluster2);

        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        waitingForBundleFinish(cluster3, oldBundleId);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);
        AssertUtil.checkNotStatus(cluster3OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);

    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessAddNewInputInEachColoWithOneColoDown() throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(3);
        String endTime = TimeUtil.getTimeWrtSystemTime(10);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        String originalProcess = bundles[1].getProcessData();
        String newFeedName = bundles[1].getInputFeedNameFromBundle() + "2";
        String inputFeed = bundles[1].getInputFeedFromBundle();
        bundles[1].addProcessInput(newFeedName, "inputData2");
        inputFeed = Util.setFeedName(inputFeed, newFeedName);
        String updatedProcess = bundles[1].getProcessData();

        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(originalProcess));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, originalProcess, 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(originalProcess), EntityType.PROCESS);

        InstanceUtil.waitTillInstancesAreCreated(cluster3, originalProcess, 0, 10);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        //submit new feed
        AssertUtil.assertSucceeded(prism.getFeedHelper().submitEntity(inputFeed));

        Util.shutDownService(cluster3.getProcessHelper());

        AssertUtil.assertPartial(
                prism.getProcessHelper()
                        .update(updatedProcess, updatedProcess));

        Util.startService(cluster3.getProcessHelper());
        bundles[1].verifyDependencyListing(cluster2);

        dualComparison(prism, cluster3, bundles[1].getProcessData());
        Assert.assertFalse(Util.isDefinitionSame(cluster2, prism, originalProcess));

        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), false, false);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1],
                Job.Status.RUNNING);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        while (Util.parseResponse(
                prism.getProcessHelper().update(updatedProcess, updatedProcess)).getStatus()
                != APIResult.Status.SUCCEEDED) {
            LOGGER.info("update didnt SUCCEED in last attempt");
            TimeUtil.sleepSeconds(10);
        }
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        Assert.assertTrue(Util.isDefinitionSame(cluster2, prism, originalProcess));
        bundles[1].verifyDependencyListing(cluster2);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                updatedProcess, true, false);
        waitingForBundleFinish(cluster3, oldBundleId);

        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 1, 10);

        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1],
                Job.Status.RUNNING);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessDecreaseValidityInEachColoWithOneProcessRunning() throws Exception {
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        String newEndTime = TimeUtil.addMinsToTime(TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getEnd()
        ), -2);
        bundles[1].setProcessValidity(TimeUtil.dateToOozieDate(
                        bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                                .getStart()),
                newEndTime);
        while (Util.parseResponse(
                (prism.getProcessHelper()
                        .update(bundles[1].getProcessData(), bundles[1].getProcessData())))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            LOGGER.info("update didnt SUCCEED in last attempt");
            TimeUtil.sleepSeconds(10);
        }
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), false, true);

        bundles[1].verifyDependencyListing(cluster2);

        dualComparison(prism, cluster3, bundles[1].getProcessData());
        waitingForBundleFinish(cluster3, oldBundleId);
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        int finalNumberOfInstances = InstanceUtil
                .getProcessInstanceList(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS)
                .size();
        Assert.assertEquals(finalNumberOfInstances,
                getExpectedNumberOfWorkflowInstances(bundles[1]
                                .getProcessObject().getClusters().getClusters().get(0).getValidity()
                                .getStart(),
                        bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                .getValidity().getEnd()));
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
        int expectedNumberOfWorkflows =
                getExpectedNumberOfWorkflowInstances(TimeUtil.dateToOozieDate(
                                bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                        .getValidity().getStart()),
                        newEndTime);
        Assert.assertEquals(OozieUtil.getNumberOfWorkflowInstances(cluster3, oldBundleId),
                expectedNumberOfWorkflows);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessIncreaseValidityInEachColoWithOneProcessSuspended() throws Exception {
        String startTime = TimeUtil.getTimeWrtSystemTime(-1);
        String endTime = TimeUtil.getTimeWrtSystemTime(3);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        TimeUtil.sleepSeconds(30);
        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        String newEndTime = TimeUtil.addMinsToTime(TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getEnd()
        ), 4);
        bundles[1].setProcessValidity(TimeUtil.dateToOozieDate(
                        bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                                .getStart()),
            newEndTime);

        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().suspend(bundles[1].getProcessData()));
        while (Util.parseResponse(
                prism.getProcessHelper()
                        .update((bundles[1].getProcessData()), bundles[1].getProcessData()))
                .getStatus() != APIResult.Status.SUCCEEDED) {
            LOGGER.info("update didnt SUCCEED in last attempt");
            TimeUtil.sleepSeconds(10);
        }
        AssertUtil.assertSucceeded(cluster3.getProcessHelper().resume(bundles[1].getProcessData()));

        dualComparison(prism, cluster2, bundles[1].getProcessData());

        dualComparison(prism, cluster3, bundles[1].getProcessData());
        waitingForBundleFinish(cluster3, oldBundleId);
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        int finalNumberOfInstances = InstanceUtil
                .getProcessInstanceList(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS)
                .size();
        Assert.assertEquals(finalNumberOfInstances,
                getExpectedNumberOfWorkflowInstances(bundles[1]
                                .getProcessObject().getClusters().getClusters().get(0).getValidity()
                                .getStart(),
                        bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                .getValidity().getEnd()));
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    private void setBundleWFPath(Bundle... bundles) {
        for (Bundle bundle : bundles) {
            bundle.setProcessWorkflow(workflowPath);
        }
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessFrequencyInEachColoWithOneProcessRunningDaily() throws Exception {
        //set daily process
        final String startTime = TimeUtil.getTimeWrtSystemTime(-20);
        String endTime = TimeUtil.getTimeWrtSystemTime(4000);
        bundles[1].setProcessPeriodicity(1, TimeUnit.days);
        bundles[1].setOutputFeedPeriodicity(1, TimeUnit.days);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        List<String> oldNominalTimes =
                OozieUtil.getActionsNominalTime(cluster3, oldBundleId, EntityType.PROCESS);

        LOGGER.info("original process: " + Util.prettyPrintXml(bundles[1].getProcessData()));

        String updatedProcess = InstanceUtil
                .setProcessFrequency(bundles[1].getProcessData(),
                        new Frequency("" + 5, TimeUnit.minutes));

        LOGGER.info("updated process: " + updatedProcess);

        //now to update
        ServiceResponse response =
                prism.getProcessHelper().update(updatedProcess, updatedProcess);
        AssertUtil.assertSucceeded(response);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 1, 10);

        String prismString = dualComparison(prism, cluster2, bundles[1].getProcessData());
        Assert.assertEquals(Util.getProcessObject(prismString).getFrequency(),
                new Frequency("" + 5, TimeUnit.minutes));
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated
        // correctly.
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void
    updateProcessFrequencyInEachColoWithOneProcessRunningDailyToMonthlyWithStartChange()
        throws Exception {
        //set daily process
        final String startTime = TimeUtil.getTimeWrtSystemTime(-20);
        String endTime = TimeUtil.getTimeWrtSystemTime(4000 * 60);
        bundles[1].setProcessPeriodicity(1, TimeUnit.days);
        bundles[1].setOutputFeedPeriodicity(1, TimeUnit.days);
        bundles[1].setProcessValidity(startTime, endTime);

        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        LOGGER.info("original process: " + Util.prettyPrintXml(bundles[1].getProcessData()));

        String updatedProcess = InstanceUtil
                .setProcessFrequency(bundles[1].getProcessData(),
                        new Frequency("" + 1, TimeUnit.months));
        updatedProcess = InstanceUtil
                .setProcessValidity(updatedProcess, TimeUtil.getTimeWrtSystemTime(10),
                        endTime);

        LOGGER.info("updated process: " + updatedProcess);

        //now to update
        ServiceResponse response =
                prism.getProcessHelper().update(updatedProcess, updatedProcess);
        AssertUtil.assertSucceeded(response);
        String prismString = dualComparison(prism, cluster3, bundles[1].getProcessData());
        Assert.assertEquals(Util.getProcessObject(prismString).getFrequency(),
                new Frequency("" + 1, TimeUnit.months));
        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessRollStartTimeBackwardsToPastInEachColoWithOneProcessRunning()
        throws Exception {
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        TimeUtil.sleepSeconds(30);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);

        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster3, oldBundleId,
                EntityType.PROCESS);

        String newStartTime = TimeUtil.addMinsToTime(TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getStart()
        ), -3);
        bundles[1].setProcessValidity(newStartTime, TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getEnd()
        ));

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);
        OozieUtil.createMissingDependencies(cluster3, EntityType.PROCESS, bundles[1].getProcessName(), 0);
        AssertUtil.assertSucceeded(
                prism.getProcessHelper()
                        .update(bundles[1].getProcessData(), bundles[1].getProcessData()));

        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, true);
        bundles[1].verifyDependencyListing(cluster2);
        dualComparison(prism, cluster3, bundles[1].getProcessData());
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessRollStartTimeForwardInEachColoWithOneProcessSuspended()
        throws Exception {
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        TimeUtil.sleepSeconds(30);

        OozieUtil.getNumberOfWorkflowInstances(cluster3, oldBundleId);
        String oldStartTime = TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getStart()
        );
        String newStartTime = TimeUtil.addMinsToTime(TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getStart()
        ), 3);
        bundles[1].setProcessValidity(newStartTime, TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getEnd()
        ));

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().suspend(bundles[1].getProcessData()));

        AssertUtil.assertSucceeded(
                prism.getProcessHelper()
                        .update(bundles[1].getProcessData(), bundles[1].getProcessData()));

        dualComparison(prism, cluster2, bundles[1].getProcessData());

        bundles[1].verifyDependencyListing(cluster2);

        dualComparison(prism, cluster3, bundles[1].getProcessData());
        //ensure that the running process has new coordinators created; while the submitted
        // one is updated correctly.
        int finalNumberOfInstances =
                InstanceUtil.getProcessInstanceListFromAllBundles(cluster3,
                        Util.getProcessName(bundles[1].getProcessData()), EntityType.PROCESS).size();
        Assert.assertEquals(finalNumberOfInstances,
                getExpectedNumberOfWorkflowInstances(oldStartTime,
                        bundles[1].getProcessObject().getClusters().getClusters().get(0)
                                .getValidity().getEnd()));
        Assert.assertEquals(InstanceUtil
                .getProcessInstanceList(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS)
                .size(), getExpectedNumberOfWorkflowInstances(newStartTime,
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity().getEnd()));

        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    @Test(groups = { "multiCluster" }, timeOut = 1200000)
    public void updateProcessRollStartTimeBackwardsInEachColoWithOneProcessSuspended()
        throws Exception {
        bundles[1].submitBundle(prism);
        //now to schedule in 1 colo and let it remain in another
        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().schedule(bundles[1].getProcessData()));
        String oldBundleId = InstanceUtil
                .getLatestBundleID(cluster3,
                        Util.readEntityName(bundles[1].getProcessData()), EntityType.PROCESS);
        TimeUtil.sleepSeconds(30);

        String newStartTime = TimeUtil.addMinsToTime(TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getStart()
        ), -3);
        bundles[1].setProcessValidity(newStartTime, TimeUtil.dateToOozieDate(
                bundles[1].getProcessObject().getClusters().getClusters().get(0).getValidity()
                        .getEnd()
        ));
        InstanceUtil.waitTillInstancesAreCreated(cluster3, bundles[1].getProcessData(), 0, 10);

        waitForProcessToReachACertainState(cluster3, bundles[1], Job.Status.RUNNING);

        AssertUtil.assertSucceeded(
                cluster3.getProcessHelper().suspend(bundles[1].getProcessData()));
        AssertUtil.assertSucceeded(
                prism.getProcessHelper()
                        .update(bundles[1].getProcessData(), bundles[1].getProcessData()));
        AssertUtil.assertSucceeded(cluster3.getProcessHelper().resume(bundles[1].getProcessData()));
        List<String> oldNominalTimes =
                OozieUtil.getActionsNominalTime(cluster3, oldBundleId, EntityType.PROCESS);

        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleId, oldNominalTimes,
                bundles[1].getProcessData(), true, false);

        bundles[1].verifyDependencyListing(cluster2);

        dualComparison(prism, cluster3, bundles[1].getProcessData());
        AssertUtil.checkNotStatus(cluster2OC, EntityType.PROCESS, bundles[1], Job.Status.RUNNING);
    }

    @Test(timeOut = 1200000)
    public void
    updateProcessWorkflowXml() throws URISyntaxException, JAXBException,
            IOException, OozieClientException, AuthenticationException {
        Bundle b = BundleUtil.readELBundle();
        HadoopFileEditor hadoopFileEditor = null;
        try {

            b = new Bundle(b, cluster1);
            b.setProcessWorkflow(workflowPath);
            b.generateUniqueBundle();

            b.setProcessValidity(TimeUtil.getTimeWrtSystemTime(-10),
                    TimeUtil.getTimeWrtSystemTime(15));
            b.submitFeedsScheduleProcess(prism);

            InstanceUtil.waitTillInstancesAreCreated(cluster1, b.getProcessData(), 0, 10);
            OozieUtil.createMissingDependencies(cluster1, EntityType.PROCESS,
                    b.getProcessName(), 0);
            InstanceUtil.waitTillInstanceReachState(serverOC.get(0),
                    Util.readEntityName(b.getProcessData()), 0, CoordinatorAction.Status.RUNNING,
                    EntityType.PROCESS);

            //save old data
            String oldBundleID = InstanceUtil
                    .getLatestBundleID(cluster1,
                            Util.readEntityName(b.getProcessData()), EntityType.PROCESS);

            List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster1,
                    oldBundleID,
                    EntityType.PROCESS);

            //update workflow.xml
            hadoopFileEditor = new HadoopFileEditor(cluster1FS);
            hadoopFileEditor.edit(new ProcessMerlin(b
                            .getProcessData()).getWorkflow().getPath() + "/workflow.xml", "</workflow-app>",
                    "<!-- some comment -->");

            //update
            prism.getProcessHelper().update(b.getProcessData(),
                    b.getProcessData());

            TimeUtil.sleepSeconds(20);
            //verify new bundle creation
            OozieUtil.verifyNewBundleCreation(cluster1, oldBundleID, oldNominalTimes,
                    b.getProcessData(), true, true);

        } finally {
            b.deleteBundle(prism);
            if (hadoopFileEditor != null) {
                hadoopFileEditor.restore();
            }
        }

    }

    public ServiceResponse updateProcessConcurrency(Bundle bundle, int concurrency)
        throws Exception {
        String oldData = bundle.getProcessData();
        ProcessMerlin updatedProcess = new ProcessMerlin(bundle.getProcessObject());
        updatedProcess.setParallel(concurrency);

        return prism.getProcessHelper()
                .update(oldData, updatedProcess.toString());
    }

    /**
     * this method compares process xml definition from 2 falcon servers / prism and expects them to
     * be identical. If the definitions are identical then the definition from @param coloHelper1
     * is @return are response.
     */
    private String dualComparison(ColoHelper coloHelper1, ColoHelper coloHelper2,
            String processData) throws Exception {
        String colo1Response = getResponse(coloHelper1, processData, true);
        String colo2Response = getResponse(coloHelper2, processData, true);
        Assert.assertTrue(XmlUtil.isIdentical(colo1Response, colo2Response),
                "Process definition should have been identical");
        return getResponse(coloHelper1, processData, true);
    }

    /**
     * this method compares process xml definition from 2 falcon servers / prism and expects them to
     * be different.
     */
    private void dualComparisonFailure(ColoHelper coloHelper1, ColoHelper coloHelper2,
            String processData) throws Exception {
        Assert.assertFalse(XmlUtil.isIdentical(getResponse(coloHelper1, processData, true),
                getResponse(coloHelper2, processData, true)), "Process definition should not have been "
                + "identical");
    }

    private String getResponse(ColoHelper prism, String processData, boolean bool)
        throws Exception {
        ServiceResponse response = prism.getProcessHelper().getEntityDefinition(processData);
        if (bool) {
            AssertUtil.assertSucceeded(response);
        } else {
            AssertUtil.assertFailed(response);
        }
        String result = response.getMessage();
        Assert.assertNotNull(result);

        return result;

    }

    private void waitForProcessToReachACertainState(ColoHelper coloHelper, Bundle bundle,
            Job.Status state)
        throws Exception {

        while (OozieUtil.getOozieJobStatus(coloHelper.getFeedHelper().getOozieClient(),
                Util.readEntityName(bundle.getProcessData()), EntityType.PROCESS) != state) {
            //keep waiting
            TimeUtil.sleepSeconds(10);
        }

        //now check if the coordinator is in desired state
        CoordinatorJob coord = getDefaultOozieCoord(coloHelper, InstanceUtil
                .getLatestBundleID(coloHelper, Util.readEntityName(bundle.getProcessData()),
                        EntityType.PROCESS));

        while (coord.getStatus() != state) {
            TimeUtil.sleepSeconds(10);
            coord = getDefaultOozieCoord(coloHelper, InstanceUtil
                    .getLatestBundleID(coloHelper, Util.readEntityName(bundle.getProcessData()),
                            EntityType.PROCESS));
        }
    }

    private Bundle usualGrind(Bundle b) throws Exception {
        b.setInputFeedDataPath(inputFeedPath);
        String prefix = b.getFeedDataPathPrefix();
        HadoopUtil.deleteDirIfExists(prefix.substring(1), cluster1FS);
        final String starTime = TimeUtil.getTimeWrtSystemTime(3);
        String endTime = TimeUtil.getTimeWrtSystemTime(7);
        b.setProcessPeriodicity(1, TimeUnit.minutes);
        b.setOutputFeedPeriodicity(1, TimeUnit.minutes);
        b.setProcessValidity(starTime, endTime);
        return b;
    }

    private ExecutionType getRandomExecutionType(Bundle bundle) throws Exception {
        ExecutionType current = bundle.getProcessObject().getOrder();
        Random r = new Random();
        ExecutionType[] values = ExecutionType.values();
        int i;
        do {

            i = r.nextInt(values.length);
        } while (current == values[i]);
        return values[i];
    }

    public ServiceResponse updateProcessFrequency(Bundle bundle,
            org.apache.falcon.entity.v0.Frequency frequency)
        throws Exception {
        String oldData = bundle.getProcessData();
        ProcessMerlin updatedProcess = new ProcessMerlin(bundle.getProcessObject());
        updatedProcess.setFrequency(frequency);
        return prism.getProcessHelper()
                .update(oldData, updatedProcess.toString());
    }

    //need to expand this function more later
    private int getExpectedNumberOfWorkflowInstances(String start, String end) {
        DateTime startDate = new DateTime(start);
        DateTime endDate = new DateTime(end);
        Minutes minutes = Minutes.minutesBetween((startDate), (endDate));
        return minutes.getMinutes();
    }

    private int getExpectedNumberOfWorkflowInstances(Date start, Date end) {
        DateTime startDate = new DateTime(start);
        DateTime endDate = new DateTime(end);
        Minutes minutes = Minutes.minutesBetween((startDate), (endDate));
        return minutes.getMinutes();
    }

    private int getExpectedNumberOfWorkflowInstances(String start, Date end) {
        DateTime startDate = new DateTime(start);
        DateTime endDate = new DateTime(end);
        Minutes minutes = Minutes.minutesBetween((startDate), (endDate));
        return minutes.getMinutes();
    }

    private void waitingForBundleFinish(ColoHelper coloHelper, String bundleId, int minutes)
        throws Exception {
        int wait = 0;
        while (!OozieUtil.isBundleOver(coloHelper, bundleId)) {
            //create missing dependencies if new instance have come up
            OozieUtil.createMissingDependenciesForBundle(coloHelper, bundleId);

            //keep waiting
            LOGGER.info("bundle " + bundleId + " not over .. waiting");
            TimeUtil.sleepSeconds(60);
            wait++;
            if (wait == minutes) {
                Assert.assertTrue(false);
                break;
            }
        }
    }

    private void waitingForBundleFinish(ColoHelper coloHelper, String bundleId) throws Exception {
        waitingForBundleFinish(coloHelper, bundleId, 15);
    }

    private CoordinatorJob getDefaultOozieCoord(ColoHelper coloHelper, String bundleId)
        throws Exception {
        OozieClient client = coloHelper.getFeedHelper().getOozieClient();
        BundleJob bundlejob = client.getBundleJobInfo(bundleId);

        for (CoordinatorJob coord : bundlejob.getCoordinators()) {
            if (coord.getAppName().contains("DEFAULT")) {
                return client.getCoordJobInfo(coord.getId());
            }
        }
        return null;
    }

    @AfterClass(alwaysRun = true)
    public void tearDownClass() throws IOException {
        cleanTestDirs();
    }
}