...

In order to fix these issues, we should introduce a new RPC named incrementalAlterConfigs.  The new RPC should operate incrementally, modifying only the configuration values that are specified.  We should deprecate AlterConfigs.

Proposed Changes

The new IncrementalAlterConfigs API in AdminClient will take a collection of operations describing the configuration modifications.  There are four types of operations.

• Set: set a configuration to a value.  The value must not be null.
• Delete: delete a configuration key
• Append: if a configuration key is a list of values, add to the list.
• Subtract: if a configuration key is a list of values, subtract from the list.

Public Interfaces

@InterfaceStability.Evolving
public interfaceclass AlterConfigOp {

   ConfigResource resource();
}

public class SetConfigOp extends AlterConfigOpenum OpType {
  public SetConfigOp(ConfigResource resource, String value);
}

public class DeleteConfigOp extends AlterConfigOp {
  public DeleteConfigOp(SConfigResource resource);
}

public class AppendConfigOp extends AlterConfigOp {
  public AppendConfigOp(ConfigResource resource, String value);
}

public class SubtractConfigOp extends AlterConfigOp {
  public SubtractConfigOp(ConfigResource resource, String value);
}

public IncrementalAlterConfigsResult incrementalAlterConfigs(
    Collection<AlterConfigOp> ops,
    final IncrementalAlterConfigsOptions options);

If a configuration key is specified multiple times, that is an error, and an error code will be returned for each instance of the configuration key.

Similar to AlterConfigsResult, IncrementalAlterConfigsResult will have a values function which allows callers to determine the result of a specific configuration resource modifications, and an all function which will throw an exception if any operation failed.

@InterfaceStability.Evolving
public class IncrementalAlterConfigsResult {
    private final Map<ConfigResource, KafkaFuture<Void>> futuresSET((byte) 0), DELETE((byte) 1), APPEND((byte) 2), SUBTRACT((byte) 3);

        private static final Map<Byte, OpType> OP_TYPES = Collections.unmodifiableMap(
                Arrays.stream(values()).collect(Collectors.toMap(OpType::id, Function.identity()))
        );

    IncrementalAlterConfigsResult(Map<ConfigResource, KafkaFuture<Void>> futures) {
 private final      this.futures = futures;
 byte id;

   }

    /**
     * Return a map from resources to futures which can be used to check the status of the operation on each resource.
     */
 OpType(final byte id) {
            this.id = id;
        }

        public Map<ConfigResource, KafkaFuture<Void>> valuesbyte id() {
        return futures;
   return }id;

    /**
    }

 * Return a future which succeeds only ifpublic allstatic theOpType alterforId(final configsbyte operations succeed.
     */
    public KafkaFuture<Void> all() {
id) {
            return KafkaFutureOP_TYPES.allOf(futures.values().toArray(new KafkaFuture[0]))get(id);
    }
}

Similar to AlterConfigsOptions, IncrementalAlterConfigsOptions will include a timeout value and a dry-run flag.

@InterfaceStability.Evolving
public class IncrementalAlterConfigsOptions extends AbstractOptions<IncrementalAlterConfigsOptions> {}
    private boolean validateOnly = false;}

    /**
private final    * Set the request timeout in milliseconds for this operation or {@code null} if the default request timeout for theConfigEntry configEntry;
    private *final AdminClient should be used.
     */OpType opType;

    public IncrementalAlterConfigsOptions timeoutMs(Integer timeoutMsAlterConfigOp(ConfigEntry configEntry, OpType operationType) {
        this.timeoutMsconfigEntry = timeoutMsconfigEntry;
        this.opType = return thisoperationType;
    }

    /**
public ConfigEntry configEntry() {
  * Return true if the request shouldreturn beconfigEntry;
 validated without altering the configs.
     */ };

    public booleanOpType shouldValidateOnlyopType() {
        return validateOnlyopType;
    };
}


public AlterConfigsResult incrementalAlterConfigs(
		Map<ConfigResource, Collection<AlterConfigOp>> /**configs,
     * Set to truefinal if the request should be validated without altering the configs.
     */
    public IncrementalAlterConfigsOptions validateOnly(boolean validateOnly) {
        this.validateOnly = validateOnly;
        return this;
    }
}AlterConfigsOptions options);

Proposed Changes

The new IncrementalAlterConfigs API in AdminClient will take a collection of operations describing the configuration modifications.  There are four types of operations.

• Set: set a configuration to a value.  The value must not be null.
• Delete: delete a configuration key
• Append: if a configuration key is a list of values, add to the list.
• Subtract: if a configuration key is a list of values, subtract from the list.

We will use existing AlterConfigsOptions, AlterConfigsResult API classes to pass the  API config options and to return the result of  configuration resource modifications.

Similar to existing alterConfigs API, we will to keep the "transactionality" of updating several configs  for the same ConfigResource at once. We guarantee that we never do a partial update of a collection of configs
for a ConfigResource from a single request. On validation/update error, we will return the error for the ConfigResource.

Protocol APIs

There will be a new IncrementalAlterConfigsRequest Incremental AlterConfigsRequest.

IncrementalAlterConfigsOp => INT8
0: SET
1: REMOVE
2: APPEND
3: SUBTRACT

IncrementalAlterConfigsRequest (Version: 0) => [resources] validate_only
 validate_only => BOOLEAN
 resources => resource_type resource_name [configs]
 resource_type => INT8
 resource_name => STRING
 configs => config_name config_op config_value
 config_name => STRING
 config_op => INT8
 config_value => NULLABLE_STRING

The IncrementalAlterConfigsResponse Incremental AlterConfigsResponse is the same as the AlterConfigsResponse.

...

• If a configuration key was specified more than once in the request for given ConfigResource, we fail ConfigResource update with the INVALID_REQUEST will be returned for all occurrences of the configuration key.
• If APPEND or SUBTRACT was specified for a configuration key that does not take a list of elements, we fail ConfigResource update with the INVALID_REQUEST will be returned for the configuration key.

Compatibility, Deprecation, and Migration Plan

...