[fix][broker] Fix future of clear delayed message can't complete

[coderzc]

PIP: #16763

Motivation

If newTracker() throws an exception, then the future will can't complete forever, this will lead to the topic can't be deleted until restart broker.

Modifications

Use cleanResidualSnapshots instead of creating a tracker to clear residual snapshots if the tracker does not exist.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Documentation

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

Matching PR in forked repository

PR in forked repository:

[cbornet]

@coderzc we are now in code freeze for 3.0. So unless this is a blocker for the release, this change should target 3.0.1/3.1. Please change the milestone again to 3.0 if you believe that's a blocker.

[coderzc]

@coderzc we are now in code freeze for 3.0. So unless this is a blocker for the release, this change should target 3.0.1/3.1. Please change the milestone again to 3.0 if you believe that's a blocker.

This bug will lead to a topic can't be deleted until restart broker, so I think it is a blocker.

[poorbarcode]

Could you add a test?

[poorbarcode]

If we never touch the method trackDelayedDelivery(), then the variable delayedDeliveryTracker is always is empty, right?

If yes, why need we initialize it when we should clear delayed messages? Can we just return a completed future?

[coderzc]

@poorbarcode Because the dispatcher was closed before unsubscribe was called, this lead to the tracker reference being lost, so we need to reinitialize the tracker to clean up residual persistent data.

Please see: line-1320、line-1356

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

Lines 1320 to 1357 in 4f503fd

	subscriptions.forEach((s, sub) -> futures.add(sub.disconnect()));
	if (closeIfClientsConnected) {
	replicators.forEach((cluster, replicator) -> futures.add(replicator.disconnect()));
	shadowReplicators.forEach((__, replicator) -> futures.add(replicator.disconnect()));
	producers.values().forEach(producer -> futures.add(producer.disconnect()));
	}
	FutureUtil.waitForAll(futures).thenRunAsync(() -> {
	closeClientFuture.complete(null);
	}, getOrderedExecutor()).exceptionally(ex -> {
	log.error("[{}] Error closing clients", topic, ex);
	unfenceTopicToResume();
	closeClientFuture.completeExceptionally(ex);
	return null;
	});
	closeClientFuture.thenAccept(__ -> {
	CompletableFuture<Void> deleteTopicAuthenticationFuture = new CompletableFuture<>();
	brokerService.deleteTopicAuthenticationWithRetry(topic, deleteTopicAuthenticationFuture, 5);
	deleteTopicAuthenticationFuture.thenCompose(ignore -> deleteSchema())
	.thenCompose(ignore -> {
	if (!SystemTopicNames.isTopicPoliciesSystemTopic(topic)
	&& brokerService.getPulsar().getConfiguration().isSystemTopicEnabled()) {
	return deleteTopicPolicies();
	} else {
	return CompletableFuture.completedFuture(null);
	}
	})
	.thenCompose(ignore -> transactionBufferCleanupAndClose())
	.whenComplete((v, ex) -> {
	if (ex != null) {
	log.error("[{}] Error deleting topic", topic, ex);
	unfenceTopicToResume();
	deleteFuture.completeExceptionally(ex);
	} else {
	List<CompletableFuture<Void>> subsDeleteFutures = new ArrayList<>();
	subscriptions.forEach((sub, p) -> subsDeleteFutures.add(unsubscribe(sub)));

[poorbarcode]

Because the dispatcher was closed before unsubscribe was called, this lead to the tracker reference being lost, so we need to reinitialize the tracker to clean up residual persistent data.

I see. Another question: the method new tracker will call recover internally, right? If yes, I feel it is an expensive task. Can we only set the Tracker to an unavailable state when closing the dispatcher to avoid recover the second time?

[coderzc]

You are right, I think we should use another way to clean up data to avoid unnecessary recover, such as add a clean method in the factory.

[coderzc]

@poorbarcode I use cleanResidualSnapshots instead of create a tracker to clear residual snapshots if the tracker does not exist, please take a look again.

[codecov-commenter]

Codecov Report

Merging #20075 (4220f69) into master (a332fea) will increase coverage by 37.34%.
The diff coverage is 84.00%.

@@              Coverage Diff              @@
##             master   #20075       +/-   ##
=============================================
+ Coverage     35.58%   72.93%   +37.34%     
- Complexity    12423    31901    +19478     
=============================================
  Files          1691     1868      +177     
  Lines        128771   138309     +9538     
  Branches      14044    15211     +1167     
=============================================
+ Hits          45822   100874    +55052     
+ Misses        76931    29422    -47509     
- Partials       6018     8013     +1995     

Flag	Coverage Δ
inttests	24.29% <0.00%> (?)
systests	24.88% <0.00%> (-0.02%)	⬇️
unittests	72.23% <84.00%> (+39.08%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...rg/apache/pulsar/broker/delayed/bucket/Bucket.java	88.70% <ø> (+88.70%)	⬆️
...r/delayed/bucket/BucketDelayedDeliveryTracker.java	82.90% <ø> (+82.90%)	⬆️
...sar/broker/service/persistent/PersistentTopic.java	79.39% <70.00%> (+28.57%)	⬆️
...r/delayed/BucketDelayedDeliveryTrackerFactory.java	93.54% <90.90%> (+93.54%)	⬆️
...sistent/PersistentDispatcherMultipleConsumers.java	75.83% <100.00%> (+23.42%)	⬆️

... and 1438 files with indirect coverage changes