[fix][broker]fail to update partition meta of topic due to ConflictException: subscription already exists for topic

[HQebupt]

fixed #10374

Motivation

#10374 wants to handle the existing subscriptions when updating partition with force command parameter. However, it could not update the partition metadata if the Subscription already exists for topic exception comes. Because it miss updating when handling such exception.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

Lines 4360 to 4392 in 3160edc

	private CompletableFuture<Void> updatePartitionedTopic(TopicName topicName, int numPartitions, boolean force) {
	CompletableFuture<Void> result = new CompletableFuture<>();
	createSubscriptions(topicName, numPartitions).thenCompose(__ -> {
	CompletableFuture<Void> future = namespaceResources().getPartitionedTopicResources()
	.updatePartitionedTopicAsync(topicName, p -> new PartitionedTopicMetadata(numPartitions));
	future.exceptionally(ex -> {
	// If the update operation fails, clean up the partitions that were created
	getPartitionedTopicMetadataAsync(topicName, false, false).thenAccept(metadata -> {
	int oldPartition = metadata.partitions;
	for (int i = oldPartition; i < numPartitions; i++) {
	topicResources().deletePersistentTopicAsync(topicName.getPartition(i)).exceptionally(ex1 -> {
	log.warn("[{}] Failed to clean up managedLedger {}", clientAppId(), topicName,
	ex1.getCause());
	return null;
	});
	}
	}).exceptionally(e -> {
	log.warn("[{}] Failed to clean up managedLedger", topicName, e);
	return null;
	});
	return null;
	});
	return future;
	}).thenAccept(__ -> result.complete(null)).exceptionally(ex -> {
	if (force && ex.getCause() instanceof PulsarAdminException.ConflictException) {
	result.complete(null);
	return null;
	}
	result.completeExceptionally(ex);
	return null;
	});
	return result;
	}

The line 4364 updatePartitionedTopicAsync could not reach if the line 4362 createSubscriptions occur an exception of PulsarAdminException$ConflictException: Subscription already exists for topic.

Caused by: org.apache.pulsar.client.admin.PulsarAdminException$ConflictException: Subscription already exists for topic
	at org.apache.pulsar.client.admin.internal.BaseResource.getApiException(BaseResource.java:219) ~[classes/:?]
	at org.apache.pulsar.client.admin.internal.BaseResource$1.failed(BaseResource.java:129) ~[classes/:?]
	at org.glassfish.jersey.client.JerseyInvocation$1.failed(JerseyInvocation.java:839) ~[jersey-client-2.31.jar:?]
	at org.glassfish.jersey.client.JerseyInvocation$1.completed(JerseyInvocation.java:820) ~[jersey-client-2.31.jar:?]
	at org.glassfish.jersey.client.ClientRuntime.processResponse(ClientRuntime.java:229) ~[jersey-client-2.31.jar:?]
	at org.glassfish.jersey.client.ClientRuntime.access$200(ClientRuntime.java:62) ~[jersey-client-2.31.jar:?]
	at org.glassfish.jersey.client.ClientRuntime$2.lambda$response$0(ClientRuntime.java:173) ~[jersey-client-2.31.jar:?]
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248) ~[jersey-common-2.31.jar:?]
	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244) ~[jersey-common-2.31.jar:?]
	at org.glassfish.jersey.internal.Errors.process(Errors.java:292) ~[jersey-common-2.31.jar:?]
	at org.glassfish.jersey.internal.Errors.process(Errors.java:274) ~[jersey-common-2.31.jar:?]
	at org.glassfish.jersey.internal.Errors.process(Errors.java:244) ~[jersey-common-2.31.jar:?]
	at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:288) ~[jersey-common-2.31.jar:?]
	at org.glassfish.jersey.client.ClientRuntime$2.response(ClientRuntime.java:173) ~[jersey-client-2.31.jar:?]
	at org.apache.pulsar.client.admin.internal.http.AsyncHttpConnector.lambda$0(AsyncHttpConnector.java:212) ~[classes/:?]

Modifications

update topic partition metadata when handling ConflictException.

Verifying this change

• Make sure that the change passes the CI checks.

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

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

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

Check the box below and label this PR (if you have committer privilege).

Need to update docs?

• doc-not-needed

[HQebupt]

/pulsarbot run-failure-checks

[HQebupt]

/pulsarbot run-failure-checks

[Jason918]

This fix LGTM.
We can also consider add a "force" parameter in admin.topics().createSubscription() to ignore conflict exception when the subscription already exists.

[AnonHxy]

LGTM now, left minor commnet :) @HQebupt

[HQebupt]

This fix LGTM.
We can also consider add a "force" parameter in admin.topics().createSubscription() to ignore conflict exception when the subscription already exists.

I will continue to pay attention to this point.

[HQebupt]

/pulsarbot run-failure-checks

[HQebupt]

/pulsarbot run-failure-checks

[HQebupt]

/pulsarbot run-failure-checks

[Jason918]

Can you help open a new PR to branch-2.10? There are a lot conflict when cherry-pick directly. @HQebupt

[HQebupt]

Can you help open a new PR to branch-2.10? There are a lot conflict when cherry-pick directly. @HQebupt

Sure

[Jason918]

Move release/2.10.2 to #17488