Add bookkeeperDeleted in topic internal stats.

Author: druidliu

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

@@ -4312,6 +4312,7 @@ public CompletableFuture<ManagedLedgerInternalStats> getManagedLedgerInternalSta
                 info.entries = li.getEntries();
                 info.size = li.getSize();
                 info.offloaded = li.hasOffloadContext() && li.getOffloadContext().getComplete();
+                info.bookkeeperDeleted = li.hasOffloadContext() && li.getOffloadContext().getBookkeeperDeleted();
                 if (includeLedgerMetadata) {
                     // lookup metadata from the hashmap which contains completed async operations
                     info.metadata = ledgerMetadataFutures.get(li.getLedgerId()).getNow(null);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -2383,6 +2383,8 @@ public CompletableFuture<PersistentTopicInternalStats> getInternalStats(boolean
                             info.entries = li.getEntries();
                             info.size = li.getSize();
                             info.offloaded = li.hasOffloadContext() && li.getOffloadContext().getComplete();
+                            info.bookkeeperDeleted =
+                                    li.hasOffloadContext() && li.getOffloadContext().getBookkeeperDeleted();
                             stats.ledgers.add(info);
                             if (includeLedgerMetadata) {
                                 futures.add(ml.getLedgerMetadata(li.getLedgerId()).handle((lMetadata, ex) -> {

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/ManagedLedgerInternalStats.java

@@ -77,6 +77,7 @@ public static class LedgerInfo {
         public boolean offloaded;
         public String metadata;
         public boolean underReplicated;
+        public boolean bookkeeperDeleted;
     }
 
     /**

tests/integration/src/test/java/org/apache/pulsar/tests/integration/offload/TestBaseOffload.java

@@ -32,6 +32,7 @@
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats;
 import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
 import org.apache.pulsar.tests.integration.suites.PulsarTieredStorageTestSuite;
 import org.awaitility.Awaitility;
@@ -304,12 +305,21 @@ protected void testPublishOffloadAndConsumeDeletionLag(String serviceUrl, String
                 "namespaces", "get-offload-deletion-lag", namespace).getStdout();
         Assert.assertTrue(output.contains("Unset for namespace"));
 
+        PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(adminUrl).build();
+
         long offloadedLedger = writeAndWaitForOffload(serviceUrl, adminUrl, topic);
         // give it up to 5 seconds to delete, it shouldn't
         // so we wait this every time
         Thread.sleep(5000);
         Assert.assertTrue(ledgerExistsInBookKeeper(offloadedLedger));
 
+        long finalOffloadedLedger1 = offloadedLedger;
+        ManagedLedgerInternalStats.LedgerInfo offloadedLedgerInfo =
+                admin.topics().getInternalStats(topic).ledgers.stream()
+                        .filter((x) -> x.ledgerId == finalOffloadedLedger1).findFirst().get();
+        Assert.assertTrue(offloadedLedgerInfo.offloaded);
+        Assert.assertFalse(offloadedLedgerInfo.bookkeeperDeleted);
+
         pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "set-offload-deletion-lag", namespace,
                 "--lag", "0m");
         output = pulsarCluster.runAdminCommandOnAnyBroker(
@@ -324,6 +334,12 @@ protected void testPublishOffloadAndConsumeDeletionLag(String serviceUrl, String
         }
         Assert.assertFalse(ledgerExistsInBookKeeper(offloadedLedger));
 
+        long finalOffloadedLedger2 = offloadedLedger;
+        offloadedLedgerInfo = admin.topics().getInternalStats(topic).ledgers.stream()
+                .filter((x) -> x.ledgerId == finalOffloadedLedger2).findFirst().get();
+        Assert.assertTrue(offloadedLedgerInfo.offloaded);
+        Assert.assertTrue(offloadedLedgerInfo.bookkeeperDeleted);
+
         pulsarCluster.runAdminCommandOnAnyBroker("namespaces", "clear-offload-deletion-lag", namespace);
 
         Thread.sleep(5); // wait 5 seconds to allow broker to see update
@@ -338,6 +354,14 @@ protected void testPublishOffloadAndConsumeDeletionLag(String serviceUrl, String
         // so we wait this every time
         Thread.sleep(5000);
         Assert.assertTrue(ledgerExistsInBookKeeper(offloadedLedger));
+
+        long finalOffloadedLedger3 = offloadedLedger;
+        offloadedLedgerInfo = admin.topics().getInternalStats(topic).ledgers.stream()
+                .filter((x) -> x.ledgerId == finalOffloadedLedger3).findFirst().get();
+        Assert.assertTrue(offloadedLedgerInfo.offloaded);
+        Assert.assertFalse(offloadedLedgerInfo.bookkeeperDeleted);
+
+        admin.close();
     }
 
     protected void testDeleteOffloadedTopic(String serviceUrl, String adminUrl,