| /** |
| * 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.hcat; |
| |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.falcon.regression.core.bundle.Bundle; |
| import org.apache.falcon.entity.v0.cluster.Interfacetype; |
| 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.util.AssertUtil; |
| import org.apache.falcon.regression.core.util.BundleUtil; |
| import org.apache.falcon.regression.core.util.HCatUtil; |
| 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.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.fs.Path; |
| import org.apache.hive.hcatalog.api.HCatAddPartitionDesc; |
| import org.apache.hive.hcatalog.api.HCatClient; |
| import org.apache.hive.hcatalog.api.HCatCreateTableDesc; |
| import org.apache.hive.hcatalog.common.HCatException; |
| import org.apache.hive.hcatalog.data.schema.HCatFieldSchema; |
| import org.apache.log4j.Logger; |
| import org.apache.oozie.client.CoordinatorAction; |
| import org.apache.oozie.client.OozieClient; |
| import org.joda.time.format.DateTimeFormat; |
| 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.DataProvider; |
| import org.testng.annotations.Test; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| |
| @Test(groups = "embedded") |
| public class HCatReplicationTest extends BaseTestClass { |
| |
| private static final Logger logger = Logger.getLogger(HCatReplicationTest.class); |
| ColoHelper cluster = servers.get(0); |
| FileSystem clusterFS = serverFS.get(0); |
| HCatClient clusterHC; |
| |
| ColoHelper cluster2 = servers.get(1); |
| FileSystem cluster2FS = serverFS.get(1); |
| OozieClient cluster2OC = serverOC.get(1); |
| HCatClient cluster2HC; |
| |
| ColoHelper cluster3 = servers.get(2); |
| FileSystem cluster3FS = serverFS.get(2); |
| OozieClient cluster3OC = serverOC.get(2); |
| HCatClient cluster3HC; |
| |
| final String baseTestHDFSDir = baseHDFSDir + "/HCatReplicationTest"; |
| |
| final String dbName = "default"; |
| private static final String localHCatData = OSUtil.getPath(OSUtil.RESOURCES, "hcat", "data"); |
| private static final double TIMEOUT = 15; |
| |
| @BeforeClass(alwaysRun = true) |
| public void beforeClass() throws IOException { |
| clusterHC = cluster.getClusterHelper().getHCatClient(); |
| cluster2HC = cluster2.getClusterHelper().getHCatClient(); |
| cluster3HC = cluster3.getClusterHelper().getHCatClient(); |
| // create the base dir on all clusters. |
| HadoopUtil.recreateDir(serverFS, baseTestHDFSDir); |
| } |
| |
| @BeforeMethod(alwaysRun = true) |
| public void setUp() throws Exception { |
| Bundle bundle = BundleUtil.readHCatBundle(); |
| bundles[0] = new Bundle(bundle, cluster.getPrefix()); |
| bundles[0].generateUniqueBundle(); |
| bundles[0].setClusterInterface(Interfacetype.REGISTRY, |
| cluster.getClusterHelper().getHCatEndpoint()); |
| |
| bundles[1] = new Bundle(bundle, cluster2.getPrefix()); |
| bundles[1].generateUniqueBundle(); |
| bundles[1].setClusterInterface(Interfacetype.REGISTRY, cluster2.getClusterHelper() |
| .getHCatEndpoint()); |
| |
| bundles[2] = new Bundle(bundle, cluster3.getPrefix()); |
| bundles[2].generateUniqueBundle(); |
| bundles[2].setClusterInterface(Interfacetype.REGISTRY, cluster3.getClusterHelper() |
| .getHCatEndpoint()); |
| |
| } |
| |
| @DataProvider |
| public String[][] generateSeparators() { |
| //disabling till FALCON-372 is fixed |
| //return new String[][] {{"-"}, {"/"}}; |
| return new String[][]{{"-"},}; |
| } |
| |
| // make sure oozie changes mentioned FALCON-389 are done on the clusters. Otherwise the test |
| // will fail. |
| // Noticed with hive 0.13 we need the following issues resolved to work HIVE-6848 and |
| // HIVE-6868. Also oozie share libs need to have hive jars that have these jira's resolved and |
| // the maven depenendcy you are using to run the tests has to have hcat that has these fixed. |
| @Test(dataProvider = "generateSeparators") |
| public void oneSourceOneTarget(String separator) throws Exception { |
| String tcName = "HCatReplication_oneSourceOneTarget"; |
| if (separator.equals("-")) { |
| tcName += "_hyphen"; |
| } else { |
| tcName += "_slash"; |
| } |
| String tblName = tcName; |
| String testHdfsDir = baseTestHDFSDir + "/" + tcName; |
| HadoopUtil.recreateDir(serverFS, testHdfsDir); |
| final String startDate = "2010-01-01T20:00Z"; |
| final String endDate = "2099-01-01T00:00Z"; |
| final String dataEndDate = "2010-01-01T21:00Z"; |
| final String tableUriPartitionFragment = StringUtils |
| .join(new String[]{"#dt=${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator); |
| String tableUri = "catalog:" + dbName + ":" + tblName + tableUriPartitionFragment; |
| final String datePattern = |
| StringUtils.join(new String[]{"yyyy", "MM", "dd", "HH"}, separator); |
| // use the start date for both as this will only generate 2 partitions. |
| List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startDate, dataEndDate, 60, |
| DateTimeFormat.forPattern(datePattern)); |
| |
| final List<String> dataset = HadoopUtil.flattenAndPutDataInFolder(clusterFS, |
| localHCatData, testHdfsDir, dataDates); |
| final String col1Name = "id"; |
| final String col2Name = "value"; |
| final String partitionColumn = "dt"; |
| |
| ArrayList<HCatFieldSchema> cols = new ArrayList<HCatFieldSchema>(); |
| cols.add(HCatUtil.getStringSchema(col1Name, col1Name + " comment")); |
| cols.add(HCatUtil.getStringSchema(col2Name, col2Name + " comment")); |
| ArrayList<HCatFieldSchema> partitionCols = new ArrayList<HCatFieldSchema>(); |
| |
| // create table on cluster 1 and add data to it. |
| partitionCols.add(HCatUtil.getStringSchema(partitionColumn, partitionColumn + " partition")); |
| createTable(clusterHC, dbName, tblName, cols, partitionCols, testHdfsDir); |
| addPartitionsToTable(dataDates, dataset, "dt", dbName, tblName, clusterHC); |
| |
| // create table on target cluster. |
| createTable(cluster2HC, dbName, tblName, cols, partitionCols, testHdfsDir); |
| |
| Bundle.submitCluster(bundles[0], bundles[1]); |
| |
| bundles[0].setInputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setInputFeedValidity(startDate, endDate); |
| bundles[0].setInputFeedTableUri(tableUri); |
| |
| String feed = bundles[0].getDataSets().get(0); |
| // set the cluster 2 as the target. |
| feed = InstanceUtil.setFeedClusterWithTable(feed, |
| XmlUtil.createValidity(startDate, endDate), |
| XmlUtil.createRtention("months(9000)", ActionType.DELETE), |
| Util.readEntityName(bundles[1].getClusters().get(0)), ClusterType.TARGET, null, |
| tableUri); |
| |
| AssertUtil.assertSucceeded(prism.getFeedHelper().submitAndSchedule(feed)); |
| TimeUtil.sleepSeconds(TIMEOUT); |
| //check if all coordinators exist |
| Assert.assertEquals(InstanceUtil |
| .checkIfFeedCoordExist(cluster2.getFeedHelper(), Util.readEntityName(feed), |
| "REPLICATION"), 1); |
| |
| //replication should start, wait while it ends |
| // we will check for 2 instances so that both partitions are copied over. |
| InstanceUtil.waitTillInstanceReachState(cluster2OC, Util.readEntityName(feed), 2, |
| CoordinatorAction.Status.SUCCEEDED, EntityType.FEED); |
| |
| //check if data was replicated correctly |
| List<Path> cluster1ReplicatedData = HadoopUtil |
| .getAllFilesRecursivelyHDFS(clusterFS, new Path(testHdfsDir)); |
| logger.info("Data on source cluster: " + cluster1ReplicatedData); |
| List<Path> cluster2ReplicatedData = HadoopUtil |
| .getAllFilesRecursivelyHDFS(cluster2FS, new Path(testHdfsDir)); |
| logger.info("Data on target cluster: " + cluster2ReplicatedData); |
| AssertUtil.checkForListSizes(cluster1ReplicatedData, cluster2ReplicatedData); |
| |
| } |
| |
| // make sure oozie changes mentioned FALCON-389 are done on the clusters. Otherwise the test |
| // will fail. |
| // Noticed with hive 0.13 we need the following issues resolved to work HIVE-6848 and |
| // HIVE-6868. Also oozie share libs need to have hive jars that have these jira's resolved and |
| // the maven depenendcy you are using to run the tests has to have hcat that has these fixed. |
| // This test can fail randomly because of https://issues.apache.org/jira/browse/FALCON-401 |
| @Test(dataProvider = "generateSeparators") |
| public void oneSourceTwoTarget(String separator) throws Exception { |
| String tcName = "HCatReplication_oneSourceTwoTarget"; |
| if (separator.equals("-")) { |
| tcName += "_hyphen"; |
| } else { |
| tcName += "_slash"; |
| } |
| String tblName = tcName; |
| String testHdfsDir = baseTestHDFSDir + "/" + tcName; |
| HadoopUtil.recreateDir(serverFS, testHdfsDir); |
| final String startDate = "2010-01-01T20:00Z"; |
| final String endDate = "2099-01-01T00:00Z"; |
| final String dataEndDate = "2010-01-01T21:00Z"; |
| final String tableUriPartitionFragment = StringUtils |
| .join(new String[]{"#dt=${YEAR}", "${MONTH}", "${DAY}", "${HOUR}"}, separator); |
| String tableUri = "catalog:" + dbName + ":" + tblName + tableUriPartitionFragment; |
| final String datePattern = |
| StringUtils.join(new String[]{"yyyy", "MM", "dd", "HH"}, separator); |
| // use the start date for both as this will only generate 2 partitions. |
| List<String> dataDates = TimeUtil.getMinuteDatesOnEitherSide(startDate, dataEndDate, 60, |
| DateTimeFormat.forPattern(datePattern)); |
| |
| final List<String> dataset = HadoopUtil.flattenAndPutDataInFolder(clusterFS, |
| localHCatData, testHdfsDir, dataDates); |
| final String col1Name = "id"; |
| final String col2Name = "value"; |
| final String partitionColumn = "dt"; |
| |
| ArrayList<HCatFieldSchema> cols = new ArrayList<HCatFieldSchema>(); |
| cols.add(HCatUtil.getStringSchema(col1Name, col1Name + " comment")); |
| cols.add(HCatUtil.getStringSchema(col2Name, col2Name + " comment")); |
| ArrayList<HCatFieldSchema> partitionCols = new ArrayList<HCatFieldSchema>(); |
| |
| // create table on cluster 1 and add data to it. |
| partitionCols.add(HCatUtil.getStringSchema(partitionColumn, partitionColumn + " partition")); |
| createTable(clusterHC, dbName, tblName, cols, partitionCols, testHdfsDir); |
| addPartitionsToTable(dataDates, dataset, "dt", dbName, tblName, clusterHC); |
| |
| // create table on target cluster. |
| createTable(cluster2HC, dbName, tblName, cols, partitionCols, testHdfsDir); |
| createTable(cluster3HC, dbName, tblName, cols, partitionCols, testHdfsDir); |
| |
| Bundle.submitCluster(bundles[0], bundles[1], bundles[2]); |
| |
| bundles[0].setInputFeedPeriodicity(1, Frequency.TimeUnit.hours); |
| bundles[0].setInputFeedValidity(startDate, endDate); |
| bundles[0].setInputFeedTableUri(tableUri); |
| |
| String feed = bundles[0].getDataSets().get(0); |
| // set the cluster 2 as the target. |
| feed = InstanceUtil.setFeedClusterWithTable(feed, |
| XmlUtil.createValidity(startDate, endDate), |
| XmlUtil.createRtention("months(9000)", ActionType.DELETE), |
| Util.readEntityName(bundles[1].getClusters().get(0)), ClusterType.TARGET, null, |
| tableUri); |
| // set the cluster 3 as the target. |
| feed = InstanceUtil.setFeedClusterWithTable(feed, |
| XmlUtil.createValidity(startDate, endDate), |
| XmlUtil.createRtention("months(9000)", ActionType.DELETE), |
| Util.readEntityName(bundles[2].getClusters().get(0)), ClusterType.TARGET, null, |
| tableUri); |
| |
| AssertUtil.assertSucceeded(prism.getFeedHelper().submitAndSchedule(feed)); |
| TimeUtil.sleepSeconds(TIMEOUT); |
| //check if all coordinators exist |
| Assert.assertEquals(InstanceUtil |
| .checkIfFeedCoordExist(cluster2.getFeedHelper(), Util.readEntityName(feed), |
| "REPLICATION"), 1); |
| |
| //check if all coordinators exist |
| Assert.assertEquals(InstanceUtil |
| .checkIfFeedCoordExist(cluster3.getFeedHelper(), Util.readEntityName(feed), |
| "REPLICATION"), 1); |
| |
| //replication should start, wait while it ends |
| // we will check for 2 instances so that both partitions are copied over. |
| InstanceUtil.waitTillInstanceReachState(cluster2OC, Util.readEntityName(feed), 2, |
| CoordinatorAction.Status.SUCCEEDED, EntityType.FEED); |
| |
| //replication should start, wait while it ends |
| // we will check for 2 instances so that both partitions are copied over. |
| InstanceUtil.waitTillInstanceReachState(cluster3OC, Util.readEntityName(feed), 2, |
| CoordinatorAction.Status.SUCCEEDED, EntityType.FEED); |
| |
| //check if data was replicated correctly |
| List<Path> srcData = HadoopUtil |
| .getAllFilesRecursivelyHDFS(clusterFS, new Path(testHdfsDir)); |
| logger.info("Data on source cluster: " + srcData); |
| List<Path> cluster2TargetData = HadoopUtil |
| .getAllFilesRecursivelyHDFS(cluster2FS, new Path(testHdfsDir)); |
| logger.info("Data on target cluster: " + cluster2TargetData); |
| AssertUtil.checkForListSizes(srcData, cluster2TargetData); |
| List<Path> cluster3TargetData = HadoopUtil |
| .getAllFilesRecursivelyHDFS(cluster3FS, new Path(testHdfsDir)); |
| logger.info("Data on target cluster: " + cluster3TargetData); |
| AssertUtil.checkForListSizes(srcData, cluster3TargetData); |
| } |
| |
| //TODO: More tests need to be added such as |
| // Tests to make sure new partitions that are added are replicated |
| // Tests to make sure partitions that do no match the pattern are not copied |
| |
| private void addPartitionsToTable(List<String> partitions, List<String> partitionLocations, |
| String partitionCol, |
| String dbName, String tableName, HCatClient hc) throws |
| HCatException { |
| Assert.assertEquals(partitions.size(), partitionLocations.size(), |
| "Number of locations is not same as number of partitions."); |
| final List<HCatAddPartitionDesc> partitionDesc = new ArrayList<HCatAddPartitionDesc>(); |
| for (int i = 0; i < partitions.size(); ++i) { |
| final String partition = partitions.get(i); |
| final Map<String, String> onePartition = new HashMap<String, String>(); |
| onePartition.put(partitionCol, partition); |
| final String partitionLoc = partitionLocations.get(i); |
| partitionDesc |
| .add(HCatAddPartitionDesc.create(dbName, tableName, partitionLoc, onePartition) |
| .build()); |
| } |
| logger.info("adding partitions: " + partitionDesc); |
| hc.addPartitions(partitionDesc); |
| } |
| |
| private static void createTable(HCatClient hcatClient, String dbName, String tblName, |
| List<HCatFieldSchema> cols, List<HCatFieldSchema> partitionCols, |
| String hdfsDir) throws HCatException { |
| hcatClient.dropTable(dbName, tblName, true); |
| hcatClient.createTable(HCatCreateTableDesc |
| .create(dbName, tblName, cols) |
| .partCols(partitionCols) |
| .ifNotExists(true) |
| .isTableExternal(true) |
| .location(hdfsDir) |
| .build()); |
| } |
| |
| @AfterMethod(alwaysRun = true) |
| public void tearDown() { |
| removeBundles(); |
| } |
| |
| @AfterClass(alwaysRun = true) |
| public void tearDownClass() throws IOException { |
| cleanTestDirs(); |
| } |
| } |
