You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/10/20 23:43:10 UTC
(kafka) branch trunk updated: MINOR: Validate operation types in incremental config updates (#23822)
This is an automated email from the ASF dual-hosted git repository.
chia7712 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 6adbf155a6a MINOR: Validate operation types in incremental config updates (#23822)
6adbf155a6a is described below
commit 6adbf155a6aeda2bbc3dcf0893dd6ffe4a333ed5
Author: S.Y. Wang <jo...@gmail.com>
AuthorDate: Wed Oct 21 08:43:03 2026 +0900
MINOR: Validate operation types in incremental config updates (#23822)
There are already similar checks. It will be better to add the same
check accordingly.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>, Luke Chen
<sh...@gmail.com>
---
.../main/scala/kafka/server/ControllerApis.scala | 29 +++++++---
.../unit/kafka/server/ControllerApisTest.scala | 67 ++++++++++++++++++++++
2 files changed, 87 insertions(+), 9 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ControllerApis.scala b/core/src/main/scala/kafka/server/ControllerApis.scala
index 909b113c578..6ff4b010a2a 100644
--- a/core/src/main/scala/kafka/server/ControllerApis.scala
+++ b/core/src/main/scala/kafka/server/ControllerApis.scala
@@ -743,7 +743,7 @@ class ControllerApis(
val alterConfigsRequest = request.body(classOf[IncrementalAlterConfigsRequest])
val context = new ControllerRequestContext(request.context.header.data, request.context.principal,
OptionalLong.empty())
- val duplicateResources = new util.HashSet[ConfigResource]
+ val invalidResources = new util.HashSet[ConfigResource]
val configChanges = new util.HashMap[ConfigResource,
util.Map[String, Entry[AlterConfigOp.OpType, String]]]()
val brokerLoggerResponses = new util.ArrayList[AlterConfigsResourceResponse](1)
@@ -771,18 +771,29 @@ class ControllerApis(
setErrorMessage("Unknown resource type " + resource.resourceType() + ".").
setResourceName(resource.resourceName()).
setResourceType(resource.resourceType()))
- } else if (!duplicateResources.contains(configResource)) {
+ } else if (!invalidResources.contains(configResource)) {
val altersByName = new util.HashMap[String, Entry[AlterConfigOp.OpType, String]]()
- resource.configs.forEach { config =>
- altersByName.put(config.name, new util.AbstractMap.SimpleEntry[AlterConfigOp.OpType, String](
- AlterConfigOp.OpType.forId(config.configOperation), config.value))
+ val configs = resource.configs.iterator()
+ var apiError = ApiError.NONE
+ while (configs.hasNext && !apiError.isFailure) {
+ val config = configs.next()
+ val opType = AlterConfigOp.OpType.forId(config.configOperation)
+ if (opType == null) {
+ apiError = new ApiError(INVALID_REQUEST, s"Unknown operation type ${config.configOperation}")
+ } else {
+ altersByName.put(config.name, new util.AbstractMap.SimpleEntry[AlterConfigOp.OpType, String](
+ opType, config.value))
+ }
}
- if (configChanges.put(configResource, altersByName) != null) {
- duplicateResources.add(configResource)
+ if (apiError.isSuccess && configChanges.put(configResource, altersByName) != null) {
+ apiError = new ApiError(INVALID_REQUEST, "Duplicate resource.")
+ }
+ if (apiError.isFailure) {
+ invalidResources.add(configResource)
configChanges.remove(configResource)
response.responses().add(new AlterConfigsResourceResponse().
- setErrorCode(INVALID_REQUEST.code()).
- setErrorMessage("Duplicate resource.").
+ setErrorCode(apiError.error().code()).
+ setErrorMessage(apiError.message()).
setResourceName(resource.resourceName()).
setResourceType(resource.resourceType()))
}
diff --git a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
index 1a5656f8e07..e9a3fc710aa 100644
--- a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
@@ -553,6 +553,73 @@ class ControllerApisTest {
response.data().responses().asScala.toSet)
}
+ @ParameterizedTest
+ @ValueSource(ints = Array(-1, 4, 127))
+ def testInvalidIncrementalAlterConfigsOperation(invalidOperation: Int): Unit = {
+ val invalidResource = new ConfigResource(ConfigResource.Type.TOPIC, "invalid")
+ val validResource = new ConfigResource(ConfigResource.Type.TOPIC, "valid")
+ val requestData = new IncrementalAlterConfigsRequestData().setResources(
+ new AlterConfigsResourceCollection(util.Arrays.asList(
+ new AlterConfigsResource().
+ setResourceName(invalidResource.name()).
+ setResourceType(invalidResource.`type`().id()).
+ setConfigs(new AlterableConfigCollection(util.Arrays.asList(
+ new AlterableConfig().
+ setName(TopicConfig.FLUSH_MS_CONFIG).
+ setValue("1000").
+ setConfigOperation(AlterConfigOp.OpType.SET.id()),
+ new AlterableConfig().
+ setName(TopicConfig.RETENTION_MS_CONFIG).
+ setValue("2000").
+ setConfigOperation(invalidOperation.toByte),
+ new AlterableConfig().
+ setName(TopicConfig.RETENTION_MS_CONFIG).
+ setValue("3000").
+ setConfigOperation(AlterConfigOp.OpType.SET.id()),
+ new AlterableConfig().
+ setName(TopicConfig.FLUSH_MS_CONFIG).
+ setValue("4000").
+ setConfigOperation(126.toByte)))),
+ new AlterConfigsResource().
+ setResourceName(validResource.name()).
+ setResourceType(validResource.`type`().id()).
+ setConfigs(new AlterableConfigCollection(util.Arrays.asList(new AlterableConfig().
+ setName(TopicConfig.FLUSH_MS_CONFIG).
+ setValue("1000").
+ setConfigOperation(AlterConfigOp.OpType.SET.id())))))))
+ requestData.resources().add(requestData.resources().asScala.last.duplicate().setResourceName(invalidResource.name()))
+ val request = buildRequest(new IncrementalAlterConfigsRequest.Builder(requestData).build(0))
+ val controller = spy(new MockController.Builder().build())
+ controllerApis = createControllerApis(None, controller)
+ controllerApis.handleIncrementalAlterConfigs(request).get()
+
+ val capturedChanges = ArgumentCaptor.forClass(
+ classOf[util.Map[ConfigResource, util.Map[String, util.Map.Entry[AlterConfigOp.OpType, String]]]])
+ verify(controller).incrementalAlterConfigs(any(classOf[ControllerRequestContext]),
+ capturedChanges.capture(), ArgumentMatchers.eq(false), ArgumentMatchers.eq(false))
+ assertEquals(Set(validResource), capturedChanges.getValue.keySet().asScala.toSet)
+ assertEquals(Map(TopicConfig.FLUSH_MS_CONFIG ->
+ new util.AbstractMap.SimpleEntry[AlterConfigOp.OpType, String](AlterConfigOp.OpType.SET, "1000")),
+ capturedChanges.getValue.get(validResource).asScala.toMap)
+
+ val capturedResponse = ArgumentCaptor.forClass(classOf[AbstractResponse])
+ verify(requestChannel).sendResponse(ArgumentMatchers.eq(request), capturedResponse.capture())
+ val response = capturedResponse.getValue.asInstanceOf[IncrementalAlterConfigsResponse]
+ assertEquals(2, response.data().responses().size())
+ assertEquals(Set(
+ new AlterConfigsResourceResponse().
+ setResourceName(invalidResource.name()).
+ setResourceType(invalidResource.`type`().id()).
+ setErrorCode(INVALID_REQUEST.code()).
+ setErrorMessage(s"Unknown operation type $invalidOperation"),
+ new AlterConfigsResourceResponse().
+ setResourceName(validResource.name()).
+ setResourceType(validResource.`type`().id()).
+ setErrorCode(NONE.code()).
+ setErrorMessage(null)),
+ response.data().responses().asScala.toSet)
+ }
+
@ParameterizedTest
@ValueSource(booleans = Array(false, true))
def testInvalidIncrementalAlterConfigsResources(denyAllAuthorizer: Boolean): Unit = {