code: @Override
  	public S3Recoverable snapshotAndGetRecoverable(@Nullable final RefCountedFSOutputStream incompletePartFile) throws IOException {
  
  		final String incompletePartObjectName = safelyUploadSmallPart(incompletePartFile);
  
  		// make sure all other uploads are complete
  		// this currently makes the method blocking,
  		// to be made non-blocking in the future
  		awaitPendingPartsUpload();
  
  		final String objectName = currentUploadInfo.getObjectName();
  		final String uploadId = currentUploadInfo.getUploadId();
  		final List<PartETag> completedParts = currentUploadInfo.getCopyOfEtagsOfCompleteParts();
  		final long sizeInBytes = currentUploadInfo.getExpectedSizeInBytes();
  
  		if (incompletePartObjectName == null) {
  			return new S3Recoverable(objectName, uploadId, completedParts, sizeInBytes);
  		} else {
  			return new S3Recoverable(objectName, uploadId, completedParts, sizeInBytes, incompletePartObjectName, incompletePartFile.getPos());
  		}
  	}
docs: Creates a snapshot of this MultiPartUpload, from which the upload can be resumed.
  
  <p>Data buffered locally which is less than
  {@link org.apache.flink.fs.s3.common.FlinkS3FileSystem#S3_MULTIPART_MIN_PART_SIZE S3_MULTIPART_MIN_PART_SIZE},
  and cannot be uploaded as part of the MPU and set to S3 as independent objects.
  
  <p>This implementation currently blocks until all part uploads are complete and returns
  a completed future.