You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yo...@apache.org on 2021/12/23 14:41:07 UTC
[pulsar] 01/08: Close old compacted ledger when open new. (#13210)
This is an automated email from the ASF dual-hosted git repository.
yong pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3c253ea52c13763f60bf8d15b32e2eee5c6c5840
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Mon Dec 13 17:03:20 2021 +0800
Close old compacted ledger when open new. (#13210)
(cherry picked from commit 07ef9231db8b844586b9217ee2d59237eb9c54b7)
---
.../pulsar/broker/service/persistent/CompactorSubscription.java | 8 ++++++--
.../org/apache/pulsar/broker/service/PersistentTopicTest.java | 2 ++
2 files changed, 8 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java
index 76e54b4..f7279968 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/CompactorSubscription.java
@@ -47,8 +47,12 @@ public class CompactorSubscription extends PersistentSubscription {
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
index 9f21e46..27b5c87 100644
--- 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
@@ -1877,6 +1877,8 @@ public class PersistentTopicTest extends MockedBookKeeperTestCase {
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);
}