@Override
        public void accept(ConfigResource configResource) {
            switch (configResource.type()) {
                case BROKER_LOGGER:
                    // BROKER_LOGGER are always allowed.
                    break;
                case BROKER:
                    // Cluster configs are always allowed.
                    if (configResource.name().isEmpty()) break;

                    // Otherwise, check that the node ID is valid.
                    int nodeId;
                    try {
                        nodeId = Integer.parseInt(configResource.name());
                    } catch (NumberFormatException e) {
                        throw new InvalidRequestException("Invalid broker name " +
                            configResource.name());
                    }
                    if (!(clusterControl.brokerRegistrations().containsKey(nodeId) ||
                            featureControl.isControllerId(nodeId))) {
                        throw new BrokerIdNotRegisteredException("No node with id " +
                            nodeId + " found.");
                    }
                    break;
                case TOPIC:
                    if (replicationControl.getTopicId(configResource.name()) == null) {
                        throw new UnknownTopicOrPartitionException("The topic '" +
                            configResource.name() + "' does not exist.");
                    }
                    break;
                default:
                    break;
            }
        }
Checks that a configuration resource exists.
<p>
This object must be used only from the controller event thread.
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java