HDDS-14417. Skip wrapping allocated but unused blocks for empty file as pseudo file and save in deletedTable

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

@@ -1155,7 +1155,7 @@ protected OzoneLockStrategy getOzoneLockStrategy(OzoneManager ozoneManager) {
    */
   protected OmKeyInfo wrapUncommittedBlocksAsPseudoKey(
       List<OmKeyLocationInfo> uncommitted, OmKeyInfo omKeyInfo) {
-    if (uncommitted.isEmpty()) {
+    if (uncommitted.isEmpty() || omKeyInfo.getDataSize() == 0) {
       return null;
     }
     LOG.debug("Detect allocated but uncommitted blocks {} in key {}.",

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

@@ -814,6 +814,41 @@ public void testValidateAndUpdateCacheOnOverwriteWithUncommittedBlocks() throws
     assertThat(key).doesNotEndWith(String.valueOf(omKeyInfoList.get(0).getObjectID()));
   }
 
+  @Test
+  public void testPreAllocatedUncommittedBlockForEmptyFile() throws Exception {
+    dataSize = 0;
+    List<KeyLocation> emptyKeyLocations = new ArrayList<>();
+    List<KeyLocation> oneKeyLocations = getKeyLocation(1);
+    OMRequest omRequest = createCommitKeyRequest(emptyKeyLocations, false);
+    OMRequest modifiedOmRequest = doPreExecute(omRequest);
+    OMKeyCommitRequest omKeyCommitRequest = getOmKeyCommitRequest(modifiedOmRequest);
+
+    OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName,
+        omMetadataManager, omKeyCommitRequest.getBucketLayout());
+
+    List<OmKeyLocationInfo> allocatedLocationList =
+        oneKeyLocations.stream().map(OmKeyLocationInfo::getFromProtobuf).collect(Collectors.toList());
+    String openKey = addKeyToOpenKeyTable(allocatedLocationList);
+    OmKeyInfo omKeyInfo = omMetadataManager.getOpenKeyTable(omKeyCommitRequest.getBucketLayout()).get(openKey);
+    assertNotNull(omKeyInfo);
+
+    String ozoneKey = getOzonePathKey();
+    omKeyInfo = omMetadataManager.getKeyTable(omKeyCommitRequest.getBucketLayout()).get(ozoneKey);
+    assertNull(omKeyInfo);
+
+    OMClientResponse omClientResponse = omKeyCommitRequest.validateAndUpdateCache(ozoneManager, 100L);
+    assertEquals(OK, omClientResponse.getOMResponse().getStatus());
+
+    Map<String, RepeatedOmKeyInfo> toDeleteKeyList = ((OMKeyCommitResponse) omClientResponse).getKeysToDelete();
+    assertNull(toDeleteKeyList);
+
+    omKeyInfo = omMetadataManager.getOpenKeyTable(omKeyCommitRequest.getBucketLayout()).get(openKey);
+    assertNull(omKeyInfo);
+
+    omKeyInfo = omMetadataManager.getKeyTable(omKeyCommitRequest.getBucketLayout()).get(ozoneKey);
+    assertNotNull(omKeyInfo);
+  }
+
   /**
    * This method calls preExecute and verify the modified request.
    * @param originalOMRequest

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/s3/multipart/TestS3MultipartUploadCommitPartRequest.java

@@ -556,6 +556,65 @@ public void testValidateAndUpdateCacheOnOverWriteWithUncommittedBlocks() throws
     assertEquals(2, toDeleteKeyMap.size());
   }
 
+  @Test
+  public void testValidateAndUpdateCacheWithUncommittedBlockForEmptyPart() throws Exception {
+    String volumeName = UUID.randomUUID().toString();
+    String bucketName = UUID.randomUUID().toString();
+    String keyName = getKeyName();
+
+    OMRequestTestUtils.addVolumeAndBucketToDB(volumeName, bucketName,
+        omMetadataManager, getBucketLayout());
+
+    createParentPath(volumeName, bucketName);
+
+    // Create key to be overwritten
+    OMRequest initiateMPURequest = doPreExecuteInitiateMPU(volumeName,
+        bucketName, keyName);
+
+    S3InitiateMultipartUploadRequest s3InitiateMultipartUploadRequest =
+        getS3InitiateMultipartUploadReq(initiateMPURequest);
+
+    OMClientResponse omClientResponse =
+        s3InitiateMultipartUploadRequest.validateAndUpdateCache(ozoneManager, 1L);
+
+    long clientID = Time.now();
+    String multipartUploadID = omClientResponse.getOMResponse()
+        .getInitiateMultiPartUploadResponse().getMultipartUploadID();
+
+    List<KeyLocation> emptyKeyLocationInfos = new ArrayList<>();
+    List<KeyLocation> originalKeyLocationList = getKeyLocation(1);
+    List<OmKeyLocationInfo> originalKeyLocationInfos = originalKeyLocationList
+        .stream().map(OmKeyLocationInfo::getFromProtobuf)
+        .collect(Collectors.toList());
+
+    OMRequest commitMultipartRequest = doPreExecuteCommitMPU(volumeName,
+        bucketName, keyName, clientID, multipartUploadID, 1, emptyKeyLocationInfos);
+
+    S3MultipartUploadCommitPartRequest s3MultipartUploadCommitPartRequest =
+        getS3MultipartUploadCommitReq(commitMultipartRequest);
+
+    addKeyToOpenKeyTable(volumeName, bucketName, keyName, clientID, originalKeyLocationInfos);
+
+    omClientResponse =
+        s3MultipartUploadCommitPartRequest.validateAndUpdateCache(ozoneManager, 2L);
+
+    assertSame(OzoneManagerProtocolProtos.Status.OK,
+        omClientResponse.getOMResponse().getStatus());
+
+    String multipartKey = omMetadataManager.getMultipartKey(volumeName,
+        bucketName, keyName, multipartUploadID);
+
+    OmMultipartKeyInfo multipartKeyInfo = omMetadataManager.getMultipartInfoTable().get(multipartKey);
+    assertNotNull(multipartKeyInfo);
+    assertEquals(1, multipartKeyInfo.getPartKeyInfoMap().size());
+    PartKeyInfo partKeyInfo = multipartKeyInfo.getPartKeyInfo(1);
+    assertNotNull(partKeyInfo);
+
+    Map<String, RepeatedOmKeyInfo> toDeleteKeyMap =
+        ((S3MultipartUploadCommitPartResponse) omClientResponse).getKeyToDelete();
+    assertNull(toDeleteKeyMap);
+  }
+
   protected void addKeyToOpenKeyTable(String volumeName, String bucketName,
       String keyName, long clientID) throws Exception {
     OMRequestTestUtils.addKeyToTable(true, true, volumeName, bucketName,