code: private Table<String, Integer, ProgramController> createFlowlets(Program program, RunId runId,
                                                                     FlowSpecification flowSpec) {
      Table<String, Integer, ProgramController> flowlets = HashBasedTable.create();
  
      try {
        for (Map.Entry<String, FlowletDefinition> entry : flowSpec.getFlowlets().entrySet()) {
          int instanceCount = entry.getValue().getInstances();
          for (int instanceId = 0; instanceId < instanceCount; instanceId++) {
            flowlets.put(entry.getKey(), instanceId,
                         startFlowlet(program, createFlowletOptions(entry.getKey(), instanceId, instanceCount, runId)));
          }
        }
      } catch (Throwable t) {
        try {
          // Need to stop all started flowlets
          Futures.successfulAsList(Iterables.transform(flowlets.values(),
            new Function<ProgramController, ListenableFuture<?>>() {
              @Override
              public ListenableFuture<?> apply(ProgramController controller) {
                return controller.stop();
              }
          })).get();
        } catch (Exception e) {
          LOG.error("Fail to stop all flowlets on failure.");
        }
        throw Throwables.propagate(t);
      }
      return flowlets;
    }
docs: Starts all flowlets in the flow program.
  @param program Program to run
  @param flowSpec The {@link FlowSpecification}.
  @return A {@link com.google.common.collect.Table} with row as flowlet id, column as instance id,
  cell as the {@link ProgramController} for the flowlet.