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;
  }
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 { 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 ; }
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.