@VisibleForTesting
  CompleteMultipartUploadResponse multipartCopy(
      S3ResourceId sourcePath, S3ResourceId destinationPath, HeadObjectResponse sourceObjectHead)
      throws SdkServiceException {
    CreateMultipartUploadRequest initiateUploadRequest =
        CreateMultipartUploadRequest.builder()
            .bucket(destinationPath.getBucket())
            .key(destinationPath.getKey())
            .storageClass(config.getS3StorageClass())
            .metadata(sourceObjectHead.metadata())
            .serverSideEncryption(config.getSSEAlgorithm())
            .ssekmsKeyId(config.getSSEKMSKeyId())
            .sseCustomerKey(config.getSSECustomerKey().getKey())
            .sseCustomerAlgorithm(config.getSSECustomerKey().getAlgorithm())
            .sseCustomerKeyMD5(config.getSSECustomerKey().getMD5())
            .build();

    CreateMultipartUploadResponse createMultipartUploadResponse =
        s3Client.get().createMultipartUpload(initiateUploadRequest);
    final String uploadId = createMultipartUploadResponse.uploadId();

    List<CompletedPart> completedParts = new ArrayList<>();

    final long objectSize = sourceObjectHead.contentLength();
    CopyPartResult copyPartResult;
    CompletedPart completedPart;
    // extra validation in case a caller calls directly S3FileSystem.multipartCopy
    // without using S3FileSystem.copy in the future
    if (objectSize == 0) {
      final UploadPartCopyRequest uploadPartCopyRequest =
          UploadPartCopyRequest.builder()
              .destinationBucket(destinationPath.getBucket())
              .destinationKey(destinationPath.getKey())
              .sourceBucket(sourcePath.getBucket())
              .sourceKey(sourcePath.getKey())
              .uploadId(uploadId)
              .partNumber(1)
              .sseCustomerKey(config.getSSECustomerKey().getKey())
              .sseCustomerAlgorithm(config.getSSECustomerKey().getAlgorithm())
              .sseCustomerKeyMD5(config.getSSECustomerKey().getMD5())
              .copySourceSSECustomerKey(config.getSSECustomerKey().getKey())
              .copySourceSSECustomerAlgorithm(config.getSSECustomerKey().getAlgorithm())
              .copySourceSSECustomerKeyMD5(config.getSSECustomerKey().getMD5())
              .build();

      copyPartResult = s3Client.get().uploadPartCopy(uploadPartCopyRequest).copyPartResult();
      completedPart = CompletedPart.builder().partNumber(1).eTag(copyPartResult.eTag()).build();
      completedParts.add(completedPart);
    } else {
      long bytePosition = 0;
      // Amazon parts are 1-indexed, not zero-indexed.
      for (int partNumber = 1; bytePosition < objectSize; partNumber++) {
        final UploadPartCopyRequest uploadPartCopyRequest =
            UploadPartCopyRequest.builder()
                .destinationBucket(destinationPath.getBucket())
                .destinationKey(destinationPath.getKey())
                .sourceBucket(sourcePath.getBucket())
                .sourceKey(sourcePath.getKey())
                .uploadId(uploadId)
                .partNumber(partNumber)
                .copySourceRange(
                    String.format(
                        "bytes=%s-%s",
                        bytePosition,
                        Math.min(objectSize - 1, bytePosition + MAX_COPY_OBJECT_SIZE_BYTES - 1)))
                .sseCustomerKey(config.getSSECustomerKey().getKey())
                .sseCustomerAlgorithm(config.getSSECustomerKey().getAlgorithm())
                .sseCustomerKeyMD5(config.getSSECustomerKey().getMD5())
                .copySourceSSECustomerKey(config.getSSECustomerKey().getKey())
                .copySourceSSECustomerAlgorithm(config.getSSECustomerKey().getAlgorithm())
                .copySourceSSECustomerKeyMD5(config.getSSECustomerKey().getMD5())
                .build();

        copyPartResult = s3Client.get().uploadPartCopy(uploadPartCopyRequest).copyPartResult();
        completedPart =
            CompletedPart.builder().partNumber(partNumber).eTag(copyPartResult.eTag()).build();
        completedParts.add(completedPart);

        bytePosition += MAX_COPY_OBJECT_SIZE_BYTES;
      }
    }
    CompletedMultipartUpload completedMultipartUpload =
        CompletedMultipartUpload.builder().parts(completedParts).build();

    CompleteMultipartUploadRequest completeUploadRequest =
        CompleteMultipartUploadRequest.builder()
            .bucket(destinationPath.getBucket())
            .key(destinationPath.getKey())
            .uploadId(uploadId)
            .multipartUpload(completedMultipartUpload)
            .build();
    return s3Client.get().completeMultipartUpload(completeUploadRequest);
  }
@Test
  public void testMultipartCopy() throws IOException {
    testMultipartCopy(s3Config("s3"));
    testMultipartCopy(s3Config("other"));
    testMultipartCopy(s3ConfigWithSSECustomerKey("s3"));
    testMultipartCopy(s3ConfigWithSSECustomerKey("other"));
  }