[pulsar-admin] Add remove-subscription-types-enabled command for topic

[yuruguo]

Motivation

CLI bin/pulsar-admin topics supports set-subscription-types-enabled and get-subscription-types-enabled command, but lacks corresponding remove-subscription-types-enabled commands, the purpose of this PR is to add this command for topic.

Documentation

Automatically generate doc through code

• doc

[github-actions]

@yuruguo:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[github-actions]

@yuruguo:Thanks for providing doc info!

[eolivelli]

We should not modify a object that could be the result of a read from a cache

[yuruguo]

Sorry, I don't fully understand what you mean. I'm still a little confused about it. Can you give me more information？
I understand that this is the process of making TopicPolicies.subscriptionTypesEnabled as empty list(default initial value), right?
Maybe according to your meaning, how should I modify it?

[eolivelli]

you are using the returned object from getTopicPoliciesAsyncWithRetry using op.get()
if pulsar().getTopicPoliciesService().updateTopicPoliciesAsync fails then the local copy of the variable in memory is update but the value is not write to persistent metadata storage.

We should copy/clone the object and not mutate it.
It would be better to have "immutable" objects, but that would be another story

[yuruguo]

Many topic policy set&remove operations are implemented in this way, so they all need to be adjusted, as below:

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

Lines 3012 to 3020 in aa97d39

	protected CompletableFuture<Void> internalSetPersistence(PersistencePolicies persistencePolicies) {
	validatePersistencePolicies(persistencePolicies);
	return getTopicPoliciesAsyncWithRetry(topicName)
	.thenCompose(op -> {
	TopicPolicies topicPolicies = op.orElseGet(TopicPolicies::new);
	topicPolicies.setPersistence(persistencePolicies);
	return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, topicPolicies);
	});
	}

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

Lines 3022 to 3031 in aa97d39

	protected CompletableFuture<Void> internalRemovePersistence() {
	return getTopicPoliciesAsyncWithRetry(topicName)
	.thenCompose(op -> {
	if (!op.isPresent()) {
	return CompletableFuture.completedFuture(null);
	}
	op.get().setPersistence(null);
	return pulsar().getTopicPoliciesService().updateTopicPoliciesAsync(topicName, op.get());
	});
	}

[eolivelli]

I am approving this patch.

thank you @yuruguo for checking about my comments.

I believe we have to follow up and fix potential local cache consistency problems.

would you like to follow up ?

cc @merlimat

[yuruguo]

I am approving this patch.

thank you @yuruguo for checking about my comments.

I believe we have to follow up and fix potential local cache consistency problems.

would you like to follow up ?

cc @merlimat

Thanks for your guidance, I will create an issue to track this problem and try to submit a PR to solve it :)