[fix] [broker] The broker has two identical Persitenttopics #16247

poorbarcode · 2022-06-27T16:25:43Z

Motivation

Problem occur

With the transaction feature, we send and receive messages, and at the same time, execute admin API: unload namespace 1000 times. Then the problem occur: Consumer could not receive any message, and there has no error log. After that we tried admin API: get topic stats, and the response showed only producers are registered on topic, and no consumers are registered on topic, but consumer stat is Ready in the client. This means that the state of the consumer is inconsistent between the broker and the client.

Location problem

Then we found the problem: Two PersistentTopic which have the same name registered at a broker node, consumer registered on one (aka topic-c), and producer registered on another one (aka topic-p). At this time, when we send messages, the data flow like this :

client: producer sends a message

broker: handle cmd-send

broker: find the topic by name, it is "topic-p"

broker: find all subscriptions registered on "topic-p"

broker: found one subscription, but it has no consumers registered

broker: no need to send the message to the client

But the consumer exactly registered on another topic: topic-c, so consumer could not receive any message.

Repreduce

How to reproduce two topics registered at the same broker node ?

Make transaction buffer recover, admin unload namespace, client create consumer, client create producer executed at the same time, the process flow like this (at the step-11, the problem occurs ):

Time	`transaction buffer recoverr`	`admin unload namespace`	`client create consumer`	`client create producer`
1	TB recover
2	TB recover failure	topic.unload
3	topic.close(false)	topic.close(true)
4	brokerService.topics.remove(topicName)
5	remove topic finish		lookup
6			create `topic-c`
7			consumer registered on `topic-c`
8		brokerService.topics.remove(topic)
9		remove `topic-c` finish		lookup
10				create `topic-p`
11				producer registered on `topic-p`

Each column means the individual process. e.g. client create consumer, client create producer.
Multiple processes are going on at the same time, and all effet the brokerService.topics.
Column Time is used only to indicate the order of each step, not the actual time.
The important steps are explained below:

step 3

Even if persistent topic propertyisClosingOrDeleting have already changed to true, it still can be executed another once, see line-1247::

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

