From ffc6cf46fc4a2b18bfff902c2790b7cef241cd7a Mon Sep 17 00:00:00 2001 From: Gabriella Fu <2234862823@qq.com> Date: Mon, 10 Aug 2026 17:51:19 -0400 Subject: [PATCH] KAFKA-20790: adding assignor broker side validation Move the streams.assignor.name registry check from the active controller to the broker that receives the IncrementalAlterConfigs request. The assignors are registered on brokers, which are the nodes that run the group coordinator, so the controller was validating a group's selection against its own group.streams.assignors. KIP-1357 states that the check relies on "all brokers having the same group.streams.assignors configuration" - controllers are not brokers, so the implementation silently required controller nodes to carry a broker-side plugin config and the custom assignor classes on their classpath. A controller-only node left unconfigured rejected every custom assignor name, while DescribeConfigs, which is served by a broker, reported that same name as the group's effective assignor. - GroupConfig.validateAssignorName is extracted from validateValues so that it can be called with just a name and the registered names. It trims its input because it now runs before ConfigDef parses the value. - ConfigAdminManager.preprocess checks SET operations on streams.assignor.name and rejects them with INVALID_CONFIG, so an unregistered name is never forwarded to the controller. - ControllerConfigurationValidator no longer consults the assignor registry. It keeps validating the remaining group configs, which need the merged post-operation view that only the controller has. The runtime fallback in GroupMetadataManager is unchanged and remains the safety net for a name that is not available on the coordinator's broker. Co-Authored-By: Claude Opus 5 (1M context) --- .../kafka/server/ConfigAdminManager.scala | 28 ++++++-- .../ControllerConfigurationValidator.scala | 3 + .../kafka/server/ConfigAdminManagerTest.scala | 63 +++++++++++++++++- .../StreamsGroupHeartbeatRequestTest.scala | 61 +++++++++++++++++ .../kafka/coordinator/group/GroupConfig.java | 31 +++++---- .../coordinator/group/GroupConfigTest.java | 66 +++++-------------- 6 files changed, 186 insertions(+), 66 deletions(-) diff --git a/core/src/main/scala/kafka/server/ConfigAdminManager.scala b/core/src/main/scala/kafka/server/ConfigAdminManager.scala index 68210276e564f..761dc4b48f8ab 100644 --- a/core/src/main/scala/kafka/server/ConfigAdminManager.scala +++ b/core/src/main/scala/kafka/server/ConfigAdminManager.scala @@ -33,6 +33,7 @@ import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.{Alte import org.apache.kafka.common.protocol.Errors.{INVALID_REQUEST, UNKNOWN_SERVER_ERROR} import org.apache.kafka.common.requests.ApiError import org.apache.kafka.common.resource.{Resource, ResourceType} +import org.apache.kafka.coordinator.group.GroupConfig import org.apache.kafka.metadata.ConfigRepository import org.apache.kafka.server.config.AbstractKafkaConfig import org.apache.kafka.server.logger.RuntimeLoggerManager @@ -63,15 +64,16 @@ import scala.jdk.CollectionConverters._ * * Configuration processing is split into two parts. * - The first step, called "preprocessing," handles setting KIP-412 log levels, validating - * BROKER configurations. We also filter out some other things here like UNKNOWN resource - * types, etc. + * BROKER configurations, and validating the streams task assignor selected by a GROUP + * configuration. We also filter out some other things here like UNKNOWN resource types, etc. * - The second step is "persistence," and handles storing the configurations durably to our * metadata store. * * The active controller performs its own configuration validation step in * [[kafka.server.ControllerConfigurationValidator]]. This is mainly important for - * TOPIC resources, since we already validated changes to BROKER resources on the - * forwarding broker. The controller is also responsible for enforcing the configured + * TOPIC resources, since we already validated changes to BROKER resources, and the + * selected streams task assignor of GROUP resources, on the forwarding broker. The + * controller is also responsible for enforcing the configured * [[org.apache.kafka.server.policy.AlterConfigPolicy]]. */ class ConfigAdminManager(nodeId: Int, @@ -143,7 +145,14 @@ class ConfigAdminManager(nodeId: Int, validateResourceNameIsCurrentNodeId(resource.resourceName()) } validateBrokerConfigChange(resource, configResource) - case TOPIC | CLIENT_METRICS | GROUP => + case GROUP => + resource.configs().forEach { config => + if (config.name() == GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG && + config.configOperation() == OpType.SET.id()) { + validateStreamsAssignorNameChange(config.value()) + } + } + case TOPIC | CLIENT_METRICS => // Nothing to do. case _ => throw new InvalidRequestException(s"Unknown resource type ${resource.resourceType().toInt}") @@ -199,6 +208,15 @@ class ConfigAdminManager(nodeId: Int, } } + /** + * Validate a change to the task assignor selected by a streams group. This is done here rather + * than on the controller because the assignors are registered on the brokers, which are the + * nodes that run the group coordinator. + */ + private def validateStreamsAssignorNameChange(assignorName: String): Unit = { + GroupConfig.validateAssignorName(assignorName, conf.groupCoordinatorConfig.streamsGroupAssignorNames()) + } + /** * Preprocess a legacy configuration operation on the broker. * diff --git a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala index 7fdf7b00a98b0..1049df39cab1c 100644 --- a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala +++ b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala @@ -39,6 +39,9 @@ import scala.collection.mutable * the controller. Therefore, the validation here is just a kind of sanity check, which * should never fail under normal conditions. * + * The streams task assignor selected by a GROUP resource is intentionally not validated here, so + * that a controller does not need the assignors on its classpath. See ConfigAdminManager. + * * This validator does not handle changes to BROKER_LOGGER resources. Despite being bundled * in the same RPC, BROKER_LOGGER is not really a dynamic configuration in the same sense * as the others. It is not persisted to the metadata log. diff --git a/core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala b/core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala index 053dc9e0e1abc..9ff1d1122e0f6 100644 --- a/core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala @@ -22,7 +22,7 @@ import java.util.Collections import kafka.utils.TestUtils import org.apache.kafka.clients.admin.AlterConfigOp.OpType -import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, BROKER_LOGGER, TOPIC, UNKNOWN} +import org.apache.kafka.common.config.ConfigResource.Type.{BROKER, BROKER_LOGGER, GROUP, TOPIC, UNKNOWN} import org.apache.kafka.clients.admin.{AlterConfigOp, ConfigEntry} import org.apache.kafka.common.config.ConfigDef.ConfigKey import org.apache.kafka.common.errors.{InvalidConfigurationException, InvalidRequestException} @@ -40,6 +40,7 @@ import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.{Alte import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.protocol.Errors.{INVALID_REQUEST, NONE} import org.apache.kafka.common.requests.ApiError +import org.apache.kafka.coordinator.group.{GroupConfig, GroupCoordinatorConfig} import org.apache.kafka.metadata.MockConfigRepository import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} import org.junit.jupiter.api.{Assertions, Test} @@ -48,8 +49,12 @@ import org.slf4j.LoggerFactory class ConfigAdminManagerTest { val logger = LoggerFactory.getLogger(classOf[ConfigAdminManagerTest]) - def newConfigAdminManager(brokerId: Integer): ConfigAdminManager = { + def newConfigAdminManager(brokerId: Integer): ConfigAdminManager = + newConfigAdminManager(brokerId, Map.empty) + + def newConfigAdminManager(brokerId: Integer, overrides: Map[String, String]): ConfigAdminManager = { val config = TestUtils.createBrokerConfig(nodeId = brokerId) + overrides.foreach { case (key, value) => config.setProperty(key, value) } new ConfigAdminManager(brokerId, new KafkaConfig(config), new MockConfigRepository()) } @@ -413,6 +418,60 @@ class ConfigAdminManagerTest { unknown))))) } + def groupIncremental(configName: String, value: String, opType: OpType): IAlterConfigsResource = + new IAlterConfigsResource(). + setResourceName("group"). + setResourceType(GROUP.id). + setConfigs(new IAlterableConfigCollection( + util.Arrays.asList(new IAlterableConfig().setName(configName). + setValue(value). + setConfigOperation(opType.id())))) + + @Test + def testPreprocessIncrementalWithStreamsAssignorName(): Unit = { + // A built-in assignor and a custom one, registered by short name and by class name respectively. + val manager = newConfigAdminManager(1, + Map(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG -> + s"sticky,${classOf[CustomStreamsTaskAssignor].getName}")) + + // Both are selectable by the name the assignor reports, and neither resource is preprocessed, + // so both requests are forwarded to the controller. + Seq("sticky", CustomStreamsTaskAssignor.NAME).foreach { name => + val group = groupIncremental(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, name, OpType.SET) + assertEquals(Collections.emptyMap(), + manager.preprocess(new IncrementalAlterConfigsRequestData(). + setResources(new IAlterConfigsResourceCollection(util.Arrays.asList( + group))), + (_, _) => true)) + } + + // A name that no registered assignor reports is rejected before the request is forwarded. + val unknown = groupIncremental(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "does-not-exist", OpType.SET) + assertEquals(Collections.singletonMap(unknown, + new ApiError(Errors.INVALID_CONFIG, "streams.assignor.name 'does-not-exist' is not a " + + "registered task assignor. Registered assignors are: [sticky, custom].")), + manager.preprocess(new IncrementalAlterConfigsRequestData(). + setResources(new IAlterConfigsResourceCollection(util.Arrays.asList( + unknown))), + (_, _) => true)) + } + + @Test + def testPreprocessIncrementalWithUnregisteredBuiltinStreamsAssignorName(): Unit = { + // Only the custom assignor is registered. A built-in is not implicitly available, so it can no + // longer be selected either. + val manager = newConfigAdminManager(1, + Map(GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG -> classOf[CustomStreamsTaskAssignor].getName)) + val sticky = groupIncremental(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "sticky", OpType.SET) + assertEquals(Collections.singletonMap(sticky, + new ApiError(Errors.INVALID_CONFIG, "streams.assignor.name 'sticky' is not a " + + "registered task assignor. Registered assignors are: [custom].")), + manager.preprocess(new IncrementalAlterConfigsRequestData(). + setResources(new IAlterConfigsResourceCollection(util.Arrays.asList( + sticky))), + (_, _) => true)) + } + @Test def testContainsDuplicates(): Unit = { assertFalse(ConfigAdminManager.containsDuplicates(Seq())) diff --git a/core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala b/core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala index 23240726a9d16..4b92c1b36bdd0 100644 --- a/core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala @@ -1043,6 +1043,67 @@ class StreamsGroupHeartbeatRequestTest(cluster: ClusterInstance) extends GroupCo } } + @ClusterTest( + types = Array(Type.KRAFT), + serverProperties = Array( + // Registered on the broker only, so that the controller has never heard of this assignor. + // The class name has to be spelled out because annotation values must be compile-time constants. + new ClusterConfigProperty( + id = 0, + key = GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG, + value = "kafka.server.CustomStreamsTaskAssignor" + ) + ) + ) + def testAlterStreamsAssignorNameGroupConfigIsValidatedOnTheBroker(): Unit = { + val admin = cluster.admin() + val groupId = "test-group" + + try { + TestUtils.createOffsetsTopicWithAdmin( + admin = admin, + brokers = cluster.brokers.values().asScala.toSeq, + controllers = cluster.controllers().values().asScala.toSeq + ) + + val groupConfigResource = new ConfigResource(ConfigResource.Type.GROUP, groupId) + + // An assignor registered on the broker is accepted even though the controller does not have it, + // because the name is checked by the broker that receives the request. + val customAlterOp = new AlterConfigOp( + new ConfigEntry(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, CustomStreamsTaskAssignor.NAME), + AlterConfigOp.OpType.SET + ) + admin.incrementalAlterConfigs( + Map(groupConfigResource -> List(customAlterOp).asJavaCollection).asJava + ).all().get() + + TestUtils.waitUntilTrue(() => { + val describedConfigs = admin.describeConfigs(List(groupConfigResource).asJava).all().get() + describedConfigs.get(groupConfigResource).get(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG).value() == + CustomStreamsTaskAssignor.NAME + }, s"${GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG} was not updated to the expected value within the timeout period.") + + // Conversely, the built-in assignor is rejected: it is not registered on this broker, even though + // it is the assignor that the controller itself has registered. + val stickyAlterOp = new AlterConfigOp( + new ConfigEntry(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "sticky"), + AlterConfigOp.OpType.SET + ) + val executionException = assertThrows(classOf[ExecutionException], () => + admin.incrementalAlterConfigs( + Map(groupConfigResource -> List(stickyAlterOp).asJavaCollection).asJava + ).all().get() + ) + assertTrue(executionException.getCause.isInstanceOf[InvalidConfigurationException], + s"Expected InvalidConfigurationException but got ${executionException.getCause}") + assertTrue(executionException.getCause.getMessage.contains("'sticky' is not a registered task assignor"), + s"Unexpected error message: ${executionException.getCause.getMessage}") + } finally { + admin.close() + } + } + @ClusterTest( serverProperties = Array( // The class name has to be spelled out because annotation values must be compile-time constants. diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java index f9e678f2005f6..4323a2475ef7b 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java @@ -477,9 +477,28 @@ public static void validateNames(Map newGroupConfig) { } } + /** + * Check that the selected task assignor is one of the assignors registered on the broker. + * Validated separately from {@link #validate} because the registry of assignors only exists on + * brokers, which are the nodes that run the group coordinator. + * + * @param assignorName The requested assignor name, as an unparsed config value. + * @param registeredAssignorNames The names of the assignors registered on the broker. + */ + public static void validateAssignorName(String assignorName, List registeredAssignorNames) { + // ConfigDef trims STRING values while parsing, so trim here too in order to accept the same + // values as the rest of the validation. + String trimmed = assignorName.trim(); + if (!registeredAssignorNames.contains(trimmed)) { + throw new InvalidConfigurationException(STREAMS_ASSIGNOR_NAME_CONFIG + " '" + trimmed + + "' is not a registered task assignor. Registered assignors are: " + registeredAssignorNames + "."); + } + } + /** * Check that the given properties contain only valid group config names and that - * all values can be parsed and are valid. + * all values can be parsed and are valid. Does not cover + * {@link #STREAMS_ASSIGNOR_NAME_CONFIG}, see {@link #validateAssignorName}. * * @param newGroupConfig The new unparsed group config overrides. * @param groupCoordinatorConfig The group coordinator config. @@ -628,16 +647,6 @@ private static void validateValues( groupCoordinatorConfig.streamsGroupMaxWarmupReplicas() ); - // The selected streams assignor must be one of the assignors registered on the broker. - if (parsed.containsKey(STREAMS_ASSIGNOR_NAME_CONFIG)) { - String assignorName = (String) parsed.get(STREAMS_ASSIGNOR_NAME_CONFIG); - List registeredAssignors = groupCoordinatorConfig.streamsGroupAssignorNames(); - if (!registeredAssignors.contains(assignorName)) { - throw new InvalidConfigurationException(STREAMS_ASSIGNOR_NAME_CONFIG + " '" + assignorName + - "' is not a registered task assignor. Registered assignors are: " + registeredAssignors + "."); - } - } - // Cross-field validations: session timeout must be greater than heartbeat interval. validateSessionExceedsHeartbeat( parsed, diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java index db64d0945b61d..2bdf9fbf416e6 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java @@ -140,7 +140,7 @@ public void testFromPropsInvalid() { } else if (!GroupConfig.ERRORS_DEADLETTERQUEUE_TOPIC_NAME_CONFIG.equals(name) && !GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG.equals(name)) { // Free-form string configs (no ConfigDef validator) accept any value at construction - // time; their values are validated separately in GroupConfig.validate. + // time; their values are validated separately. assertPropertyInvalid(name, "not_a_number", "-0.1"); } }); @@ -384,48 +384,24 @@ public void testStreamsRackAwareAssignmentTagsValidation() { @Test public void testStreamsAssignorNameValidation() { - // A registered assignor name is accepted. - Map props = createValidGroupConfig(); - props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "sticky"); - doTestValidProps(props); - - // An unknown assignor name is rejected with INVALID_CONFIG. - props = createValidGroupConfig(); - props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "does-not-exist"); - doTestInvalidProps(props, InvalidConfigurationException.class); - } + List registered = List.of("sticky", "custom"); - @Test - public void testStreamsAssignorNameSelectsCustomAssignor() { - // A custom assignor registered on the broker can be selected by its name. - GroupCoordinatorConfig groupCoordinatorConfig = createGroupCoordinatorConfig(Map.of( - GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNORS_CONFIG, - "sticky," + GroupCoordinatorConfigTest.CustomTaskAssignor.class.getName() - )); - - Map props = createValidGroupConfig(); - props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "CustomTaskAssignor"); - assertDoesNotThrow(() -> GroupConfig.validate(props, groupCoordinatorConfig, createShareGroupConfig())); + // Every registered name is accepted, and values are trimmed the way ConfigDef parses them. + assertDoesNotThrow(() -> GroupConfig.validateAssignorName("sticky", registered)); + assertDoesNotThrow(() -> GroupConfig.validateAssignorName("custom", registered)); + assertDoesNotThrow(() -> GroupConfig.validateAssignorName(" custom ", registered)); - // The built-in assignor is still selectable alongside it. - props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "sticky"); - assertDoesNotThrow(() -> GroupConfig.validate(props, groupCoordinatorConfig, createShareGroupConfig())); - - // The custom assignor's class name is not a valid selector; only its name() is. - props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, GroupCoordinatorConfigTest.CustomTaskAssignor.class.getName()); - assertThrows(InvalidConfigurationException.class, - () -> GroupConfig.validate(props, groupCoordinatorConfig, createShareGroupConfig())); + // An unregistered name is rejected. The receiving broker turns this into INVALID_CONFIG. + assertEquals("streams.assignor.name 'does-not-exist' is not a registered task assignor. " + + "Registered assignors are: [sticky, custom].", + assertThrows(InvalidConfigurationException.class, + () -> GroupConfig.validateAssignorName("does-not-exist", registered)).getMessage()); } @Test public void testStreamsAssignorNameEvaluateIsLenient() { - // The Admin path (validate) rejects an unknown assignor name... - Map props = createValidGroupConfig(); - props.put(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "does-not-exist"); - doTestInvalidProps(props, InvalidConfigurationException.class); - - // ...but the metadata-replay path (evaluate) accepts it, so a value that was valid when set - // survives a broker restart even if the assignor was later removed from the broker config. + // The metadata-replay path accepts an unregistered assignor name, so a value that was valid when + // set survives a broker restart even if the assignor was later removed from the broker config. Properties replayed = new Properties(); replayed.setProperty(GroupConfig.STREAMS_ASSIGNOR_NAME_CONFIG, "does-not-exist"); assertDoesNotThrow(() -> GroupConfig.evaluate(replayed, "group", createGroupCoordinatorConfig(), createShareGroupConfig())); @@ -946,21 +922,15 @@ private Map createValidGroupConfig() { } private GroupCoordinatorConfig createGroupCoordinatorConfig() { - return createGroupCoordinatorConfig(Map.of()); - } - - private GroupCoordinatorConfig createGroupCoordinatorConfig(Map overrides) { - Map configs = new HashMap<>(Map.of( - GroupCoordinatorConfig.CONSUMER_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000, - GroupCoordinatorConfig.SHARE_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000, - GroupCoordinatorConfig.STREAMS_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000 - )); - configs.putAll(overrides); return GroupCoordinatorConfigTest.createGroupCoordinatorConfig( OFFSET_METADATA_MAX_SIZE, OFFSETS_RETENTION_CHECK_INTERVAL_MS, OFFSETS_RETENTION_MINUTES, - configs + Map.of( + GroupCoordinatorConfig.CONSUMER_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000, + GroupCoordinatorConfig.SHARE_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000, + GroupCoordinatorConfig.STREAMS_GROUP_MIN_ASSIGNMENT_INTERVAL_MS_CONFIG, 1000 + ) ); }