GitHub PullRequestReviewCommentEvent at 2014-03-27 20:47:46
Repository: falcon-regression
Actor: raghavgautam
Payload: {"comment":{"url":"https://api.github.com/repos/samarthgupta437/falcon-regression/pulls/comments/11044201","id":11044201,"diff_hunk":"@@ -0,0 +1,328 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License.  You may obtain a copy of the License at\n+ *\n+ *     http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.falcon.regression;\n+\n+import org.apache.falcon.regression.core.bundle.Bundle;\n+import org.apache.falcon.regression.core.enumsAndConstants.ENTITY_TYPE;\n+import org.apache.falcon.regression.core.generated.feed.ActionType;\n+import org.apache.falcon.regression.core.generated.feed.ClusterType;\n+import org.apache.falcon.regression.core.generated.feed.Feed;\n+import org.apache.falcon.regression.core.helpers.ColoHelper;\n+import org.apache.falcon.regression.core.response.ProcessInstancesResult;\n+import org.apache.falcon.regression.core.util.AssertUtil;\n+import org.apache.falcon.regression.core.util.HadoopUtil;\n+import org.apache.falcon.regression.core.util.InstanceUtil;\n+import org.apache.falcon.regression.core.util.Util;\n+import org.apache.falcon.regression.core.util.XmlUtil;\n+import org.apache.falcon.regression.testHelper.BaseTestClass;\n+import org.apache.hadoop.fs.FileSystem;\n+import org.apache.hadoop.fs.Path;\n+import org.apache.hadoop.security.authentication.client.AuthenticationException;\n+import org.apache.oozie.client.CoordinatorAction;\n+import org.apache.oozie.client.OozieClient;\n+import org.apache.oozie.client.OozieClientException;\n+import org.joda.time.DateTime;\n+import org.joda.time.DateTimeZone;\n+import org.joda.time.format.DateTimeFormat;\n+import org.joda.time.format.DateTimeFormatter;\n+import org.testng.Assert;\n+import org.testng.annotations.BeforeMethod;\n+import org.testng.annotations.Test;\n+\n+import javax.xml.bind.JAXBException;\n+import java.io.IOException;\n+import java.lang.reflect.Method;\n+import java.net.URISyntaxException;\n+import java.text.ParseException;\n+import java.util.List;\n+\n+@Test(groups = \"embedded\")\n+public class FeedReplicationTest extends BaseTestClass {\n+\n+    ColoHelper cluster1 = servers.get(0);\n+    ColoHelper cluster2 = servers.get(1);\n+    ColoHelper cluster3 = servers.get(2);\n+    FileSystem cluster1FS = serverFS.get(0);\n+    OozieClient cluster2OC = serverOC.get(1);\n+    OozieClient cluster3OC = serverOC.get(2);\n+    String aggregateWorkflowDir = baseWorkflowDir + \"/aggregator\";\n+    String dateTemplate = \"/${YEAR}/${MONTH}/${DAY}/${HOUR}/${MINUTE}\";\n+    String baseTestDir = baseHDFSDir + \"/FeedReplicationTest\";\n+    String sourcePath = baseTestDir + \"/source\";\n+    String feedDataLocation = baseTestDir + \"/source\" + dateTemplate;\n+    String targetPath = baseTestDir + \"/target\";\n+    String targetDataLocation = targetPath + dateTemplate;\n+    int timeout = 3;\n+\n+    @BeforeMethod(alwaysRun = true)\n+    public void setUp(Method method) throws JAXBException, IOException {\n+        Util.print(\"test name: \" + method.getName());\n+        Bundle bundle = (Bundle) Bundle.readBundle(\"LocalDC_feedReplicaltion_BillingRC\")[0][0];\n+\n+        bundles[0] = new Bundle(bundle, cluster1.getEnvFileName(), cluster1.getPrefix());\n+        bundles[1] = new Bundle(bundle, cluster2.getEnvFileName(), cluster2.getPrefix());\n+        bundles[2] = new Bundle(bundle, cluster3.getEnvFileName(), cluster3.getPrefix());\n+\n+        bundles[0].generateUniqueBundle();\n+        bundles[1].generateUniqueBundle();\n+        bundles[2].generateUniqueBundle();\n+    }\n+\n+    @Test\n+    public void replicate1Source1Target()\n+    throws AuthenticationException, IOException, URISyntaxException, ParseException, JAXBException,\n+    InterruptedException, OozieClientException {\n+        Bundle.submitCluster(bundles[0], bundles[1]);\n+        String startTime = InstanceUtil.getTimeWrtSystemTime(0);\n+        String endTime = InstanceUtil.addMinsToTime(startTime, 5);\n+        Util.print(\"Time range between : \" + startTime + \" and \" + endTime);\n+\n+        //configure feed\n+        String feed = bundles[0].getDataSets().get(0);\n+        feed = InstanceUtil.setFeedFilePath(feed, feedDataLocation);\n+        //erase all clusters from feed definition\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(\"2012-10-01T12:00Z\", \"2010-01-01T00:00Z\"),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE), null,\n+                ClusterType.SOURCE, null);\n+        //set cluster1 as source\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(startTime, endTime),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE),\n+                Util.readClusterName(bundles[0].getClusters().get(0)),\n+                ClusterType.SOURCE, null);\n+        //set cluster2 as target\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(startTime, endTime),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE),\n+                Util.readClusterName(bundles[1].getClusters().get(0)),\n+                ClusterType.TARGET, null, targetDataLocation);\n+\n+        //submit and schedule feed\n+        Util.print(\"Feed : \" + feed);\n+        AssertUtil.assertSucceeded(\n+                prism.getFeedHelper().submitAndSchedule(Util.URLS.SUBMIT_AND_SCHEDULE_URL,\n+                        feed));\n+\n+        //upload necessary data\n+        DateTime date = new DateTime(startTime, DateTimeZone.UTC);\n+        DateTimeFormatter fmt = DateTimeFormat.forPattern(\"yyyy'/'MM'/'dd'/'HH'/'mm'\");\n+        String timePattern = fmt.print(date);\n+        String sourceLocation = sourcePath + \"/\" + timePattern + \"/\";\n+        String targetLocation = targetPath + \"/\" + timePattern + \"/\";\n+        HadoopUtil.createDir(sourceLocation, cluster1FS);\n+\n+        Path toSource = new Path(sourceLocation);\n+        Path toTarget = new Path(targetLocation);\n+        HadoopUtil.copyDataToFolder(cluster1, toSource, \"feed-s4Replication.xml\");\n+        HadoopUtil.copyDataToFolder(cluster1, toSource, \"log_01.txt\");\n+\n+        //check if all coordinators exist\n+        Assert.assertEquals(InstanceUtil\n+                .checkIfFeedCoordExist(cluster2.getFeedHelper(), Util.readDatasetName(feed),\n+                        \"REPLICATION\"), 1);\n+\n+        //replication should start, wait while it ends\n+        InstanceUtil.waitTillInstanceReachState(cluster2OC, Util.readEntityName(feed), 1,\n+                CoordinatorAction.Status.SUCCEEDED, timeout, ENTITY_TYPE.FEED);\n+\n+        //check if data has been replicated correctly\n+        List<Path> cluster1ReplicatedData = HadoopUtil\n+                .getAllFilesRecursivelyHDFS(cluster1, toSource);\n+        List<Path> cluster2ReplicatedData = HadoopUtil\n+                .getAllFilesRecursivelyHDFS(cluster2, toTarget, \"_SUCCESS\");\n+\n+        AssertUtil.checkForPathsSizes(cluster1ReplicatedData, cluster2ReplicatedData);\n+    }\n+\n+    @Test\n+    public void replicate1Source2Targets() throws Exception {\n+        Bundle.submitCluster(bundles[0], bundles[1], bundles[2]);\n+        String startTime = InstanceUtil.getTimeWrtSystemTime(0);\n+        String endTime = InstanceUtil.addMinsToTime(startTime, 5);\n+        Util.print(\"Time range between : \" + startTime + \" and \" + endTime);\n+\n+        //configure feed\n+        String feed = bundles[0].getDataSets().get(0);\n+        feed = InstanceUtil.setFeedFilePath(feed, feedDataLocation);\n+        //erase all clusters from feed definition\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(\"2012-10-01T12:00Z\", \"2010-01-01T00:00Z\"),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE), null,\n+                ClusterType.SOURCE, null);\n+        //set cluster1 as source\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(startTime, endTime),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE),\n+                Util.readClusterName(bundles[0].getClusters().get(0)),\n+                ClusterType.SOURCE, null);\n+        //set cluster2 as target\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(startTime, endTime),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE),\n+                Util.readClusterName(bundles[1].getClusters().get(0)),\n+                ClusterType.TARGET, null, targetDataLocation);\n+        //set cluster3 as target\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(startTime, endTime),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE),\n+                Util.readClusterName(bundles[2].getClusters().get(0)),\n+                ClusterType.TARGET, null, targetDataLocation);\n+\n+        //submit and schedule feed\n+        Util.print(\"Feed : \" + feed);\n+        AssertUtil.assertSucceeded(\n+                prism.getFeedHelper().submitAndSchedule(Util.URLS.SUBMIT_AND_SCHEDULE_URL,\n+                        feed));\n+\n+        //upload necessary data\n+        DateTime date = new DateTime(startTime, DateTimeZone.UTC);\n+        DateTimeFormatter fmt = DateTimeFormat.forPattern(\"yyyy'/'MM'/'dd'/'HH'/'mm'\");\n+        String timePattern = fmt.print(date);\n+        String sourceLocation = sourcePath + \"/\" + timePattern + \"/\";\n+        String targetLocation = targetPath + \"/\" + timePattern + \"/\";\n+        HadoopUtil.createDir(sourceLocation, cluster1FS);\n+\n+        Path toSource = new Path(sourceLocation);\n+        Path toTarget = new Path(targetLocation);\n+        HadoopUtil.copyDataToFolder(cluster1, toSource, \"feed-s4Replication.xml\");\n+        HadoopUtil.copyDataToFolder(cluster1, toSource, \"log_01.txt\");\n+\n+        //check if all coordinators exist\n+        Assert.assertEquals(InstanceUtil\n+                .checkIfFeedCoordExist(cluster2.getFeedHelper(), Util.readDatasetName(feed),\n+                        \"REPLICATION\"), 1);\n+        Assert.assertEquals(InstanceUtil\n+                .checkIfFeedCoordExist(cluster3.getFeedHelper(), Util.readDatasetName(feed),\n+                        \"REPLICATION\"), 1);\n+\n+        //replication on cluster 2 should start, wait till it ends\n+        InstanceUtil.waitTillInstanceReachState(cluster2OC, Util.readEntityName(feed), 1,\n+                CoordinatorAction.Status.SUCCEEDED, timeout, ENTITY_TYPE.FEED);\n+\n+        //replication on cluster 3 should start, wait till it ends\n+        InstanceUtil.waitTillInstanceReachState(cluster3OC, Util.readEntityName(feed), 1,\n+                CoordinatorAction.Status.SUCCEEDED, timeout, ENTITY_TYPE.FEED);\n+\n+        //check if data has been replicated correctly\n+        List<Path> cluster1ReplicatedData = HadoopUtil\n+                .getAllFilesRecursivelyHDFS(cluster1, toSource);\n+        List<Path> cluster2ReplicatedData = HadoopUtil\n+                .getAllFilesRecursivelyHDFS(cluster2, toTarget, \"_SUCCESS\");\n+        List<Path> cluster3ReplicatedData = HadoopUtil\n+                .getAllFilesRecursivelyHDFS(cluster3, toTarget, \"_SUCCESS\");\n+\n+        AssertUtil.checkForPathsSizes(cluster1ReplicatedData, cluster2ReplicatedData);\n+        AssertUtil.checkForPathsSizes(cluster1ReplicatedData, cluster3ReplicatedData);\n+    }\n+\n+    @Test\n+    public void availabilityFlagTest() throws Exception {\n+        //replicate1Source1Target scenario + set availability flag but don't upload required file\n+        Bundle.submitCluster(bundles[0], bundles[1]);\n+        String startTime = InstanceUtil.getTimeWrtSystemTime(0);\n+        String endTime = InstanceUtil.addMinsToTime(startTime, 5);\n+        Util.print(\"Time range between : \" + startTime + \" and \" + endTime);\n+\n+        //configure feed\n+        String availabilityFlagName = \"README.md\";\n+        String feedName = Util.readEntityName(bundles[0].getDataSets().get(0));\n+        Feed feedElement = InstanceUtil.getFeedElement(bundles[0], feedName);\n+        feedElement.setAvailabilityFlag(availabilityFlagName);\n+        InstanceUtil.writeFeedElement(bundles[0], feedElement, feedName);\n+        String feed = bundles[0].getDataSets().get(0);\n+        feed = InstanceUtil.setFeedFilePath(feed, feedDataLocation);\n+        //erase all clusters from feed definition\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(\"2012-10-01T12:00Z\", \"2010-01-01T00:00Z\"),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE), null,\n+                ClusterType.SOURCE, null);\n+        //set cluster1 as source\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(startTime, endTime),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE),\n+                Util.readClusterName(bundles[0].getClusters().get(0)),\n+                ClusterType.SOURCE, null);\n+        //set cluster2 as target\n+        feed = InstanceUtil.setFeedCluster(feed,\n+                XmlUtil.createValidity(startTime, endTime),\n+                XmlUtil.createRtention(\"days(1000000)\", ActionType.DELETE),\n+                Util.readClusterName(bundles[1].getClusters().get(0)),\n+                ClusterType.TARGET, null, targetDataLocation);\n+\n+        //submit and schedule feed\n+        Util.print(\"Feed : \" + feed);\n+        AssertUtil.assertSucceeded(\n+                prism.getFeedHelper().submitAndSchedule(Util.URLS.SUBMIT_AND_SCHEDULE_URL,\n+                        feed));\n+\n+        //upload necessary data\n+        DateTime date = new DateTime(startTime, DateTimeZone.UTC);\n+        DateTimeFormatter fmt = DateTimeFormat.forPattern(\"yyyy'/'MM'/'dd'/'HH'/'mm'\");\n+        String timePattern = fmt.print(date);\n+        String sourceLocation = sourcePath + \"/\" + timePattern + \"/\";\n+        String targetLocation = targetPath + \"/\" + timePattern + \"/\";\n+        HadoopUtil.createDir(sourceLocation, cluster1FS);\n+\n+        Path toSource = new Path(sourceLocation);\n+        Path toTarget = new Path(targetLocation);\n+        HadoopUtil.copyDataToFolder(cluster1, toSource, \"feed-s4Replication.xml\");\n+        HadoopUtil.copyDataToFolder(cluster1, toSource, \"log_01.txt\");\n+\n+        //check if all coordinators exist\n+        Assert.assertEquals(InstanceUtil\n+                .checkIfFeedCoordExist(cluster2.getFeedHelper(), feedName, \"REPLICATION\"), 1);\n+\n+        //replication should not start\n+        boolean started = true;\n+        try{\n+            InstanceUtil.waitTillInstanceReachState(cluster2OC, Util.readEntityName(feed), 1,\n+                    CoordinatorAction.Status.RUNNING, timeout, ENTITY_TYPE.FEED);\n+        }catch (AssertionError error){","path":"merlin/src/test/java/org/apache/falcon/regression/FeedReplicationTest.java","position":297,"original_position":297,"commit_id":"a3027c0a454d6d3c2d4ba193808f79a3904623b8","original_commit_id":"a3027c0a454d6d3c2d4ba193808f79a3904623b8","user":{"login":"raghavgautam","id":1521534,"avatar_url":"https://avatars.githubusercontent.com/u/1521534?","gravatar_id":"3349a18e5a27e065017fe12a880daf11","url":"https://api.github.com/users/raghavgautam","html_url":"https://github.com/raghavgautam","followers_url":"https://api.github.com/users/raghavgautam/followers","following_url":"https://api.github.com/users/raghavgautam/following{/other_user}","gists_url":"https://api.github.com/users/raghavgautam/gists{/gist_id}","starred_url":"https://api.github.com/users/raghavgautam/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/raghavgautam/subscriptions","organizations_url":"https://api.github.com/users/raghavgautam/orgs","repos_url":"https://api.github.com/users/raghavgautam/repos","events_url":"https://api.github.com/users/raghavgautam/events{/privacy}","received_events_url":"https://api.github.com/users/raghavgautam/received_events","type":"User","site_admin":false},"body":"catching AssertionError is not a good idea. I think what you are trying to accomplish is to wait for 1 minute and then ensure that Coordinator action is not running after that which can be done by just sleeping for 1 min.","created_at":"2014-03-27T20:47:46Z","updated_at":"2014-03-27T20:47:46Z","html_url":"https://github.com/samarthgupta437/falcon-regression/pull/116#discussion_r11044201","pull_request_url":"https://api.github.com/repos/samarthgupta437/falcon-regression/pulls/116","_links":{"self":{"href":"https://api.github.com/repos/samarthgupta437/falcon-regression/pulls/comments/11044201"},"html":{"href":"https://github.com/samarthgupta437/falcon-regression/pull/116#discussion_r11044201"},"pull_request":{"href":"https://api.github.com/repos/samarthgupta437/falcon-regression/pulls/116"}}}}