You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2023/03/09 02:58:27 UTC
[pulsar] branch branch-2.10 updated: [fix] [broker] Topic close failure leaves subscription in a permanent fence state (#19692)
This is an automated email from the ASF dual-hosted git repository.
yubiao pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 751eeea172e [fix] [broker] Topic close failure leaves subscription in a permanent fence state (#19692)
751eeea172e is described below
commit 751eeea172e125a7e44baf66b3728db5c348d279
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Mar 9 10:38:21 2023 +0800
[fix] [broker] Topic close failure leaves subscription in a permanent fence state (#19692)
Motivation : After a Topic close failure or a delete failure, the fence state will be reset to get the topic back to work,but it will not reset the fence state of the subscription, which will result in the consumer never being able to connect to the broker.
Modifications: Reset the fence state of subscriptions if the operation of topic close is failed.
(cherry picked from commit cdeef00c5f6a5bd3197b4ca6de0a0505b18835d8)
---
.../service/persistent/PersistentSubscription.java | 39 +++++++++++----
.../broker/service/persistent/PersistentTopic.java | 1 +
.../service/persistent/PersistentTopicTest.java | 55 ++++++++++++++++++++++
3 files changed, 87 insertions(+), 8 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index ff1c654f1d8..3352fb78d51 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -114,6 +114,7 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
private volatile ReplicatedSubscriptionSnapshotCache replicatedSubscriptionSnapshotCache;
private final PendingAckHandle pendingAckHandle;
private volatile Map<String, String> subscriptionProperties;
+ private volatile CompletableFuture<Void> fenceFuture;
static {
REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES.put(REPLICATED_SUBSCRIPTION_PROPERTY, 1L);
@@ -777,7 +778,10 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
*/
@Override
public synchronized CompletableFuture<Void> disconnect() {
- CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();
+ if (fenceFuture != null){
+ return fenceFuture;
+ }
+ fenceFuture = new CompletableFuture<>();
// block any further consumers on this subscription
IS_FENCED_UPDATER.set(this, TRUE);
@@ -785,19 +789,38 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
(dispatcher != null ? dispatcher.close() : CompletableFuture.completedFuture(null))
.thenCompose(v -> close()).thenRun(() -> {
log.info("[{}][{}] Successfully disconnected and closed subscription", topicName, subName);
- disconnectFuture.complete(null);
+ fenceFuture.complete(null);
}).exceptionally(exception -> {
- IS_FENCED_UPDATER.set(this, FALSE);
- if (dispatcher != null) {
- dispatcher.reset();
- }
log.error("[{}][{}] Error disconnecting consumers from subscription", topicName, subName,
exception);
- disconnectFuture.completeExceptionally(exception);
+ fenceFuture.completeExceptionally(exception);
+ resumeAfterFence();
return null;
});
+ return fenceFuture;
+ }
- return disconnectFuture;
+ /**
+ * Resume subscription after topic deletion or close failure.
+ */
+ public synchronized void resumeAfterFence() {
+ // If "fenceFuture" is null, it means that "disconnect" has never been called.
+ if (fenceFuture != null) {
+ fenceFuture.whenComplete((ignore, ignoreEx) -> {
+ synchronized (PersistentSubscription.this) {
+ try {
+ if (IS_FENCED_UPDATER.compareAndSet(this, TRUE, FALSE)) {
+ if (dispatcher != null) {
+ dispatcher.reset();
+ }
+ }
+ fenceFuture = null;
+ } catch (Exception ex) {
+ log.error("[{}] Resume subscription [{}] failure", topicName, subName, ex);
+ }
+ }
+ });
+ }
}
/**
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 020b9616980..246998872b8 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
@@ -2956,6 +2956,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
}
private void unfenceTopicToResume() {
+ subscriptions.values().forEach(sub -> sub.resumeAfterFence());
isFenced = false;
isClosingOrDeleting = false;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
index 056208c6798..50c92a74430 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java
@@ -19,7 +19,10 @@
package org.apache.pulsar.broker.service.persistent;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
@@ -36,8 +39,12 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import lombok.Data;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.mledger.ManagedLedger;
+import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
@@ -337,4 +344,52 @@ public class PersistentTopicTest extends BrokerTestBase {
private String name;
private int age;
}
+
+ @Test
+ public void testDeleteTopicFail() throws Exception {
+ final String fullyTopicName = "persistent://prop/ns-abc/" + "tp_"
+ + UUID.randomUUID().toString().replaceAll("-", "");
+ // Mock topic.
+ BrokerService brokerService = spy(pulsar.getBrokerService());
+ doReturn(brokerService).when(pulsar).getBrokerService();
+
+ // Create a sub, and send one message.
+ Consumer consumer1 = pulsarClient.newConsumer(Schema.STRING).topic(fullyTopicName).subscriptionName("sub1")
+ .subscribe();
+ consumer1.close();
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(fullyTopicName).create();
+ producer.send("1");
+ producer.close();
+
+ // Make a failed delete operation.
+ AtomicBoolean makeDeletedFailed = new AtomicBoolean(true);
+ PersistentTopic persistentTopic = (PersistentTopic) brokerService.getTopic(fullyTopicName, false).get().get();
+ doAnswer(invocation -> {
+ CompletableFuture future = (CompletableFuture) invocation.getArguments()[1];
+ if (makeDeletedFailed.get()) {
+ future.completeExceptionally(new RuntimeException("mock ex for test"));
+ } else {
+ future.complete(null);
+ }
+ return null;
+ }).when(brokerService)
+ .deleteTopicAuthenticationWithRetry(any(String.class), any(CompletableFuture.class), anyInt());
+ try {
+ persistentTopic.delete().get();
+ } catch (Exception e) {
+ org.testng.Assert.assertTrue(e instanceof ExecutionException);
+ org.testng.Assert.assertTrue(e.getCause() instanceof java.lang.RuntimeException);
+ org.testng.Assert.assertEquals(e.getCause().getMessage(), "mock ex for test");
+ }
+
+ // Assert topic works after deleting failure.
+ Consumer consumer2 = pulsarClient.newConsumer(Schema.STRING).topic(fullyTopicName).subscriptionName("sub1")
+ .subscribe();
+ org.testng.Assert.assertEquals("1", consumer2.receive(2, TimeUnit.SECONDS).getValue());
+ consumer2.close();
+
+ // Make delete success.
+ makeDeletedFailed.set(false);
+ persistentTopic.delete().get();
+ }
}