You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2019/04/16 23:26:46 UTC
[kafka] branch trunk updated: KAFKA-7466: Add
IncrementalAlterConfigs API (KIP-339) (#6247)
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 3b1524c KAFKA-7466: Add IncrementalAlterConfigs API (KIP-339) (#6247)
3b1524c is described below
commit 3b1524c5dfd2a94f3fb919dad0de70984963772b
Author: Manikumar Reddy <ma...@gmail.com>
AuthorDate: Wed Apr 17 04:56:33 2019 +0530
KAFKA-7466: Add IncrementalAlterConfigs API (KIP-339) (#6247)
Reviewers: Colin P. McCabe <cm...@apache.org>, Viktor Somogyi <vi...@gmail.com>, Stanislav Kozlovski <st...@outlook.com>, Rajini Sivaram <ra...@googlemail.com>, Ismael Juma <is...@juma.me.uk>
---
checkstyle/suppressions.xml | 2 +-
.../apache/kafka/clients/admin/AdminClient.java | 46 ++++++
.../apache/kafka/clients/admin/AlterConfigOp.java | 96 +++++++++++
.../kafka/clients/admin/KafkaAdminClient.java | 90 +++++++++++
.../org/apache/kafka/common/config/ConfigDef.java | 4 +
.../org/apache/kafka/common/protocol/ApiKeys.java | 6 +-
.../kafka/common/requests/AbstractRequest.java | 2 +
.../kafka/common/requests/AbstractResponse.java | 2 +
.../requests/IncrementalAlterConfigsRequest.java | 91 +++++++++++
.../requests/IncrementalAlterConfigsResponse.java | 99 ++++++++++++
.../message/IncrementalAlterConfigsRequest.json | 41 +++++
.../message/IncrementalAlterConfigsResponse.json | 36 +++++
.../kafka/clients/admin/KafkaAdminClientTest.java | 56 +++++++
.../kafka/clients/admin/MockAdminClient.java | 7 +
.../kafka/common/requests/RequestResponseTest.java | 35 ++++
core/src/main/scala/kafka/log/LogConfig.scala | 2 +
.../src/main/scala/kafka/server/AdminManager.scala | 180 +++++++++++++++------
core/src/main/scala/kafka/server/KafkaApis.scala | 31 ++++
.../kafka/api/AdminClientIntegrationTest.scala | 178 ++++++++++++++++++++
.../kafka/api/AuthorizerIntegrationTest.scala | 33 +++-
.../server/DynamicBrokerReconfigurationTest.scala | 16 +-
.../scala/unit/kafka/server/RequestQuotaTest.scala | 10 +-
.../test/scala/unit/kafka/utils/TestUtils.scala | 28 +++-
23 files changed, 1023 insertions(+), 68 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index ce2706d..ca103a2 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -43,7 +43,7 @@
files="Sender.java"/>
<suppress checks="ClassDataAbstractionCoupling"
- files="(KafkaConsumer|ConsumerCoordinator|Fetcher|KafkaProducer|AbstractRequest|AbstractResponse|TransactionManager|KafkaAdminClient).java"/>
+ files="(KafkaConsumer|ConsumerCoordinator|Fetcher|KafkaProducer|AbstractRequest|AbstractResponse|TransactionManager|AdminClient|KafkaAdminClient).java"/>
<suppress checks="ClassDataAbstractionCoupling"
files="(Errors|SaslAuthenticatorTest|AgentTest|CoordinatorTest).java"/>
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
index b823cdc..8826f83 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/AdminClient.java
@@ -368,7 +368,9 @@ public abstract class AdminClient implements AutoCloseable {
* @param configs The resources with their configs (topic is the only resource type with configs that can
* be updated currently)
* @return The AlterConfigsResult
+ * @deprecated Since 2.3. Use {@link #incrementalAlterConfigs(Map)}.
*/
+ @Deprecated
public AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs) {
return alterConfigs(configs, new AlterConfigsOptions());
}
@@ -385,10 +387,54 @@ public abstract class AdminClient implements AutoCloseable {
* be updated currently)
* @param options The options to use when describing configs
* @return The AlterConfigsResult
+ * @deprecated Since 2.3. Use {@link #incrementalAlterConfigs(Map, AlterConfigsOptions)}.
*/
+ @Deprecated
public abstract AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, AlterConfigsOptions options);
/**
+ * Incrementally updates the configuration for the specified resources with default options.
+ *
+ * This is a convenience method for #{@link AdminClient#incrementalAlterConfigs(Map, AlterConfigsOptions)} with default options.
+ * See the overload for more details.*
+ *
+ * This operation is supported by brokers with version 2.3.0 or higher.
+ *
+ * @param configs The resources with their configs
+ * @return The IncrementalAlterConfigsResult
+ */
+ public AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs) {
+ return incrementalAlterConfigs(configs, new AlterConfigsOptions());
+ }
+
+
+ /**
+ * Incrementally update the configuration for the specified resources.
+ *
+ * Updates are not transactional so they may succeed for some resources while fail for others. The configs for
+ * a particular resource are updated atomically.
+ *
+ * <p>The following exceptions can be anticipated when calling {@code get()} on the futures obtained from
+ * the returned {@code IncrementalAlterConfigsResult}:</p>
+ * <ul>
+ * <li>{@link org.apache.kafka.common.errors.ClusterAuthorizationException}
+ * if the authenticated user didn't have alter access to the cluster.</li>
+ * <li>{@link org.apache.kafka.common.errors.TopicAuthorizationException}
+ * if the authenticated user didn't have alter access to the Topic.</li>
+ * <li>{@link org.apache.kafka.common.errors.InvalidRequestException}
+ * if the request details are invalid. e.g., a configuration key was specified more than once for a resource</li>
+ * </ul>*
+ *
+ * This operation is supported by brokers with version 2.3.0 or higher.
+ *
+ * @param configs The resources with their configs
+ * @param options The options to use when altering configs
+ * @return The IncrementalAlterConfigsResult
+ */
+ public abstract AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource,
+ Collection<AlterConfigOp>> configs, AlterConfigsOptions options);
+
+ /**
* Change the log directory for the specified replicas. If the replica does not exist on the broker, the result
* shows REPLICA_NOT_AVAILABLE for the given replica and the replica will be created in the given log directory on the
* broker when it is created later. If the replica already exists on the broker, the replica will be moved to the given
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigOp.java b/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigOp.java
new file mode 100644
index 0000000..367c842
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigOp.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * A class representing a alter configuration entry containing name, value and operation type.
+ *
+ * The API of this class is evolving, see {@link AdminClient} for details.
+ */
+@InterfaceStability.Evolving
+public class AlterConfigOp {
+
+ public enum OpType {
+ SET((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()))
+ );
+
+ private final byte id;
+
+ OpType(final byte id) {
+ this.id = id;
+ }
+
+ public byte id() {
+ return id;
+ }
+
+ public static OpType forId(final byte id) {
+ return OP_TYPES.get(id);
+ }
+ }
+
+ private final ConfigEntry configEntry;
+ private final OpType opType;
+
+ public AlterConfigOp(ConfigEntry configEntry, OpType operationType) {
+ this.configEntry = configEntry;
+ this.opType = operationType;
+ }
+
+ public ConfigEntry configEntry() {
+ return configEntry;
+ };
+
+ public OpType opType() {
+ return opType;
+ };
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ final AlterConfigOp that = (AlterConfigOp) o;
+ return opType == that.opType &&
+ Objects.equals(configEntry, that.configEntry);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(opType, configEntry);
+ }
+
+ @Override
+ public String toString() {
+ return "AlterConfigOp{" +
+ "opType=" + opType +
+ ", configEntry=" + configEntry +
+ '}';
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 336597f..23d7fd5 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -69,6 +69,10 @@ import org.apache.kafka.common.message.DescribeGroupsRequestData;
import org.apache.kafka.common.message.DescribeGroupsResponseData.DescribedGroup;
import org.apache.kafka.common.message.DescribeGroupsResponseData.DescribedGroupMember;
import org.apache.kafka.common.message.MetadataRequestData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterableConfigSet;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterableConfig;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterConfigsResource;
import org.apache.kafka.common.metrics.JmxReporter;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
@@ -121,6 +125,8 @@ import org.apache.kafka.common.requests.ExpireDelegationTokenRequest;
import org.apache.kafka.common.requests.ExpireDelegationTokenResponse;
import org.apache.kafka.common.requests.FindCoordinatorRequest;
import org.apache.kafka.common.requests.FindCoordinatorResponse;
+import org.apache.kafka.common.requests.IncrementalAlterConfigsRequest;
+import org.apache.kafka.common.requests.IncrementalAlterConfigsResponse;
import org.apache.kafka.common.requests.ListGroupsRequest;
import org.apache.kafka.common.requests.ListGroupsResponse;
import org.apache.kafka.common.requests.MetadataRequest;
@@ -1885,6 +1891,7 @@ public class KafkaAdminClient extends AdminClient {
}
@Override
+ @Deprecated
public AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, final AlterConfigsOptions options) {
final Map<ConfigResource, KafkaFutureImpl<Void>> allFutures = new HashMap<>();
// We must make a separate AlterConfigs request for every BROKER resource we want to alter
@@ -1949,6 +1956,89 @@ public class KafkaAdminClient extends AdminClient {
}
@Override
+ public AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs,
+ final AlterConfigsOptions options) {
+ final Map<ConfigResource, KafkaFutureImpl<Void>> allFutures = new HashMap<>();
+ // We must make a separate AlterConfigs request for every BROKER resource we want to alter
+ // and send the request to that specific broker. Other resources are grouped together into
+ // a single request that may be sent to any broker.
+ final Collection<ConfigResource> unifiedRequestResources = new ArrayList<>();
+
+ for (ConfigResource resource : configs.keySet()) {
+ if (resource.type() == ConfigResource.Type.BROKER && !resource.isDefault()) {
+ NodeProvider nodeProvider = new ConstantNodeIdProvider(Integer.parseInt(resource.name()));
+ allFutures.putAll(incrementalAlterConfigs(configs, options, Collections.singleton(resource), nodeProvider));
+ } else
+ unifiedRequestResources.add(resource);
+ }
+ if (!unifiedRequestResources.isEmpty())
+ allFutures.putAll(incrementalAlterConfigs(configs, options, unifiedRequestResources, new LeastLoadedNodeProvider()));
+
+ return new AlterConfigsResult(new HashMap<>(allFutures));
+ }
+
+ private Map<ConfigResource, KafkaFutureImpl<Void>> incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs,
+ final AlterConfigsOptions options,
+ Collection<ConfigResource> resources,
+ NodeProvider nodeProvider) {
+ final Map<ConfigResource, KafkaFutureImpl<Void>> futures = new HashMap<>();
+ for (ConfigResource resource : resources)
+ futures.put(resource, new KafkaFutureImpl<>());
+
+ final long now = time.milliseconds();
+ runnable.call(new Call("incrementalAlterConfigs", calcDeadlineMs(now, options.timeoutMs()), nodeProvider) {
+
+ @Override
+ public AbstractRequest.Builder createRequest(int timeoutMs) {
+ return new IncrementalAlterConfigsRequest.Builder(
+ toIncrementalAlterConfigsRequestData(resources, configs, options.shouldValidateOnly()));
+ }
+
+ @Override
+ public void handleResponse(AbstractResponse abstractResponse) {
+ IncrementalAlterConfigsResponse response = (IncrementalAlterConfigsResponse) abstractResponse;
+ Map<ConfigResource, ApiError> errors = IncrementalAlterConfigsResponse.fromResponseData(response.data());
+ for (Map.Entry<ConfigResource, KafkaFutureImpl<Void>> entry : futures.entrySet()) {
+ KafkaFutureImpl<Void> future = entry.getValue();
+ ApiException exception = errors.get(entry.getKey()).exception();
+ if (exception != null) {
+ future.completeExceptionally(exception);
+ } else {
+ future.complete(null);
+ }
+ }
+ }
+
+ @Override
+ void handleFailure(Throwable throwable) {
+ completeAllExceptionally(futures.values(), throwable);
+ }
+ }, now);
+ return futures;
+ }
+
+ private IncrementalAlterConfigsRequestData toIncrementalAlterConfigsRequestData(final Collection<ConfigResource> resources,
+ final Map<ConfigResource, Collection<AlterConfigOp>> configs,
+ final boolean validateOnly) {
+ IncrementalAlterConfigsRequestData requestData = new IncrementalAlterConfigsRequestData();
+ requestData.setValidateOnly(validateOnly);
+ for (ConfigResource resource : resources) {
+ AlterableConfigSet alterableConfigSet = new AlterableConfigSet();
+ for (AlterConfigOp configEntry : configs.get(resource))
+ alterableConfigSet.add(new AlterableConfig().
+ setName(configEntry.configEntry().name()).
+ setValue(configEntry.configEntry().value()).
+ setConfigOperation(configEntry.opType().id()));
+
+ AlterConfigsResource alterConfigsResource = new AlterConfigsResource();
+ alterConfigsResource.setResourceType(resource.type().id()).
+ setResourceName(resource.name()).setConfigs(alterableConfigSet);
+ requestData.resources().add(alterConfigsResource);
+ }
+ return requestData;
+ }
+
+ @Override
public AlterReplicaLogDirsResult alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment, final AlterReplicaLogDirsOptions options) {
final Map<TopicPartitionReplica, KafkaFutureImpl<Void>> futures = new HashMap<>(replicaAssignment.size());
diff --git a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
index 1259882..9f28b56 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
@@ -1078,6 +1078,10 @@ public class ConfigDef {
public boolean hasDefault() {
return !NO_DEFAULT_VALUE.equals(this.defaultValue);
}
+
+ public Type type() {
+ return type;
+ }
}
protected List<String> headers() {
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
index 33d6736..5391135 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
@@ -38,6 +38,8 @@ import org.apache.kafka.common.message.SaslAuthenticateRequestData;
import org.apache.kafka.common.message.SaslAuthenticateResponseData;
import org.apache.kafka.common.message.SaslHandshakeRequestData;
import org.apache.kafka.common.message.SaslHandshakeResponseData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData;
import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.SchemaException;
import org.apache.kafka.common.protocol.types.Struct;
@@ -189,7 +191,9 @@ public enum ApiKeys {
DESCRIBE_DELEGATION_TOKEN(41, "DescribeDelegationToken", DescribeDelegationTokenRequest.schemaVersions(), DescribeDelegationTokenResponse.schemaVersions()),
DELETE_GROUPS(42, "DeleteGroups", DeleteGroupsRequest.schemaVersions(), DeleteGroupsResponse.schemaVersions()),
ELECT_PREFERRED_LEADERS(43, "ElectPreferredLeaders", ElectPreferredLeadersRequestData.SCHEMAS,
- ElectPreferredLeadersResponseData.SCHEMAS);
+ ElectPreferredLeadersResponseData.SCHEMAS),
+ INCREMENTAL_ALTER_CONFIGS(44, "IncrementalAlterConfigs", IncrementalAlterConfigsRequestData.SCHEMAS,
+ IncrementalAlterConfigsResponseData.SCHEMAS);
private static final ApiKeys[] ID_TO_TYPE;
private static final int MIN_API_KEY = 0;
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
index c069bc9..c199f8e 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
@@ -231,6 +231,8 @@ public abstract class AbstractRequest extends AbstractRequestResponse {
return new DeleteGroupsRequest(struct, apiVersion);
case ELECT_PREFERRED_LEADERS:
return new ElectPreferredLeadersRequest(struct, apiVersion);
+ case INCREMENTAL_ALTER_CONFIGS:
+ return new IncrementalAlterConfigsRequest(struct, apiVersion);
default:
throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
"code should be updated to do so.", apiKey));
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
index 50ae0b5..c21fa2b 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
@@ -158,6 +158,8 @@ public abstract class AbstractResponse extends AbstractRequestResponse {
return new DeleteGroupsResponse(struct);
case ELECT_PREFERRED_LEADERS:
return new ElectPreferredLeadersResponse(struct, version);
+ case INCREMENTAL_ALTER_CONFIGS:
+ return new IncrementalAlterConfigsResponse(struct, version);
default:
throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
"code should be updated to do so.", apiKey));
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/IncrementalAlterConfigsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/IncrementalAlterConfigsRequest.java
new file mode 100644
index 0000000..3a87cdb
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/requests/IncrementalAlterConfigsRequest.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterConfigsResource;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.AlterConfigsResourceResult;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.types.Struct;
+
+import java.nio.ByteBuffer;
+
+public class IncrementalAlterConfigsRequest extends AbstractRequest {
+
+ public static class Builder extends AbstractRequest.Builder<IncrementalAlterConfigsRequest> {
+ private final IncrementalAlterConfigsRequestData data;
+
+ public Builder(IncrementalAlterConfigsRequestData data) {
+ super(ApiKeys.INCREMENTAL_ALTER_CONFIGS);
+ this.data = data;
+ }
+
+ @Override
+ public IncrementalAlterConfigsRequest build(short version) {
+ return new IncrementalAlterConfigsRequest(data, version);
+ }
+
+ @Override
+ public String toString() {
+ return data.toString();
+ }
+ }
+
+ private final IncrementalAlterConfigsRequestData data;
+ private final short version;
+
+ private IncrementalAlterConfigsRequest(IncrementalAlterConfigsRequestData data, short version) {
+ super(ApiKeys.INCREMENTAL_ALTER_CONFIGS, version);
+ this.data = data;
+ this.version = version;
+ }
+
+ IncrementalAlterConfigsRequest(final Struct struct, final short version) {
+ super(ApiKeys.INCREMENTAL_ALTER_CONFIGS, version);
+ this.data = new IncrementalAlterConfigsRequestData(struct, version);
+ this.version = version;
+ }
+
+ public static IncrementalAlterConfigsRequest parse(ByteBuffer buffer, short version) {
+ return new IncrementalAlterConfigsRequest(ApiKeys.INCREMENTAL_ALTER_CONFIGS.parseRequest(version, buffer), version);
+ }
+
+ public IncrementalAlterConfigsRequestData data() {
+ return data;
+ }
+
+ @Override
+ protected Struct toStruct() {
+ return data.toStruct(version);
+ }
+
+ @Override
+ public AbstractResponse getErrorResponse(final int throttleTimeMs, final Throwable e) {
+ IncrementalAlterConfigsResponseData response = new IncrementalAlterConfigsResponseData();
+ ApiError apiError = ApiError.fromThrowable(e);
+ for (AlterConfigsResource resource : data.resources()) {
+ response.responses().add(new AlterConfigsResourceResult()
+ .setResourceName(resource.resourceName())
+ .setResourceType(resource.resourceType())
+ .setErrorCode(apiError.error().code())
+ .setErrorMessage(apiError.message()));
+ }
+ return new IncrementalAlterConfigsResponse(response);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/IncrementalAlterConfigsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/IncrementalAlterConfigsResponse.java
new file mode 100644
index 0000000..1e5aea1
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/requests/IncrementalAlterConfigsResponse.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.AlterConfigsResourceResult;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.types.Struct;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+
+public class IncrementalAlterConfigsResponse extends AbstractResponse {
+
+ public static IncrementalAlterConfigsResponseData toResponseData(final int requestThrottleMs,
+ final Map<ConfigResource, ApiError> results) {
+ IncrementalAlterConfigsResponseData responseData = new IncrementalAlterConfigsResponseData();
+ responseData.setThrottleTimeMs(requestThrottleMs);
+ for (Map.Entry<ConfigResource, ApiError> entry : results.entrySet()) {
+ responseData.responses().add(new AlterConfigsResourceResult().
+ setResourceName(entry.getKey().name()).
+ setResourceType(entry.getKey().type().id()).
+ setErrorCode(entry.getValue().error().code()).
+ setErrorMessage(entry.getValue().message()));
+ }
+ return responseData;
+ }
+
+ public static Map<ConfigResource, ApiError> fromResponseData(final IncrementalAlterConfigsResponseData data) {
+ Map<ConfigResource, ApiError> map = new HashMap<>();
+ for (AlterConfigsResourceResult result : data.responses()) {
+ map.put(new ConfigResource(ConfigResource.Type.forId(result.resourceType()), result.resourceName()),
+ new ApiError(Errors.forCode(result.errorCode()), result.errorMessage()));
+ }
+ return map;
+ }
+
+ private final IncrementalAlterConfigsResponseData data;
+
+ public IncrementalAlterConfigsResponse(IncrementalAlterConfigsResponseData data) {
+ this.data = data;
+ }
+
+ public IncrementalAlterConfigsResponse(final Struct struct, final short version) {
+ this.data = new IncrementalAlterConfigsResponseData(struct, version);
+ }
+
+ public IncrementalAlterConfigsResponseData data() {
+ return data;
+ }
+
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ HashMap<Errors, Integer> counts = new HashMap<>();
+ for (AlterConfigsResourceResult result : data.responses()) {
+ Errors error = Errors.forCode(result.errorCode());
+ counts.put(error, counts.getOrDefault(error, 0) + 1);
+ }
+ return counts;
+ }
+
+ @Override
+ protected Struct toStruct(final short version) {
+ return data.toStruct(version);
+ }
+
+ @Override
+ public boolean shouldClientThrottle(short version) {
+ return version >= 0;
+ }
+
+ @Override
+ public int throttleTimeMs() {
+ return data.throttleTimeMs();
+ }
+
+ public static IncrementalAlterConfigsResponse parse(ByteBuffer buffer, short version) {
+ return new IncrementalAlterConfigsResponse(
+ ApiKeys.INCREMENTAL_ALTER_CONFIGS.responseSchema(version).read(buffer), version);
+ }
+}
diff --git a/clients/src/main/resources/common/message/IncrementalAlterConfigsRequest.json b/clients/src/main/resources/common/message/IncrementalAlterConfigsRequest.json
new file mode 100644
index 0000000..d808c04
--- /dev/null
+++ b/clients/src/main/resources/common/message/IncrementalAlterConfigsRequest.json
@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+ "apiKey": 44,
+ "type": "request",
+ "name": "IncrementalAlterConfigsRequest",
+ "validVersions": "0",
+ "fields": [
+ { "name": "Resources", "type": "[]AlterConfigsResource", "versions": "0+",
+ "about": "The incremental updates for each resource.", "fields": [
+ { "name": "ResourceType", "type": "int8", "versions": "0+", "mapKey": true,
+ "about": "The resource type." },
+ { "name": "ResourceName", "type": "string", "versions": "0+", "mapKey": true,
+ "about": "The resource name." },
+ { "name": "Configs", "type": "[]AlterableConfig", "versions": "0+",
+ "about": "The configurations.", "fields": [
+ { "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
+ "about": "The configuration key name." },
+ { "name": "ConfigOperation", "type": "int8", "versions": "0+", "mapKey": true,
+ "about": "The type (Set, Delete, Append, Subtract) of operation." },
+ { "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+",
+ "about": "The value to set for the configuration key."}
+ ]}
+ ]},
+ { "name": "ValidateOnly", "type": "bool", "versions": "0+",
+ "about": "True if we should validate the request, but not change the configurations."}
+ ]
+}
diff --git a/clients/src/main/resources/common/message/IncrementalAlterConfigsResponse.json b/clients/src/main/resources/common/message/IncrementalAlterConfigsResponse.json
new file mode 100644
index 0000000..71aa997
--- /dev/null
+++ b/clients/src/main/resources/common/message/IncrementalAlterConfigsResponse.json
@@ -0,0 +1,36 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+ "apiKey": 44,
+ "type": "response",
+ "name": "IncrementalAlterConfigsResponse",
+ "validVersions": "0",
+ "fields": [
+ { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
+ "about": "Duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+ { "name": "responses", "type": "[]AlterConfigsResourceResult", "versions": "0+",
+ "about": "The responses for each resource.", "fields": [
+ { "name": "ErrorCode", "type": "int16", "versions": "0+",
+ "about": "The resource error code." },
+ { "name": "ErrorMessage", "type": "string", "nullableVersions": "0+", "versions": "0+",
+ "about": "The resource error message, or null if there was no error." },
+ { "name": "ResourceType", "type": "int8", "versions": "0+",
+ "about": "The resource type." },
+ { "name": "ResourceName", "type": "string", "versions": "0+",
+ "about": "The resource name." }
+ ]}
+ ]
+}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index 220fc50..1367b94 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -59,6 +59,8 @@ import org.apache.kafka.common.message.DescribeGroupsResponseData;
import org.apache.kafka.common.message.ElectPreferredLeadersResponseData;
import org.apache.kafka.common.message.ElectPreferredLeadersResponseData.PartitionResult;
import org.apache.kafka.common.message.ElectPreferredLeadersResponseData.ReplicaElectionResult;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.AlterConfigsResourceResult;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.ApiError;
import org.apache.kafka.common.requests.CreateAclsResponse;
@@ -78,6 +80,7 @@ import org.apache.kafka.common.requests.DescribeConfigsResponse;
import org.apache.kafka.common.requests.DescribeGroupsResponse;
import org.apache.kafka.common.requests.ElectPreferredLeadersResponse;
import org.apache.kafka.common.requests.FindCoordinatorResponse;
+import org.apache.kafka.common.requests.IncrementalAlterConfigsResponse;
import org.apache.kafka.common.requests.ListGroupsResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
@@ -1203,6 +1206,59 @@ public class KafkaAdminClientTest {
}
}
+ @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 AlterConfigsResourceResult()
+ .setResourceName("")
+ .setResourceType(ConfigResource.Type.BROKER.id())
+ .setErrorCode(Errors.CLUSTER_AUTHORIZATION_FAILED.code())
+ .setErrorMessage("authorization error"));
+
+ responseData.responses().add(new AlterConfigsResourceResult()
+ .setResourceName("topic1")
+ .setResourceType(ConfigResource.Type.TOPIC.id())
+ .setErrorCode(Errors.INVALID_REQUEST.code())
+ .setErrorMessage("Config value append is not allowed for config"));
+
+ env.kafkaClient().prepareResponse(new IncrementalAlterConfigsResponse(responseData));
+
+ ConfigResource brokerResource = new ConfigResource(ConfigResource.Type.BROKER, "");
+ ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, "topic1");
+
+ 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);
+
+ final Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
+ configs.put(brokerResource, Collections.singletonList(alterConfigOp1));
+ configs.put(topicResource, Collections.singletonList(alterConfigOp2));
+
+ AlterConfigsResult result = env.adminClient().incrementalAlterConfigs(configs);
+ TestUtils.assertFutureError(result.values().get(brokerResource), ClusterAuthorizationException.class);
+ TestUtils.assertFutureError(result.values().get(topicResource), InvalidRequestException.class);
+
+ // Test a call where there are no errors.
+ responseData = new IncrementalAlterConfigsResponseData();
+ responseData.responses().add(new AlterConfigsResourceResult()
+ .setResourceName("")
+ .setResourceType(ConfigResource.Type.BROKER.id())
+ .setErrorCode(Errors.NONE.code())
+ .setErrorMessage(ApiError.NONE.message()));
+
+ env.kafkaClient().prepareResponse(new IncrementalAlterConfigsResponse(responseData));
+ env.adminClient().incrementalAlterConfigs(Collections.singletonMap(brokerResource, asList(alterConfigOp1))).all().get();
+ }
+ }
+
@SafeVarargs
private static <T> void assertCollectionIs(Collection<T> collection, T... elements) {
for (T element : elements) {
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
index b669a32..9709372 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
@@ -375,11 +375,18 @@ public class MockAdminClient extends AdminClient {
}
@Override
+ @Deprecated
public AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, AlterConfigsOptions options) {
throw new UnsupportedOperationException("Not implemented yet");
}
@Override
+ public AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs,
+ AlterConfigsOptions options) {
+ throw new UnsupportedOperationException("Not implemented yet");
+ }
+
+ @Override
public AlterReplicaLogDirsResult alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment, AlterReplicaLogDirsOptions options) {
throw new UnsupportedOperationException("Not implemented yet");
}
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index ca695c7..e595195 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -62,6 +62,12 @@ import org.apache.kafka.common.message.SaslAuthenticateRequestData;
import org.apache.kafka.common.message.SaslAuthenticateResponseData;
import org.apache.kafka.common.message.SaslHandshakeRequestData;
import org.apache.kafka.common.message.SaslHandshakeResponseData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterConfigsResource;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterableConfigSet;
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterableConfig;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData;
+import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.AlterConfigsResourceResult;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.network.Send;
import org.apache.kafka.common.protocol.ApiKeys;
@@ -338,6 +344,9 @@ public class RequestResponseTest {
checkRequest(createElectPreferredLeadersRequestNullPartitions());
checkErrorResponse(createElectPreferredLeadersRequest(), new UnknownServerException());
checkResponse(createElectPreferredLeadersResponse(), 0);
+ checkRequest(createIncrementalAlterConfigsRequest());
+ checkErrorResponse(createIncrementalAlterConfigsRequest(), new UnknownServerException());
+ checkResponse(createIncrementalAlterConfigsResponse(), 0);
}
@Test
@@ -1477,4 +1486,30 @@ public class RequestResponseTest {
return new ElectPreferredLeadersResponse(data);
}
+ private IncrementalAlterConfigsRequest createIncrementalAlterConfigsRequest() {
+ IncrementalAlterConfigsRequestData data = new IncrementalAlterConfigsRequestData();
+ AlterableConfig alterableConfig = new AlterableConfig()
+ .setName("retention.ms")
+ .setConfigOperation((byte) 0)
+ .setValue("100");
+ AlterableConfigSet alterableConfigs = new AlterableConfigSet();
+ alterableConfigs.add(alterableConfig);
+
+ data.resources().add(new AlterConfigsResource()
+ .setResourceName("testtopic")
+ .setResourceType(ResourceType.TOPIC.code())
+ .setConfigs(alterableConfigs));
+ return new IncrementalAlterConfigsRequest.Builder(data).build((short) 0);
+ }
+
+ private IncrementalAlterConfigsResponse createIncrementalAlterConfigsResponse() {
+ IncrementalAlterConfigsResponseData data = new IncrementalAlterConfigsResponseData();
+
+ data.responses().add(new AlterConfigsResourceResult()
+ .setResourceName("testtopic")
+ .setResourceType(ResourceType.TOPIC.code())
+ .setErrorCode(Errors.INVALID_REQUEST.code())
+ .setErrorMessage("Duplicate Keys"));
+ return new IncrementalAlterConfigsResponse(data);
+ }
}
diff --git a/core/src/main/scala/kafka/log/LogConfig.scala b/core/src/main/scala/kafka/log/LogConfig.scala
index ab58949..c1e5c62 100755
--- a/core/src/main/scala/kafka/log/LogConfig.scala
+++ b/core/src/main/scala/kafka/log/LogConfig.scala
@@ -297,6 +297,8 @@ object LogConfig {
throw new InvalidConfigurationException(s"Unknown topic config name: $name")
}
+ private[kafka] def configKeys: Map[String, ConfigKey] = configDef.configKeys.asScala
+
/**
* Check that the given properties contain only valid log config names and that all values can be parsed and are valid
*/
diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index 0cdaad6..d424700 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -24,6 +24,9 @@ import kafka.log.LogConfig
import kafka.metrics.KafkaMetricsGroup
import kafka.utils._
import kafka.zk.{AdminZkClient, KafkaZkClient}
+import org.apache.kafka.clients.admin.AlterConfigOp
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
+import org.apache.kafka.common.config.ConfigDef.ConfigKey
import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, ConfigResource}
import org.apache.kafka.common.errors.{ApiException, InvalidConfigurationException, InvalidPartitionsException, InvalidReplicaAssignmentException, InvalidRequestException, ReassignmentInProgressException, UnknownTopicOrPartitionException}
import org.apache.kafka.common.internals.Topic
@@ -38,7 +41,7 @@ import org.apache.kafka.common.requests.{AlterConfigsRequest, ApiError, Describe
import org.apache.kafka.server.policy.{AlterConfigPolicy, CreateTopicPolicy}
import org.apache.kafka.server.policy.CreateTopicPolicy.RequestMetadata
-import scala.collection.{mutable, _}
+import scala.collection.{Map, mutable, _}
import scala.collection.JavaConverters._
class AdminManager(val config: KafkaConfig,
@@ -364,59 +367,113 @@ class AdminManager(val config: KafkaConfig,
def alterConfigs(configs: Map[ConfigResource, AlterConfigsRequest.Config], validateOnly: Boolean): Map[ConfigResource, ApiError] = {
configs.map { case (resource, config) =>
- def validateConfigPolicy(resourceType: ConfigResource.Type): Unit = {
- alterConfigPolicy match {
- case Some(policy) =>
- val configEntriesMap = config.entries.asScala.map(entry => (entry.name, entry.value)).toMap
- policy.validate(new AlterConfigPolicy.RequestMetadata(
- new ConfigResource(resourceType, resource.name), configEntriesMap.asJava))
- case None =>
- }
- }
try {
+ val configEntriesMap = config.entries.asScala.map(entry => (entry.name, entry.value)).toMap
+
+ val configProps = new Properties
+ config.entries.asScala.foreach { configEntry =>
+ configProps.setProperty(configEntry.name, configEntry.value)
+ }
+
resource.`type` match {
- case ConfigResource.Type.TOPIC =>
- val topic = resource.name
+ case ConfigResource.Type.TOPIC => alterTopicConfigs(resource, validateOnly, configProps, configEntriesMap)
+ case ConfigResource.Type.BROKER => alterBrokerConfigs(resource, validateOnly, configProps, configEntriesMap)
+ case resourceType =>
+ throw new InvalidRequestException(s"AlterConfigs is only supported for topics and brokers, but resource type is $resourceType")
+ }
+ } catch {
+ case e @ (_: ConfigException | _: IllegalArgumentException) =>
+ val message = s"Invalid config value for resource $resource: ${e.getMessage}"
+ info(message)
+ resource -> ApiError.fromThrowable(new InvalidRequestException(message, e))
+ case e: Throwable =>
+ // Log client errors at a lower level than unexpected exceptions
+ val message = s"Error processing alter configs request for resource $resource, config $config"
+ if (e.isInstanceOf[ApiException])
+ info(message, e)
+ else
+ error(message, e)
+ resource -> ApiError.fromThrowable(e)
+ }
+ }.toMap
+ }
- val properties = new Properties
- config.entries.asScala.foreach { configEntry =>
- properties.setProperty(configEntry.name, configEntry.value)
- }
+ private def alterTopicConfigs(resource: ConfigResource, validateOnly: Boolean,
+ configProps: Properties, configEntriesMap: Map[String, String]): (ConfigResource, ApiError) = {
+ val topic = resource.name
+ adminZkClient.validateTopicConfig(topic, configProps)
+ validateConfigPolicy(resource, configEntriesMap)
+ if (!validateOnly) {
+ info(s"Updating topic $topic with new configuration $config")
+ adminZkClient.changeTopicConfig(topic, configProps)
+ }
- adminZkClient.validateTopicConfig(topic, properties)
- validateConfigPolicy(ConfigResource.Type.TOPIC)
- if (!validateOnly) {
- info(s"Updating topic $topic with new configuration $config")
- adminZkClient.changeTopicConfig(topic, properties)
- }
+ resource -> ApiError.NONE
+ }
- resource -> ApiError.NONE
+ private def alterBrokerConfigs(resource: ConfigResource, validateOnly: Boolean,
+ configProps: Properties, configEntriesMap: Map[String, String]): (ConfigResource, ApiError) = {
+ val brokerId = getBrokerId(resource)
+ val perBrokerConfig = brokerId.nonEmpty
+ this.config.dynamicConfig.validate(configProps, perBrokerConfig)
+ validateConfigPolicy(resource, configEntriesMap)
+ if (!validateOnly) {
+ if (perBrokerConfig)
+ this.config.dynamicConfig.reloadUpdatedFilesWithoutConfigChange(configProps)
+ adminZkClient.changeBrokerConfig(brokerId,
+ this.config.dynamicConfig.toPersistentProps(configProps, perBrokerConfig))
+ }
- case ConfigResource.Type.BROKER =>
- val brokerId = if (resource.name == null || resource.name.isEmpty)
- None
- else {
- val id = resourceNameToBrokerId(resource.name)
- if (id != this.config.brokerId)
- throw new InvalidRequestException(s"Unexpected broker id, expected ${this.config.brokerId}, but received $resource.name")
- Some(id)
- }
- val configProps = new Properties
- config.entries.asScala.foreach { configEntry =>
- configProps.setProperty(configEntry.name, configEntry.value)
- }
+ resource -> ApiError.NONE
+ }
+
+ private def getBrokerId(resource: ConfigResource) = {
+ if (resource.name == null || resource.name.isEmpty)
+ None
+ else {
+ val id = resourceNameToBrokerId(resource.name)
+ if (id != this.config.brokerId)
+ throw new InvalidRequestException(s"Unexpected broker id, expected ${this.config.brokerId}, but received $resource.name")
+ Some(id)
+ }
+ }
+
+ private def validateConfigPolicy(resource: ConfigResource, configEntriesMap: Map[String, String]): Unit = {
+ alterConfigPolicy match {
+ case Some(policy) =>
+ policy.validate(new AlterConfigPolicy.RequestMetadata(
+ new ConfigResource(resource.`type`(), resource.name), configEntriesMap.asJava))
+ case None =>
+ }
+ }
+ def incrementalAlterConfigs(configs: Map[ConfigResource, List[AlterConfigOp]], validateOnly: Boolean): Map[ConfigResource, ApiError] = {
+ configs.map { case (resource, alterConfigOps) =>
+ try {
+ //throw InvalidRequestException if any duplicate keys
+ val duplicateKeys = alterConfigOps.groupBy(config => config.configEntry().name())
+ .mapValues(_.size).filter(_._2 > 1).keys.toSet
+ if (duplicateKeys.nonEmpty)
+ throw new InvalidRequestException(s"Error due to duplicate config keys : ${duplicateKeys.mkString(",")}")
+
+ val configEntriesMap = alterConfigOps.map(entry => (entry.configEntry().name(), entry.configEntry().value())).toMap
+
+ resource.`type` match {
+ case ConfigResource.Type.TOPIC =>
+ val configProps = adminZkClient.fetchEntityConfig(ConfigType.Topic, resource.name)
+ prepareIncrementalConfigs(alterConfigOps, configProps, LogConfig.configKeys)
+ alterTopicConfigs(resource, validateOnly, configProps, configEntriesMap)
+
+ case ConfigResource.Type.BROKER =>
+ val brokerId = getBrokerId(resource)
val perBrokerConfig = brokerId.nonEmpty
- this.config.dynamicConfig.validate(configProps, perBrokerConfig)
- validateConfigPolicy(ConfigResource.Type.BROKER)
- if (!validateOnly) {
- if (perBrokerConfig)
- this.config.dynamicConfig.reloadUpdatedFilesWithoutConfigChange(configProps)
- adminZkClient.changeBrokerConfig(brokerId,
- this.config.dynamicConfig.toPersistentProps(configProps, perBrokerConfig))
- }
- resource -> ApiError.NONE
+ val persistentProps = if (perBrokerConfig) adminZkClient.fetchEntityConfig(ConfigType.Broker, brokerId.get.toString)
+ else adminZkClient.fetchEntityConfig(ConfigType.Broker, ConfigEntityName.Default)
+
+ val configProps = this.config.dynamicConfig.fromPersistentProps(persistentProps, perBrokerConfig)
+ prepareIncrementalConfigs(alterConfigOps, configProps, KafkaConfig.configKeys)
+ alterBrokerConfigs(resource, validateOnly, configProps, configEntriesMap)
case resourceType =>
throw new InvalidRequestException(s"AlterConfigs is only supported for topics and brokers, but resource type is $resourceType")
}
@@ -427,7 +484,7 @@ class AdminManager(val config: KafkaConfig,
resource -> ApiError.fromThrowable(new InvalidRequestException(message, e))
case e: Throwable =>
// Log client errors at a lower level than unexpected exceptions
- val message = s"Error processing alter configs request for resource $resource, config $config"
+ val message = s"Error processing alter configs request for resource $resource, config $alterConfigOps"
if (e.isInstanceOf[ApiException])
info(message, e)
else
@@ -437,6 +494,37 @@ class AdminManager(val config: KafkaConfig,
}.toMap
}
+ private def prepareIncrementalConfigs(alterConfigOps: List[AlterConfigOp], configProps: Properties, configKeys: Map[String, ConfigKey]): Unit = {
+
+ def listType(configName: String, configKeys: Map[String, ConfigKey]): Boolean = {
+ val configKey = configKeys(configName)
+ if (configKey == null)
+ throw new InvalidConfigurationException(s"Unknown topic config name: $configName")
+ configKey.`type` == ConfigDef.Type.LIST
+ }
+
+ alterConfigOps.foreach { alterConfigOp =>
+ alterConfigOp.opType() match {
+ case OpType.SET => configProps.setProperty(alterConfigOp.configEntry().name(), alterConfigOp.configEntry().value())
+ case OpType.DELETE => configProps.remove(alterConfigOp.configEntry().name())
+ case OpType.APPEND => {
+ if (!listType(alterConfigOp.configEntry().name(), configKeys))
+ throw new InvalidRequestException(s"Config value append is not allowed for config key: ${alterConfigOp.configEntry().name()}")
+ val oldValueList = configProps.getProperty(alterConfigOp.configEntry().name()).split(",").toList
+ val newValueList = oldValueList ::: alterConfigOp.configEntry().value().split(",").toList
+ configProps.setProperty(alterConfigOp.configEntry().name(), newValueList.mkString(","))
+ }
+ case OpType.SUBTRACT => {
+ if (!listType(alterConfigOp.configEntry().name(), configKeys))
+ throw new InvalidRequestException(s"Config value subtract is not allowed for config key: ${alterConfigOp.configEntry().name()}")
+ val oldValueList = configProps.getProperty(alterConfigOp.configEntry().name()).split(",").toList
+ val newValueList = oldValueList.diff(alterConfigOp.configEntry().value().split(",").toList)
+ configProps.setProperty(alterConfigOp.configEntry().name(), newValueList.mkString(","))
+ }
+ }
+ }
+ }
+
def shutdown() {
topicPurgatory.shutdown()
CoreUtils.swallow(createTopicPolicy.foreach(_.close()), this)
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 140fdd4..56adc9b 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -39,6 +39,8 @@ import kafka.security.auth.{Resource, _}
import kafka.server.QuotaFactory.{QuotaManagers, UnboundedQuota}
import kafka.utils.{CoreUtils, Logging}
import kafka.zk.{AdminZkClient, KafkaZkClient}
+import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry}
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
import org.apache.kafka.common.acl.{AccessControlEntry, AclBinding}
import org.apache.kafka.common.config.ConfigResource
import org.apache.kafka.common.errors._
@@ -151,6 +153,7 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.DESCRIBE_DELEGATION_TOKEN => handleDescribeTokensRequest(request)
case ApiKeys.DELETE_GROUPS => handleDeleteGroupsRequest(request)
case ApiKeys.ELECT_PREFERRED_LEADERS => handleElectPreferredReplicaLeader(request)
+ case ApiKeys.INCREMENTAL_ALTER_CONFIGS => handleIncrementalAlterConfigsRequest(request)
}
} catch {
case e: FatalExitError => throw e
@@ -2155,6 +2158,34 @@ class KafkaApis(val requestChannel: RequestChannel,
new ApiError(error, null)
}
+ def handleIncrementalAlterConfigsRequest(request: RequestChannel.Request): Unit = {
+ val alterConfigsRequest = request.body[IncrementalAlterConfigsRequest]
+
+ val configs = alterConfigsRequest.data().resources().iterator().asScala.map { alterConfigResource =>
+ val configResource = new ConfigResource(ConfigResource.Type.forId(alterConfigResource.resourceType()), alterConfigResource.resourceName())
+ configResource -> alterConfigResource.configs().iterator().asScala.map {
+ alterConfig => new AlterConfigOp(new ConfigEntry(alterConfig.name(), alterConfig.value()), OpType.forId(alterConfig.configOperation())) }.toList
+ }.toMap
+
+ val (authorizedResources, unauthorizedResources) = configs.partition { case (resource, _) =>
+ resource.`type` match {
+ case ConfigResource.Type.BROKER =>
+ authorize(request.session, AlterConfigs, Resource.ClusterResource)
+ case ConfigResource.Type.TOPIC =>
+ authorize(request.session, AlterConfigs, Resource(Topic, resource.name, LITERAL))
+ case rt => throw new InvalidRequestException(s"Unexpected resource type $rt")
+ }
+ }
+
+ val authorizedResult = adminManager.incrementalAlterConfigs(authorizedResources, alterConfigsRequest.data().validateOnly())
+ val unauthorizedResult = unauthorizedResources.keys.map { resource =>
+ resource -> configsAuthorizationApiError(request.session, resource)
+ }
+ sendResponseMaybeThrottle(request, requestThrottleMs =>
+ new IncrementalAlterConfigsResponse(IncrementalAlterConfigsResponse.toResponseData(requestThrottleMs,
+ (authorizedResult ++ unauthorizedResult).asJava)))
+ }
+
def handleDescribeConfigsRequest(request: RequestChannel.Request): Unit = {
val describeConfigsRequest = request.body[DescribeConfigsRequest]
val (authorizedResources, unauthorizedResources) = describeConfigsRequest.resources.asScala.partition { resource =>
diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
index dbb6213..9e35f40 100644
--- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
@@ -1406,6 +1406,184 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
assertEquals(2, currentLeader(partition1))
assertEquals(2, currentLeader(partition2))
}
+
+ @Test
+ def testValidIncrementalAlterConfigs(): Unit = {
+ client = AdminClient.create(createConfig)
+
+ // Create topics
+ val topic1 = "incremental-alter-configs-topic-1"
+ val topic1Resource = new ConfigResource(ConfigResource.Type.TOPIC, topic1)
+ val topic1CreateConfigs = new Properties
+ topic1CreateConfigs.setProperty(LogConfig.RetentionMsProp, "60000000")
+ topic1CreateConfigs.setProperty(LogConfig.CleanupPolicyProp, LogConfig.Compact)
+ createTopic(topic1, numPartitions = 1, replicationFactor = 1, topic1CreateConfigs)
+
+ val topic2 = "incremental-alter-configs-topic-2"
+ val topic2Resource = new ConfigResource(ConfigResource.Type.TOPIC, topic2)
+ createTopic(topic2)
+
+ // Alter topic configs
+ var topic1AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.FlushMsProp, "1000"), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry(LogConfig.CleanupPolicyProp, LogConfig.Delete), AlterConfigOp.OpType.APPEND),
+ new AlterConfigOp(new ConfigEntry(LogConfig.RetentionMsProp, ""), AlterConfigOp.OpType.DELETE)
+ ).asJavaCollection
+
+ val topic2AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.MinCleanableDirtyRatioProp, "0.9"), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry(LogConfig.CompressionTypeProp, "lz4"), AlterConfigOp.OpType.SET)
+ ).asJavaCollection
+
+ var alterResult = client.incrementalAlterConfigs(Map(
+ topic1Resource -> topic1AlterConfigs,
+ topic2Resource -> topic2AlterConfigs
+ ).asJava)
+
+ assertEquals(Set(topic1Resource, topic2Resource).asJava, alterResult.values.keySet)
+ alterResult.all.get
+
+ // Verify that topics were updated correctly
+ var describeResult = client.describeConfigs(Seq(topic1Resource, topic2Resource).asJava)
+ var configs = describeResult.all.get
+
+ assertEquals(2, configs.size)
+
+ assertEquals("1000", configs.get(topic1Resource).get(LogConfig.FlushMsProp).value)
+ assertEquals("compact,delete", configs.get(topic1Resource).get(LogConfig.CleanupPolicyProp).value)
+ assertEquals((Defaults.LogRetentionHours * 60 * 60 * 1000).toString, configs.get(topic1Resource).get(LogConfig.RetentionMsProp).value)
+
+ assertEquals("0.9", configs.get(topic2Resource).get(LogConfig.MinCleanableDirtyRatioProp).value)
+ assertEquals("lz4", configs.get(topic2Resource).get(LogConfig.CompressionTypeProp).value)
+
+ //verify subtract operation
+ topic1AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.CleanupPolicyProp, LogConfig.Compact), AlterConfigOp.OpType.SUBTRACT)
+ ).asJava
+
+ alterResult = client.incrementalAlterConfigs(Map(
+ topic1Resource -> topic1AlterConfigs
+ ).asJava)
+ alterResult.all.get
+
+ // Verify that topics were updated correctly
+ describeResult = client.describeConfigs(Seq(topic1Resource).asJava)
+ configs = describeResult.all.get
+
+ assertEquals("delete", configs.get(topic1Resource).get(LogConfig.CleanupPolicyProp).value)
+ assertEquals("1000", configs.get(topic1Resource).get(LogConfig.FlushMsProp).value) // verify previous change is still intact
+
+ // Alter topics with validateOnly=true
+ topic1AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.CleanupPolicyProp, LogConfig.Compact), AlterConfigOp.OpType.APPEND)
+ ).asJava
+
+ alterResult = client.incrementalAlterConfigs(Map(
+ topic1Resource -> topic1AlterConfigs
+ ).asJava, new AlterConfigsOptions().validateOnly(true))
+ alterResult.all.get
+
+ // Verify that topics were not updated due to validateOnly = true
+ describeResult = client.describeConfigs(Seq(topic1Resource).asJava)
+ configs = describeResult.all.get
+
+ assertEquals("delete", configs.get(topic1Resource).get(LogConfig.CleanupPolicyProp).value)
+
+ //Alter topics with validateOnly=true with invalid configs
+ topic1AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.CompressionTypeProp, "zip"), AlterConfigOp.OpType.SET)
+ ).asJava
+
+ alterResult = client.incrementalAlterConfigs(Map(
+ topic1Resource -> topic1AlterConfigs
+ ).asJava, new AlterConfigsOptions().validateOnly(true))
+
+ assertFutureExceptionTypeEquals(alterResult.values().get(topic1Resource), classOf[InvalidRequestException],
+ Some("Invalid config value for resource"))
+ }
+
+ @Test
+ def testInvalidIncrementalAlterConfigs(): Unit = {
+ client = AdminClient.create(createConfig)
+
+ // Create topics
+ val topic1 = "incremental-alter-configs-topic-1"
+ val topic1Resource = new ConfigResource(ConfigResource.Type.TOPIC, topic1)
+ createTopic(topic1)
+
+ val topic2 = "incremental-alter-configs-topic-2"
+ val topic2Resource = new ConfigResource(ConfigResource.Type.TOPIC, topic2)
+ createTopic(topic2)
+
+ //Add duplicate Keys for topic1
+ var topic1AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.MinCleanableDirtyRatioProp, "0.75"), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry(LogConfig.MinCleanableDirtyRatioProp, "0.65"), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry(LogConfig.CompressionTypeProp, "gzip"), AlterConfigOp.OpType.SET) // valid entry
+ ).asJavaCollection
+
+ //Add valid config for topic2
+ var topic2AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.MinCleanableDirtyRatioProp, "0.9"), AlterConfigOp.OpType.SET)
+ ).asJavaCollection
+
+ var alterResult = client.incrementalAlterConfigs(Map(
+ topic1Resource -> topic1AlterConfigs,
+ topic2Resource -> topic2AlterConfigs
+ ).asJava)
+ assertEquals(Set(topic1Resource, topic2Resource).asJava, alterResult.values.keySet)
+
+ //InvalidRequestException error for topic1
+ assertFutureExceptionTypeEquals(alterResult.values().get(topic1Resource), classOf[InvalidRequestException],
+ Some("Error due to duplicate config keys"))
+
+ //operation should succeed for topic2
+ alterResult.values().get(topic2Resource).get()
+
+ // Verify that topic1 is not config not updated, and topic2 config is updated
+ val describeResult = client.describeConfigs(Seq(topic1Resource, topic2Resource).asJava)
+ val configs = describeResult.all.get
+ assertEquals(2, configs.size)
+
+ assertEquals(Defaults.LogCleanerMinCleanRatio.toString, configs.get(topic1Resource).get(LogConfig.MinCleanableDirtyRatioProp).value)
+ assertEquals(Defaults.CompressionType.toString, configs.get(topic1Resource).get(LogConfig.CompressionTypeProp).value)
+ assertEquals("0.9", configs.get(topic2Resource).get(LogConfig.MinCleanableDirtyRatioProp).value)
+
+ //check invalid use of append/subtract operation types
+ topic1AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.CompressionTypeProp, "gzip"), AlterConfigOp.OpType.APPEND)
+ ).asJavaCollection
+
+ topic2AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.CompressionTypeProp, "snappy"), AlterConfigOp.OpType.SUBTRACT)
+ ).asJavaCollection
+
+ alterResult = client.incrementalAlterConfigs(Map(
+ topic1Resource -> topic1AlterConfigs,
+ topic2Resource -> topic2AlterConfigs
+ ).asJava)
+ assertEquals(Set(topic1Resource, topic2Resource).asJava, alterResult.values.keySet)
+
+ assertFutureExceptionTypeEquals(alterResult.values().get(topic1Resource), classOf[InvalidRequestException],
+ Some("Config value append is not allowed for config"))
+
+ assertFutureExceptionTypeEquals(alterResult.values().get(topic2Resource), classOf[InvalidRequestException],
+ Some("Config value subtract is not allowed for config"))
+
+
+ //try to add invalid config
+ topic1AlterConfigs = Seq(
+ new AlterConfigOp(new ConfigEntry(LogConfig.MinCleanableDirtyRatioProp, "1.1"), AlterConfigOp.OpType.SET)
+ ).asJavaCollection
+
+ alterResult = client.incrementalAlterConfigs(Map(
+ topic1Resource -> topic1AlterConfigs
+ ).asJava)
+ assertEquals(Set(topic1Resource).asJava, alterResult.values.keySet)
+
+ assertFutureExceptionTypeEquals(alterResult.values().get(topic1Resource), classOf[InvalidRequestException],
+ Some("Invalid config value for resource"))
+ }
}
object AdminClientIntegrationTest {
diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
index 9b15389..0bc1045 100644
--- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
@@ -25,7 +25,7 @@ import kafka.network.SocketServer
import kafka.security.auth._
import kafka.server.{BaseRequestTest, KafkaConfig}
import kafka.utils.TestUtils
-import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig}
+import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig, AlterConfigOp}
import org.apache.kafka.clients.consumer._
import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener
import org.apache.kafka.clients.producer._
@@ -33,8 +33,9 @@ import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter
import org.apache.kafka.common.config.ConfigResource
import org.apache.kafka.common.errors._
import org.apache.kafka.common.internals.Topic.GROUP_METADATA_TOPIC_NAME
-import org.apache.kafka.common.message.{ControlledShutdownRequestData, CreateTopicsRequestData, DeleteTopicsRequestData, DescribeGroupsRequestData, JoinGroupRequestData, LeaveGroupRequestData}
+import org.apache.kafka.common.message._
import org.apache.kafka.common.message.CreateTopicsRequestData.{CreatableTopic, CreatableTopicSet}
+import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.{AlterConfigsResource, AlterableConfig, AlterableConfigSet}
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record.{CompressionType, MemoryRecords, Records, SimpleRecord}
@@ -148,8 +149,9 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
ApiKeys.ALTER_REPLICA_LOG_DIRS -> classOf[AlterReplicaLogDirsResponse],
ApiKeys.DESCRIBE_LOG_DIRS -> classOf[DescribeLogDirsResponse],
ApiKeys.CREATE_PARTITIONS -> classOf[CreatePartitionsResponse],
- ApiKeys.ELECT_PREFERRED_LEADERS -> classOf[ElectPreferredLeadersResponse]
- )
+ ApiKeys.ELECT_PREFERRED_LEADERS -> classOf[ElectPreferredLeadersResponse],
+ ApiKeys.INCREMENTAL_ALTER_CONFIGS -> classOf[IncrementalAlterConfigsResponse]
+ )
val requestKeyToError = Map[ApiKeys, Nothing => Errors](
ApiKeys.METADATA -> ((resp: requests.MetadataResponse) => resp.errors.asScala.find(_._1 == topic).getOrElse(("test", Errors.NONE))._2),
@@ -194,7 +196,9 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
if (resp.logDirInfos.size() > 0) resp.logDirInfos.asScala.head._2.error else Errors.CLUSTER_AUTHORIZATION_FAILED),
ApiKeys.CREATE_PARTITIONS -> ((resp: CreatePartitionsResponse) => resp.errors.asScala.find(_._1 == topic).get._2.error),
ApiKeys.ELECT_PREFERRED_LEADERS -> ((resp: ElectPreferredLeadersResponse) =>
- ElectPreferredLeadersRequest.fromResponseData(resp.data()).get(tp).error())
+ ElectPreferredLeadersRequest.fromResponseData(resp.data()).get(tp).error()),
+ ApiKeys.INCREMENTAL_ALTER_CONFIGS -> ((resp: IncrementalAlterConfigsResponse) =>
+ IncrementalAlterConfigsResponse.fromResponseData(resp.data()).get(new ConfigResource(ConfigResource.Type.TOPIC, tp.topic)).error)
)
val requestKeysToAcls = Map[ApiKeys, Map[Resource, Set[Acl]]](
@@ -233,7 +237,8 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
ApiKeys.ALTER_REPLICA_LOG_DIRS -> clusterAlterAcl,
ApiKeys.DESCRIBE_LOG_DIRS -> clusterDescribeAcl,
ApiKeys.CREATE_PARTITIONS -> topicAlterAcl,
- ApiKeys.ELECT_PREFERRED_LEADERS -> clusterAlterAcl
+ ApiKeys.ELECT_PREFERRED_LEADERS -> clusterAlterAcl,
+ ApiKeys.INCREMENTAL_ALTER_CONFIGS -> topicAlterConfigsAcl
)
@Before
@@ -392,6 +397,19 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
new AlterConfigsRequest.ConfigEntry(LogConfig.MaxMessageBytesProp, "1000000")
))), true).build()
+ private def incrementalAlterConfigsRequest = {
+ val data = new IncrementalAlterConfigsRequestData
+ val alterableConfig = new AlterableConfig
+ alterableConfig.setName(LogConfig.MaxMessageBytesProp).
+ setValue("1000000").setConfigOperation(AlterConfigOp.OpType.SET.id())
+ val alterableConfigSet = new AlterableConfigSet
+ alterableConfigSet.add(alterableConfig)
+ data.resources().add(new AlterConfigsResource().
+ setResourceName(tp.topic).setResourceType(ConfigResource.Type.TOPIC.id()).
+ setConfigs(alterableConfigSet))
+ new IncrementalAlterConfigsRequest.Builder(data).build()
+ }
+
private def describeAclsRequest = new DescribeAclsRequest.Builder(AclBindingFilter.ANY).build()
private def createAclsRequest = new CreateAclsRequest.Builder(
@@ -449,7 +467,8 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
ApiKeys.ADD_OFFSETS_TO_TXN -> addOffsetsToTxnRequest,
// Check StopReplica last since some APIs depend on replica availability
ApiKeys.STOP_REPLICA -> stopReplicaRequest,
- ApiKeys.ELECT_PREFERRED_LEADERS -> electPreferredLeadersRequest
+ ApiKeys.ELECT_PREFERRED_LEADERS -> electPreferredLeadersRequest,
+ ApiKeys.INCREMENTAL_ALTER_CONFIGS -> incrementalAlterConfigsRequest
)
for ((key, request) <- requestKeyToRequest) {
diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
index fb7b539..760644e 100644
--- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
@@ -39,6 +39,7 @@ import kafka.utils._
import kafka.utils.Implicits._
import kafka.zk.{ConfigEntityChangeNotificationZNode, ZooKeeperTestHarness}
import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
import org.apache.kafka.clients.admin.ConfigEntry.{ConfigSource, ConfigSynonym}
import org.apache.kafka.clients.admin._
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, ConsumerRecords, KafkaConsumer}
@@ -329,7 +330,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
Files.copy(Paths.get(combinedStoreProps.getProperty(SSL_TRUSTSTORE_LOCATION_CONFIG)),
Paths.get(sslProperties1.getProperty(SSL_TRUSTSTORE_LOCATION_CONFIG)),
StandardCopyOption.REPLACE_EXISTING)
- TestUtils.alterConfigs(servers, adminClients.head, oldTruststoreProps, perBrokerConfig = true).all.get()
+ TestUtils.incrementalAlterConfigs(servers, adminClients.head, oldTruststoreProps, perBrokerConfig = true).all.get()
verifySslProduceConsume(sslProperties1, "alter-truststore-4")
verifySslProduceConsume(sslProperties2, "alter-truststore-5")
}
@@ -509,7 +510,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
// Enable unclean leader election
val newProps = new Properties
newProps.put(KafkaConfig.UncleanLeaderElectionEnableProp, "true")
- TestUtils.alterConfigs(servers, adminClients.head, newProps, perBrokerConfig = false).all.get
+ TestUtils.incrementalAlterConfigs(servers, adminClients.head, newProps, perBrokerConfig = false).all.get
waitForConfigOnServer(controller, KafkaConfig.UncleanLeaderElectionEnableProp, "true")
// Verify that the old follower with missing records is elected as the new leader
@@ -908,7 +909,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
val unknownConfig = "some.config"
props.put(unknownConfig, "some.config.value")
- TestUtils.alterConfigs(servers, adminClients.head, props, perBrokerConfig = true).all.get
+ TestUtils.incrementalAlterConfigs(servers, adminClients.head, props, perBrokerConfig = true).all.get
TestUtils.waitUntilTrue(() => servers.forall(server => server.config.listeners.size == existingListenerCount + 1),
"Listener config not updated")
@@ -971,11 +972,14 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
.mkString(",")
val props = fetchBrokerConfigsFromZooKeeper(servers.head)
- val listenerProps = props.asScala.keySet.filter(_.startsWith(listenerPrefix(listenerName)))
- listenerProps.foreach(props.remove)
+ val deleteListenerProps = new Properties()
+ deleteListenerProps ++= props.asScala.filter(entry => entry._1.startsWith(listenerPrefix(listenerName)))
+ TestUtils.incrementalAlterConfigs(servers, adminClients.head, deleteListenerProps, perBrokerConfig = true, opType = OpType.DELETE).all.get
+
+ props.clear()
props.put(KafkaConfig.ListenersProp, listeners)
props.put(KafkaConfig.ListenerSecurityProtocolMapProp, listenerMap)
- TestUtils.alterConfigs(servers, adminClients.head, props, perBrokerConfig = true).all.get
+ TestUtils.incrementalAlterConfigs(servers, adminClients.head, props, perBrokerConfig = true).all.get
TestUtils.waitUntilTrue(() => servers.forall(server => server.config.listeners.size == existingListenerCount - 1),
"Listeners not updated")
diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
index 05f4bcd..710ed57 100644
--- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
+++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
@@ -22,9 +22,9 @@ import kafka.log.LogConfig
import kafka.network.RequestChannel.Session
import kafka.security.auth._
import kafka.utils.TestUtils
-import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter, AclBinding, AclBindingFilter, AclOperation, AclPermissionType}
+import org.apache.kafka.common.acl._
import org.apache.kafka.common.config.ConfigResource
-import org.apache.kafka.common.message.{CreateTopicsRequestData, DeleteTopicsRequestData, DescribeGroupsRequestData, ElectPreferredLeadersRequestData, InitProducerIdRequestData, JoinGroupRequestData, LeaveGroupRequestData, SaslAuthenticateRequestData, SaslHandshakeRequestData}
+import org.apache.kafka.common.message._
import org.apache.kafka.common.resource.{PatternType, ResourcePattern, ResourcePatternFilter, ResourceType => AdminResourceType}
import org.apache.kafka.common.{Node, TopicPartition}
import org.apache.kafka.common.message.ControlledShutdownRequestData
@@ -402,6 +402,10 @@ class RequestQuotaTest extends BaseRequestTest {
.setTimeoutMs(0)
.setTopicPartitions(Collections.singletonList(partition)))
+ case ApiKeys.INCREMENTAL_ALTER_CONFIGS =>
+ new IncrementalAlterConfigsRequest.Builder(
+ new IncrementalAlterConfigsRequestData())
+
case _ =>
throw new IllegalArgumentException("Unsupported API key " + apiKey)
}
@@ -501,6 +505,8 @@ class RequestQuotaTest extends BaseRequestTest {
case ApiKeys.DELETE_GROUPS => new DeleteGroupsResponse(response).throttleTimeMs
case ApiKeys.OFFSET_FOR_LEADER_EPOCH => new OffsetsForLeaderEpochResponse(response).throttleTimeMs
case ApiKeys.ELECT_PREFERRED_LEADERS => new ElectPreferredLeadersResponse(response).throttleTimeMs
+ case ApiKeys.INCREMENTAL_ALTER_CONFIGS =>
+ new IncrementalAlterConfigsResponse(response, ApiKeys.INCREMENTAL_ALTER_CONFIGS.latestVersion()).throttleTimeMs
case requestId => throw new IllegalArgumentException(s"No throttle time for $requestId")
}
}
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index c84d75e..f1a5cca 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -26,8 +26,8 @@ import java.security.cert.X509Certificate
import java.time.Duration
import java.util.{Collections, Properties}
import java.util.concurrent.{Callable, ExecutionException, Executors, TimeUnit}
-import javax.net.ssl.X509TrustManager
+import javax.net.ssl.X509TrustManager
import kafka.api._
import kafka.cluster.{Broker, EndPoint}
import kafka.log._
@@ -38,7 +38,8 @@ import Implicits._
import kafka.controller.LeaderIsrAndControllerEpoch
import kafka.zk._
import org.apache.kafka.clients.CommonClientConfigs
-import org.apache.kafka.clients.admin.{AdminClient, AlterConfigsResult, Config, ConfigEntry}
+import org.apache.kafka.clients.admin.AlterConfigOp.OpType
+import org.apache.kafka.clients.admin._
import org.apache.kafka.clients.consumer._
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.{KafkaFuture, TopicPartition}
@@ -1410,6 +1411,20 @@ object TestUtils extends Logging {
adminClient.alterConfigs(configs)
}
+ def incrementalAlterConfigs(servers: Seq[KafkaServer], adminClient: AdminClient, props: Properties,
+ perBrokerConfig: Boolean, opType: OpType = OpType.SET): AlterConfigsResult = {
+ val configEntries = props.asScala.map { case (k, v) => new AlterConfigOp(new ConfigEntry(k, v), opType) }.toList.asJavaCollection
+ val configs = if (perBrokerConfig) {
+ servers.map { server =>
+ val resource = new ConfigResource(ConfigResource.Type.BROKER, server.config.brokerId.toString)
+ (resource, configEntries)
+ }.toMap.asJava
+ } else {
+ Map(new ConfigResource(ConfigResource.Type.BROKER, "") -> configEntries).asJava
+ }
+ adminClient.incrementalAlterConfigs(configs)
+ }
+
def alterTopicConfigs(adminClient: AdminClient, topic: String, topicConfigs: Properties): AlterConfigsResult = {
val configEntries = topicConfigs.asScala.map { case (k, v) => new ConfigEntry(k, v) }.toList.asJava
val newConfig = new Config(configEntries)
@@ -1451,15 +1466,18 @@ object TestUtils extends Logging {
(out.toString, err.toString)
}
- def assertFutureExceptionTypeEquals(future: KafkaFuture[_], clazz: Class[_ <: Throwable]): Unit = {
+ def assertFutureExceptionTypeEquals(future: KafkaFuture[_], clazz: Class[_ <: Throwable],
+ expectedErrorMessage: Option[String] = None): Unit = {
try {
future.get()
fail("Expected CompletableFuture.get to return an exception")
} catch {
case e: ExecutionException =>
- val cause = e.getCause()
+ val cause = e.getCause
assertTrue("Expected an exception of type " + clazz.getName + "; got type " +
- cause.getClass().getName, clazz.isInstance(cause))
+ cause.getClass.getName, clazz.isInstance(cause))
+ expectedErrorMessage.foreach(message => assertTrue(s"Received error message : ${cause.getMessage}" +
+ s" does not contain expected error message : $message", cause.getMessage.contains(message)))
}
}