Skip to content

KAFKA-20790: adding assignor broker side validation #23121

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gabriellefu wants to merge 1 commit into
base: trunk
Choose a base branch
from
Open

KAFKA-20790: adding assignor broker side validation #23121

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 23 additions & 5 deletions core/src/main/scala/kafka/server/ConfigAdminManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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}")
Expand Down Expand Up @@ -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.
*
Expand Down
3 changes: 3 additions & 0 deletions core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
63 changes: 61 additions & 2 deletions core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand All @@ -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}
Expand All @@ -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())
}

Expand Down Expand Up @@ -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()))
Expand Down
61 changes: 61 additions & 0 deletions core/src/test/scala/unit/kafka/server/StreamsGroupHeartbeatRequestTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
31 changes: 20 additions & 11 deletions group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -477,9 +477,28 @@ public static void validateNames(Map<String, String> 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<String> 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.
Expand Down Expand Up @@ -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<String> 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,
Expand Down
66 changes: 18 additions & 48 deletions group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
}
});
Expand Down Expand Up @@ -384,48 +384,24 @@ public void testStreamsRackAwareAssignmentTagsValidation() {

@Test
public void testStreamsAssignorNameValidation() {
// A registered assignor name is accepted.
Map<String, String> 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<String> 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<String, String> 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<String, String> 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()));
Expand Down Expand Up @@ -946,21 +922,15 @@ private Map<String, String> createValidGroupConfig() {
}

private GroupCoordinatorConfig createGroupCoordinatorConfig() {
return createGroupCoordinatorConfig(Map.of());
}

private GroupCoordinatorConfig createGroupCoordinatorConfig(Map<String, Object> overrides) {
Map<String, Object> 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
)
);
}

Expand Down