Revert "fix(issues1140): fix S3ObjectControlManager NPE" (#1157)

Revert "fix(issues1140): fix S3ObjectControlManager NPE (#1151)"

This reverts commit d501b81.

metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java

@@ -27,8 +27,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.OptionalLong;
+import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -84,7 +86,8 @@ public class S3ObjectControlManager {
 
     private final TimelineHashSet<Long /* objectId */> preparedObjects;
 
-    private final TimelineHashSet<Long/*objectId*/> markDestroyedObjects;
+    // TODO: support different deletion policies, based on time dimension or space dimension?
+    private final Queue<Long/*objectId*/> markDestroyedObjects;
 
     private final S3Operator operator;
 
@@ -112,7 +115,7 @@ public S3ObjectControlManager(
         this.nextAssignedObjectId = new TimelineLong(snapshotRegistry);
         this.objectsMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
         this.preparedObjects = new TimelineHashSet<>(snapshotRegistry, 0);
-        this.markDestroyedObjects = new TimelineHashSet<>(snapshotRegistry, 0);
+        this.markDestroyedObjects = new LinkedBlockingDeque<>();
         this.operator = operator;
         this.lifecycleListeners = new ArrayList<>();
         this.lifecycleCheckTimer = Executors.newSingleThreadScheduledExecutor(