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:55:08 UTC
[pulsar] branch branch-2.9 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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new 21d85fca559 [fix] [broker] Topic close failure leaves subscription in a permanent fence state (#19692)
21d85fca559 is described below
commit 21d85fca559c9088701455060d7f0613f454aa6f
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 | 61 ++++++++++++++++++++--
3 files changed, 89 insertions(+), 12 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 b93410fe472..7aac21c8994 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
@@ -112,6 +112,7 @@ public class PersistentSubscription extends AbstractSubscription implements Subs
private volatile ReplicatedSubscriptionSnapshotCache replicatedSubscriptionSnapshotCache;
private final PendingAckHandle pendingAckHandle;
+ private volatile CompletableFuture<Void> fenceFuture;
static {
REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES.put(REPLICATED_SUBSCRIPTION_PROPERTY, 1L);
@@ -769,7 +770,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);
@@ -777,19 +781,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 cc90770f423..3a83d3d1e4c 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
@@ -3059,6 +3059,7 @@ public class PersistentTopic extends AbstractTopic
}
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 18eddcb7465..22beba0d366 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,26 +19,31 @@
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;
import static org.mockito.Mockito.verify;
-import static org.testng.Assert.assertTrue;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
-
+import static org.testng.Assert.assertTrue;
+import com.google.common.collect.Sets;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
-
-import com.google.common.collect.Sets;
+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;
@@ -271,4 +276,52 @@ public class PersistentTopicTest extends BrokerTestBase {
producer.close();
}
}
+
+ @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();
+ }
}