diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 66361499ee321..9c42e75cb175b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -242,8 +242,9 @@ public static boolean isDedupCursorName(String name) { protected final MessageDeduplication messageDeduplication; private static final Long COMPACTION_NEVER_RUN = -0xfebecffeL; - private volatile CompletableFuture currentCompaction = CompletableFuture.completedFuture( + volatile CompletableFuture currentCompaction = CompletableFuture.completedFuture( COMPACTION_NEVER_RUN); + final AtomicBoolean disablingCompaction = new AtomicBoolean(false); private TopicCompactionService topicCompactionService; // TODO: Create compaction strategy from topic policy when exposing strategic compaction to users. @@ -1340,18 +1341,42 @@ private void asyncDeleteCursorWithCleanCompactionLedger(PersistentSubscription s return; } - currentCompaction.handle((__, e) -> { - if (e != null) { - log.warn("[{}][{}] Last compaction task failed", topic, subscriptionName); + // Avoid concurrently execute compaction and unsubscribing. + synchronized (this) { + if (!disablingCompaction.compareAndSet(false, true)) { + unsubscribeFuture.completeExceptionally( + new SubscriptionBusyException("the subscription is deleting by another task")); + return; } - return ((PulsarCompactorSubscription) subscription).cleanCompactedLedger(); - }).whenComplete((__, ex) -> { - if (ex != null) { - log.error("[{}][{}] Error cleaning compacted ledger", topic, subscriptionName, ex); - unsubscribeFuture.completeExceptionally(ex); + } + // Unsubscribe compaction cursor and delete compacted ledger. + currentCompaction.thenCompose(__ -> { + asyncDeleteCursor(subscriptionName, unsubscribeFuture); + return unsubscribeFuture; + }).thenAccept(__ -> { + try { + ((PulsarCompactorSubscription) subscription).cleanCompactedLedger(); + } catch (Exception ex) { + Long compactedLedger = null; + Optional compactedTopicContext = getCompactedTopicContext(); + if (compactedTopicContext.isPresent() && compactedTopicContext.get().getLedger() != null) { + compactedLedger = compactedTopicContext.get().getLedger().getId(); + } + log.error("[{}][{}][{}] Error cleaning compacted ledger", topic, subscriptionName, compactedLedger, ex); + } finally { + // Reset the variable: disablingCompaction, + disablingCompaction.compareAndSet(true, false); + } + }).exceptionally(ex -> { + if (currentCompaction.isCompletedExceptionally()) { + log.warn("[{}][{}] Last compaction task failed", topic, subscriptionName); } else { - asyncDeleteCursor(subscriptionName, unsubscribeFuture); + log.warn("[{}][{}] Failed to delete cursor task failed", topic, subscriptionName); } + // Reset the variable: disablingCompaction, + disablingCompaction.compareAndSet(true, false); + unsubscribeFuture.completeExceptionally(ex); + return null; }); } @@ -3941,6 +3966,10 @@ public synchronized void triggerCompaction() log.info("[{}] Topic is closing or deleting, skip triggering compaction", topic); return; } + if (disablingCompaction.get()) { + log.info("[{}] Compaction is disabling, skip triggering compaction", topic); + return; + } if (strategicCompactionMap.containsKey(topic)) { currentCompaction = brokerService.pulsar().getStrategicCompactor() 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 141320b54e7a4..5105889a25972 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 @@ -83,8 +83,15 @@ public CompletableFuture newCompactedLedger(Position p, l compactionHorizon = p; // delete the ledger from the old context once the new one is open - return compactedTopicContext.thenCompose( - __ -> previousContext != null ? previousContext : CompletableFuture.completedFuture(null)); + return compactedTopicContext.thenCompose(ctx -> { + if (ctx != null && ctx.getLedger() != null && ctx.getLedger().getId() == compactedLedgerId) { + // Print an error log here, which is not expected. + log.error("[__compaction] Using the same compacted ledger to override the old one, which is not" + + " expected and it may cause a ledger lost error. {} -> {}", compactedLedgerId, + ctx.getLedger().getId()); + } + return previousContext != null ? previousContext : CompletableFuture.completedFuture(null); + }); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/CompactionConcurrencyTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/CompactionConcurrencyTest.java new file mode 100644 index 0000000000000..7f7a32e6015e3 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/CompactionConcurrencyTest.java @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service.persistent; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.bookkeeper.mledger.Position; +import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.util.FutureUtil; +import org.apache.pulsar.compaction.Compactor; +import org.apache.zookeeper.MockZooKeeper; +import org.awaitility.Awaitility; +import org.awaitility.reflect.WhiteboxImpl; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +@Test(groups = "broker") +public class CompactionConcurrencyTest extends ProducerConsumerBase { + // don't make this over 2000ms, otherwise the test will be flaky due to ZKSessionWatcher + static final int DELETE_OPERATION_DELAY_MS = 1900; + + @BeforeClass + @Override + protected void setup() throws Exception { + super.internalSetup(); + super.producerBaseSetup(); + } + + @AfterClass + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + @Override + protected void doInitConf() throws Exception { + super.doInitConf(); + // Disable the scheduled task: compaction. + conf.setBrokerServiceCompactionMonitorIntervalInSeconds(Integer.MAX_VALUE); + // Disable the scheduled task: retention. + conf.setRetentionCheckIntervalInSeconds(Integer.MAX_VALUE); + } + + private void triggerCompactionAndWait(String topicName) throws Exception { + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + persistentTopic.triggerCompaction(); + Awaitility.await().untilAsserted(() -> { + Position lastConfirmPos = persistentTopic.getManagedLedger().getLastConfirmedEntry(); + Position markDeletePos = persistentTopic + .getSubscription(Compactor.COMPACTION_SUBSCRIPTION).getCursor().getMarkDeletedPosition(); + assertEquals(markDeletePos.getLedgerId(), lastConfirmPos.getLedgerId()); + assertEquals(markDeletePos.getEntryId(), lastConfirmPos.getEntryId()); + }); + } + + @Test + public void testDisableCompactionConcurrently() throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + admin.topics().createNonPartitionedTopic(topicName); + admin.topicPolicies().setCompactionThreshold(topicName, 1); + admin.topics().createSubscription(topicName, "s1", MessageId.earliest); + var producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).enableBatching(false).create(); + producer.newMessage().key("k0").value("v0").send(); + triggerCompactionAndWait(topicName); + admin.topics().deleteSubscription(topicName, "s1"); + PersistentTopic persistentTopic = + (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).get().get(); + AtomicBoolean disablingCompaction = persistentTopic.disablingCompaction; + + // Disable compaction. + // Inject a delay when the first time of deleting cursor. + AtomicInteger times = new AtomicInteger(); + String cursorPath = String.format("/managed-ledgers/%s/__compaction", + TopicName.get(topicName).getPersistenceNamingEncoding()); + admin.topicPolicies().removeCompactionThreshold(topicName); + mockZooKeeper.delay(DELETE_OPERATION_DELAY_MS, (op, path) -> { + return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; + }); + mockZooKeeperGlobal.delay(DELETE_OPERATION_DELAY_MS, (op, path) -> { + return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; + }); + AtomicReference> f1 = new AtomicReference>(); + AtomicReference> f2 = new AtomicReference>(); + new Thread(() -> { + f1.set(admin.topics().deleteSubscriptionAsync(topicName, "__compaction")); + }).start(); + new Thread(() -> { + f2.set(admin.topics().deleteSubscriptionAsync(topicName, "__compaction")); + }).start(); + + // Verify: the next compaction will be skipped. + Awaitility.await().untilAsserted(() -> { + assertTrue(disablingCompaction.get()); + }); + producer.newMessage().key("k1").value("v1").send(); + producer.newMessage().key("k2").value("v2").send(); + CompletableFuture currentCompaction1 = persistentTopic.currentCompaction; + WhiteboxImpl.getInternalState(persistentTopic, "currentCompaction"); + persistentTopic.triggerCompaction(); + CompletableFuture currentCompaction2 = persistentTopic.currentCompaction; + assertTrue(currentCompaction1 == currentCompaction2); + + // Verify: one of the requests should fail. + Awaitility.await().untilAsserted(() -> { + assertTrue(f1.get() != null); + assertTrue(f2.get() != null); + assertTrue(f1.get().isDone()); + assertTrue(f2.get().isDone()); + assertTrue(f1.get().isCompletedExceptionally() || f2.get().isCompletedExceptionally()); + assertTrue(!f1.get().isCompletedExceptionally() || !f2.get().isCompletedExceptionally()); + }); + try { + f1.get().join(); + f2.get().join(); + fail("Should fail"); + } catch (Exception ex) { + Throwable actEx = FutureUtil.unwrapCompletionException(ex); + assertTrue(actEx instanceof PulsarAdminException.PreconditionFailedException); + } + + // cleanup. + producer.close(); + admin.topics().delete(topicName, false); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java index a28392fb99d2a..252ee939aace3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java @@ -21,16 +21,20 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotEquals; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.CompressionType; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; @@ -38,11 +42,15 @@ import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerBuilder; import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.BatchMessageIdImpl; import org.apache.pulsar.client.impl.MessageIdImpl; import org.apache.pulsar.client.impl.ReaderImpl; +import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.util.FutureUtil; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.MockZooKeeper; import org.awaitility.Awaitility; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -308,6 +316,73 @@ public void testGetLastMessageIdAfterCompactionWithCompression(boolean enabledBa admin.topics().delete(topicName, false); } + @DataProvider + public Object[][] isInjectedCursorDeleteError() { + return new Object[][] { + {false}, + {true} + }; + } + + @Test(dataProvider = "isInjectedCursorDeleteError") + public void testReadMsgsAfterDisableCompaction(boolean isInjectedCursorDeleteError) throws Exception { + String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); + admin.topics().createNonPartitionedTopic(topicName); + admin.topicPolicies().setCompactionThreshold(topicName, 1); + admin.topics().createSubscription(topicName, "s1", MessageId.earliest); + var producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).enableBatching(false).create(); + producer.newMessage().key("k0").value("v0").send(); + producer.newMessage().key("k1").value("v1").send(); + producer.newMessage().key("k2").value("v2").send(); + triggerCompactionAndWait(topicName); + admin.topics().deleteSubscription(topicName, "s1"); + + // Disable compaction. + // Inject a failure that the first time to delete cursor will fail. + if (isInjectedCursorDeleteError) { + AtomicInteger times = new AtomicInteger(); + String cursorPath = String.format("/managed-ledgers/%s/__compaction", + TopicName.get(topicName).getPersistenceNamingEncoding()); + admin.topicPolicies().removeCompactionThreshold(topicName); + mockZooKeeper.failConditional(KeeperException.Code.SESSIONEXPIRED, (op, path) -> { + return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; + }); + mockZooKeeperGlobal.failConditional(KeeperException.Code.SESSIONEXPIRED, (op, path) -> { + return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; + }); + try { + admin.topics().deleteSubscription(topicName, "__compaction"); + fail("Should fail"); + } catch (Exception ex) { + assertTrue(ex instanceof PulsarAdminException.ServerSideErrorException); + } + } + + // Create a reader with start at earliest. + // Verify: the reader will receive 3 messages. + admin.topics().unload(topicName); + Reader reader = pulsarClient.newReader(Schema.STRING).topic(topicName).readCompacted(true) + .startMessageId(MessageId.earliest).create(); + producer.newMessage().key("k3").value("v3").send(); + assertTrue(reader.hasMessageAvailable()); + Message m0 = reader.readNext(10, TimeUnit.SECONDS); + assertEquals(m0.getValue(), "v0"); + assertTrue(reader.hasMessageAvailable()); + Message m1 = reader.readNext(10, TimeUnit.SECONDS); + assertEquals(m1.getValue(), "v1"); + assertTrue(reader.hasMessageAvailable()); + Message m2 = reader.readNext(10, TimeUnit.SECONDS); + assertEquals(m2.getValue(), "v2"); + assertTrue(reader.hasMessageAvailable()); + Message m3 = reader.readNext(10, TimeUnit.SECONDS); + assertEquals(m3.getValue(), "v3"); + + // cleanup. + producer.close(); + reader.close(); + admin.topics().delete(topicName, false); + } + @Test(dataProvider = "enabledBatch") public void testGetLastMessageIdAfterCompactionEndWithNullMsg(boolean enabledBatch) throws Exception { String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp");