StartupProperties.get().setProperty(SecurityUtil.AUTHENTICATION_TYPE, secureOption);
OozieFeedWorkflowBuilder builder = new OozieFeedWorkflowBuilder(tableFeed);
List<COORDINATORAPP> coords = builder.getCoordinators(
trgCluster, new Path("/projects/falcon/"));
COORDINATORAPP coord = coords.get(0);
Assert.assertEquals("2010-01-01T00:40Z", coord.getStart());
Assert.assertEquals(getWorkflowAppPath(),
coord.getAction().getWorkflow().getAppPath());
Assert.assertEquals("FALCON_FEED_REPLICATION_" + tableFeed.getName() + "_"
+ srcCluster.getName(), coord.getName());
Assert.assertEquals("${coord:minutes(20)}", coord.getFrequency());
SYNCDATASET inputDataset = (SYNCDATASET) coord.getDatasets()
.getDatasetOrAsyncDataset().get(0);
Assert.assertEquals("${coord:minutes(20)}", inputDataset.getFrequency());
Assert.assertEquals("input-dataset", inputDataset.getName());
String sourceRegistry = ClusterHelper.getInterface(srcCluster, Interfacetype.REGISTRY).getEndpoint();
sourceRegistry = sourceRegistry.replace("thrift", "hcat");
Assert.assertEquals(inputDataset.getUriTemplate(),
sourceRegistry + "/source_db/source_clicks_table/ds=${YEAR}${MONTH}${DAY};region=${region}");
SYNCDATASET outputDataset = (SYNCDATASET) coord.getDatasets()
.getDatasetOrAsyncDataset().get(1);
Assert.assertEquals(outputDataset.getFrequency(), "${coord:minutes(20)}");
Assert.assertEquals("output-dataset", outputDataset.getName());
String targetRegistry = ClusterHelper.getInterface(trgCluster, Interfacetype.REGISTRY).getEndpoint();
targetRegistry = targetRegistry.replace("thrift", "hcat");
Assert.assertEquals(outputDataset.getUriTemplate(),
targetRegistry + "/target_db/target_clicks_table/ds=${YEAR}${MONTH}${DAY};region=${region}");
String inEventName =coord.getInputEvents().getDataIn().get(0).getName();
String inEventDataset =coord.getInputEvents().getDataIn().get(0).getDataset();
String inEventInstance = coord.getInputEvents().getDataIn().get(0).getInstance().get(0);
Assert.assertEquals("input", inEventName);
Assert.assertEquals("input-dataset", inEventDataset);
Assert.assertEquals("${now(0,-40)}", inEventInstance);
String outEventInstance = coord.getOutputEvents().getDataOut().get(0).getInstance();
Assert.assertEquals("${now(0,-40)}", outEventInstance);
// assert FS staging area
String wfPath = coord.getAction().getWorkflow().getAppPath().replace("${nameNode}", "");
final FileSystem fs = trgMiniDFS.getFileSystem();
Assert.assertTrue(fs.exists(new Path(wfPath + "/scripts")));
Assert.assertTrue(fs.exists(new Path(wfPath + "/scripts/falcon-table-export.hql")));
Assert.assertTrue(fs.exists(new Path(wfPath + "/scripts/falcon-table-import.hql")));
Assert.assertTrue(fs.exists(new Path(wfPath + "/conf")));
Assert.assertTrue(fs.exists(new Path(wfPath + "/conf/falcon-source-hive-site.xml")));
Assert.assertTrue(fs.exists(new Path(wfPath + "/conf/falcon-target-hive-site.xml")));
HashMap<String, String> props = new HashMap<String, String>();
for (Property prop : coord.getAction().getWorkflow().getConfiguration().getProperty()) {
props.put(prop.getName(), prop.getValue());
}
final CatalogStorage srcStorage = (CatalogStorage) FeedHelper.createStorage(srcCluster, tableFeed);
final CatalogStorage trgStorage = (CatalogStorage) FeedHelper.createStorage(trgCluster, tableFeed);
// verify the replication param that feed replicator depends on
Assert.assertEquals(props.get("sourceRelativePaths"), "IGNORE");
Assert.assertTrue(props.containsKey("distcpSourcePaths"));
Assert.assertEquals(props.get("distcpSourcePaths"),
FeedHelper.getStagingPath(srcCluster, tableFeed, srcStorage, Tag.REPLICATION,
"${coord:formatTime(coord:nominalTime(), 'yyyy-MM-dd-HH-mm')}" + "/" + trgCluster.getName()));
Assert.assertTrue(props.containsKey("distcpTargetPaths"));
Assert.assertEquals(props.get("distcpTargetPaths"),
FeedHelper.getStagingPath(trgCluster, tableFeed, trgStorage, Tag.REPLICATION,
"${coord:formatTime(coord:nominalTime(), 'yyyy-MM-dd-HH-mm')}" + "/" + trgCluster.getName()));
Assert.assertEquals(props.get("falconFeedStorageType"), Storage.TYPE.TABLE.name());
// verify table props
assertTableStorageProperties(srcCluster, srcStorage, props, "falconSource");
assertTableStorageProperties(trgCluster, trgStorage, props, "falconTarget");
// verify the late data params
Assert.assertEquals(props.get("falconInputFeeds"), tableFeed.getName());
Assert.assertEquals(props.get("falconInPaths"), "${coord:dataIn('input')}");
Assert.assertEquals(props.get("falconInputFeedStorageTypes"), Storage.TYPE.TABLE.name());
// verify the post processing params
Assert.assertEquals(props.get("feedNames"), tableFeed.getName());
Assert.assertEquals(props.get("feedInstancePaths"), "${coord:dataOut('output')}");
Assert.assertTrue(Storage.TYPE.TABLE == FeedHelper.getStorageType(tableFeed, trgCluster));
assertReplicationHCatCredentials(getWorkflowapp(coord),
coord.getAction().getWorkflow().getAppPath().replace("${nameNode}", ""));
}