HDDS-16161. Prevent failed OM requests from changing bucket usage

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMDirectoryCreateRequest.java

@@ -177,7 +177,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         List<String> missingParents = omPathInfo.getMissingParents();
         long baseObjId = ozoneManager.getObjectIdFromTxId(trxnLogIndex);
         OmBucketInfo omBucketInfo =
-            getBucketInfo(omMetadataManager, volumeName, bucketName);
+            getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
 
         dirKeyInfo = createDirectoryKeyInfoWithACL(keyName, keyArgs, baseObjId,
             omBucketInfo, omPathInfo, trxnLogIndex,
@@ -193,7 +193,10 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         OMFileRequest.addKeyTableCacheEntries(omMetadataManager, volumeName,
             bucketName, omBucketInfo.getBucketLayout(),
             dirKeyInfo, missingParentInfos, trxnLogIndex);
-
+
+        omMetadataManager.getBucketTable().addCacheEntry(
+            omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
         result = Result.SUCCESS;
         omClientResponse = new OMDirectoryCreateResponse(omResponse.build(),
             dirKeyInfo, missingParentInfos, result, getBucketLayout(),

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMDirectoryCreateRequestWithFSO.java

@@ -133,7 +133,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           omDirectoryResult == NONE) {
 
         OmBucketInfo omBucketInfo =
-            getBucketInfo(omMetadataManager, volumeName, bucketName);
+            getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
         // prepare all missing parents
         missingParentInfos = getAllMissingParentDirInfo(
                 ozoneManager, keyArgs, omBucketInfo, omPathInfo, trxnLogIndex);
@@ -157,6 +157,10 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
             volumeId, bucketId, trxnLogIndex,
             missingParentInfos, dirInfo);
 
+        // Publish only here: createDirectoryInfoWithACL above can still fail with UNAUTHORIZED.
+        omMetadataManager.getBucketTable().addCacheEntry(
+            omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
         result = OMDirectoryCreateRequest.Result.SUCCESS;
         omClientResponse =
             new OMDirectoryCreateResponseWithFSO(omResponse.build(),

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java

@@ -227,7 +227,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
 
       // do open key
       omBucketInfo =
-          getBucketInfo(omMetadataManager, volumeName, bucketName);
+          getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
       final ReplicationConfig repConfig = OzoneConfigUtil
           .resolveReplicationConfigPreference(keyArgs.getType(),
               keyArgs.getFactor(), keyArgs.getEcReplicationConfig(),
@@ -277,6 +277,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           bucketName, omBucketInfo.getBucketLayout(),
           null, missingParentInfos, trxnLogIndex);
 
+      omMetadataManager.getBucketTable().addCacheEntry(
+          omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
       // Prepare response
       omResponse.setCreateFileResponse(CreateFileResponse.newBuilder()
           .setKeyInfo(omKeyInfo.getNetworkProtobuf(getOmRequest().getVersion(),

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequestWithFSO.java

@@ -184,7 +184,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           .collect(Collectors.toList());
       omFileInfo.appendNewBlocks(newLocationList, false);
 
-      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
+      omBucketInfo = getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
       // check bucket and volume quota
       long preAllocatedSpace =
           newLocationList.size() * ozoneManager.getScmBlockSize() * repConfig
@@ -205,6 +205,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       OMFileRequest.addDirectoryTableCacheEntries(omMetadataManager, volumeId,
               bucketId, trxnLogIndex, missingParentInfos, null);
 
+      omMetadataManager.getBucketTable().addCacheEntry(
+          omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
       // Prepare response. Sets user given full key name in the 'keyName'
       // attribute in response object.
       int clientVersion = getOmRequest().getVersion();

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

@@ -42,6 +42,7 @@
 import org.apache.hadoop.ozone.audit.AuditLoggerType;
 import org.apache.hadoop.ozone.audit.OMSystemAction;
 import org.apache.hadoop.ozone.om.DeletingServiceMetrics;
+import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OMMetadataManager.VolumeBucketId;
 import org.apache.hadoop.ozone.om.OMMetrics;
 import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
@@ -147,7 +148,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
 
           omMetrics.decNumKeys();
           omMetrics.incNumKeyDeletesInternal();
-          OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
+          OmBucketInfo omBucketInfo = accumulatedBucketInfo(volBucketInfoMap, omMetadataManager,
               processed.volumeName, processed.bucketName);
           // bucketInfo can be null in case of delete volume or bucket
           // or key does not belong to bucket as bucket is recreated
@@ -184,7 +185,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           omMetrics.decNumKeys();
           omMetrics.incNumKeyDeletesInternal();
           numSubFilesMoved++;
-          OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
+          OmBucketInfo omBucketInfo = accumulatedBucketInfo(volBucketInfoMap, omMetadataManager,
               processed.volumeName, processed.bucketName);
           // bucketInfo can be null in case of delete volume or bucket
           // or key does not belong to bucket as bucket is recreated
@@ -205,7 +206,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           deletedDirNames.add(path.getDeletedDir());
           BucketNameInfo bucketNameInfo = volumeBucketIdMap.get(new VolumeBucketId(path.getVolumeId(),
               path.getBucketId()));
-          OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager,
+          OmBucketInfo omBucketInfo = accumulatedBucketInfo(volBucketInfoMap, omMetadataManager,
               bucketNameInfo.getVolumeName(), bucketNameInfo.getBucketName());
           if (omBucketInfo != null && omBucketInfo.getObjectID() == path.getBucketId()) {
             omBucketInfo.purgeSnapshotUsedNamespace(1);
@@ -222,6 +223,14 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       deletingServiceMetrics.incrNumSubFilesMoved(numSubFilesMoved);
       deletingServiceMetrics.incrNumDirPurged(numDirsDeleted);
 
+      // All bucket locks are still held here, and every purge path has been processed, so the
+      // accumulated copies can be published as a whole.
+      for (Map.Entry<Pair<String, String>, OmBucketInfo> entry : volBucketInfoMap.entrySet()) {
+        omMetadataManager.getBucketTable().addCacheEntry(
+            omMetadataManager.getBucketKey(entry.getKey().getLeft(), entry.getKey().getRight()),
+            entry.getValue(), context.getIndex());
+      }
+
       TransactionInfo transactionInfo = TransactionInfo.valueOf(context.getTermIndex());
       if (fromSnapshotInfo != null) {
         fromSnapshotInfo.setLastTransactionInfo(transactionInfo.toByteString());
@@ -269,6 +278,18 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         getBucketLayout(), volBucketInfoMap, fromSnapshotInfo, openKeyInfoMap);
   }
 
+  /**
+   * Return the copy this request accumulates deltas on, fetching it on first use. All three purge
+   * paths can touch the same bucket, so they must share one copy or earlier deltas are lost.
+   */
+  private OmBucketInfo accumulatedBucketInfo(Map<Pair<String, String>, OmBucketInfo> volBucketInfoMap,
+      OMMetadataManager omMetadataManager, String volumeName, String bucketName) {
+    OmBucketInfo omBucketInfo = volBucketInfoMap.get(Pair.of(volumeName, bucketName));
+
+    return omBucketInfo != null ? omBucketInfo
+        : getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
+  }
+
   /**
    * Helper class to hold processed key information.
    */

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

@@ -195,7 +195,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       bucketLockAcquired = getOmLockDetails().isLockAcquired();
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
-      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
+      omBucketInfo = getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
 
       // Check for directory exists with same name, if it exists throw error.
       if (LOG.isDebugEnabled()) {
@@ -407,6 +407,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
 
       omBucketInfo.incrUsedBytes(correctedSpace);
 
+      omMetadataManager.getBucketTable().addCacheEntry(
+          omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
       omResponse.setCommitKeyResponse(CommitKeyResponse.newBuilder()
           .setModificationTime(commitKeyArgs.getModificationTime())
           .build());

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

@@ -130,7 +130,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       bucketLockAcquired = getOmLockDetails().isLockAcquired();
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
-      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
+      omBucketInfo = getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
 
       String errMsg = "Cannot create file : " + keyName
               + " as parent directory doesn't exist";
@@ -351,6 +351,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
 
       omBucketInfo.incrUsedBytes(correctedSpace);
 
+      omMetadataManager.getBucketTable().addCacheEntry(
+          omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
       omResponse.setCommitKeyResponse(CommitKeyResponse.newBuilder()
           .setModificationTime(commitKeyArgs.getModificationTime())
           .build());

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

@@ -247,7 +247,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       keyArgs = validateAndRewriteIfMatchAsExpectedGeneration(keyArgs, dbKeyInfo);
 
       OmBucketInfo bucketInfo =
-          getBucketInfo(omMetadataManager, volumeName, bucketName);
+          getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
 
       // If FILE_EXISTS we just override like how we used to do for Key Create.
       if (LOG.isDebugEnabled()) {
@@ -333,6 +333,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         OMFileRequest.addKeyTableCacheEntries(omMetadataManager, volumeName,
             bucketName, bucketInfo.getBucketLayout(),
             null, missingParentInfos, trxnLogIndex);
+
+        // Parent directory creation holds the bucket write lock; key path locking leaves
+        // numMissingParents at 0.
+        omMetadataManager.getBucketTable().addCacheEntry(
+            omMetadataManager.getBucketKey(volumeName, bucketName), bucketInfo, trxnLogIndex);
       }
 
       // Add to cache entry can be done outside of lock for this openKey.

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

@@ -175,7 +175,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
               .collect(Collectors.toList());
       omFileInfo.appendNewBlocks(newLocationList, false);
 
-      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
+      omBucketInfo = getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
       // check bucket and volume quota
       long preAllocatedSpace =
           newLocationList.size() * ozoneManager.getScmBlockSize() * repConfig
@@ -201,6 +201,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
               volumeId, bucketId, trxnLogIndex,
               missingParentInfos, null);
 
+      omMetadataManager.getBucketTable().addCacheEntry(
+          omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
       // Prepare response. Sets user given full key name in the 'keyName'
       // attribute in response object.
       int clientVersion = getOmRequest().getVersion();

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

@@ -160,7 +160,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
           CacheValue.get(trxnLogIndex));
 
       OmBucketInfo omBucketInfo =
-          getBucketInfo(omMetadataManager, volumeName, bucketName);
+          getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
 
       long quotaReleased = sumBlockLengths(omKeyInfo);
       // Empty entries won't be added to deleted table so this key shouldn't get added to snapshotUsed space.
@@ -186,6 +186,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         }
       }
 
+      omMetadataManager.getBucketTable().addCacheEntry(
+          omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
       omClientResponse = new OMKeyDeleteResponse(
           omResponse.setDeleteKeyResponse(DeleteKeyResponse.newBuilder())
               .build(), omKeyInfo,

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

@@ -157,7 +157,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
                 CacheValue.get(trxnLogIndex));
       }
 
-      omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
+      omBucketInfo = getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
 
       long quotaReleased = sumBlockLengths(omKeyInfo);
       // Empty entries won't be added to deleted table so this key shouldn't get added to snapshotUsed space.
@@ -188,6 +188,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         auditMap.put(OzoneConsts.REPLICATION_CONFIG, omKeyInfo.getReplicationConfig().toString());
       }
 
+      omMetadataManager.getBucketTable().addCacheEntry(
+          omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
       omClientResponse = new OMKeyDeleteResponseWithFSO(omResponse
           .setDeleteKeyResponse(DeleteKeyResponse.newBuilder()).build(),
           keyName, omKeyInfo,

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

@@ -142,7 +142,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
         deletingServiceMetrics.setLastAOSTransactionInfo(transactionInfo);
       }
       List<OmBucketInfo> bucketInfoList = updateBucketSize(purgeKeysRequest.getBucketPurgeKeysSizeList(),
-          omMetadataManager);
+          omMetadataManager, context.getIndex());
 
       if (LOG.isDebugEnabled()) {
         Map<String, String> auditParams = new LinkedHashMap<>();
@@ -168,7 +168,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
   }
 
   private List<OmBucketInfo> updateBucketSize(List<BucketPurgeKeysSize> bucketPurgeKeysSizeList,
-      OMMetadataManager omMetadataManager) throws OMException {
+      OMMetadataManager omMetadataManager, long trxnLogIndex) throws OMException {
     Map<String, Map<String, List<BucketPurgeKeysSize>>> bucketPurgeKeysSizes = new HashMap<>();
     List<String[]> bucketKeyList = new ArrayList<>();
     for (BucketPurgeKeysSize bucketPurgeKey : bucketPurgeKeysSizeList) {
@@ -192,7 +192,7 @@ private List<OmBucketInfo> updateBucketSize(List<BucketPurgeKeysSize> bucketPurg
         String volumeName = volEntry.getKey();
         for (Map.Entry<String, List<BucketPurgeKeysSize>> bucketEntry : volEntry.getValue().entrySet()) {
           String bucketName = bucketEntry.getKey();
-          OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager, volumeName, bucketName);
+          OmBucketInfo omBucketInfo = getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
           // Check null if bucket has been deleted.
           if (omBucketInfo != null) {
             boolean bucketUpdated = false;
@@ -205,6 +205,8 @@ private List<OmBucketInfo> updateBucketSize(List<BucketPurgeKeysSize> bucketPurg
               }
             }
             if (bucketUpdated) {
+              omMetadataManager.getBucketTable().addCacheEntry(
+                  omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
               bucketInfoList.add(omBucketInfo.copyObject());
             }
           }

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

@@ -945,13 +945,11 @@ public static long sumBlockLengths(OmKeyInfo omKeyInfo) {
   }
 
   /**
-   * Return bucket info for the specified bucket.
+   * Return bucket info for the specified bucket, for read-only use.
    * <p>
    * The returned {@link OmBucketInfo} is the cached instance, returned by
-   * reference. A caller that mutates it (for example quota accounting) before a
-   * point where the request may still fail must first take a
-   * {@link OmBucketInfo#copyObject()} and publish that copy only on success,
-   * otherwise a failed request leaks the mutation into the cache.
+   * reference. Callers that mutate it must use
+   * {@link #getBucketInfoForUpdate(OMMetadataManager, String, String)} instead.
    */
   @Nullable
   public static OmBucketInfo getBucketInfo(OMMetadataManager omMetadataManager,
@@ -964,6 +962,24 @@ public static OmBucketInfo getBucketInfo(OMMetadataManager omMetadataManager,
     return value != null ? value.getCacheValue() : null;
   }
 
+  /**
+   * Return a copy of the cached bucket info for callers that mutate it.
+   * <p>
+   * Mutations stay invisible until the caller publishes the copy with
+   * {@code getBucketTable().addCacheEntry(...)}, after all fallible work and
+   * only on the path that persists the response. A caller that publishes must
+   * hold the bucket write lock for the whole read-modify-publish, not just the
+   * publish, or a concurrent writer can be lost. A caller under key path
+   * locking holds only the bucket read lock, so it must not publish.
+   */
+  @Nullable
+  public static OmBucketInfo getBucketInfoForUpdate(OMMetadataManager omMetadataManager,
+      String volume, String bucket) {
+    OmBucketInfo omBucketInfo = getBucketInfo(omMetadataManager, volume, bucket);
+
+    return omBucketInfo != null ? omBucketInfo.copyObject() : null;
+  }
+
   /**
    * Prepare OmKeyInfo which will be persisted to openKeyTable.
    * @return OmKeyInfo

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

@@ -237,7 +237,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       }
 
       OmBucketInfo omBucketInfo =
-          getBucketInfo(omMetadataManager, volumeName, bucketName);
+          getBucketInfoForUpdate(omMetadataManager, volumeName, bucketName);
 
       Map<String, OmKeyInfo> openKeyInfoMap = new HashMap<>();
       // Mark all keys which can be deleted, in cache as deleted.
@@ -260,6 +260,12 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
       }
 
       final long volumeId = omMetadataManager.getVolumeId(volumeName);
+
+      // A partial delete still persists the accepted keys and the bucket row, so publish here
+      // rather than only when every key was deleted.
+      omMetadataManager.getBucketTable().addCacheEntry(
+          omMetadataManager.getBucketKey(volumeName, bucketName), omBucketInfo, trxnLogIndex);
+
       omClientResponse =
           getOmClientResponse(ozoneManager, omKeyInfoList, dirList, omResponse,
               unDeletedKeys, keyToError, deleteStatus, omBucketInfo, volumeId, openKeyInfoMap, state);