@Override
    public AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs,
                                                      final AlterConfigsOptions options) {
        final Map<ConfigResource, KafkaFutureImpl<Void>> allFutures = new HashMap<>();
        // BROKER_LOGGER requests always go to a specific, constant broker or controller node.
        //
        // BROKER resource changes for a specific (non-default) resource go to either that specific
        // node (if using bootstrap.servers), or directly to the active controller (if using
        // bootstrap.controllers)
        //
        // All other requests go to the least loaded broker (if using bootstrap.servers) or the
        // active controller (if using bootstrap.controllers)
        final Collection<ConfigResource> unifiedRequestResources = new ArrayList<>();

        for (ConfigResource resource : configs.keySet()) {
            Integer node = nodeFor(resource);
            if (metadataManager.usingBootstrapControllers()) {
                if (!resource.type().equals(ConfigResource.Type.BROKER_LOGGER)) {
                    node = null;
                }
            }
            if (node != null) {
                NodeProvider nodeProvider = new ConstantNodeIdProvider(node, true);
                allFutures.putAll(incrementalAlterConfigs(configs, options, Collections.singleton(resource), nodeProvider));
            } else
                unifiedRequestResources.add(resource);
        }
        if (!unifiedRequestResources.isEmpty())
            allFutures.putAll(incrementalAlterConfigs(configs, options, unifiedRequestResources, new LeastLoadedBrokerOrActiveKController()));

        return new AlterConfigsResult(new HashMap<>(allFutures));
    }
Fail futures in the given Map which were retried due to exceeding quota. We propagate
the initial error back to the caller if the request timed out.
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java