HDDS-14968. Concurrent S3 Conditional PUT Commit Conflict Detection (#10043)

Author: peterxcli

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/exceptions/OMException.java

@@ -279,5 +279,7 @@ public enum ResultCodes {
     ETAG_MISMATCH,
 
     ETAG_NOT_AVAILABLE,
+
+    ATOMIC_WRITE_CONFLICT,
   }
 }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/OzoneRpcClientTests.java

@@ -42,6 +42,8 @@
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_URI_DELIMITER;
 import static org.apache.hadoop.ozone.client.OzoneClientTestUtils.assertKeyContent;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL;
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.ATOMIC_WRITE_CONFLICT;
+import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.ETAG_MISMATCH;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.KEY_ALREADY_EXISTS;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.KEY_NOT_FOUND;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.NO_SUCH_MULTIPART_UPLOAD_ERROR;
@@ -1424,7 +1426,7 @@ void rewriteFailsDueToOutdatedGenerationAtCommit(BucketLayout layout) throws IOE
       keyInfo = ozoneManager.lookupKey(keyArgs);
 
       OMException e = assertThrows(OMException.class, out::close);
-      assertEquals(KEY_NOT_FOUND, e.getResult());
+      assertEquals(ATOMIC_WRITE_CONFLICT, e.getResult());
       assertThat(e).hasMessageContaining("does not match the expected generation to rewrite");
     } finally {
       if (out != null) {
@@ -1561,7 +1563,7 @@ void testRewriteKeyIfMatchFailsWithWrongETag(BucketLayout layout) throws IOExcep
         out.write(newContent);
       }
     });
