private String persistWorkUnit(final Path workUnitFileDir, final WorkUnit workUnit, ParallelRunner stateSerDeRunner)
      throws IOException {
    final StateStore stateStore;
    String workUnitFileName = workUnit.getId();

    if (workUnit instanceof MultiWorkUnit) {
      workUnitFileName += MULTI_WORK_UNIT_FILE_EXTENSION;
      stateStore = stateStores.getMwuStateStore();
    } else {
      workUnitFileName += WORK_UNIT_FILE_EXTENSION;
      stateStore = stateStores.getWuStateStore();
    }

    Path workUnitFile = new Path(workUnitFileDir, workUnitFileName);
    final String fileName = workUnitFile.getName();
    final String storeName = workUnitFile.getParent().getName();
    stateSerDeRunner.submitCallable(new Callable<Void>() {
      @Override
      public Void call() throws Exception {
        stateStore.put(storeName, fileName, workUnit);
        return null;
      }
    }, "Serialize state to store " + storeName + " file " + fileName);

    return workUnitFile.toString();
  }
private String persistWorkUnit ( final Path workUnitFileDir , final WorkUnit workUnit , ParallelRunner stateSerDeRunner ) throws IOException { final StateStore stateStore ; String workUnitFileName = workUnit . getId ( ) ; if ( workUnit instanceof MultiWorkUnit ) { workUnitFileName += MULTI_WORK_UNIT_FILE_EXTENSION ; stateStore = stateStores . getMwuStateStore ( ) ; } else { workUnitFileName += WORK_UNIT_FILE_EXTENSION ; stateStore = stateStores . getWuStateStore ( ) ; } Path workUnitFile = new Path ( workUnitFileDir , workUnitFileName ) ; final String fileName = workUnitFile . getName ( ) ; final String storeName = workUnitFile . getParent ( ) . getName ( ) ; stateSerDeRunner . submitCallable ( new Callable < Void > ( ) { @ Override public Void call ( ) throws Exception { stateStore . put ( storeName , fileName , workUnit ) ; return null ; } } , "Serialize state to store " + storeName + " file " + fileName ) ; return workUnitFile . toString ( ) ; }
Persist a single {@link WorkUnit} (flattened) to a file.