code: 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();
    }
docs: Persist a single {@link WorkUnit} (flattened) to a file.