fix review comment

Author: bharatviswa504

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/Table.java

@@ -123,6 +123,14 @@ default void addCacheEntry(CacheKey<KEY> cacheKey,
     throw new NotImplementedException("addCacheEntry is not implemented");
   }
 
+  /**
+   * Get the cache value from table cache.
+   * @param cacheKey
+   */
+  default CacheValue<VALUE> getCacheValue(CacheKey<KEY> cacheKey) {
+    throw new NotImplementedException("getCacheValue is not implemented");
+  }
+
   /**
    * Removes all the entries from the table cache which are having epoch value
    * less

hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/TypedTable.java

@@ -158,6 +158,11 @@ public void addCacheEntry(CacheKey<KEY> cacheKey,
     cache.put(cacheKey, cacheValue);
   }
 
+  @Override
+  public CacheValue<VALUE> getCacheValue(CacheKey<KEY> cacheKey) {
+    return cache.get(cacheKey);
+  }
+
   public Iterator<Map.Entry<CacheKey<KEY>, CacheValue<VALUE>>> cacheIterator() {
     return cache.iterator();
   }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java

@@ -473,9 +473,18 @@ public boolean isVolumeEmpty(String volume) throws IOException {
     try (TableIterator<String, ? extends KeyValue<String, OmBucketInfo>>
         bucketIter = bucketTable.iterator()) {
       KeyValue<String, OmBucketInfo> kv = bucketIter.seek(volumePrefix);
-      if (kv != null && kv.getKey().startsWith(volumePrefix)) {
-        return false; // we found at least one bucket with this volume prefix.
+
+      if (kv != null) {
+        // Check the entry in db is not marked for delete. This can happen
+        // while entry is marked for delete, but it is not flushed to DB.
+        CacheValue<OmBucketInfo> cacheValue =
+            bucketTable.getCacheValue(new CacheKey(kv.getKey()));
+        if (kv.getKey().startsWith(volumePrefix)
+            && cacheValue != null && cacheValue.getCacheValue() != null) {
+          return false; // we found at least one bucket with this volume prefix.
+        }
       }
+
     }
     return true;
   }
@@ -509,9 +518,19 @@ public boolean isBucketEmpty(String volume, String bucket)
     try (TableIterator<String, ? extends KeyValue<String, OmKeyInfo>> keyIter =
         keyTable.iterator()) {
       KeyValue<String, OmKeyInfo> kv = keyIter.seek(keyPrefix);
-      if (kv != null && kv.getKey().startsWith(keyPrefix)) {
-        return false; // we found at least one key with this vol/bucket prefix.
+
+      if (kv != null) {
+        // Check the entry in db is not marked for delete. This can happen
+        // while entry is marked for delete, but it is not flushed to DB.
+        CacheValue<OmKeyInfo> cacheValue =
+            keyTable.getCacheValue(new CacheKey(kv.getKey()));
+        if (kv.getKey().startsWith(keyPrefix)
+            && cacheValue != null && cacheValue.getCacheValue() != null) {
+          return false; // we found at least one key with this vol/bucket
+          // prefix.
+        }
       }
+
     }
     return true;
   }