docs: @Override
  	public void snapshotState(StateSnapshotContext context) throws Exception {
  		super.snapshotState(context);
  
  		checkState(checkpointedState != null,
  			"The operator state has not been properly initialized.");
  
  		int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
  
  		checkpointedState.clear();
  
  		List<TimestampedFileInputSplit> readerState = reader.getReaderState();
  
  		try {
  			for (TimestampedFileInputSplit split : readerState) {
  				// create a new partition for each entry.
  				checkpointedState.add(split);
  			}
  		} catch (Exception e) {
  			checkpointedState.clear();
  
  			throw new Exception("Could not add timestamped file input splits to to operator " +
  				"state backend of operator " + getOperatorName() + '.', e);
  		}
  
  		if (LOG.isDebugEnabled()) {
  			LOG.debug("{} (taskIdx={}) checkpointed {} splits: {}.",
  				getClass().getSimpleName(), subtaskIdx, readerState.size(), readerState);
  		}
  	}