Close old compacted ledger when open new. (#13210)

(cherry picked from commit 07ef923)
apache · Dec 13, 2021 · 2055eca · 2055eca
1 parent 66b58dc
commit 2055eca
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 2 deletions.
diff --git a/...oker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java b/...oker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java
@@ -47,8 +47,12 @@ public CompactorSubscription(PersistentTopic topic, CompactedTopic compactedTopi
         Map<String, Long> properties = cursor.getProperties();
         if (properties.containsKey(Compactor.COMPACTED_TOPIC_LEDGER_PROPERTY)) {
             long compactedLedgerId = properties.get(Compactor.COMPACTED_TOPIC_LEDGER_PROPERTY);
-            compactedTopic.newCompactedLedger(cursor.getMarkDeletedPosition(),
-                                              compactedLedgerId);
+            compactedTopic.newCompactedLedger(cursor.getMarkDeletedPosition(), compactedLedgerId)
+                    .thenAccept(previousContext -> {
+                        if (previousContext != null) {
+                            compactedTopic.deleteCompactedLedger(previousContext.getLedger().getId());
+                        }
+                    });
         }
     }
 

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -1873,6 +1873,8 @@ public void testCompactorSubscriptionUpdatedOnInit() throws Exception {
 
         PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
         CompactedTopic compactedTopic = mock(CompactedTopic.class);
+        when(compactedTopic.newCompactedLedger(any(Position.class), anyLong()))
+                .thenReturn(CompletableFuture.completedFuture(null));
         new CompactorSubscription(topic, compactedTopic, Compactor.COMPACTION_SUBSCRIPTION, cursorMock);
         verify(compactedTopic, Mockito.times(1)).newCompactedLedger(position, ledgerId);
     }