整体来说多个处理流程是解偶的，这样可以在面对多数据源情况下，能更加的灵活。
下面只展示了，主要的一些流程

下面补充一点，读取文件状态的保存，切分信息用了一个 ListState 来保存。具体要保存的信息，由数据源自定义。保存操作默认由SourceOperator 来实现

   /** The state that holds the currently assigned splits. */
    private ListState<SplitT> readerState;


   @Override
    public void snapshotState(StateSnapshotContext context) throws Exception {
        long checkpointId = context.getCheckpointId();
        LOG.debug("Taking a snapshot for checkpoint {}", checkpointId);
        readerState.update(sourceReader.snapshotState(checkpointId));
    }