code: @Override
  	public void snapshotState(FunctionSnapshotContext context) throws Exception {
  		Preconditions.checkState(this.checkpointedState != null,
  			"The " + getClass().getSimpleName() + " has not been properly initialized.");
  
  		if (LOG.isDebugEnabled()) {
  			LOG.debug("{} checkpointing: Messages: {}, checkpoint id: {}, timestamp: {}",
  				idsForCurrentCheckpoint, context.getCheckpointId(), context.getCheckpointTimestamp());
  		}
  
  		pendingCheckpoints.addLast(new Tuple2<>(context.getCheckpointId(), idsForCurrentCheckpoint));
  		idsForCurrentCheckpoint = new HashSet<>(64);
  
  		this.checkpointedState.clear();
  		this.checkpointedState.add(SerializedCheckpointData.fromDeque(pendingCheckpoints, idSerializer));
  	}
docs: ------------------------------------------------------------------------