switch (operation) {
    case LOOKUP:
    case GET_STATS:
    case GET_METADATA:
        // 需要生产或消费权限
        return canLookupAsync(topicName, role, authData);
    case PRODUCE:
        // 需要生产权限
        return canProduceAsync(topicName, role, authData);
    case GET_SUBSCRIPTIONS:
    case CONSUME:
    case SUBSCRIBE:
    case UNSUBSCRIBE:
    case SKIP:
    case EXPIRE_MESSAGES:
    case PEEK_MESSAGES:
    case RESET_CURSOR:
    case GET_BACKLOG_SIZE:
    case SET_REPLICATED_SUBSCRIPTION_STATUS:
    case GET_REPLICATED_SUBSCRIPTION_STATUS:
        // 需要消费权限
        return canConsumeAsync(topicName, role, authData, authData.getSubscription());
    case TERMINATE:
    case COMPACT:
    case OFFLOAD:
    case UNLOAD:
    case TRIM_TOPIC:
    case DELETE_METADATA:
    case UPDATE_METADATA:
    case ADD_BUNDLE_RANGE:
    case GET_BUNDLE_RANGE:
    case DELETE_BUNDLE_RANGE:
        // 需要更高权限即 adminRole/superUser 权限
        return CompletableFuture.completedFuture(false);
    default:
        return FutureUtil.failedFuture(new IllegalStateException("TopicOperation [" + operation.name() + "] is not supported."));
}