code: @Override
  	public void snapshotState(FunctionSnapshotContext context) throws Exception {
  		Preconditions.checkState(this.checkpointedState != null,
  			"The " + getClass().getSimpleName() + " has not been properly initialized.");
  
  		this.checkpointedState.clear();
  		this.checkpointedState.add(this.numElementsEmitted);
  	}
docs: ------------------------------------------------------------------------