HDDS-15067. Conditional CompleteMultipartUpload

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/OzoneBucket.java

@@ -950,6 +950,27 @@ public OmMultipartUploadCompleteInfo completeMultipartUpload(String key,
         partsMap);
   }
 
+  /**
+   * Complete Multipart upload with conditional write support.
+   * This will combine all the parts and make the key visible in ozone,
+   * but only if the specified preconditions are met.
+   *
+   * @param key key name
+   * @param uploadID multipart upload ID
+   * @param partsMap map of part numbers to ETags
+   * @param expectedDataGeneration expected data generation for conditional write
+   *        (use OzoneConsts.EXPECTED_GEN_CREATE_IF_NOT_EXISTS for If-None-Match: *)
+   * @param expectedETag expected ETag for conditional write (for If-Match)
+   * @return OmMultipartUploadCompleteInfo
+   * @throws IOException if precondition fails or other I/O error occurs
+   */
+  public OmMultipartUploadCompleteInfo completeMultipartUpload(String key,
+      String uploadID, Map<Integer, String> partsMap,
+      Long expectedDataGeneration, String expectedETag) throws IOException {
+    return proxy.completeMultipartUpload(volumeName, name, key, uploadID,
+        partsMap, expectedDataGeneration, expectedETag);
+  }
+
   /**
    * Abort multipart upload request.
    * @param keyName

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -727,6 +727,27 @@ OmMultipartUploadCompleteInfo completeMultipartUpload(String volumeName,
       String bucketName, String keyName, String uploadID,
       Map<Integer, String> partsMap) throws IOException;
 
+  /**
+   * Complete Multipart upload with conditional write support.
+   * This will combine all the parts and make the key visible in ozone,
+   * but only if the specified preconditions are met.
+   *
+   * @param volumeName volume name
+   * @param bucketName bucket name
+   * @param keyName key name
+   * @param uploadID multipart upload ID
+   * @param partsMap map of part numbers to ETags
+   * @param expectedDataGeneration expected data generation for conditional write
+   *        (use OzoneConsts.EXPECTED_GEN_CREATE_IF_NOT_EXISTS for If-None-Match: *)
+   * @param expectedETag expected ETag for conditional write (for If-Match)
+   * @return OmMultipartUploadCompleteInfo
+   * @throws IOException if precondition fails or other I/O error occurs
+   */
+  OmMultipartUploadCompleteInfo completeMultipartUpload(String volumeName,
+      String bucketName, String keyName, String uploadID,
+      Map<Integer, String> partsMap,
+      Long expectedDataGeneration, String expectedETag) throws IOException;
+
   /**
    * Abort Multipart upload request for the given key with given uploadID.
    * @param volumeName

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -2154,6 +2154,39 @@ public OmMultipartUploadCompleteInfo completeMultipartUpload(
 
   }
 
+  @Override
+  public OmMultipartUploadCompleteInfo completeMultipartUpload(
+      String volumeName, String bucketName, String keyName, String uploadID,
+      Map<Integer, String> partsMap,
+      Long expectedDataGeneration, String expectedETag) throws IOException {
+    verifyVolumeName(volumeName);
+    verifyBucketName(bucketName);
+    HddsClientUtils.checkNotNull(keyName, uploadID);
+    String ownerName = getRealUserInfo().getShortUserName();
+
+    OmKeyArgs.Builder builder = new OmKeyArgs.Builder()
+        .setVolumeName(volumeName)
+        .setBucketName(bucketName)
+        .setKeyName(keyName)
+        .setMultipartUploadID(uploadID)
+        .setOwnerName(ownerName);
+
+    if (expectedDataGeneration != null) {
+      builder.setExpectedDataGeneration(expectedDataGeneration);
+    }
+    if (expectedETag != null) {
+      builder.setExpectedETag(expectedETag);
+    }
+
+    OmKeyArgs keyArgs = builder.build();
+
+    OmMultipartUploadCompleteList omMultipartUploadCompleteList =
+        new OmMultipartUploadCompleteList(partsMap);
+
+    return ozoneManagerClient.completeMultipartUpload(keyArgs,
+        omMultipartUploadCompleteList);
+  }
+
   @Override
   public void abortMultipartUpload(String volumeName,
        String bucketName, String keyName, String uploadID) throws IOException {

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -1805,6 +1805,13 @@ public OmMultipartUploadCompleteInfo completeMultipartUpload(
           OzoneAcl.toProtobuf(a)).collect(Collectors.toList()));
     }
 
+    if (omKeyArgs.getExpectedDataGeneration() != null) {
+      keyArgs.setExpectedDataGeneration(omKeyArgs.getExpectedDataGeneration());
+    }
+    if (omKeyArgs.getExpectedETag() != null) {
+      keyArgs.setExpectedETag(omKeyArgs.getExpectedETag());
+    }
+
     multipartUploadCompleteRequest.setKeyArgs(keyArgs.build());
     multipartUploadCompleteRequest.addAllPartsList(multipartUploadList
         .getPartsList());

hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java

@@ -655,6 +655,208 @@ public void testMultipartUploadPartWithWrongMD5Header(String wrongMd5Base64, Str
     assertFalse(s3Client.doesObjectExist(bucketName, keyName));
   }
 
+  @Test
+  public void testCompleteMultipartUploadIfNoneMatch() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    s3Client.createBucket(bucketName);
+
+    // Initiate and complete multipart upload with If-None-Match: *
+    InitiateMultipartUploadRequest initRequest =
+        new InitiateMultipartUploadRequest(bucketName, keyName);
+    InitiateMultipartUploadResult initResponse = s3Client.initiateMultipartUpload(initRequest);
+    String uploadId = initResponse.getUploadId();
+
+    // Upload a part
+    String partContent = "part1data";
+    byte[] partBytes = partContent.getBytes(StandardCharsets.UTF_8);
+    UploadPartRequest uploadRequest = new UploadPartRequest()
+        .withBucketName(bucketName)
+        .withKey(keyName)
+        .withUploadId(uploadId)
+        .withPartNumber(1)
+        .withInputStream(new ByteArrayInputStream(partBytes))
+        .withPartSize(partBytes.length);
+    UploadPartResult uploadResult = s3Client.uploadPart(uploadRequest);
+
+    // Complete with If-None-Match: * (key doesn't exist, should succeed)
+    List<PartETag> partETags = Collections.singletonList(uploadResult.getPartETag());
+    CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(
+        bucketName, keyName, uploadId, partETags);
+    completeRequest.putCustomRequestHeader("If-None-Match", "*");
+
+    CompleteMultipartUploadResult result = s3Client.completeMultipartUpload(completeRequest);
+    assertNotNull(result.getETag());
+
+    // Verify object was created
+    assertTrue(s3Client.doesObjectExist(bucketName, keyName));
+  }
+
+  @Test
+  public void testCompleteMultipartUploadIfNoneMatchFail() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    s3Client.createBucket(bucketName);
+
+    // First, create an existing key
+    s3Client.putObject(bucketName, keyName,
+        new ByteArrayInputStream("existing".getBytes(StandardCharsets.UTF_8)),
+        new ObjectMetadata());
+
+    // Initiate multipart upload for same key
+    InitiateMultipartUploadRequest initRequest =
+        new InitiateMultipartUploadRequest(bucketName, keyName);
+    InitiateMultipartUploadResult initResponse = s3Client.initiateMultipartUpload(initRequest);
+    String uploadId = initResponse.getUploadId();
+
+    // Upload a part
+    String partContent = "part1data";
+    byte[] partBytes = partContent.getBytes(StandardCharsets.UTF_8);
+    UploadPartRequest uploadRequest = new UploadPartRequest()
+        .withBucketName(bucketName)
+        .withKey(keyName)
+        .withUploadId(uploadId)
+        .withPartNumber(1)
+        .withInputStream(new ByteArrayInputStream(partBytes))
+        .withPartSize(partBytes.length);
+    UploadPartResult uploadResult = s3Client.uploadPart(uploadRequest);
+
+    // Complete with If-None-Match: * (key exists, should fail)
+    List<PartETag> partETags = Collections.singletonList(uploadResult.getPartETag());
+    CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(
+        bucketName, keyName, uploadId, partETags);
+    completeRequest.putCustomRequestHeader("If-None-Match", "*");
+
+    AmazonServiceException ase = assertThrows(AmazonServiceException.class,
+        () -> s3Client.completeMultipartUpload(completeRequest));
+
+    assertEquals(ErrorType.Client, ase.getErrorType());
+    assertEquals(412, ase.getStatusCode());
+    assertEquals("PreconditionFailed", ase.getErrorCode());
+  }
+
+  @Test
+  public void testCompleteMultipartUploadIfMatch() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    s3Client.createBucket(bucketName);
+
+    // First, create an existing key and get its ETag
+    PutObjectResult existingResult = s3Client.putObject(bucketName, keyName,
+        new ByteArrayInputStream("existing".getBytes(StandardCharsets.UTF_8)),
+        new ObjectMetadata());
+    String existingETag = existingResult.getETag();
+
+    // Initiate multipart upload for same key
+    InitiateMultipartUploadRequest initRequest =
+        new InitiateMultipartUploadRequest(bucketName, keyName);
+    InitiateMultipartUploadResult initResponse = s3Client.initiateMultipartUpload(initRequest);
+    String uploadId = initResponse.getUploadId();
+
+    // Upload a part
+    String partContent = "newcontent";
+    byte[] partBytes = partContent.getBytes(StandardCharsets.UTF_8);
+    UploadPartRequest uploadRequest = new UploadPartRequest()
+        .withBucketName(bucketName)
+        .withKey(keyName)
+        .withUploadId(uploadId)
+        .withPartNumber(1)
+        .withInputStream(new ByteArrayInputStream(partBytes))
+        .withPartSize(partBytes.length);
+    UploadPartResult uploadResult = s3Client.uploadPart(uploadRequest);
+
+    // Complete with If-Match: <existingETag> (should succeed)
+    List<PartETag> partETags = Collections.singletonList(uploadResult.getPartETag());
+    CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(
+        bucketName, keyName, uploadId, partETags);
+    completeRequest.putCustomRequestHeader("If-Match", "\"" + existingETag + "\"");
+
+    CompleteMultipartUploadResult result = s3Client.completeMultipartUpload(completeRequest);
+    assertNotNull(result.getETag());
+    assertNotEquals(existingETag, stripQuotes(result.getETag()));
+  }
+
+  @Test
+  public void testCompleteMultipartUploadIfMatchFail() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    s3Client.createBucket(bucketName);
+
+    // First, create an existing key
+    s3Client.putObject(bucketName, keyName,
+        new ByteArrayInputStream("existing".getBytes(StandardCharsets.UTF_8)),
+        new ObjectMetadata());
+
+    // Initiate multipart upload for same key
+    InitiateMultipartUploadRequest initRequest =
+        new InitiateMultipartUploadRequest(bucketName, keyName);
+    InitiateMultipartUploadResult initResponse = s3Client.initiateMultipartUpload(initRequest);
+    String uploadId = initResponse.getUploadId();
+
+    // Upload a part
+    String partContent = "newcontent";
+    byte[] partBytes = partContent.getBytes(StandardCharsets.UTF_8);
+    UploadPartRequest uploadRequest = new UploadPartRequest()
+        .withBucketName(bucketName)
+        .withKey(keyName)
+        .withUploadId(uploadId)
+        .withPartNumber(1)
+        .withInputStream(new ByteArrayInputStream(partBytes))
+        .withPartSize(partBytes.length);
+    UploadPartResult uploadResult = s3Client.uploadPart(uploadRequest);
+
+    // Complete with If-Match: wrong-etag (should fail)
+    List<PartETag> partETags = Collections.singletonList(uploadResult.getPartETag());
+    CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(
+        bucketName, keyName, uploadId, partETags);
+    completeRequest.putCustomRequestHeader("If-Match", "\"wrong-etag\"");
+
+    AmazonServiceException ase = assertThrows(AmazonServiceException.class,
+        () -> s3Client.completeMultipartUpload(completeRequest));
+
+    assertEquals(ErrorType.Client, ase.getErrorType());
+    assertEquals(412, ase.getStatusCode());
+    assertEquals("PreconditionFailed", ase.getErrorCode());
+  }
+
+  @Test
+  public void testCompleteMultipartUploadIfMatchMissingKeyFail() throws Exception {
+    final String bucketName = getBucketName();
+    final String keyName = getKeyName();
+    s3Client.createBucket(bucketName);
+
+    // Initiate multipart upload (key doesn't exist)
+    InitiateMultipartUploadRequest initRequest =
+        new InitiateMultipartUploadRequest(bucketName, keyName);
+    InitiateMultipartUploadResult initResponse = s3Client.initiateMultipartUpload(initRequest);
+    String uploadId = initResponse.getUploadId();
+
+    // Upload a part
+    String partContent = "newcontent";
+    byte[] partBytes = partContent.getBytes(StandardCharsets.UTF_8);
+    UploadPartRequest uploadRequest = new UploadPartRequest()
+        .withBucketName(bucketName)
+        .withKey(keyName)
+        .withUploadId(uploadId)
+        .withPartNumber(1)
+        .withInputStream(new ByteArrayInputStream(partBytes))
+        .withPartSize(partBytes.length);
+    UploadPartResult uploadResult = s3Client.uploadPart(uploadRequest);
+
+    // Complete with If-Match on non-existent key (should fail)
+    List<PartETag> partETags = Collections.singletonList(uploadResult.getPartETag());
+    CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest(
+        bucketName, keyName, uploadId, partETags);
+    completeRequest.putCustomRequestHeader("If-Match", "\"some-etag\"");
+
+    AmazonServiceException ase = assertThrows(AmazonServiceException.class,
+        () -> s3Client.completeMultipartUpload(completeRequest));
+
+    assertEquals(ErrorType.Client, ase.getErrorType());
+    assertEquals(412, ase.getStatusCode());
+    assertEquals("PreconditionFailed", ase.getErrorCode());
+  }
+
   @Test
   public void testPutDoubleSlashPrefixObject() throws IOException {
     final String bucketName = getBucketName();