HDDS-12787. Make expired key move to trash configurable (#9594)

Author: xichen01

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -4915,6 +4915,15 @@
     <tag>OZONE</tag>
     <description>It specifies whether to enable lifecycle management service.</description>
   </property>
+
+  <property>
+    <name>ozone.lifecycle.service.move.to.trash.enabled</name>
+    <value>true</value>
+    <tag>OZONE</tag>
+    <description>When enabled KeyLifecycleService will move expired keys/dirs to trash if trash is available.
+      When disabled, it will delete them directly.
+    </description>
+  </property>
   <property>
     <name>ozone.lifecycle.service.delete.batch-size</name>
     <value>1000</value>

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

@@ -188,6 +188,11 @@ public final class OMConfigKeys {
       "ozone.lifecycle.service.delete.cached.directory.max-count";
   public static final long OZONE_KEY_LIFECYCLE_SERVICE_DELETE_CACHED_DIRECTORY_MAX_COUNT_DEFAULT = 1000000;
 
+  public static final String OZONE_KEY_LIFECYCLE_SERVICE_MOVE_TO_TRASH_ENABLED =
+      "ozone.lifecycle.service.move.to.trash.enabled";
+  public static final boolean
+      OZONE_KEY_LIFECYCLE_SERVICE_MOVE_TO_TRASH_ENABLED_DEFAULT = true;
+
   /**
    * OM Ratis related configurations.
    */

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

@@ -26,6 +26,8 @@
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_DELETE_CACHED_DIRECTORY_MAX_COUNT_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_ENABLED;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_ENABLED_DEFAULT;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_MOVE_TO_TRASH_ENABLED;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_KEY_LIFECYCLE_SERVICE_MOVE_TO_TRASH_ENABLED_DEFAULT;
 import static org.apache.hadoop.ozone.om.helpers.BucketLayout.OBJECT_STORE;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -109,6 +111,7 @@ public class KeyLifecycleService extends BackgroundService {
   private long cachedDirMaxCount;
   private final AtomicBoolean suspended;
   private final AtomicBoolean isServiceEnabled;
+  private final AtomicBoolean moveToTrashEnabled;
   private KeyLifecycleServiceMetrics metrics;
   // A set of bucket name that have LifecycleActionTask scheduled
   private final ConcurrentHashMap<String, LifecycleActionTask> inFlight;
@@ -137,6 +140,8 @@ public KeyLifecycleService(OzoneManager ozoneManager,
     this.metrics = KeyLifecycleServiceMetrics.create();
     this.isServiceEnabled = new AtomicBoolean(conf.getBoolean(OZONE_KEY_LIFECYCLE_SERVICE_ENABLED,
         OZONE_KEY_LIFECYCLE_SERVICE_ENABLED_DEFAULT));
+    this.moveToTrashEnabled = new AtomicBoolean(conf.getBoolean(OZONE_KEY_LIFECYCLE_SERVICE_MOVE_TO_TRASH_ENABLED,
+        OZONE_KEY_LIFECYCLE_SERVICE_MOVE_TO_TRASH_ENABLED_DEFAULT));
     this.inFlight = new ConcurrentHashMap();
     this.omMetadataManager = ozoneManager.getMetadataManager();
     int limit = (int) conf.getStorageSize(
@@ -812,7 +817,7 @@ private void onSuccess(String bucketName) {
     }
 
     private void handleAndClearFullList(OmBucketInfo bucket, LimitedExpiredObjectList keysList, boolean dir) {
-      if (bucket.getBucketLayout() != OBJECT_STORE && ozoneTrash != null) {
+      if (moveToTrashEnabled.get() && bucket.getBucketLayout() != OBJECT_STORE && ozoneTrash != null) {
         moveToTrash(bucket, keysList, dir);
       } else {
         sendDeleteKeysRequestAndClearList(bucket.getVolumeName(), bucket.getBucketName(), keysList, dir);
@@ -1082,6 +1087,11 @@ public void setOzoneTrash(OzoneTrash ozoneTrash) {
     this.ozoneTrash = ozoneTrash;
   }
 
+  @VisibleForTesting
+  public void setMoveToTrashEnabled(boolean enabled) {
+    this.moveToTrashEnabled.set(enabled);
+  }
+
   /**
    * An in-memory list with limited size to hold expired object infos, including object name and current update ID.
    */

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/service/TestKeyLifecycleService.java

@@ -214,6 +214,7 @@ void setup(@TempDir File testDir) throws Exception {
     @AfterEach
     void resume() {
       keyLifecycleService.setOzoneTrash(null);
+      keyLifecycleService.setMoveToTrashEnabled(true);
     }
 
     @AfterAll
@@ -1578,6 +1579,42 @@ void testGetLifecycleServiceStatus() throws Exception {
 
       deleteLifecyclePolicy(volumeName, bucketName);
     }
+
+    @Test
+    void testDisableMoveToTrashDeletesDirectly() throws Exception {
+      final String volumeName = getTestName();
+      final String bucketName = uniqueObjectName("bucket");
+      final String prefix = "key";
+      long initialDeletedKeyCount = getDeletedKeyCount();
+      long initialKeyCount = getKeyCount(FILE_SYSTEM_OPTIMIZED);
+      long initialRenamedKeyCount = metrics.getNumKeyRenamed().value();
+
+      // Create keys
+      createKeys(volumeName, bucketName, FILE_SYSTEM_OPTIMIZED, KEY_COUNT, 1, prefix, null);
+      Thread.sleep(SERVICE_INTERVAL);
+      GenericTestUtils.waitFor(() -> getKeyCount(FILE_SYSTEM_OPTIMIZED) - initialKeyCount == KEY_COUNT,
+          WAIT_CHECK_INTERVAL, 1000);
+
+      // Make trash available, but disable move.to.trash in KeyLifecycleService.
+      keyLifecycleService.setMoveToTrashEnabled(false);
+      final float trashInterval = 0.5f; // 30 seconds
+      conf.setFloat(FS_TRASH_INTERVAL_KEY, trashInterval);
+      FileSystem fs = SecurityUtil.doAsLoginUser(
+          (PrivilegedExceptionAction<FileSystem>)
+              () -> new TrashOzoneFileSystem(om));
+      keyLifecycleService.setOzoneTrash(new OzoneTrash(fs, conf, om));
+
+      // Expire keys
+      ZonedDateTime date = ZonedDateTime.now(ZoneOffset.UTC).plusSeconds(EXPIRE_SECONDS);
+      createLifecyclePolicy(volumeName, bucketName, FILE_SYSTEM_OPTIMIZED, "", null, date.toString(), true);
+
+      // With move.to.trash disabled, keys should be deleted directly (not renamed).
+      GenericTestUtils.waitFor(() ->
+          (getDeletedKeyCount() - initialDeletedKeyCount) == KEY_COUNT, WAIT_CHECK_INTERVAL, 10000);
+      assertEquals(initialRenamedKeyCount, metrics.getNumKeyRenamed().value());
+      assertEquals(0, getKeyCount(FILE_SYSTEM_OPTIMIZED) - initialKeyCount);
+      deleteLifecyclePolicy(volumeName, bucketName);
+    }
   }
 
   /**