public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topicName,
                                                               TopicOperation operation,
                                                               String role,
                                                               AuthenticationDataSource authData) {
        if (log.isDebugEnabled()) {
            log.debug("Check if role {} is allowed to execute topic operation {} on topic {}",
                    role, operation, topicName);
        }
        if (!this.conf.isAuthorizationEnabled()) {
            return CompletableFuture.completedFuture(true);
        }

        CompletableFuture<Boolean> allowFuture =
                provider.allowTopicOperationAsync(topicName, role, operation, authData);
        if (log.isDebugEnabled()) {
            return allowFuture.whenComplete((allowed, exception) -> {
                if (exception == null) {
                    if (allowed) {
                        log.debug("Topic operation {} on topic {} is allowed: role = {}",
                                operation, topicName, role);
                    } else {
                        log.debug("Topic operation {} on topic {} is NOT allowed: role = {}",
                                operation, topicName, role);
                    }
                } else {
                    log.debug("Failed to check if topic operation {} on topic {} is allowed:"
                                    + " role = {}",
                            operation, topicName, role, exception);
                }
            });
        } else {
            return allowFuture;
        }
    }
@Test(dataProvider = "roles")
    public void testTopicOperationAsync(String role, String originalRole, boolean shouldPass) throws Exception {
        boolean isAuthorized = authorizationService.allowTopicOperationAsync(TopicName.get("topic"),
                TopicOperation.PRODUCE, originalRole, role, null).get();
        checkResult(shouldPass, isAuthorized);
    }