code: @Override
  	public void serialize(SavepointV2 checkpointMetadata, DataOutputStream dos) throws IOException {
  		// first: checkpoint ID
  		dos.writeLong(checkpointMetadata.getCheckpointId());
  
  		// second: master state
  		final Collection<MasterState> masterStates = checkpointMetadata.getMasterStates();
  		dos.writeInt(masterStates.size());
  		for (MasterState ms : masterStates) {
  			serializeMasterState(ms, dos);
  		}
  
  		// third: operator states
  		Collection<OperatorState> operatorStates = checkpointMetadata.getOperatorStates();
  		dos.writeInt(operatorStates.size());
  
  		for (OperatorState operatorState : operatorStates) {
  			// Operator ID
  			dos.writeLong(operatorState.getOperatorID().getLowerPart());
  			dos.writeLong(operatorState.getOperatorID().getUpperPart());
  
  			// Parallelism
  			int parallelism = operatorState.getParallelism();
  			dos.writeInt(parallelism);
  			dos.writeInt(operatorState.getMaxParallelism());
  			dos.writeInt(1);
  
  			// Sub task states
  			Map<Integer, OperatorSubtaskState> subtaskStateMap = operatorState.getSubtaskStates();
  			dos.writeInt(subtaskStateMap.size());
  			for (Map.Entry<Integer, OperatorSubtaskState> entry : subtaskStateMap.entrySet()) {
  				dos.writeInt(entry.getKey());
  				serializeSubtaskState(entry.getValue(), dos);
  			}
  		}
  	}
docs: ------------------------------------------------------------------------