[fix][broker] The topic might reference a closed ledger

[shibd]

Motivation

We observe that a normal topic might reference a closed ledger and it never auto recover. will cause the producer and customer stuck.

The root cause is related to topic create timeout. Assuming we have two concurrent requests to create a topic: firstCreate(Thread-1), secondCreate(Thread-2)

Thread-1	Thread-2
firstCreate topic timeout
	remove this old topic and recreate a new topic
	Open ledger from cache, and get an old ledger that create by firstCreate topic
call topic.close
old ledger close and remove it from cache
	but this old ledger being referenced to new topic and that stats is close

• When the firstCreate topic timeout, will call topic.close. it will close the ledger, and remove it from the ledger cache.
  

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

  Lines 1745 to 1752 in f07b3a0

  	executor().submit(() -> {
  	persistentTopic.close().whenComplete((ignore, ex) -> {
  	if (ex != null) {
  	log.warn("[{}] Get an error when closing topic.",
  	topic, ex);
  	}
  	});
  	});

• If the secondCreate request successfully creates a topic before the old ledger closes, the reference will be made to the old ledger.

Modifications

• When creating a topic, if a topic future exists and it completes with a timeout exception, do not remove it; instead, return the exception directly.
• Once the topic closure is complete, remove the topic from the topics.

Verifying this change

• Add testCloseLedgerThatTopicAfterCreateTimeout to cover this case.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: shibd#37

[dao-jun]

Great work!

[shibd]

This unit test assumes a manager ledger that can never be completed to mock topic create timeout, which does not make sense in a production scenario.

Remove this assertion, and add an assertion afterward: Once ml is completed, the topic should be removed due to timeout.

[codelipenghui]

I would like to suggest to follow the solution from this PR #22283

1. The broker should clean up all the topicFutures for any exceptions finally
2. The new get topic operation should not remove the topicFuture that created by previous get topic operation
3. The topicFuture should only be created if there is no previous topicFuture. If the previous topicFuture is not removed from the map yet, the broker should always use the existing topicFuture (waiting for completion or return error to the client side directly)

The existing code has mixed them which will introduce contention for the get topic operation. Now, we checked the topic load timeout exception, but don't know how many others we also need to check.

[shibd]

Agree with these points. Actually, I tried do that: https://github.com/shibd/pulsar/pull/38/files

But, I got some tests that not pass, So, I used this short-term solution.

I'm going to continue to look into it.

[shibd]

A new solution refer to this PR: #22860

[shibd]

Close with this comments: #22739 (comment)