/**
 * 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 com.jcraft.jsch.JSchException;
import org.apache.falcon.regression.core.bundle.Bundle;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.Frequency;
import org.apache.falcon.entity.v0.feed.ActionType;
import org.apache.falcon.entity.v0.feed.ClusterType;
import org.apache.falcon.regression.core.helpers.ColoHelper;
import org.apache.falcon.regression.core.response.ServiceResponse;
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.oozie.client.CoordinatorAction;
import org.apache.oozie.client.OozieClientException;
import org.custommonkey.xmlunit.Diff;
import org.custommonkey.xmlunit.XMLUnit;
import org.joda.time.DateTime;
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 org.xml.sax.SAXException;
import org.apache.log4j.Logger;

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


public class UpdateAtSpecificTimeTest extends BaseTestClass {

    private static final Logger logger = Logger.getLogger(UpdateAtSpecificTimeTest.class);

    Bundle processBundle;
    ColoHelper cluster1 = servers.get(0);
    ColoHelper cluster2 = servers.get(1);
    ColoHelper cluster3 = servers.get(2);
    FileSystem cluster2FS = serverFS.get(1);
    private final String baseTestDir = baseHDFSDir + "/UpdateAtSpecificTimeTest-data";
    String aggregateWorkflowDir = baseHDFSDir + "/aggregator";

    @BeforeClass(alwaysRun = true)
    public void uploadWorkflow() throws Exception {
        uploadDirToClusters(aggregateWorkflowDir, OSUtil.RESOURCES_OOZIE);
    }

    @BeforeMethod(alwaysRun = true)
    public void setup(Method method) throws IOException {
        logger.info("test name: " + method.getName());
        Bundle bundle = BundleUtil.readLocalDCBundle();
        bundles[0] = new Bundle(bundle, cluster1);
        bundles[1] = new Bundle(bundle, cluster2);
        bundles[2] = new Bundle(bundle, cluster3);

        bundles[0].generateUniqueBundle();
        bundles[1].generateUniqueBundle();
        bundles[2].generateUniqueBundle();

        processBundle = BundleUtil.readELBundle();
        processBundle = new Bundle(processBundle, cluster1);
        processBundle.generateUniqueBundle();
        processBundle.setProcessWorkflow(aggregateWorkflowDir);
    }

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

    @Test(groups = {"singleCluster", "0.3.1", "embedded"}, timeOut = 1200000, enabled = true)
    public void invalidChar_Process()
        throws JAXBException, IOException, URISyntaxException,
        AuthenticationException, OozieClientException {
        processBundle.setProcessValidity(TimeUtil.getTimeWrtSystemTime(0),
            TimeUtil.getTimeWrtSystemTime(20));
        processBundle.submitFeedsScheduleProcess(prism);
        InstanceUtil.waitTillInstancesAreCreated(cluster1, processBundle.getProcessData(), 0);
        String oldProcess =
            processBundle.getProcessData();
        processBundle.setProcessValidity(TimeUtil.getTimeWrtSystemTime(5),
            TimeUtil.getTimeWrtSystemTime(100));
        ServiceResponse r = prism.getProcessHelper().update(oldProcess,
            processBundle.getProcessData(), "abc", null);
        Assert.assertTrue(r.getMessage()
            .contains("java.lang.IllegalArgumentException: abc is not a valid UTC string"));
    }

    @Test(groups = {"singleCluster", "0.3.1", "embedded"}, timeOut = 1200000, enabled = true)
    public void invalidChar_Feed()
        throws JAXBException, IOException, URISyntaxException, AuthenticationException,
        OozieClientException {

        String feed = submitAndScheduleFeed(processBundle);
        InstanceUtil.waitTillInstancesAreCreated(cluster1, feed, 0);

        //update frequency
        Frequency f = new Frequency("" + 21, Frequency.TimeUnit.minutes);
        String updatedFeed = InstanceUtil.setFeedFrequency(feed, f);
        ServiceResponse r = prism.getFeedHelper().update(feed, updatedFeed, "abc", null);
        Assert.assertTrue(r.getMessage()
            .contains("java.lang.IllegalArgumentException: abc is not a valid UTC string"));
    }

    @Test(groups = {"singleCluster", "0.3.1", "embedded"}, timeOut = 1200000, enabled = true)
    public void updateTimeInPast_Process()
        throws JAXBException, IOException, URISyntaxException,
        OozieClientException, AuthenticationException {

        processBundle.setProcessValidity(TimeUtil.getTimeWrtSystemTime(0),
                TimeUtil.getTimeWrtSystemTime(20));
        processBundle.submitFeedsScheduleProcess(prism);

        //get old process details
        String oldProcess = processBundle.getProcessData();
        String oldBundleId = InstanceUtil.getLatestBundleID(cluster1,
            Util.readEntityName(processBundle.getProcessData()), EntityType.PROCESS);

        InstanceUtil.waitTillInstancesAreCreated(cluster1, oldProcess, 0);
        List<String> initialNominalTimes = OozieUtil.getActionsNominalTime(cluster1,
            oldBundleId, EntityType.PROCESS);

        // update process by adding property
        processBundle.setProcessProperty("someProp", "someValue");
        ServiceResponse r = prism.getProcessHelper().update(oldProcess,
            processBundle.getProcessData(), TimeUtil.getTimeWrtSystemTime(-10000), null);
        AssertUtil.assertSucceeded(r);

        //check new coord created with current time
        OozieUtil.verifyNewBundleCreation(cluster1, oldBundleId, initialNominalTimes,
            processBundle.getProcessData(), true, false);
        InstanceUtil.waitTillInstancesAreCreated(cluster1, oldProcess, 1);
        OozieUtil.verifyNewBundleCreation(cluster1, oldBundleId, initialNominalTimes,
            processBundle.getProcessData(), true, true);
    }

    @Test(groups = {"MultiCluster", "0.3.1", "embedded"}, timeOut = 1200000, enabled = true)
    public void updateTimeInPast_Feed()
        throws JAXBException, IOException, OozieClientException,
        URISyntaxException, AuthenticationException {

        String startTimeCluster_source = TimeUtil.getTimeWrtSystemTime(-10);
        String startTimeCluster_target = TimeUtil.getTimeWrtSystemTime(10);
        String feed = getMultiClusterFeed(startTimeCluster_source, startTimeCluster_target);
        logger.info("feed: " + Util.prettyPrintXml(feed));

        //submit and schedule feed
        ServiceResponse r = prism.getFeedHelper().submitEntity(feed);
        TimeUtil.sleepSeconds(10);
        AssertUtil.assertSucceeded(r);
        InstanceUtil.waitTillInstancesAreCreated(cluster1, feed, 0);

        //update frequency
        Frequency f = new Frequency("" + 7, Frequency.TimeUnit.minutes);
        String updatedFeed = InstanceUtil.setFeedFrequency(feed, f);
        r = prism.getFeedHelper().update(feed, updatedFeed,
            TimeUtil.getTimeWrtSystemTime(-10000), null);
        AssertUtil.assertSucceeded(r);
        InstanceUtil.waitTillInstancesAreCreated(cluster1, feed, 1);

        //check correct number of coord exists or not
        Assert.assertEquals(InstanceUtil
            .checkIfFeedCoordExist(cluster1.getFeedHelper(),
                Util.readEntityName(feed),
                "REPLICATION"), 2);
        Assert.assertEquals(InstanceUtil
            .checkIfFeedCoordExist(cluster2.getFeedHelper(), Util.readEntityName(feed),
                "RETENTION"), 2);
        Assert.assertEquals(InstanceUtil
            .checkIfFeedCoordExist(cluster1.getFeedHelper(), Util.readEntityName(feed),
                "RETENTION"), 2);
        Assert.assertEquals(InstanceUtil
            .checkIfFeedCoordExist(cluster3.getFeedHelper(), Util.readEntityName(feed),
                "RETENTION"), 2);
    }

    @Test(groups = {"MultiCluster", "0.3.1", "distributed"}, timeOut = 1200000, enabled = true)
    public void inNextFewMinutesUpdate_RollForward_Process()
        throws JAXBException, IOException, URISyntaxException, JSchException,
        OozieClientException, SAXException, AuthenticationException {
        /*
        submit process on 3 clusters. Schedule on 2 clusters. Bring down one of
        the scheduled cluster. Update with time 5 minutes from now. On running
        cluster new coord should be created with start time +5 and no instance
        should be missing. On 3rd cluster where process was only submit,
        definition should be updated. Bring the down cluster up. Update with same
        definition again, now the recently up cluster should also have new
        coords.
        */
        try {
            Util.startService(cluster2.getProcessHelper());
            String startTime = TimeUtil.getTimeWrtSystemTime(-15);
            processBundle.setProcessValidity(startTime,
                TimeUtil.getTimeWrtSystemTime(60));
            processBundle.addClusterToBundle(bundles[1].getClusters().get(0),
                ClusterType.SOURCE, null, null);
            processBundle.addClusterToBundle(bundles[2].getClusters().get(0),
                ClusterType.SOURCE, null, null);
            processBundle.submitBundle(prism);

            //schedule of 2 cluster
            cluster1.getProcessHelper().schedule(processBundle.getProcessData());
            cluster2.getProcessHelper().schedule(processBundle.getProcessData());
            InstanceUtil.waitTillInstancesAreCreated(cluster2, processBundle.getProcessData(), 0);

            //shut down cluster2
            Util.shutDownService(cluster2.getProcessHelper());

            // save old data before update
            String oldProcess = processBundle.getProcessData();
            String oldBundleID_cluster1 = InstanceUtil
                .getLatestBundleID(cluster1,
                    Util.readEntityName(oldProcess), EntityType.PROCESS);
            String oldBundleID_cluster2 = InstanceUtil
                .getLatestBundleID(cluster2,
                    Util.readEntityName(oldProcess), EntityType.PROCESS);
            List<String> oldNominalTimes_cluster1 = OozieUtil.getActionsNominalTime
                (cluster1,
                    oldBundleID_cluster1, EntityType.PROCESS);
            List<String> oldNominalTimes_cluster2 = OozieUtil.getActionsNominalTime
                (cluster2,
                    oldBundleID_cluster2, EntityType.PROCESS);

            //update process validity
            processBundle.setProcessProperty("someProp", "someValue");

            //send update request
            String updateTime = TimeUtil.getTimeWrtSystemTime(5);
            ServiceResponse r = prism.getProcessHelper()
                .update(oldProcess, processBundle.getProcessData(), updateTime);
            AssertUtil.assertPartial(r);
            InstanceUtil.waitTillInstancesAreCreated(cluster1, processBundle.getProcessData(), 1);

            //verify new bundle on cluster1 and definition on cluster3
            OozieUtil
                .verifyNewBundleCreation(cluster1, oldBundleID_cluster1, oldNominalTimes_cluster1,
                    oldProcess, true, false);
            OozieUtil.verifyNewBundleCreation(cluster2, oldBundleID_cluster2,
                oldNominalTimes_cluster2,
                oldProcess, false, false);
            String definition_cluster_3 = Util.getEntityDefinition(cluster3,
                processBundle.getProcessData(), true);
            Assert.assertTrue(XmlUtil.isIdentical(definition_cluster_3,
                processBundle.getProcessData()), "Process definitions should be equal");

            //start the stopped cluster2
            Util.startService(cluster2.getProcessHelper());
            TimeUtil.sleepSeconds(40);
            String newBundleID_cluster1 = InstanceUtil.getLatestBundleID(cluster1,
                Util.readEntityName(oldProcess), EntityType.PROCESS);

            //send second update request
            r = prism.getProcessHelper().update(oldProcess, processBundle.getProcessData(),
                updateTime, null);
            AssertUtil.assertSucceeded(r);
            String def_cluster_2 = Util.getEntityDefinition(cluster2,
                processBundle.getProcessData(), true);
            logger.info("def_cluster_2 : " + Util.prettyPrintXml(def_cluster_2));

            // verify new bundle in cluster2 and no new bundle in cluster1  and
            OozieUtil.verifyNewBundleCreation(cluster1, newBundleID_cluster1,
                oldNominalTimes_cluster1, oldProcess, false, false);
            OozieUtil.verifyNewBundleCreation(cluster2, oldBundleID_cluster2,
                oldNominalTimes_cluster2, oldProcess, true, false);

            //wait till update time is reached
            TimeUtil.sleepTill(updateTime);
            OozieUtil.verifyNewBundleCreation(cluster2, oldBundleID_cluster2,
                oldNominalTimes_cluster2, oldProcess, true, true);
            OozieUtil.verifyNewBundleCreation(cluster1, oldBundleID_cluster1,
                oldNominalTimes_cluster1, oldProcess, true, true);
        } finally {
            Util.restartService(cluster2.getProcessHelper());
        }
    }

    @Test(groups = {"MultiCluster", "0.3.1", "distributed"}, timeOut = 1200000, enabled = true)
    public void inNextFewMinutesUpdate_RollForward_Feed()
        throws JAXBException, IOException, URISyntaxException, JSchException, 
        OozieClientException, SAXException, AuthenticationException {
        try {
            String startTimeCluster_source = TimeUtil.getTimeWrtSystemTime(-18);
            String feed = getMultiClusterFeed(startTimeCluster_source, startTimeCluster_source);
            logger.info("feed: " + Util.prettyPrintXml(feed));

            //submit feed on all 3 clusters
            ServiceResponse r = prism.getFeedHelper().submitEntity(feed);
            AssertUtil.assertSucceeded(r);

            //schedule feed of cluster1 and cluster2
            r = cluster1.getFeedHelper().schedule(feed);
            AssertUtil.assertSucceeded(r);
            r = cluster2.getFeedHelper().schedule(feed);
            AssertUtil.assertSucceeded(r);
            InstanceUtil.waitTillInstancesAreCreated(cluster1, feed, 0);

            //shutdown cluster2
            Util.shutDownService(cluster2.getProcessHelper());

            //add some property to feed so that new bundle is created
            String updatedFeed = Util.setFeedProperty(feed, "someProp", "someVal");

            //save old data
            String oldBundle_cluster1 = InstanceUtil.getLatestBundleID(cluster1,
                Util.readEntityName(feed), EntityType.FEED);
            List<String> oldNominalTimes_cluster1 = OozieUtil.getActionsNominalTime
                (cluster1, oldBundle_cluster1, EntityType.FEED);

            //send update command with +5 mins in future
            String updateTime = TimeUtil.getTimeWrtSystemTime(5);
            r = prism.getFeedHelper().update(feed, updatedFeed, updateTime, null);
            AssertUtil.assertPartial(r);

            //verify new bundle creation on cluster1 and new definition on cluster3
            OozieUtil.verifyNewBundleCreation(cluster1, oldBundle_cluster1,
                oldNominalTimes_cluster1, feed, true, false);
            String definition = Util.getEntityDefinition(cluster3, feed, true);
            Diff diff = XMLUnit.compareXML(definition, processBundle.getProcessData());
            logger.info(diff);

            //start stopped cluster2
            Util.startService(cluster2.getProcessHelper());
            String newBundle_cluster1 = InstanceUtil.getLatestBundleID(cluster1,
                Util.readEntityName(feed), EntityType.FEED);

            //send update again
            r = prism.getFeedHelper().update(feed, updatedFeed, updateTime, null);
            AssertUtil.assertSucceeded(r);

            //verify new bundle creation on cluster2 and no new bundle on cluster1
            Assert.assertEquals(InstanceUtil
                .checkIfFeedCoordExist(cluster2.getFeedHelper(), Util.readEntityName(feed),
                    "RETENTION"), 2);
            OozieUtil.verifyNewBundleCreation(cluster1, newBundle_cluster1,
                oldNominalTimes_cluster1, feed, false, false);

            //wait till update time is reached
            TimeUtil.sleepTill(TimeUtil.getTimeWrtSystemTime(5));

            //verify new bundle creation with instance matching
            OozieUtil.verifyNewBundleCreation(cluster1, oldBundle_cluster1,
                oldNominalTimes_cluster1, feed, true, true);
        } finally {
            Util.restartService(cluster2.getProcessHelper());
        }
    }

    @Test(groups = {"multiCluster", "0.3.1", "embedded"}, timeOut = 1200000, enabled = true)
    public void updateTimeAfterEndTime_Process()
        throws JAXBException, InterruptedException, IOException, URISyntaxException,
        OozieClientException, AuthenticationException {
        /* submit and schedule process with end time after 60 mins. Set update time
           as with +60 from start mins */
        logger.info("Running test updateTimeAfterEndTime_Process");
        String startTime = TimeUtil.getTimeWrtSystemTime(-15);
        String endTime = TimeUtil.getTimeWrtSystemTime(60);
        processBundle.setProcessValidity(startTime, endTime);
        processBundle.submitFeedsScheduleProcess(prism);
        TimeUtil.sleepSeconds(10);
        InstanceUtil.waitTillInstanceReachState(serverOC.get(0),
            Util.readEntityName(processBundle.getProcessData()), 0,
            CoordinatorAction.Status.WAITING, EntityType.PROCESS);

        //save old data
        String oldProcess = processBundle.getProcessData();
        String oldBundleID = InstanceUtil
            .getLatestBundleID(cluster1,
                Util.readEntityName(oldProcess), EntityType.PROCESS);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster1, oldBundleID,
            EntityType.PROCESS);

        //update
        processBundle.setProcessProperty("someProp", "someVal");
        String updateTime = TimeUtil.addMinsToTime(endTime, 60);
        logger.info("Original Feed : " + Util.prettyPrintXml(oldProcess));
        logger.info("Updated Feed :" + Util.prettyPrintXml(processBundle.getProcessData()));
        logger.info("Update Time : " + updateTime);
        ServiceResponse r = prism.getProcessHelper().update(oldProcess,
            processBundle.getProcessData(), updateTime, null);
        AssertUtil.assertSucceeded(r);

        //verify new bundle creation with instances matching
        OozieUtil.verifyNewBundleCreation(cluster1, oldBundleID, oldNominalTimes,
            oldProcess, true, false);
        InstanceUtil.waitTillInstancesAreCreated(cluster1, processBundle.getProcessData(), 1);
        OozieUtil.verifyNewBundleCreation(cluster1, oldBundleID, oldNominalTimes,
            oldProcess, true, true);
    }

    @Test(groups = {"multiCluster", "0.3.1", "embedded"}, timeOut = 1200000, enabled = true)
    public void updateTimeAfterEndTime_Feed()
        throws JAXBException, IOException, OozieClientException,
        URISyntaxException, AuthenticationException {

        /* submit and schedule feed with end time 60 mins in future and update with +60 in future*/
        String startTime = TimeUtil.getTimeWrtSystemTime(-15);
        String endTime = TimeUtil.getTimeWrtSystemTime(60);

        String feed = processBundle.getDataSets().get(0);
        feed = InstanceUtil.setFeedCluster(feed,
            XmlUtil.createValidity("2012-10-01T12:00Z", "2010-01-01T00:00Z"),
            XmlUtil.createRtention("days(100000)", ActionType.DELETE), null,
            ClusterType.SOURCE, null);
        feed = InstanceUtil.setFeedCluster(feed, XmlUtil.createValidity(startTime, endTime),
            XmlUtil.createRtention("days(100000)", ActionType.DELETE),
            Util.readEntityName(processBundle.getClusters().get(0)), ClusterType.SOURCE,
            null, baseTestDir + "/replication" + MINUTE_DATE_PATTERN);

        ServiceResponse r = prism.getClusterHelper().submitEntity(
            processBundle.getClusters().get(0));
        AssertUtil.assertSucceeded(r);
        r = prism.getFeedHelper().submitAndSchedule(feed);
        AssertUtil.assertSucceeded(r);
        InstanceUtil.waitTillInstancesAreCreated(cluster1, feed, 0);

        //save old data
        String oldBundleID = InstanceUtil.getLatestBundleID(cluster1,
            Util.readEntityName(feed), EntityType.FEED);
        String updateTime = TimeUtil.addMinsToTime(endTime, 60);
        String updatedFeed = Util.setFeedProperty(feed, "someProp", "someVal");
        logger.info("Original Feed : " + Util.prettyPrintXml(feed));
        logger.info("Updated Feed :" + Util.prettyPrintXml(updatedFeed));
        logger.info("Update Time : " + updateTime);
        r = prism.getFeedHelper().update(feed, updatedFeed, updateTime, null);
        AssertUtil.assertSucceeded(r);
        InstanceUtil.waitTillInstancesAreCreated(cluster1, feed, 1);

        //verify new bundle creation
        OozieUtil.verifyNewBundleCreation(cluster1, oldBundleID, null, feed, true, false);
    }

    @Test(groups = {"multiCluster", "0.3.1", "embedded"}, timeOut = 1200000, enabled = true)
    public void updateTimeBeforeStartTime_Process() throws JAXBException, IOException,
        URISyntaxException, OozieClientException, AuthenticationException {

        /* submit and schedule process with start time +10 mins from now. Update with start time
        -4 and update time +2 mins */
        String startTime = TimeUtil.getTimeWrtSystemTime(10);
        String endTime = TimeUtil.getTimeWrtSystemTime(20);
        processBundle.setProcessValidity(startTime, endTime);
        processBundle.submitFeedsScheduleProcess(prism);

        //save old data
        String oldProcess = processBundle.getProcessData();
        String oldBundleID = InstanceUtil.getLatestBundleID(cluster1,
            Util.readEntityName(oldProcess), EntityType.PROCESS);
        List<String> oldNominalTimes = OozieUtil.getActionsNominalTime(cluster1, oldBundleID,
            EntityType.PROCESS);
        processBundle.setProcessValidity(TimeUtil.addMinsToTime(startTime, -4), endTime);
        String updateTime = TimeUtil.getTimeWrtSystemTime(2);
        ServiceResponse r = prism.getProcessHelper().update(oldProcess,
            processBundle.getProcessData(), updateTime, null);
        AssertUtil.assertSucceeded(r);
        TimeUtil.sleepSeconds(10);

        //verify new bundle creation
        OozieUtil.verifyNewBundleCreation(cluster1, oldBundleID, oldNominalTimes,
            oldProcess, true, false);
    }

    @Test(groups = {"MultiCluster", "0.3.1"}, timeOut = 1200000, enabled = true)
    public void updateDiffClusterDiffValidity_Process()
        throws JAXBException, IOException, URISyntaxException, OozieClientException,
        AuthenticationException {

        //set start end process time for 3 clusters
        String startTime_cluster1 = TimeUtil.getTimeWrtSystemTime(-40);
        String endTime_cluster1 = TimeUtil.getTimeWrtSystemTime(3);
        String startTime_cluster2 = TimeUtil.getTimeWrtSystemTime(120);
        String endTime_cluster2 = TimeUtil.getTimeWrtSystemTime(240);
        String startTime_cluster3 = TimeUtil.getTimeWrtSystemTime(-30);
        String endTime_cluster3 = TimeUtil.getTimeWrtSystemTime(180);

        //create multi cluster bundle
        processBundle.setProcessValidity(startTime_cluster1, endTime_cluster1);
        processBundle.addClusterToBundle(bundles[1].getClusters().get(0),
            ClusterType.SOURCE, startTime_cluster2, endTime_cluster2);
        processBundle.addClusterToBundle(bundles[2].getClusters().get(0),
            ClusterType.SOURCE, startTime_cluster3, endTime_cluster3);

        //submit and schedule
        processBundle.submitFeedsScheduleProcess(prism);

        //wait for coord to be in running state
        InstanceUtil.waitTillInstancesAreCreated(cluster1, processBundle.getProcessData(), 0);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, processBundle.getProcessData(), 0);

        //save old info
        String oldBundleID_cluster1 = InstanceUtil.getLatestBundleID(cluster1,
            Util.readEntityName(processBundle.getProcessData()), EntityType.PROCESS);
        List<String> nominalTimes_cluster1 = OozieUtil.getActionsNominalTime(cluster1,
            oldBundleID_cluster1, EntityType.PROCESS);
        String oldBundleID_cluster2 = InstanceUtil.getLatestBundleID(cluster2,
            Util.readEntityName(processBundle.getProcessData()), EntityType.PROCESS);
        String oldBundleID_cluster3 = InstanceUtil.getLatestBundleID(cluster3,
            Util.readEntityName(processBundle.getProcessData()), EntityType.PROCESS);
        List<String> nominalTimes_cluster3 = OozieUtil.getActionsNominalTime(cluster3,
            oldBundleID_cluster3, EntityType.PROCESS);

        //update process
        String updateTime = TimeUtil.addMinsToTime(endTime_cluster1, 3);
        processBundle.setProcessProperty("someProp", "someVal");
        ServiceResponse r = prism.getProcessHelper().update(processBundle.getProcessData(),
            processBundle.getProcessData(), updateTime, null);
        AssertUtil.assertSucceeded(r);

        //check for new bundle to be created
        OozieUtil.verifyNewBundleCreation(cluster1, oldBundleID_cluster1,
            nominalTimes_cluster1, processBundle.getProcessData(), true, false);
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleID_cluster3,
            nominalTimes_cluster3, processBundle.getProcessData(), true, false);
        OozieUtil.verifyNewBundleCreation(cluster2, oldBundleID_cluster2,
            nominalTimes_cluster3, processBundle.getProcessData(), true, false);

        //wait till new coord are running on cluster1
        InstanceUtil.waitTillInstancesAreCreated(cluster1, processBundle.getProcessData(), 1);
        OozieUtil.verifyNewBundleCreation(cluster1, oldBundleID_cluster1,
            nominalTimes_cluster1, processBundle.getProcessData(), true, true);

        //verify
        String coordStartTime_cluster3 = OozieUtil.getCoordStartTime(cluster3,
            processBundle.getProcessData(), 1);
        String coordStartTime_cluster2 = OozieUtil.getCoordStartTime(cluster2,
            processBundle.getProcessData(), 1);

        DateTime updateTimeOozie = TimeUtil.oozieDateToDate(updateTime);
        Assert.assertTrue(TimeUtil.oozieDateToDate(coordStartTime_cluster3).isAfter(updateTimeOozie)
            || TimeUtil.oozieDateToDate(coordStartTime_cluster3).isEqual(updateTimeOozie),
            "new coord start time is not correct");
        Assert.assertFalse(
            TimeUtil.oozieDateToDate(coordStartTime_cluster2).isEqual(updateTimeOozie),
            "new coord start time is not correct");
        TimeUtil.sleepTill(updateTime);
        InstanceUtil.waitTillInstancesAreCreated(cluster3, processBundle.getProcessData(), 1);

        //verify that no instance are missing
        OozieUtil.verifyNewBundleCreation(cluster3, oldBundleID_cluster3,
            nominalTimes_cluster3, processBundle.getProcessData(), true, true);
    }

    private String submitAndScheduleFeed(Bundle b)
        throws JAXBException, IOException, URISyntaxException, AuthenticationException {
        String feed = b.getDataSets().get(0);
        feed = InstanceUtil.setFeedCluster(feed,
            XmlUtil.createValidity("2012-10-01T12:00Z", "2010-01-01T00:00Z"),
            XmlUtil.createRtention("days(1000000)", ActionType.DELETE), null,
            ClusterType.SOURCE, null);
        feed = InstanceUtil.setFeedCluster(feed, XmlUtil.createValidity
                ("2012-10-01T12:10Z", "2099-10-01T12:10Z"),
            XmlUtil.createRtention("days(1000000)", ActionType.DELETE),
            Util.readEntityName(b.getClusters().get(0)), ClusterType.SOURCE, "",
            "/someTestPath" + MINUTE_DATE_PATTERN);
        ServiceResponse r = prism.getClusterHelper().submitEntity(
            b.getClusters().get(0));
        AssertUtil.assertSucceeded(r);
        r = prism.getFeedHelper().submitAndSchedule(feed);
        AssertUtil.assertSucceeded(r);
        return feed;
    }

    private String getMultiClusterFeed(String startTimeCluster_source,
                                       String startTimeCluster_target)
        throws IOException, URISyntaxException, AuthenticationException {
        String testDataDir = baseTestDir + "/replication";

        //create desired feed
        String feed = bundles[0].getDataSets().get(0);

        //cluster1 is target, cluster2 is source and cluster3 is neutral
        feed = InstanceUtil.setFeedCluster(feed,
            XmlUtil.createValidity("2012-10-01T12:00Z", "2010-01-01T00:00Z"),
            XmlUtil.createRtention("days(100000)", ActionType.DELETE), null,
            ClusterType.SOURCE, null);
        feed = InstanceUtil.setFeedCluster(feed,
            XmlUtil.createValidity(startTimeCluster_source, "2099-10-01T12:10Z"),
            XmlUtil.createRtention("days(100000)", ActionType.DELETE),
            Util.readEntityName(bundles[2].getClusters().get(0)), null, null);
        feed = InstanceUtil.setFeedCluster(feed,
            XmlUtil.createValidity(startTimeCluster_target, "2099-10-01T12:25Z"),
            XmlUtil.createRtention("days(100000)", ActionType.DELETE),
            Util.readEntityName(bundles[0].getClusters().get(0)), ClusterType.TARGET,
            null, testDataDir + MINUTE_DATE_PATTERN);
        feed = InstanceUtil.setFeedCluster(feed,
            XmlUtil.createValidity(startTimeCluster_source, "2099-01-01T00:00Z"),
            XmlUtil.createRtention("days(100000)", ActionType.DELETE),
            Util.readEntityName(bundles[1].getClusters().get(0)), ClusterType.SOURCE,
            null, testDataDir + MINUTE_DATE_PATTERN);

        //submit clusters
        Bundle.submitCluster(bundles[0], bundles[1], bundles[2]);

        //create test data on cluster2
        List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startTimeCluster_source,
            TimeUtil.getTimeWrtSystemTime(60), 1);
        HadoopUtil.flattenAndPutDataInFolder(cluster2FS, OSUtil.SINGLE_FILE,
            testDataDir + "/", dataDates);
        return feed;
    }

    @AfterMethod(alwaysRun = true)
    public void tearDown(Method method) {
        logger.info("tearDown " + method.getName());
        processBundle.deleteBundle(prism);
        bundles[0].deleteBundle(prism);
        processBundle.deleteBundle(prism);
    }

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