From 1ca07c1b2c24b965b7c5d60d144355021b75a8c1 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 13 Feb 2025 17:17:51 +0800 Subject: [PATCH 1/5] [fix][broker]Consumer stuck when delete subscription __compaction failed --- .../service/persistent/PersistentTopic.java | 47 ++++-- .../pulsar/compaction/CompactedTopicImpl.java | 11 +- .../GetLastMessageIdCompactedTest.java | 151 ++++++++++++++++++ 3 files changed, 198 insertions(+), 11 deletions(-) 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 ed05e47ed38e3..39eaeb8ced05d 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 @@ -245,6 +245,7 @@ public static boolean isDedupCursorName(String name) { private static final Long COMPACTION_NEVER_RUN = -0xfebecffeL; private volatile CompletableFuture currentCompaction = CompletableFuture.completedFuture( COMPACTION_NEVER_RUN); + private volatile 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; }); } @@ -3934,6 +3959,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/compaction/GetLastMessageIdCompactedTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java index a28392fb99d2a..2cd95a3901da0 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,23 @@ 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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.bookkeeper.client.api.BookKeeper; 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,12 +45,17 @@ 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.awaitility.reflect.WhiteboxImpl; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; import org.testng.annotations.DataProvider; @@ -308,6 +320,145 @@ public void testGetLastMessageIdAfterCompactionWithCompression(boolean enabledBa admin.topics().delete(topicName, false); } + @DataProvider + public Object[][] isInjectedCursorDeleteError() { + return new Object[][] { + {false}, + {true} + }; + } + + @Test(dataProvider = "isInjectedCursorDeleteError") + public void testReadMsgsAfterUnloading(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 + 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 = + WhiteboxImpl.getInternalState(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(5000, (op, path) -> { + return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; + }); + mockZooKeeperGlobal.delay(5000, (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 = + WhiteboxImpl.getInternalState(persistentTopic, "currentCompaction"); + persistentTopic.triggerCompaction(); + CompletableFuture currentCompaction2 = + WhiteboxImpl.getInternalState(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); + } + @Test(dataProvider = "enabledBatch") public void testGetLastMessageIdAfterCompactionEndWithNullMsg(boolean enabledBatch) throws Exception { String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); From 2fe0647fa899b73280ae8aac01204a9f8a59c389 Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Thu, 13 Feb 2025 17:28:38 +0800 Subject: [PATCH 2/5] - --- .../pulsar/compaction/GetLastMessageIdCompactedTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 2cd95a3901da0..b01e6a8362d84 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 @@ -31,7 +31,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; -import org.apache.bookkeeper.client.api.BookKeeper; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.pulsar.broker.BrokerTestUtil; @@ -329,7 +328,7 @@ public Object[][] isInjectedCursorDeleteError() { } @Test(dataProvider = "isInjectedCursorDeleteError") - public void testReadMsgsAfterUnloading(boolean isInjectedCursorDeleteError) throws Exception { + public void testReadMsgsAfterDisableCompaction(boolean isInjectedCursorDeleteError) throws Exception { String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); admin.topics().createNonPartitionedTopic(topicName); admin.topicPolicies().setCompactionThreshold(topicName, 1); From 22fab21b31ba9b76665e2b322525b74c55ed2edd Mon Sep 17 00:00:00 2001 From: fengyubiao Date: Fri, 14 Feb 2025 10:44:54 +0800 Subject: [PATCH 3/5] address comments --- .../service/persistent/PersistentTopic.java | 4 +- .../persistent/CompactionConcurrencyTest.java | 151 ++++++++++++++++++ .../GetLastMessageIdCompactedTest.java | 74 --------- 3 files changed, 153 insertions(+), 76 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/CompactionConcurrencyTest.java 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 39eaeb8ced05d..6c93b2631ac2b 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 @@ -243,9 +243,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); - private volatile AtomicBoolean disablingCompaction = new AtomicBoolean(false); + final AtomicBoolean disablingCompaction = new AtomicBoolean(false); private TopicCompactionService topicCompactionService; // TODO: Create compaction strategy from topic policy when exposing strategic compaction to users. 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..7b43d3e2f40f5 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/CompactionConcurrencyTest.java @@ -0,0 +1,151 @@ +/* + * 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 { + + @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(5000, (op, path) -> { + return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; + }); + mockZooKeeperGlobal.delay(5000, (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 b01e6a8362d84..e4d721fdab1a4 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 @@ -28,9 +28,7 @@ import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -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.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.pulsar.broker.BrokerTestUtil; @@ -386,78 +384,6 @@ public void testReadMsgsAfterDisableCompaction(boolean isInjectedCursorDeleteErr admin.topics().delete(topicName, false); } - @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 = - WhiteboxImpl.getInternalState(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(5000, (op, path) -> { - return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; - }); - mockZooKeeperGlobal.delay(5000, (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 = - WhiteboxImpl.getInternalState(persistentTopic, "currentCompaction"); - persistentTopic.triggerCompaction(); - CompletableFuture currentCompaction2 = - WhiteboxImpl.getInternalState(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); - } - @Test(dataProvider = "enabledBatch") public void testGetLastMessageIdAfterCompactionEndWithNullMsg(boolean enabledBatch) throws Exception { String topicName = "persistent://public/default/" + BrokerTestUtil.newUniqueName("tp"); From babbb0e894136d8b2adb88e681f58fc54f79a440 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 9 Apr 2025 18:52:12 +0300 Subject: [PATCH 4/5] Fix checkstyle --- .../apache/pulsar/compaction/GetLastMessageIdCompactedTest.java | 1 - 1 file changed, 1 deletion(-) 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 e4d721fdab1a4..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 @@ -52,7 +52,6 @@ import org.apache.zookeeper.KeeperException; 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.DataProvider; From 13a62931bb957206d1d58bdb482567074363bf9a Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 9 Apr 2025 18:58:36 +0300 Subject: [PATCH 5/5] Fix issue with metadata store connection lost due to injected delay over 2000ms --- .../service/persistent/CompactionConcurrencyTest.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 index 7b43d3e2f40f5..7f7a32e6015e3 100644 --- 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 @@ -43,6 +43,8 @@ @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 @@ -99,10 +101,10 @@ public void testDisableCompactionConcurrently() throws Exception { String cursorPath = String.format("/managed-ledgers/%s/__compaction", TopicName.get(topicName).getPersistenceNamingEncoding()); admin.topicPolicies().removeCompactionThreshold(topicName); - mockZooKeeper.delay(5000, (op, path) -> { + mockZooKeeper.delay(DELETE_OPERATION_DELAY_MS, (op, path) -> { return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; }); - mockZooKeeperGlobal.delay(5000, (op, path) -> { + mockZooKeeperGlobal.delay(DELETE_OPERATION_DELAY_MS, (op, path) -> { return op == MockZooKeeper.Op.DELETE && cursorPath.equals(path) && times.incrementAndGet() == 1; }); AtomicReference> f1 = new AtomicReference>();