Lines 1240 to 1249 in f230d15

    
           public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) { 
        
               CompletableFuture<Void> closeFuture = new CompletableFuture<>(); 
        
               lock.writeLock().lock(); 
        
               try { 
        
                   // closing managed-ledger waits until all producers/consumers/replicators get closed. Sometimes, broker 
        
                   // forcefully wants to close managed-ledger without waiting all resources to be closed. 
        
                   if (!isClosingOrDeleting || closeWithoutWaitingClientDisconnect) { 
        
                       fenceTopicToCloseOrDelete(); 
        
                   } else {

Whether close can be executed depends on two predicates: is closing or @param closeWithoutWaitingClientDisconnect is true. This means that method topic.close can be reentrant executed when @param closeWithoutWaitingClientDisconnect is true, and in the implementation of admin API: unload namespace the parameter closeWithoutWaitingClientDisconnect is exactly true.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

Lines 723 to 725 in f230d15

    
           public CompletableFuture<Void> unloadNamespaceBundle(NamespaceBundle bundle, long timeout, TimeUnit timeoutUnit) { 
        
               return unloadNamespaceBundle(bundle, timeout, timeoutUnit, true); 
        
           }

So when transaction buffer recover fail and admin unload namespace is executed at the same time, and transaction buffer recover fail before admin unload namespace, the topic will be removed from brokerService.topics twice.

step-4 / step-8

Because of the current implementation of BrokerService. removeTopicFromCache use cmd map.remove(key), not use map.remove(key, value), So this cmd can remove any value in the map, even if it's not the desired one.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Line 1956 in f230d15

topics.remove(topic);

To sum up: We should make these two changes:

Make method topic.close non-reentrant. Also prevent reentrant between topic.close and topic.delete.
Use cmd map.remove(key, value) instead of map.remove(key) in implementation of BrokerService. removeTopicFromCache. This change will apply to both scenes: topic.close and topic.delete.

Other Modifications

In the current implementation, if closing the ledger fails, it determines that the closing topic failed. Then will reset the topic stat to no-fenced. But it changes two states [isFenced, isClosingOrDeleting] without locking, this could not guarantee consistency between them. I will fix it in this PR too (this change may not be relevant to current subject).

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

Lines 1307 to 1312 in f230d15

    
           }).exceptionally(exception -> { 
        
               log.error("[{}] Error closing topic", topic, exception); 
        
               unfenceTopicToResume(); 
        
               closeFuture.completeExceptionally(exception); 
        
               return null; 
        
           });

PR Relations

#16240

When repeating creates the same topic at one broker node, the following phenomena occur: Transaction pending ack store reuses the cached managed cursor object when initializing constructor, the process flow like this:

Time	`client create consumer 1`	`client create consumer 2`
1	lookup	lookup
1	create `topic'`
2	create `subscription'`	create `topic''`
2	open new `pending_ackledger`	create `subscription''`
3	open new `pending_ack_cursor`	reuse the cached `pending_ackledger`
4		reuse the cached `pending_ack_cursor`

If the Transaction pending ack store reuses the managed cursor in the cache, it will cause the task replay pending ack log to loop forever. This PR solves the problem of “Repeat creates topic”, and also eliminates the possibility of reuse pending-ack-cursor: #16240.

Documentation

Check the box below or label this PR directly.

Need to update docs?

doc-required
(Your PR needs to update docs, and you will update later)
doc-not-needed
(Please explain why)
doc
(Your PR contains doc changes)
doc-complete
(Docs have been already added)

poorbarcode · 2022-06-27T16:26:21Z

@gaoran10 @congbobo184 @Technoboy- Please take a look. ^_^

poorbarcode · 2022-06-27T18:19:04Z

/pulsarbot run-failure-checks

poorbarcode · 2022-06-28T02:44:50Z

/pulsarbot run-failure-checks

poorbarcode · 2022-06-28T03:11:20Z

/pulsarbot run-failure-checks

codelipenghui

We also should consider to avoid a topic can be closed more than once.

codelipenghui · 2022-06-28T08:41:36Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

+        final CompletableFuture<Optional<Topic>> createTopicFuture = getTopic(topicNameString, false);
+        return createTopicFuture.thenCompose(optionalTopic -> {
+            if (optionalTopic.isPresent() && optionalTopic.get() == topic){
+                return removeTopicFromCache(topicNameString, createTopicFuture);
+            }
+            // If topic is not in Cache, do nothing.
+            return CompletableFuture.completedFuture(null);
+        });


We can use map.compute() to simplify the logic? And looks like we don't need to wait for the future complete, because we already have the topic reference here.

We can use map.compute() to simplify the logic? And looks like we don't need to wait for the future complete

Yes， I've rewritten the logic.

We also should consider to avoid a topic can be closed more than once.

I have appended the reason why topic.close was executed twice to the Motivation, and in this PR I've overwritten the topic.close to fix it. I also added a lock to the 'reset topic stat to UN-fenced' operation, could you review the code.

poorbarcode · 2022-06-29T15:09:07Z

Hi @codelipenghui

We also should consider to avoid a topic can be closed more than once.

I have appended the reason why topic.close was executed twice to the Motivation, and in this PR I've overwritten the topic.close to fix it. I also added a lock to the 'reset topic stat to UN-fenced' operation, could you review the code.

We can use map.compute() to simplify the logic?

Unfortunately, we can't use map.compute tto simplify the logic.

And looks like we don't need to wait for the future complete, because we already have the topic reference here.
Yes, I have fixed it. The current implementation doesn't need to wait for the future to complete.

I also rewritten the Motivation of this PR to make it easier to understand. Thanks.

poorbarcode · 2022-06-29T15:11:24Z

@eolivelli @lhotari @michaeljmarshall @gaozhangmin @Jason918 @nicoloboschi Please take a look, if you have time. Thanks.

poorbarcode · 2022-06-30T03:24:55Z

/pulsarbot run-failure-checks

poorbarcode · 2022-08-17T15:53:37Z

/pulsarbot rerun-failure-checks

eolivelli · 2022-08-17T15:59:44Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

+            return CompletableFuture.completedFuture(null);
+        }
+        // We don't need to wait for the future complete, because we already have the topic reference here.
+        if (!createTopicFuture.isDone()){


I think that we should wait for this future to complete, otherwise it may create another Topic

we can save us from using getNow and we can chain the CompletableFuture with "thenCompose"

I think that we should wait for this future to complete, otherwise it may create another Topic

One comment suggestion here: "don't need to wait for the future complete"
#16247 (comment)

we can save us from using getNow and we can chain the CompletableFuture with "thenCompose"

Already use "thenCompose" instead "getNow". Thanks

eolivelli · 2022-08-17T15:59:55Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

+        return removeTopicFromCache(topic, (CompletableFuture) null);
+    }
+
+    public CompletableFuture<Void> removeTopicFromCache(String topic, CompletableFuture createTopicFuture) {


please add generic type to CompletableFuture

Already add generic type to CompletableFuture. Thanks

eolivelli · 2022-08-17T16:01:11Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+                    return this.fullyCloseFuture;
+                } else {
+                    // Why not return this.fullyCloseFuture ?
+                    // I don't know, just keep the same implementation as before.


please remove this kind of comments "I don't know"

Already remove.Thanks

eolivelli · 2022-08-17T16:02:00Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+        // Close limiters.
+        try {
+            closeLimiters();
+        } catch (Throwable t){


why are we catching "Throwable" ? this is usually a bad practice

Already instead "Throwable" to "Exception". Thanks

eolivelli · 2022-08-17T16:02:40Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+            closePhase2Future = closeClientsFuture.thenCompose(__ -> asyncCloseLedger());
+        }
+        // Complete resultFuture. If managed ledger close failure, reset topic to resume.
+        closePhase2Future.thenApply(__ -> {


nit: we can use "whenComplete" instead of theApply/exceptionally

Already use "whenComplete" instead of "theApply". Thanks

eolivelli · 2022-08-17T16:03:02Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+        }).exceptionally(exception -> {
+            log.error("[{}] Error closing topic", topic, exception);
+            // Restart rate-limiter after close managed ledger failure. Success is not guaranteed.
+            // TODO Guarantee rate-limiter open finish.


please to not leave "TODOs", open a new GH ticket and link it

Already remove "TODO". Thanks

eolivelli · 2022-08-17T16:03:28Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+            // TODO Guarantee rate-limiter open finish.
+            try {
+                restartLimitersAfterCloseTopicFail();
+            } catch (Throwable t){


please do not catch Throwable blindly

Already instead "Throwable" to "Exception". Thanks

poorbarcode · 2022-08-18T06:52:18Z

/pulsarbot rerun-failure-checks

poorbarcode · 2022-08-18T16:10:52Z

/pulsarbot rerun-failure-checks

poorbarcode · 2022-08-19T00:39:13Z

Hi @eolivelli

Could you review this PR again?

Jason918 · 2022-08-27T08:37:39Z

@codelipenghui @eolivelli @lhotari PTAL

congbobo184 · 2022-08-30T03:21:52Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

+    }
+
+    public CompletableFuture<Void> removeTopicFromCache(String topic){
+        return removeTopicFromCache(topic, (CompletableFuture) null);


Suggested change

return removeTopicFromCache(topic, (CompletableFuture) null);

return removeTopicFromCache(topic, (CompletableFuture<Optional<Topic>>) null);

Already fixed

congbobo184 · 2022-08-30T03:30:39Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+        // Close limiters.
+        try {
+            closeLimiters();
+        } catch (Exception t){


why should catch the Exception?

Just to keep the logic the same as before: close limiters failure does not affect topic close.

congbobo184 · 2022-08-30T03:52:34Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+        }
+
+        // Close client components.
+        CompletableFuture<Void> closeClientsFuture = asyncCloseClients();


Suggested change

CompletableFuture<Void> closeClientsFuture = asyncCloseClients();

CompletableFuture<Void> closeClientsFuture = asyncCloseClients(boolean closeWithoutWaitingClientDisconnect);

The code in line:1294~line:1298 handles the logic closeWithoutWaitingClientDisconnect:

CompletableFuture<Void> closePhase2Future; if (closeWithoutWaitingClientDisconnect){ closePhase2Future = asyncCloseLedger(); } else { closePhase2Future = closeClientsFuture.thenCompose(__ -> asyncCloseLedger()); }

congbobo184 · 2022-08-30T03:53:38Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

+                log.error("[{}] Error closing topic", topic, ex);
+                // Restart rate-limiter after close managed ledger failure. Success is not guaranteed.
+                try {
+                    restartLimitersAfterCloseTopicFail();


why should catch the Exception?

Just to keep the logic the same as before: Restart rate-limiter after close managed ledger failure. Success is not guaranteed.

poorbarcode · 2022-09-07T16:14:31Z

To make this PR easy to Review, it was split into two other PR:

Make method topic.close non-reentrant. Also prevent reentrant between topic.close and topic.delete.
- fixed by PR [fix][broker] One topic can be closed multiple times concurrently #17524
Use cmd map.remove(key, value) instead of map.remove(key) in implementation of `BrokerService.
- fixed by PR [fix][broker]Consumer can't consume messages because there has two sames topics in one broker #17526

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jun 27, 2022

poorbarcode changed the title ~~[fix] [broker] Repeat create same topic~~ [fix] [broker] Repeat creates same topic Jun 27, 2022

poorbarcode mentioned this pull request Jun 27, 2022

[fix] [transaction] infinite loop task that pengding ack log replay #16240

Closed

4 tasks

poorbarcode changed the title ~~[fix] [broker] Repeat creates same topic~~ [fix] [broker] The broker has two identical Perisitenttopics Jun 28, 2022

poorbarcode changed the title ~~[fix] [broker] The broker has two identical Perisitenttopics~~ [fix] [broker] The broker has two identical Persitenttopics Jun 28, 2022

codelipenghui assigned poorbarcode Jun 28, 2022

codelipenghui added this to the 2.11.0 milestone Jun 28, 2022

codelipenghui added type/bug The PR fixed a bug or issue reported a bug area/broker release/2.10.2 release/2.9.4 labels Jun 28, 2022

codelipenghui reviewed Jun 28, 2022

View reviewed changes

poorbarcode closed this Jun 29, 2022

poorbarcode reopened this Jun 29, 2022

poorbarcode added a commit to poorbarcode/pulsar that referenced this pull request Jun 29, 2022

[fix] [broker] The broker has two identical Persitenttopics apache#16247

0707daa

poorbarcode force-pushed the fix/topic_repeat_create branch from 3a3e08e to 0707daa Compare June 29, 2022 17:20

poorbarcode added a commit to poorbarcode/pulsar that referenced this pull request Jun 29, 2022

[fix] [broker] The broker has two identical Persitenttopics apache#16247

482fdfe

poorbarcode force-pushed the fix/topic_repeat_create branch from 0707daa to 482fdfe Compare June 29, 2022 17:27

poorbarcode added a commit to poorbarcode/pulsar that referenced this pull request Jun 29, 2022

[fix] [broker] The broker has two identical Persitenttopics apache#16247

7a9a1a0

poorbarcode force-pushed the fix/topic_repeat_create branch from 482fdfe to 7a9a1a0 Compare June 29, 2022 17:47

poorbarcode added a commit to poorbarcode/pulsar that referenced this pull request Jun 30, 2022

[fix] [broker] The broker has two identical Persitenttopics apache#16247

8ce8d11

poorbarcode force-pushed the fix/topic_repeat_create branch from 7a9a1a0 to 8ce8d11 Compare June 30, 2022 01:23

Anonymitaet removed the doc-not-needed Your PR changes do not impact docs label Jun 30, 2022

Anonymitaet added doc Your PR contains doc changes, no matter whether the changes are in markdown or code files. doc-not-needed Your PR changes do not impact docs and removed doc Your PR contains doc changes, no matter whether the changes are in markdown or code files. labels Jun 30, 2022

poorbarcode removed their assignment Jul 3, 2022

poorbarcode requested a review from codelipenghui July 22, 2022 17:32

codelipenghui modified the milestones: 2.11.0, 2.12.0 Jul 26, 2022

congbobo184 assigned poorbarcode Aug 17, 2022

poorbarcode force-pushed the fix/topic_repeat_create branch from 8ce8d11 to 5337613 Compare August 17, 2022 03:08

eolivelli requested changes Aug 17, 2022

View reviewed changes

poorbarcode requested review from eolivelli and removed request for codelipenghui August 18, 2022 06:50

poorbarcode added 2 commits August 18, 2022 17:18

[fix] [broker] The broker has two identical Persitenttopics

02cfca9

do not try catch Throwable

bb68f61

poorbarcode force-pushed the fix/topic_repeat_create branch from 60180d2 to bb68f61 Compare August 18, 2022 09:18

eolivelli requested a review from lhotari August 22, 2022 07:25

congbobo184 reviewed Aug 30, 2022

View reviewed changes

fix comment: generics

c797fb1

Jason918 added release/2.10.3 and removed release/2.10.2 labels Sep 4, 2022

poorbarcode closed this Sep 7, 2022

poorbarcode deleted the fix/topic_repeat_create branch September 17, 2022 02:53

	public CompletableFuture<Void> close(boolean closeWithoutWaitingClientDisconnect) {
	CompletableFuture<Void> closeFuture = new CompletableFuture<>();

	lock.writeLock().lock();
	try {
	// closing managed-ledger waits until all producers/consumers/replicators get closed. Sometimes, broker
	// forcefully wants to close managed-ledger without waiting all resources to be closed.
	if (!isClosingOrDeleting \|\| closeWithoutWaitingClientDisconnect) {
	fenceTopicToCloseOrDelete();
	} else {

	public CompletableFuture<Void> unloadNamespaceBundle(NamespaceBundle bundle, long timeout, TimeUnit timeoutUnit) {
	return unloadNamespaceBundle(bundle, timeout, timeoutUnit, true);
	}

	}).exceptionally(exception -> {
	log.error("[{}] Error closing topic", topic, exception);
	unfenceTopicToResume();
	closeFuture.completeExceptionally(exception);
	return null;
	});

	return removeTopicFromCache(topic, (CompletableFuture) null);
	return removeTopicFromCache(topic, (CompletableFuture<Optional<Topic>>) null);

	CompletableFuture<Void> closeClientsFuture = asyncCloseClients();
	CompletableFuture<Void> closeClientsFuture = asyncCloseClients(boolean closeWithoutWaitingClientDisconnect);

[fix] [broker] The broker has two identical Persitenttopics #16247

[fix] [broker] The broker has two identical Persitenttopics #16247

Uh oh!

Conversation

poorbarcode commented Jun 27, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Problem occur

Location problem

Repreduce

Other Modifications

PR Relations

Documentation

Uh oh!

poorbarcode commented Jun 27, 2022

Uh oh!

poorbarcode commented Jun 27, 2022

Uh oh!

poorbarcode commented Jun 28, 2022

Uh oh!

poorbarcode commented Jun 28, 2022

Uh oh!

codelipenghui left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poorbarcode commented Jun 29, 2022

Uh oh!

poorbarcode commented Jun 29, 2022

Uh oh!

poorbarcode commented Jun 30, 2022

Uh oh!

poorbarcode commented Aug 17, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poorbarcode commented Aug 18, 2022

Uh oh!

poorbarcode commented Aug 18, 2022

Uh oh!

poorbarcode commented Aug 19, 2022

Uh oh!

Jason918 commented Aug 27, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

poorbarcode commented Jun 27, 2022 •

edited

Loading

poorbarcode Aug 30, 2022 •

edited

Loading