[Authorization] Role with namespace produce authz can also get topics

[yuruguo]

Motivation

From this comment
Role with namespace produce or consume authz can get topics.

Documentation

• no-need-doc

[mattisonchao]

@yuruguo

Something strange here, do we need e or ex ?
Besides, we need getCause() in the exception of CompletableFuture, otherwise, we will get CompletionException

[yuruguo]

1. we need ex in finalResult.completeExceptionally().
2. I has used getCause().

PTAL again.

[eolivelli]

good work

I left some feedback

[eolivelli]

we cannot let pass every exception
we should fall back to checking the second action only in case of missing authentication, otherwise if there is a system error (system overloaded?) we are going to generate the error twice

[yuruguo]

we cannot let pass every exception

If there is an exception or error when checking the first action and we end the subsequent checks, so even if the user has the second permission and it will not take effect. Does this not meet expectations?

we should fall back to checking the second action only in case of missing authentication

I traced the implementation of allowTheSpecifiedActionOpsAsync and not found exceptions related to authentication, so I can't tell whether to perform the second check.

In addition, the implementation logic of this PR is similar canLookupAsync, as follows:

pulsar/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java

Lines 179 to 211 in d972990

	public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
	AuthenticationDataSource authenticationData) {
	CompletableFuture<Boolean> finalResult = new CompletableFuture<Boolean>();
	canProduceAsync(topicName, role, authenticationData).whenComplete((produceAuthorized, ex) -> {
	if (ex == null) {
	if (produceAuthorized) {
	finalResult.complete(produceAuthorized);
	return;
	}
	} else {
	if (log.isDebugEnabled()) {
	log.debug(
	"Topic [{}] Role [{}] exception occurred while trying to check Produce permissions. {}",
	topicName.toString(), role, ex.getMessage());
	}
	}
	canConsumeAsync(topicName, role, authenticationData, null).whenComplete((consumeAuthorized, e)
	-> {
	if (e == null) {
	finalResult.complete(consumeAuthorized);
	} else {
	if (log.isDebugEnabled()) {
	log.debug(
	"Topic [{}] Role [{}] exception occurred while trying to check Consume permissions. {}",
	topicName.toString(), role, e.getMessage());
	}
	finalResult.completeExceptionally(e);
	}
	});
	});
	return finalResult;
	}

[yuruguo]

@codelipenghui @mattisonchao @michaeljmarshall ptal

[BewareMyPower]

Add the release/2.8.4 label since it's relied by #15501.