[WIP] introduce two phase deletion protocol.

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java

@@ -81,6 +81,10 @@ public class ManagedLedgerConfig {
     @Getter
     @Setter
     private boolean cacheEvictionByMarkDeletedPosition = false;
+    private boolean supportTwoPhaseDeletion = false;
+    private int archiveDataLimitSize = 500;
+    private int retryDeleteIntervalSeconds = 60;
+    private int maxDeleteCount = 5;
 
     public boolean isCreateIfMissing() {
         return createIfMissing;
@@ -683,4 +687,35 @@ public void setInactiveLedgerRollOverTime(int inactiveLedgerRollOverTimeMs, Time
         this.inactiveLedgerRollOverTimeMs = (int) unit.toMillis(inactiveLedgerRollOverTimeMs);
     }
 
+    public boolean isSupportTwoPhaseDeletion() {
+        return supportTwoPhaseDeletion;
+    }
+
+    public void setSupportTwoPhaseDeletion(boolean supportTwoPhaseDeletion) {
+        this.supportTwoPhaseDeletion = supportTwoPhaseDeletion;
+    }
+
+    public int getArchiveDataLimitSize() {
+        return archiveDataLimitSize;
+    }
+
+    public void setArchiveDataLimitSize(int archiveDataLimitSize) {
+        this.archiveDataLimitSize = archiveDataLimitSize;
+    }
+
+    public int getRetryDeleteIntervalSeconds() {
+        return retryDeleteIntervalSeconds;
+    }
+
+    public void setRetryDeleteIntervalSeconds(int retryDeleteIntervalSeconds) {
+        this.retryDeleteIntervalSeconds = retryDeleteIntervalSeconds;
+    }
+
+    public void setMaxDeleteCount(int maxDeleteCount) {
+        this.maxDeleteCount = maxDeleteCount;
+    }
+
+    public int getMaxDeleteCount() {
+        return maxDeleteCount;
+    }
 }

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedTrash.java

@@ -0,0 +1,106 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.bookkeeper.mledger.impl.ManagedTrashImpl;
+import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
+
+public interface ManagedTrash {
+
+    enum ManagedType {
+        MANAGED_LEDGER("managed-ledger"),
+        MANAGED_CURSOR("managed-cursor"),
+        SCHEMA("schema");
+        private final String name;
+
+        ManagedType(String name) {
+            this.name = name;
+        }
+
+        public String getName() {
+            return name;
+        }
+    }
+
+    enum LedgerType {
+        BOTH,
+        OFFLOAD_LEDGER,
+        LEDGER
+    }
+
+    /**
+     * ManagedTrash name.
+     *
+     * @return full topic name + type
+     */
+    String name();
+
+    /**
+     * Initialize.
+     */
+    CompletableFuture<?> initialize();
+
+    /**
+     * Append waiting to delete ledger.
+     *
+     * @param ledgerId ledgerId
+     * @param context ledgerInfo, if offload ledger, need offload context
+     * @param type LEDGER or OFFLOAD_LEDGER
+     * @throws ManagedLedgerException
+     */
+    void appendLedgerTrashData(long ledgerId, LedgerInfo context, LedgerType type) throws ManagedLedgerException;
+
+    /**
+     * Persist trash data to meta store.
+     */
+    CompletableFuture<?> asyncUpdateTrashData();
+
+    /**
+     * Trigger deletion procedure.
+     */
+    void triggerDeleteInBackground();
+
+    /**
+     * Get all archive index, it needs combine with getArchiveData.
+     */
+    CompletableFuture<List<Long>> getAllArchiveIndex();
+
+    /**
+     * Get archive data detail info.
+     *
+     * @param index archive index
+     * @return
+     */
+    CompletableFuture<Map<ManagedTrashImpl.TrashKey, LedgerInfo>> getArchiveData(long index);
+
+    /**
+     * Async close managedTrash, it will persist trash data to meta store.
+     * @return
+     */
+    CompletableFuture<?> asyncClose();
+
+    /**
+     * Async close managedTrash, it can ensure that all ledger least delete once (exclude offload_ledger).
+     * @return
+     */
+    CompletableFuture<?> asyncCloseAfterAllLedgerDeleteOnce();
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedTrashMXBean.java

@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.mledger;
+
+import org.apache.bookkeeper.common.annotation.InterfaceAudience;
+import org.apache.bookkeeper.common.annotation.InterfaceStability;
+
+/**
+ * JMX Bean interface for ManagedTrash stats.
+ */
+@InterfaceAudience.LimitedPrivate
+@InterfaceStability.Stable
+public interface ManagedTrashMXBean {
+
+    String getName();
+
+    long getCurrentNumberOfLedgersWaitingToDelete();
+
+    void increaseTotalNumberOfDeleteLedgers();
+
+    long getTotalNumberOfDeleteLedgers();
+
+    long getCurrentNumberOfLedgersWaitingToArchive();
+
+    void increaseTotalNumberOfArchiveLedgers();
+
+    long getTotalNumberOfArchiveLedgers();
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java

@@ -387,7 +387,7 @@ public void asyncOpen(final String name, final ManagedLedgerConfig config, final
                     bookkeeperFactory.get(
                             new EnsemblePlacementPolicyConfig(config.getBookKeeperEnsemblePlacementPolicyClassName(),
                                     config.getBookKeeperEnsemblePlacementPolicyProperties())),
-                    store, config, scheduledExecutor, name, mlOwnershipChecker);
+                    store, metadataStore, config, scheduledExecutor, name, mlOwnershipChecker);
             PendingInitializeManagedLedger pendingLedger = new PendingInitializeManagedLedger(newledger);
             pendingInitializeLedgers.put(name, pendingLedger);
             newledger.initialize(new ManagedLedgerInitializeLedgerCallback() {
@@ -480,7 +480,7 @@ public void asyncOpenReadOnlyCursor(String managedLedgerName, Position startPosi
                 bookkeeperFactory
                         .get(new EnsemblePlacementPolicyConfig(config.getBookKeeperEnsemblePlacementPolicyClassName(),
                                 config.getBookKeeperEnsemblePlacementPolicyProperties())),
-                store, config, scheduledExecutor, managedLedgerName);
+                store, metadataStore, config, scheduledExecutor, managedLedgerName);
 
         roManagedLedger.initializeAndCreateCursor((PositionImpl) startPosition)
                 .thenAccept(roCursor -> callback.openReadOnlyCursorComplete(roCursor, ctx))