private static void cleanup(HelixManager mgr, final String resourceName, WorkflowConfig cfg,
String workflowResource) {
HelixDataAccessor accessor = mgr.getHelixDataAccessor();
// Remove any DAG references in workflow
PropertyKey workflowKey = getConfigPropertyKey(accessor, workflowResource);
DataUpdater<ZNRecord> dagRemover = new DataUpdater<ZNRecord>() {
@Override
public ZNRecord update(ZNRecord currentData) {
JobDag jobDag = JobDag.fromJson(currentData.getSimpleField(WorkflowConfig.DAG));
for (String child : jobDag.getDirectChildren(resourceName)) {
jobDag.getChildrenToParents().get(child).remove(resourceName);
}
for (String parent : jobDag.getDirectParents(resourceName)) {
jobDag.getParentsToChildren().get(parent).remove(resourceName);
}
jobDag.getChildrenToParents().remove(resourceName);
jobDag.getParentsToChildren().remove(resourceName);
jobDag.getAllNodes().remove(resourceName);
try {
currentData.setSimpleField(WorkflowConfig.DAG, jobDag.toJson());
} catch (Exception e) {
LOG.equals("Could not update DAG for job " + resourceName);
}
return currentData;
}
};
accessor.getBaseDataAccessor().update(workflowKey.getPath(), dagRemover,
AccessOption.PERSISTENT);
// Delete resource configs.
PropertyKey cfgKey = getConfigPropertyKey(accessor, resourceName);
if (!accessor.removeProperty(cfgKey)) {
throw new RuntimeException(
String
.format(
"Error occurred while trying to clean up job %s. Failed to remove node %s from Helix. Aborting further clean up steps.",
resourceName, cfgKey));
}
// Delete property store information for this resource.
String propStoreKey = getRebalancerPropStoreKey(resourceName);
if (!mgr.getHelixPropertyStore().remove(propStoreKey, AccessOption.PERSISTENT)) {
throw new RuntimeException(
String
.format(
"Error occurred while trying to clean up job %s. Failed to remove node %s from Helix. Aborting further clean up steps.",
resourceName, propStoreKey));
}
// Finally, delete the ideal state itself.
PropertyKey isKey = getISPropertyKey(accessor, resourceName);
if (!accessor.removeProperty(isKey)) {
throw new RuntimeException(String.format(
"Error occurred while trying to clean up task %s. Failed to remove node %s from Helix.",
resourceName, isKey));
}
LOG.info(String.format("Successfully cleaned up job resource %s.", resourceName));
boolean lastInWorkflow = true;
for (String job : cfg.getJobDag().getAllNodes()) {
// check if property store information or resource configs exist for this job
if (mgr.getHelixPropertyStore().exists(getRebalancerPropStoreKey(job),
AccessOption.PERSISTENT)
|| accessor.getProperty(getConfigPropertyKey(accessor, job)) != null
|| accessor.getProperty(getISPropertyKey(accessor, job)) != null) {
lastInWorkflow = false;
}
}
// clean up workflow-level info if this was the last in workflow
if (lastInWorkflow && cfg.isTerminable()) {
// delete workflow config
PropertyKey workflowCfgKey = getConfigPropertyKey(accessor, workflowResource);
if (!accessor.removeProperty(workflowCfgKey)) {
throw new RuntimeException(
String
.format(
"Error occurred while trying to clean up workflow %s. Failed to remove node %s from Helix. Aborting further clean up steps.",