### Question:
VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
  EventHandler<VertexEvent> { @Nullable @Private @VisibleForTesting public OutputCommitter getOutputCommitter(String outputName) { if (this.outputCommitters != null) { return outputCommitters.get(outputName); } return null; } VertexImpl(TezVertexID vertexId, VertexPlan vertexPlan,
      String vertexName, Configuration conf, EventHandler eventHandler,
      TaskAttemptListener taskAttemptListener, Clock clock,
      TaskHeartbeatHandler thh, boolean commitVertexOutputs,
      AppContext appContext, VertexLocationHint vertexLocationHint,
      Map<String, VertexGroupInfo> dagVertexGroups, JavaProfilerOptions profilerOpts); @Override TezVertexID getVertexId(); @Override VertexPlan getVertexPlan(); @Override int getDistanceFromRoot(); @Override String getName(); @Override Task getTask(TezTaskID taskID); @Override Task getTask(int taskIndex); @Override int getTotalTasks(); @Override int getCompletedTasks(); @Override int getSucceededTasks(); @Override int getRunningTasks(); @Override TezCounters getAllCounters(); VertexStats getVertexStats(); static TezCounters incrTaskCounters(
      TezCounters counters, Collection<Task> tasks); static VertexStats updateVertexStats(
      VertexStats stats, Collection<Task> tasks); @Override List<String> getDiagnostics(); @Override float getProgress(); @Override ProgressBuilder getVertexProgress(); @Override VertexStatusBuilder getVertexStatus(
      Set<StatusGetOpts> statusOptions); @Override TaskLocationHint getTaskLocationHint(TezTaskID taskId); @Override Map<TezTaskID, Task> getTasks(); @Override VertexState getState(); @Override VertexTerminationCause getTerminationCause(); @Override AppContext getAppContext(); @Override VertexState restoreFromEvent(HistoryEvent historyEvent); @Override String getLogIdentifier(); @Override void scheduleTasks(List<TaskWithLocationHint> tasksToSchedule); @Override boolean setParallelism(int parallelism, VertexLocationHint vertexLocationHint,
      Map<String, EdgeManagerDescriptor> sourceEdgeManagers,
      Map<String, RootInputSpecUpdate> rootInputSpecUpdates); void setVertexLocationHint(VertexLocationHint vertexLocationHint); @Override /** * The only entry point to change the Vertex. */ void handle(VertexEvent event); @Private void constructFinalFullcounters(); @Override void setInputVertices(Map<Vertex, Edge> inVertices); @Override void setOutputVertices(Map<Vertex, Edge> outVertices); @Override void setAdditionalInputs(List<RootInputLeafOutputProto> inputs); @Nullable @Override Map<String, OutputCommitter> getOutputCommitters(); @Nullable @Private @VisibleForTesting OutputCommitter getOutputCommitter(String outputName); @Override void setAdditionalOutputs(List<RootInputLeafOutputProto> outputs); @Nullable @Override Map<String, RootInputLeafOutputDescriptor<InputDescriptor>> getAdditionalInputs(); @Nullable @Override Map<String, RootInputLeafOutputDescriptor<OutputDescriptor>> getAdditionalOutputs(); @Override int compareTo(Vertex other); @Override boolean equals(Object obj); @Override int hashCode(); @Override Map<Vertex, Edge> getInputVertices(); @Override Map<Vertex, Edge> getOutputVertices(); @Override int getInputVerticesCount(); @Override int getOutputVerticesCount(); @Override ProcessorDescriptor getProcessorDescriptor(); @Override DAG getDAG(); Resource getTaskResource(); @Override synchronized List<InputSpec> getInputSpecList(int taskIndex); @Override synchronized List<OutputSpec> getOutputSpecList(int taskIndex); @Override synchronized List<GroupInputSpec> getGroupInputSpecList(int taskIndex); @Override synchronized void addSharedOutputs(Set<String> outputs); @Override synchronized Set<String> getSharedOutputs();  }### Answer:
@Test(timeout = 5000) public void testVertexCommitterInit() { initAllVertices(VertexState.INITED); VertexImpl v2 = vertices.get("vertex2"); Assert.assertNull(v2.getOutputCommitter("output")); VertexImpl v6 = vertices.get("vertex6"); Assert.assertTrue(v6.getOutputCommitter("outputx") instanceof CountingOutputCommitter); }