resource = this.getClass().getResource("/config/feed/hive-table-feed-out.xml");
Feed outFeed = (Feed) EntityType.FEED.getUnmarshaller().unmarshal(resource);
ConfigurationStore.get().publish(EntityType.FEED, outFeed);
resource = this.getClass().getResource("/config/process/pig-process-table.xml");
Process process = (Process) EntityType.PROCESS.getUnmarshaller().unmarshal(resource);
ConfigurationStore.get().publish(EntityType.PROCESS, process);
Cluster cluster = ConfigurationStore.get().get(EntityType.CLUSTER, "corp");
OozieProcessMapper mapper = new OozieProcessMapper(process);
Path bundlePath = new Path("/", EntityUtil.getStagingPath(process));
mapper.map(cluster, bundlePath);
assertTrue(fs.exists(bundlePath));
BUNDLEAPP bundle = getBundle(bundlePath);
assertEquals(EntityUtil.getWorkflowName(process).toString(), bundle.getName());
assertEquals(1, bundle.getCoordinator().size());
assertEquals(EntityUtil.getWorkflowName(Tag.DEFAULT, process).toString(),
bundle.getCoordinator().get(0).getName());
String coordPath = bundle.getCoordinator().get(0).getAppPath().replace("${nameNode}", "");
COORDINATORAPP coord = getCoordinator(new Path(coordPath));
HashMap<String, String> props = new HashMap<String, String>();
for (Property prop : coord.getAction().getWorkflow().getConfiguration().getProperty()) {
props.put(prop.getName(), prop.getValue());
}
// verify table props
Map<String, String> expected = getExpectedProperties(inFeed, outFeed, process, cluster);
for (Map.Entry<String, String> entry : props.entrySet()) {
if (expected.containsKey(entry.getKey())) {
Assert.assertEquals(entry.getValue(), expected.get(entry.getKey()));
}
}
// verify the late data params
Assert.assertEquals(props.get("falconInputFeeds"), process.getInputs().getInputs().get(0).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"), process.getOutputs().getOutputs().get(0).getName());
Assert.assertEquals(props.get("feedInstancePaths"), "${coord:dataOut('output')}");
}