This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 74dd9b9 Fix the reader skips compacted data which original ledger
been removed (#12522)
74dd9b9 is described below
commit 74dd9b973f019d6975497f4cbe1bd9925e6137d1
Author: lipenghui <peng...@apache.org>
AuthorDate: Tue Nov 2 12:30:30 2021 +0800
Fix the reader skips compacted data which original ledger been removed
(#12522)
* Fix the reader skips compacted data which original ledger been removed
The compactor update the compaction cursor(mark delete) first and then
update the `compactionHorizon` of the compacted topic.
During the compaction cursor move forward, the original ledger will be
removed if no other durable cursors.
At the same time, if the reader is reading data from the original ledger,
the reader will skip the data while the original ledger
been removed, details to see https://github.com/apache/pulsar/pull/6787. So
the reader might skip the compacted data since the
`compactionHorizon` have not updated yet.
The approach is:
1. Update the `compactionHorizon` before the compaction cursor move forward,
so that the reader will not skip the original data before
`compactionHorizon` updated.
If the broker crashes before the new compacted Ledger ID been persistent,
after the topic been loaded, the compaction can be trigger again and
will not loss any data,
but we will have an orphan ledger cannot be delete in the BookKeeper
cluster.
2. Remove the previous compacted Ledger after the compaction cursor move
forward, make sure the new compacted Ledger ID been persistent,
Otherwise, we might lost compacted ledger if broker crashes.
* Fix checkstyle
* Fix tests.
* Fix test
---
.../service/persistent/CompactorSubscription.java | 28 +++++++++++++---------
.../apache/pulsar/compaction/CompactedTopic.java | 3 ++-
.../pulsar/compaction/CompactedTopicImpl.java | 21 ++++++++++------
.../pulsar/broker/service/PersistentTopicTest.java | 7 +++---
.../pulsar/compaction/CompactedTopicTest.java | 14 ++++++++++-
5 files changed, 50 insertions(+), 23 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 f76dd75..76e54b4 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
@@ -21,7 +21,6 @@ package org.apache.pulsar.broker.service.persistent;
import static com.google.common.base.Preconditions.checkArgument;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
import org.apache.bookkeeper.mledger.ManagedCursor;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
@@ -34,7 +33,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CompactorSubscription extends PersistentSubscription {
- private CompactedTopic compactedTopic;
+ private final CompactedTopic compactedTopic;
public CompactorSubscription(PersistentTopic topic, CompactedTopic
compactedTopic,
String subscriptionName, ManagedCursor
cursor) {
@@ -65,15 +64,25 @@ public class CompactorSubscription extends
PersistentSubscription {
if (log.isDebugEnabled()) {
log.debug("[{}][{}] Cumulative ack on compactor subscription {}",
topicName, subName, position);
}
- CompletableFuture<Void> future = new CompletableFuture<>();
- cursor.asyncMarkDelete(position, properties, new MarkDeleteCallback() {
+
+ // The newCompactedLedger must be called at the first step because we
need to ensure the reader can read
+ // complete data from compacted Ledger, otherwise, if the original
ledger been deleted the reader cursor
+ // might move to a subsequent original ledger if `compactionHorizon`
have not updated, this will lead to
+ // the reader skips compacted data at that time, after the
`compactionHorizon` updated, the reader able
+ // to read the complete compacted data again.
+ // And we can only delete the previous ledger after the mark delete
succeed, otherwise we will loss the
+ // compacted data if mark delete failed.
+ compactedTopic.newCompactedLedger(position,
compactedLedgerId).thenAccept(previousContext -> {
+ cursor.asyncMarkDelete(position, properties, new
MarkDeleteCallback() {
@Override
public void markDeleteComplete(Object ctx) {
if (log.isDebugEnabled()) {
log.debug("[{}][{}] Mark deleted messages until
position on compactor subscription {}",
- topicName, subName, position);
+ topicName, subName, position);
+ }
+ if (previousContext != null) {
+
compactedTopic.deleteCompactedLedger(previousContext.getLedger().getId());
}
- future.complete(null);
}
@Override
@@ -81,19 +90,16 @@ public class CompactorSubscription extends
PersistentSubscription {
// TODO: cut consumer connection on markDeleteFailed
if (log.isDebugEnabled()) {
log.debug("[{}][{}] Failed to mark delete for position
on compactor subscription {}",
- topicName, subName, ctx, exception);
+ topicName, subName, ctx, exception);
}
}
}, null);
+ });
if (topic.getManagedLedger().isTerminated() &&
cursor.getNumberOfEntriesInBacklog(false) == 0) {
// Notify all consumer that the end of topic was reached
dispatcher.getConsumers().forEach(Consumer::reachedEndOfTopic);
}
-
- // Once properties have been persisted, we can notify the compacted
topic to use
- // the new ledger
- future.thenAccept((v) -> compactedTopic.newCompactedLedger(position,
compactedLedgerId));
}
private static final Logger log =
LoggerFactory.getLogger(CompactorSubscription.class);
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopic.java
b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopic.java
index 7c96937..31955a5 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopic.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopic.java
@@ -26,7 +26,8 @@ import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.broker.service.Consumer;
public interface CompactedTopic {
- CompletableFuture<?> newCompactedLedger(Position p, long
compactedLedgerId);
+ CompletableFuture<CompactedTopicContext> newCompactedLedger(Position p,
long compactedLedgerId);
+ CompletableFuture<Void> deleteCompactedLedger(long compactedLedgerId);
void asyncReadEntriesOrWait(ManagedCursor cursor,
int numberOfEntriesToRead,
boolean isFirstRead,
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
index 4bc1664..4577540 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.compaction;
import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
import com.github.benmanes.caffeine.cache.Caffeine;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ComparisonChain;
import io.netty.buffer.ByteBuf;
import java.util.ArrayList;
@@ -64,7 +65,7 @@ public class CompactedTopicImpl implements CompactedTopic {
}
@Override
- public CompletableFuture<?> newCompactedLedger(Position p, long
compactedLedgerId) {
+ public CompletableFuture<CompactedTopicContext>
newCompactedLedger(Position p, long compactedLedgerId) {
synchronized (this) {
compactionHorizon = (PositionImpl) p;
@@ -72,16 +73,17 @@ public class CompactedTopicImpl implements CompactedTopic {
compactedTopicContext = openCompactedLedger(bk, compactedLedgerId);
// delete the ledger from the old context once the new one is open
- if (previousContext != null) {
- return compactedTopicContext.thenCompose((res) ->
previousContext)
- .thenCompose((res) -> tryDeleteCompactedLedger(bk,
res.ledger.getId()));
- } else {
- return compactedTopicContext;
- }
+ return compactedTopicContext.thenCompose(__ ->
+ previousContext != null ? previousContext :
CompletableFuture.completedFuture(null));
}
}
@Override
+ public CompletableFuture<Void> deleteCompactedLedger(long
compactedLedgerId) {
+ return tryDeleteCompactedLedger(bk, compactedLedgerId);
+ }
+
+ @Override
public void asyncReadEntriesOrWait(ManagedCursor cursor,
int numberOfEntriesToRead,
boolean isFirstRead,
@@ -303,6 +305,11 @@ public class CompactedTopicImpl implements CompactedTopic {
.compare(p.getLedgerId(), m.getLedgerId())
.compare(p.getEntryId(), m.getEntryId()).result();
}
+
+ @VisibleForTesting
+ PositionImpl getCompactionHorizon() {
+ return this.compactionHorizon;
+ }
private static final Logger log =
LoggerFactory.getLogger(CompactedTopicImpl.class);
}
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 8603404..67faebe 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
@@ -21,6 +21,7 @@ package org.apache.pulsar.broker.service;
import static
org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.createMockBookKeeper;
import static
org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.createMockZooKeeper;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
@@ -123,12 +124,10 @@ import
org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.apache.pulsar.compaction.CompactedTopic;
+import org.apache.pulsar.compaction.CompactedTopicContext;
import org.apache.pulsar.compaction.Compactor;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.impl.ZKMetadataStore;
-import org.apache.pulsar.zookeeper.ZooKeeperCache;
-import org.apache.pulsar.zookeeper.ZooKeeperDataCache;
-import org.apache.pulsar.broker.admin.AdminResource;
import org.apache.zookeeper.ZooKeeper;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
@@ -1834,6 +1833,8 @@ public class PersistentTopicTest extends
MockedBookKeeperTestCase {
public void testCompactorSubscription() throws Exception {
PersistentTopic topic = new PersistentTopic(successTopicName,
ledgerMock, brokerService);
CompactedTopic compactedTopic = mock(CompactedTopic.class);
+ when(compactedTopic.newCompactedLedger(any(Position.class), anyLong()))
+
.thenReturn(CompletableFuture.completedFuture(mock(CompactedTopicContext.class)));
PersistentSubscription sub = new CompactorSubscription(topic,
compactedTopic,
Compactor.COMPACTION_SUBSCRIPTION,
cursorMock);
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
index 2dd6f8a..608d99b 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
@@ -242,7 +242,19 @@ public class CompactedTopicTest extends
MockedPulsarServiceBaseTest {
Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD).close();
// update the compacted topic ledger
- compactedTopic.newCompactedLedger(new PositionImpl(1,2),
newCompactedLedger.getId()).get();
+ PositionImpl newHorizon = new PositionImpl(1,3);
+ compactedTopic.newCompactedLedger(newHorizon,
newCompactedLedger.getId()).get();
+
+ // Make sure the old compacted ledger still exist after the new
compacted ledger created.
+ bk.openLedger(oldCompactedLedger.getId(),
+ Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE,
+ Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD).close();
+
+
Assert.assertTrue(compactedTopic.getCompactedTopicContext().isPresent());
+
Assert.assertEquals(compactedTopic.getCompactedTopicContext().get().getLedger().getId(),
+ newCompactedLedger.getId());
+ Assert.assertEquals(compactedTopic.getCompactionHorizon(), newHorizon);
+
compactedTopic.deleteCompactedLedger(oldCompactedLedger.getId()).join();
// old ledger should be deleted, new still there
try {
