Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[fix][broker] Consumer stuck when delete subscription __compaction failed #23980

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[fix][broker] Consumer stuck when delete subscription __compaction failed #23980

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1ca07c1
[fix][broker]Consumer stuck when delete subscription __compaction failed
poorbarcode Feb 13, 2025
2fe0647
-
poorbarcode Feb 13, 2025
22fab21
address comments
poorbarcode Feb 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 38 additions & 9 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,7 @@ public static boolean isDedupCursorName(String name) {
private static final Long COMPACTION_NEVER_RUN = -0xfebecffeL;
private volatile CompletableFuture<Long> currentCompaction = CompletableFuture.completedFuture(
COMPACTION_NEVER_RUN);
private volatile AtomicBoolean disablingCompaction = new AtomicBoolean(false);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
private volatile AtomicBoolean disablingCompaction = new AtomicBoolean(false);
private final AtomicBoolean disablingCompaction = new AtomicBoolean(false);

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point, I have changed the modifier

private TopicCompactionService topicCompactionService;

// TODO: Create compaction strategy from topic policy when exposing strategic compaction to users.
Expand Down Expand Up @@ -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) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there really a need for a synchronized block here?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it is. It prevents the modifying of the variable currentCompaction.

if (!disablingCompaction.compareAndSet(false, true)) {
unsubscribeFuture.completeExceptionally(
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this should be protected with if (!unsubscribeFuture.isDone()) { so that creating the exception could be avoided.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you mention the code line? I could not find it.

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> 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;
});
}

Expand Down Expand Up @@ -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()
Expand Down
11 changes: 9 additions & 2 deletions pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,8 +83,15 @@ public CompletableFuture<CompactedTopicContext> 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);
});
}
}

Expand Down
150 changes: 150 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/compaction/GetLastMessageIdCompactedTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,29 +21,40 @@
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.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;
import org.apache.pulsar.client.api.MessageId;
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;
Expand Down Expand Up @@ -308,6 +319,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 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<String> reader = pulsarClient.newReader(Schema.STRING).topic(topicName).readCompacted(true)
.startMessageId(MessageId.earliest).create();
producer.newMessage().key("k3").value("v3").send();
assertTrue(reader.hasMessageAvailable());
Message<String> m0 = reader.readNext(10, TimeUnit.SECONDS);
assertEquals(m0.getValue(), "v0");
assertTrue(reader.hasMessageAvailable());
Message<String> m1 = reader.readNext(10, TimeUnit.SECONDS);
assertEquals(m1.getValue(), "v1");
assertTrue(reader.hasMessageAvailable());
Message<String> m2 = reader.readNext(10, TimeUnit.SECONDS);
assertEquals(m2.getValue(), "v2");
assertTrue(reader.hasMessageAvailable());
Message<String> 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");
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please avoid using reflection/WhiteboxImpl. A better approach is to have a default access method which is annotation with @VisibleForTesting.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed


// 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<CompletableFuture<Void>> f1 = new AtomicReference<CompletableFuture<Void>>();
AtomicReference<CompletableFuture<Void>> f2 = new AtomicReference<CompletableFuture<Void>>();
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<Long> currentCompaction1 =
WhiteboxImpl.getInternalState(persistentTopic, "currentCompaction");
persistentTopic.triggerCompaction();
CompletableFuture<Long> currentCompaction2 =
WhiteboxImpl.getInternalState(persistentTopic, "currentCompaction");
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Again, avoid reflection/WhiteboxImpl.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed

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");
Expand Down
Loading