From 6bb5304556474757cf64c791feb0fc1cd632318b Mon Sep 17 00:00:00 2001 From: penghui Date: Wed, 20 Oct 2021 08:31:12 +0800 Subject: [PATCH] Fix compactor skips data from last compacted Ledger The compaction task depends on the last snapshot and the incremental entries to build the new snapshot. So for the compaction cursor, we need to force seek the read position to ensure the compactor can read the complete last snapshot because of the compactor will read the data before the compaction cursor mark delete position. --- .../bookkeeper/mledger/ManagedCursor.java | 6 ++- .../mledger/impl/ManagedCursorImpl.java | 6 +-- .../impl/ManagedCursorContainerTest.java | 2 +- .../pulsar/compaction/CompactedTopicImpl.java | 9 +++- .../pulsar/compaction/CompactedTopicTest.java | 45 +++++++++++++++++++ 5 files changed, 61 insertions(+), 7 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java index 4af64550fe53b..72ee1a1f9c3ea 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java @@ -425,7 +425,11 @@ void markDelete(Position position, Map properties) * @param newReadPosition * the position where to move the cursor */ - void seek(Position newReadPosition); + default void seek(Position newReadPosition) { + seek(newReadPosition, false); + } + + void seek(Position newReadPosition, boolean force); /** * Clear the cursor backlog. diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java index 235736da760dc..dd715ff7588d3 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java @@ -2163,18 +2163,16 @@ public void rewind() { } @Override - public void seek(Position newReadPositionInt) { + public void seek(Position newReadPositionInt, boolean force) { checkArgument(newReadPositionInt instanceof PositionImpl); PositionImpl newReadPosition = (PositionImpl) newReadPositionInt; lock.writeLock().lock(); try { - if (newReadPosition.compareTo(markDeletePosition) <= 0) { + if (!force && newReadPosition.compareTo(markDeletePosition) <= 0) { // Make sure the newReadPosition comes after the mark delete position newReadPosition = ledger.getNextValidPosition(markDeletePosition); } - - PositionImpl oldReadPosition = readPosition; readPosition = newReadPosition; } finally { lock.writeLock().unlock(); diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java index 6b9c0094a127e..57e1964a2c332 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java @@ -175,7 +175,7 @@ public void rewind() { } @Override - public void seek(Position newReadPosition) { + public void seek(Position newReadPosition, boolean force) { } @Override 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 131341318e0f9..4375188d0eed5 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 @@ -18,6 +18,7 @@ */ package org.apache.pulsar.compaction; +import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION; import com.github.benmanes.caffeine.cache.AsyncLoadingCache; import com.github.benmanes.caffeine.cache.Caffeine; import com.google.common.collect.ComparisonChain; @@ -122,7 +123,13 @@ public void asyncReadEntriesOrWait(ManagedCursor cursor, return readEntries(context.ledger, startPoint, endPoint) .thenAccept((entries) -> { Entry lastEntry = entries.get(entries.size() - 1); - cursor.seek(lastEntry.getPosition().getNext()); + // The compaction task depends on the last snapshot and the incremental + // entries to build the new snapshot. So for the compaction cursor, we + // need to force seek the read position to ensure the compactor can read + // the complete last snapshot because of the compactor will read the data + // before the compaction cursor mark delete position + cursor.seek(lastEntry.getPosition().getNext(), + cursor.getName().equals(COMPACTION_SUBSCRIPTION)); callback.readEntriesComplete(entries, consumer); }); } 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 3d410884e3042..adb6f469ca864 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 @@ -18,6 +18,7 @@ */ package org.apache.pulsar.compaction; +import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION; import com.github.benmanes.caffeine.cache.AsyncLoadingCache; import com.google.common.collect.Sets; @@ -437,4 +438,48 @@ public void testLastMessageIdForCompactedLedger() throws Exception { reader.readNext(); Assert.assertFalse(reader.hasMessageAvailable()); } + + @Test + public void testDoNotLossTheLastCompactedLedgerData() throws Exception { + String topic = "persistent://my-property/use/my-ns/testDoNotLossTheLastCompactedLedgerData-" + + UUID.randomUUID(); + final int numMessages = 2000; + final int keys = 200; + final String msg = "Test"; + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .blockIfQueueFull(true) + .maxPendingMessages(numMessages) + .enableBatching(false) + .create(); + CompletableFuture lastMessage = null; + for (int i = 0; i < numMessages; ++i) { + lastMessage = producer.newMessage().key(i % keys + "").value(msg).sendAsync(); + } + producer.flush(); + lastMessage.join(); + admin.topics().triggerCompaction(topic); + Awaitility.await().untilAsserted(() -> { + PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); + Assert.assertNotEquals(stats.compactedLedger.ledgerId, -1); + Assert.assertEquals(stats.compactedLedger.entries, keys); + Assert.assertEquals(admin.topics().getStats(topic) + .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); + }); + admin.topics().unload(topic); + Awaitility.await().untilAsserted(() -> { + PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); + Assert.assertEquals(stats.ledgers.size(), 1); + Assert.assertEquals(admin.topics().getStats(topic) + .getSubscriptions().get(COMPACTION_SUBSCRIPTION).getConsumers().size(), 0); + }); + admin.topics().unload(topic); + // Send one more key to and then to trigger the compaction + producer.newMessage().key(keys + "").value(msg).send(); + admin.topics().triggerCompaction(topic); + Awaitility.await().untilAsserted(() -> { + PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic); + Assert.assertEquals(stats.compactedLedger.entries, keys + 1); + }); + } }