-    assertEquals(OMException.ResultCodes.ETAG_MISMATCH, e.getResult());
+    assertEquals(ETAG_MISMATCH, e.getResult());
   }
 
   @ParameterizedTest

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -572,6 +572,8 @@ enum Status {
     ETAG_MISMATCH = 99;
 
     ETAG_NOT_AVAILABLE = 100;
+
+    ATOMIC_WRITE_CONFLICT = 101;
 }
 
 /**

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java

@@ -628,17 +628,18 @@ protected void validateAtomicRewrite(OmKeyInfo existing, OmKeyInfo toCommit, Map
 
       if (expectedGen == OzoneConsts.EXPECTED_GEN_CREATE_IF_NOT_EXISTS) {
         if (existing != null) {
-          throw new OMException("Key already exists",
-              OMException.ResultCodes.KEY_ALREADY_EXISTS);
+          throw new OMException("Atomic create-if-not-exists conflicted with an existing key",
+              OMException.ResultCodes.ATOMIC_WRITE_CONFLICT);
         }
       } else {
         if (existing == null) {
-          throw new OMException("Atomic rewrite is not allowed for a new key", KEY_NOT_FOUND);
+          throw new OMException("Atomic rewrite conflicted because the key no longer exists",
+              OMException.ResultCodes.ATOMIC_WRITE_CONFLICT);
         }
         if (expectedGen != existing.getUpdateID()) {
           throw new OMException("Cannot commit as current generation (" + existing.getUpdateID() +
               ") does not match the expected generation to rewrite (" + expectedGen + ")",
-              KEY_NOT_FOUND);
+              OMException.ResultCodes.ATOMIC_WRITE_CONFLICT);
         }
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java

@@ -1043,6 +1043,9 @@ protected OmKeyInfo createFileInfo(
             .setUpdateID(transactionLogIndex)
             .setOwnerName(keyArgs.getOwnerName())
             .setFile(true);
+    if (keyArgs.hasExpectedDataGeneration()) {
+      builder.setExpectedDataGeneration(keyArgs.getExpectedDataGeneration());
+    }
     if (omPathInfo instanceof OMFileRequest.OMPathInfoWithFSO) {
       // FileTable metadata format
       OMFileRequest.OMPathInfoWithFSO omPathInfoFSO

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyCommitRequest.java

@@ -17,7 +17,7 @@
 
 package org.apache.hadoop.ozone.om.request.key;
 
-import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status.KEY_ALREADY_EXISTS;
+import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status.ATOMIC_WRITE_CONFLICT;
 import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status.KEY_NOT_FOUND;
 import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status.OK;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -252,7 +252,7 @@ public void testAtomicRewrite() throws Exception {
     // However there is no closed key entry, so the commit should fail.
     OMClientResponse omClientResponse =
         omKeyCommitRequest.validateAndUpdateCache(ozoneManager, 100L);
-    assertEquals(KEY_NOT_FOUND, omClientResponse.getOMResponse().getStatus());
+    assertEquals(ATOMIC_WRITE_CONFLICT, omClientResponse.getOMResponse().getStatus());
 
     // Now add the key to the key table, and try again, but with different generation
     omKeyInfoBuilder.setExpectedDataGeneration(null);
@@ -261,7 +261,7 @@ public void testAtomicRewrite() throws Exception {
     closedKeyTable.put(getOzonePathKey(), invalidKeyInfo);
     // This should fail as the updateID ia zero and the open key has rewrite generation of 1.
     omClientResponse = omKeyCommitRequest.validateAndUpdateCache(ozoneManager, 100L);
-    assertEquals(KEY_NOT_FOUND, omClientResponse.getOMResponse().getStatus());
+    assertEquals(ATOMIC_WRITE_CONFLICT, omClientResponse.getOMResponse().getStatus());
 
     omKeyInfoBuilder.setUpdateID(1L);
     OmKeyInfo closedKeyInfo = omKeyInfoBuilder.build();
@@ -347,7 +347,7 @@ public void testAtomicCreateIfNotExistsCommitKeyAlreadyExists() throws Exception
 
     OMClientResponse omClientResponse =
         omKeyCommitRequest.validateAndUpdateCache(ozoneManager, 100L);
-    assertEquals(KEY_ALREADY_EXISTS, omClientResponse.getOMResponse().getStatus());
+    assertEquals(ATOMIC_WRITE_CONFLICT, omClientResponse.getOMResponse().getStatus());
   }
 
   @Test

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMKeyCreateRequest.java

@@ -166,6 +166,13 @@ public void testCreateKeyExpectedGenCreateIfNotExistsKeyMissing(
         omKeyCreateRequest.validateAndUpdateCache(ozoneManager, 100L);
 
     checkResponse(modifiedOmRequest, response, id, false, getBucketLayout());
+
+    OmKeyInfo openKeyInfo = omMetadataManager.getOpenKeyTable(getBucketLayout())
+        .get(getOpenKey(id));
+    assertNotNull(openKeyInfo);
+    assertEquals(OzoneConsts.EXPECTED_GEN_CREATE_IF_NOT_EXISTS,
+        openKeyInfo.getExpectedDataGeneration());
+    assertNull(openKeyInfo.getExpectedETag());
   }
 
   @ParameterizedTest

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/ObjectEndpoint.java

@@ -189,6 +189,8 @@ public Response put(
         throw newError(S3ErrorTable.NO_OVERWRITE, keyPath, ex);
       } else if (ex.getResult() == ResultCodes.KEY_ALREADY_EXISTS) {
         throw newError(PRECOND_FAILED, keyPath, ex);
+      } else if (ex.getResult() == ResultCodes.ATOMIC_WRITE_CONFLICT) {
+        throw newError(S3ErrorTable.CONDITIONAL_REQUEST_CONFLICT, keyPath, ex);
       } else if (ex.getResult() == ResultCodes.ETAG_MISMATCH) {
         throw newError(PRECOND_FAILED, keyPath, ex);
       } else if (ex.getResult() == ResultCodes.ETAG_NOT_AVAILABLE) {

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/exception/S3ErrorTable.java

@@ -117,6 +117,11 @@ public final class S3ErrorTable {
   public static final OS3Exception PRECOND_FAILED = new OS3Exception(
       "PreconditionFailed", "At least one of the pre-conditions you " +
       "specified did not hold", HTTP_PRECON_FAILED);
+
+  public static final OS3Exception CONDITIONAL_REQUEST_CONFLICT =
+      new OS3Exception("ConditionalRequestConflict",
+          "A conflicting conditional operation occurred. Retry the request.",
+          HTTP_CONFLICT);
 
   public static final OS3Exception NOT_IMPLEMENTED = new OS3Exception(
       "NotImplemented", "This part of feature is not implemented yet.",