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 + ) ); }