}
if (!groupInfo.outputs.isEmpty()) {
groupInfo.committed = true;
Vertex v = getVertex(groupInfo.groupMembers.iterator().next());
for (String outputName : groupInfo.outputs) {
OutputCommitter committer = v.getOutputCommitters().get(outputName);
LOG.info("Committing output: " + outputName + " for group: " + groupInfo.groupName);
if (!commitOutput(outputName, committer)) {
failedWhileCommitting = true;
break;
}
}
}
}
// commit all other outputs
// we come here for successful dag completion and when outputs need to be
// committed at the end for all or none visibility
for (Vertex vertex : vertices.values()) {
if (failedWhileCommitting) {
break;
}
if (vertex.getOutputCommitters() == null) {
LOG.info("No output committers for vertex: " + vertex.getName());
continue;
}
Map<String, OutputCommitter> outputCommitters =
new HashMap<String, OutputCommitter>(vertex.getOutputCommitters());
Set<String> sharedOutputs = vertex.getSharedOutputs();
// remove shared outputs
if (sharedOutputs != null) {
Iterator<Map.Entry<String, OutputCommitter>> iter = outputCommitters
.entrySet().iterator();
while (iter.hasNext()) {
if (sharedOutputs.contains(iter.next().getKey())) {
iter.remove();
}
}
}
if (outputCommitters.isEmpty()) {
LOG.info("No exclusive output committers for vertex: " + vertex.getName());
continue;
}
for (Map.Entry<String, OutputCommitter> entry : outputCommitters.entrySet()) {
LOG.info("Committing output: " + entry.getKey() + " for vertex: "
+ vertex.getVertexId());
if (vertex.getState() != VertexState.SUCCEEDED) {
throw new TezUncheckedException("Vertex: " + vertex.getName() +
" not in SUCCEEDED state. State= " + vertex.getState());
}
if (!commitOutput(entry.getKey(), entry.getValue())) {
failedWhileCommitting = true;
break;
}
}
}
}
if (failedWhileCommitting) {
LOG.info("DAG: " + getID() + " failed while committing");
}
if (!dagSucceeded || failedWhileCommitting) {
// come here because dag failed or
// dag succeeded and all or none semantics were on and a commit failed
for (Vertex vertex : vertices.values()) {
Map<String, OutputCommitter> outputCommitters = vertex
.getOutputCommitters();
if (outputCommitters == null || outputCommitters.isEmpty()) {
LOG.info("No output committers for vertex: " + vertex.getName());
continue;
}
for (Map.Entry<String, OutputCommitter> entry : outputCommitters
.entrySet()) {
final OutputCommitter committer = entry.getValue();
if (commitAllOutputsOnSuccess // commit all outputs on success
|| vertex.getState() != VertexState.SUCCEEDED // never commit unsuccessful outputs
) {
LOG.info("Aborting output: " + entry.getKey() + " for vertex: "
+ vertex.getVertexId());
try {
getDagUGI().doAs(new PrivilegedExceptionAction<Void>() {
@Override
public Void run() throws Exception {
committer.abortOutput(VertexStatus.State.FAILED);
return null;
}
});
} catch (Exception e) {
LOG.info("Exception in aborting output: " + entry.getKey()