@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));
    }
@Test
    public void testIncrementalAlterConfigs()  throws Exception {
        try (AdminClientUnitTestEnv env = mockClientEnv()) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

            //test error scenarios
            IncrementalAlterConfigsResponseData responseData =  new IncrementalAlterConfigsResponseData();
            responseData.responses().add(new AlterConfigsResourceResponse()
                    .setResourceName("")
                    .setResourceType(ConfigResource.Type.BROKER.id())
                    .setErrorCode(Errors.CLUSTER_AUTHORIZATION_FAILED.code())
                    .setErrorMessage("authorization error"));

            responseData.responses().add(new AlterConfigsResourceResponse()
                    .setResourceName("metric1")
                    .setResourceType(ConfigResource.Type.CLIENT_METRICS.id())
                    .setErrorCode(Errors.INVALID_REQUEST.code())
                    .setErrorMessage("Subscription is not allowed"));

            responseData.responses().add(new AlterConfigsResourceResponse()
                    .setResourceName("topic1")
                    .setResourceType(ConfigResource.Type.TOPIC.id())
                    .setErrorCode(Errors.INVALID_REQUEST.code())
                    .setErrorMessage("Config value append is not allowed for config"));

            responseData.responses().add(new AlterConfigsResourceResponse()
                    .setResourceName("group1")
                    .setResourceType(ConfigResource.Type.GROUP.id())
                    .setErrorCode(Errors.INVALID_CONFIG.code())
                    .setErrorMessage("Unknown group config name: group.initial.rebalance.delay.ms"));

            env.kafkaClient().prepareResponse(new IncrementalAlterConfigsResponse(responseData));

            ConfigResource brokerResource = new ConfigResource(ConfigResource.Type.BROKER, "");
            ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, "topic1");
            ConfigResource metricResource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, "metric1");
            ConfigResource groupResource = new ConfigResource(ConfigResource.Type.GROUP, "group1");

            AlterConfigOp alterConfigOp1 = new AlterConfigOp(
                    new ConfigEntry("log.segment.bytes", "1073741"),
                    AlterConfigOp.OpType.SET);

            AlterConfigOp alterConfigOp2 = new AlterConfigOp(
                    new ConfigEntry("compression.type", "gzip"),
                    AlterConfigOp.OpType.APPEND);

            AlterConfigOp alterConfigOp3 = new AlterConfigOp(
                    new ConfigEntry("interval.ms", "1000"),
                    AlterConfigOp.OpType.APPEND);

            AlterConfigOp alterConfigOp4 = new AlterConfigOp(
                    new ConfigEntry("group.initial.rebalance.delay.ms", "1000"),
                    AlterConfigOp.OpType.SET);

            final Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
            configs.put(brokerResource, singletonList(alterConfigOp1));
            configs.put(topicResource, singletonList(alterConfigOp2));
            configs.put(metricResource, singletonList(alterConfigOp3));
            configs.put(groupResource, singletonList(alterConfigOp4));

            AlterConfigsResult result = env.adminClient().incrementalAlterConfigs(configs);
            TestUtils.assertFutureError(result.values().get(brokerResource), ClusterAuthorizationException.class);
            TestUtils.assertFutureError(result.values().get(topicResource), InvalidRequestException.class);
            TestUtils.assertFutureError(result.values().get(metricResource), InvalidRequestException.class);
            TestUtils.assertFutureError(result.values().get(groupResource), InvalidConfigurationException.class);

            // Test a call where there are no errors.
            responseData =  new IncrementalAlterConfigsResponseData();
            responseData.responses().add(new AlterConfigsResourceResponse()
                    .setResourceName("")
                    .setResourceType(ConfigResource.Type.BROKER.id())
                    .setErrorCode(Errors.NONE.code())
                    .setErrorMessage(ApiError.NONE.message()));
            responseData.responses().add(new AlterConfigsResourceResponse()
                    .setResourceName("metric1")
                    .setResourceType(ConfigResource.Type.CLIENT_METRICS.id())
                    .setErrorCode(Errors.NONE.code())
                    .setErrorMessage(ApiError.NONE.message()));
            responseData.responses().add(new AlterConfigsResourceResponse()
                    .setResourceName("group1")
                    .setResourceType(ConfigResource.Type.GROUP.id())
                    .setErrorCode(Errors.NONE.code())
                    .setErrorMessage(ApiError.NONE.message()));

            final Map<ConfigResource, Collection<AlterConfigOp>> successConfig = new HashMap<>();
            successConfig.put(brokerResource, singletonList(alterConfigOp1));
            successConfig.put(metricResource, singletonList(alterConfigOp3));
            successConfig.put(groupResource, singletonList(alterConfigOp4));

            env.kafkaClient().prepareResponse(new IncrementalAlterConfigsResponse(responseData));
            env.adminClient().incrementalAlterConfigs(successConfig).all().get();
        }
    }