squah-confluent commented on code in PR #23121:
URL: https://github.com/apache/kafka/pull/23121#discussion_r3858660840
##########
core/src/main/scala/kafka/server/ConfigAdminManager.scala:
##########
@@ -64,15 +65,19 @@ 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 performing the full GROUP configuration
validation (see
+ * [[org.apache.kafka.coordinator.group.GroupConfig#validateOnBroker]]). 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
+ * forwarding broker. GROUP resources are validated on the controller too, so
that the
+ * cluster stays protected while it may still contain brokers that predate the
broker-side
+ * check; see [[kafka.server.ControllerConfigurationValidator]] for the
upgrade-safety
+ * gating. The controller is also responsible for enforcing the configured
* [[org.apache.kafka.server.policy.AlterConfigPolicy]].
Review Comment:
nit: I'm assuming these javadoc updates are Claude-written and will push to
cut them down since Claude loves to cram information that's true but not
immediately helpful to the reader into them.
* "protected" is not quite the right concept.
* We already point to `ControllerConfigurationValidator` for controller
validation above.
```
* TOPIC resources, since we already validated changes to BROKER and GROUP
resources on the
* forwarding broker. When the cluster may contain brokers that predate the
broker-side GROUP
* validation, GROUP resources are validated on the controller too. The
controller is also
* responsible for enforcing the configured
[[org.apache.kafka.server.policy.AlterConfigPolicy]].
* ```
##########
core/src/main/scala/kafka/server/ConfigAdminManager.scala:
##########
@@ -201,6 +208,28 @@ class ConfigAdminManager(nodeId: Int,
}
}
+ /**
+ * Perform the full GROUP config validation on the forwarding broker, before
the change is
+ * sent to the controller. See
[[kafka.server.ControllerConfigurationValidator]] for why the
+ * controller also performs this validation while a cluster may still
contain brokers that
+ * predate this broker-side check.
+ */
+ private def validateGroupConfigChangeOnBroker(resource:
IAlterConfigsResource): Unit = {
Review Comment:
nit: We don't suffix the other methods in this class with "OnBroker". All
the validation in this class is on the broker.
##########
core/src/main/scala/kafka/server/ConfigAdminManager.scala:
##########
@@ -201,6 +208,28 @@ class ConfigAdminManager(nodeId: Int,
}
}
+ /**
+ * Perform the full GROUP config validation on the forwarding broker, before
the change is
+ * sent to the controller. See
[[kafka.server.ControllerConfigurationValidator]] for why the
+ * controller also performs this validation while a cluster may still
contain brokers that
+ * predate this broker-side check.
+ */
Review Comment:
nit: I'm assuming these javadoc updates are Claude-written and will push to
cut them down since Claude loves to cram information that's true but not
immediately helpful to the reader into them.
* There's no concept of a non-full GROUP validation before or after this PR.
* All the validation methods in this file validate before the change is sent
to the controller.
* "See [[kafka.server.ControllerConfigurationValidator]] for why the
controller also performs this validation" answers a question the reader never
asked.
```suggestion
```
##########
core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala:
##########
@@ -39,6 +40,14 @@ import scala.collection.mutable
* the controller. Therefore, the validation here is just a kind of sanity
check, which
* should never fail under normal conditions.
*
+ * GROUP resources are validated here too, via {@link
GroupConfig#validateOnController}, even
+ * though the forwarding broker already performs the same full validation in
+ * {@link kafka.server.ConfigAdminManager#preprocess()} via {@link
GroupConfig#validateOnBroker}.
+ * This is kept, gated on {@code metadataVersion}, so that the cluster is
never left without GROUP
+ * config validation while it may still contain brokers that predate the
broker-side check
+ * (KAFKA-20790). It can be dropped once upgrading directly from a pre-{@link
MetadataVersion#IBP_4_5_IV0}
+ * cluster is no longer supported.
Review Comment:
* We're expressing the same information as the previous paragraph, but for
GROUPs. Copying the sentence structure makes it more obvious.
* We can leave the implementation details and removal considerations for
comments in the validation code.
```
* For changes to GROUP resources, the forwarding broker performs validation
in
* {@link kafka.server.ConfigAdminManager#preprocess()} before sending the
change to
* the controller. The validation here is only run when the cluster may
contain brokers
* that predate the broker-side validation.
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java:
##########
@@ -384,59 +384,34 @@ 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);
- }
-
- @Test
- public void testStreamsAssignorNameSelectsCustomAssignor() {
Review Comment:
Why did we remove this test?
##########
core/src/main/scala/kafka/server/ConfigAdminManager.scala:
##########
@@ -64,15 +65,19 @@ 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 performing the full GROUP configuration
validation (see
+ * [[org.apache.kafka.coordinator.group.GroupConfig#validateOnBroker]]). We
also filter out
+ * some other things here like UNKNOWN resource types, etc.
Review Comment:
nit: I'm assuming these javadoc updates are Claude-written and will push to
cut them down since Claude loves to cram information that's true but not
immediately helpful to the reader into them.
* There's no concept of a non-full GROUP validation before or after this PR.
* We don't link to the validation method for broker validation.
```
* - The first step, called "preprocessing," handles setting KIP-412 log
levels, and validating
* BROKER and GROUP configurations. We also filter out some other things
here like UNKNOWN
* resource types, etc.
```
##########
core/src/main/scala/kafka/server/ConfigAdminManager.scala:
##########
@@ -201,6 +208,28 @@ class ConfigAdminManager(nodeId: Int,
}
}
+ /**
+ * Perform the full GROUP config validation on the forwarding broker, before
the change is
+ * sent to the controller. See
[[kafka.server.ControllerConfigurationValidator]] for why the
+ * controller also performs this validation while a cluster may still
contain brokers that
+ * predate this broker-side check.
+ */
+ private def validateGroupConfigChangeOnBroker(resource:
IAlterConfigsResource): Unit = {
+ val configResource = new ConfigResource(GROUP, resource.resourceName())
Review Comment:
We pass `configResource` as a parameter in `validateBrokerConfigChange`
instead of re-creating it.
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -219,6 +224,25 @@ class ControllerConfigurationValidatorTest {
config.put("foobar", "abc")
assertEquals("Unknown group config name: foobar",
assertThrows(classOf[InvalidConfigurationException], () =>
validator.validate(
- new ConfigResource(GROUP, "group"), config, emptyMap())).getMessage)
+ new ConfigResource(GROUP, "group"), config, emptyMap(),
preGroupBrokerValidationMv)).getMessage)
+ }
+
+ @Test
+ def testInvalidGroupConfigStillRejectedBelowGatedMetadataVersion(): Unit = {
+ val config = new util.TreeMap[String, String]()
+ config.put("foobar", "abc")
+ assertEquals("Unknown group config name: foobar",
+ assertThrows(classOf[InvalidConfigurationException], () =>
validator.validate(
+ new ConfigResource(GROUP, "group"), config, emptyMap(),
MetadataVersion.IBP_4_4_IV2)).getMessage)
+ }
+
+ @Test
+ def
testInvalidGroupConfigSkippedOnceMetadataVersionGuaranteesBrokerValidation():
Unit = {
+ // Once every broker in the cluster is guaranteed to already validate this
on the forwarding
+ // broker (see ConfigAdminManager#validateGroupConfigChangeOnBroker), the
controller no longer
+ // needs to, and an otherwise-invalid config is let through here.
+ val config = new util.TreeMap[String, String]()
+ config.put("foobar", "abc")
+ validator.validate(new ConfigResource(GROUP, "group"), config, emptyMap(),
MetadataVersion.IBP_4_5_IV0)
}
Review Comment:
Claude loves spamming new tests without regard for existing tests or overall
test strategy.
There is an existing `testInvalidGroupConfig` test - how does the coverage
overlap with the new tests?
##########
core/src/main/scala/kafka/server/ConfigAdminManager.scala:
##########
@@ -145,7 +150,9 @@ class ConfigAdminManager(nodeId: Int,
validateResourceNameIsCurrentNodeId(resource.resourceName())
}
validateBrokerConfigChange(resource, configResource)
- case TOPIC | CLIENT_METRICS | GROUP =>
+ case GROUP =>
+ validateGroupConfigChangeOnBroker(resource)
Review Comment:
Do we need to update the `preprocess(AlterConfigsRequestData, ...)` path too?
##########
core/src/test/scala/unit/kafka/server/ConfigAdminManagerTest.scala:
##########
@@ -417,6 +422,60 @@ class ConfigAdminManagerTest {
(_, _) => true))
}
+ 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))
+ }
Review Comment:
These tests no longer look appropriate. We would like to test that group
validation is done on the broker (the equivalent of the first part of
`StreamsGroupHeartbeatRequestTest.testAlterStreamsAssignorNameGroupConfigIsValidatedOnTheBroker`),
not the particulars of streams assignor validation.
##########
core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala:
##########
@@ -134,8 +144,12 @@ class ControllerConfigurationValidator(kafkaConfig:
KafkaConfig) extends Configu
ClientMetricsConfigs.validate(resource.name(), filteredConfigs)
case GROUP =>
validateGroupName(resource.name())
- val filteredConfigs = filterAndValidateNullConfigs(newConfigs, "group")
- GroupConfig.validate(filteredConfigs,
kafkaConfig.groupCoordinatorConfig, kafkaConfig.shareGroupConfig)
+ // Skip once every broker in the cluster is guaranteed to already
validate this on the
+ // forwarding broker. See the class-level doc for why this can't just
be removed outright.
Review Comment:
> See the class-level doc for why this can't just be removed outright.
Can we inline the condition for removal?
```
// Validate on the controller when the cluster may contain
forwarding brokers that do
// not validate. Can be removed once upgrading directly from a
// pre-{@link MetadataVersion#IBP_4_5_IV0} cluster is no longer
supported.
```
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -36,26 +37,30 @@ class ControllerConfigurationValidatorTest {
val config = new KafkaConfig(TestUtils.createDummyBrokerConfig())
val validator = new ControllerConfigurationValidator(config)
+ // The metadata.version in effect prior to this feature shipping
(KAFKA-20790), used for all
+ // tests that are not specifically about the upgrade-safety gate below.
+ val preGroupBrokerValidationMv: MetadataVersion =
MetadataVersion.LATEST_PRODUCTION
Review Comment:
This comment is going to be inaccurate within two Kafka releases when
`LATEST_PRODUCTION` shifts under us. The exact metadata version also doesn't
matter for almost all tests. Can we inline this?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
@@ -477,6 +477,38 @@ public static void validateNames(Map<String, String>
newGroupConfig) {
}
}
+ /**
+ * Validate a GROUP config change on the forwarding broker, before it is
sent to the
+ * controller. Runs the full set of checks in {@link #validate}.
+ *
+ * @param newGroupConfig The new unparsed group config overrides.
+ * @param groupCoordinatorConfig The group coordinator config.
+ * @param shareGroupConfig The share group config.
+ */
+ public static void validateOnBroker(
+ Map<String, String> newGroupConfig,
+ GroupCoordinatorConfig groupCoordinatorConfig,
+ ShareGroupConfig shareGroupConfig
+ ) {
+ validate(newGroupConfig, groupCoordinatorConfig, shareGroupConfig);
+ }
+
+ /**
+ * Validate a GROUP config change on the controller. Runs the full set of
checks in
+ * {@link #validate}.
Review Comment:
Here the reader will wonder why we are validating both on the broker and
controller. This is a good place to explain that the controller validation is
not run once \<blah blah blah\>.
(Please do not call it an upgrade-safety gate. I'm tired of reading
Claude-speak.)
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -219,6 +224,25 @@ class ControllerConfigurationValidatorTest {
config.put("foobar", "abc")
assertEquals("Unknown group config name: foobar",
assertThrows(classOf[InvalidConfigurationException], () =>
validator.validate(
- new ConfigResource(GROUP, "group"), config, emptyMap())).getMessage)
+ new ConfigResource(GROUP, "group"), config, emptyMap(),
preGroupBrokerValidationMv)).getMessage)
+ }
+
+ @Test
+ def testInvalidGroupConfigStillRejectedBelowGatedMetadataVersion(): Unit = {
Review Comment:
* "Still" implies a before-and-after, but it doesn't mean anything to the
reader.
```suggestion
def testInvalidGroupConfigRejectedBelowMetadataVersion4_5(): Unit = {
```
##########
core/src/test/scala/unit/kafka/server/ControllerConfigurationValidatorTest.scala:
##########
@@ -219,6 +224,25 @@ class ControllerConfigurationValidatorTest {
config.put("foobar", "abc")
assertEquals("Unknown group config name: foobar",
assertThrows(classOf[InvalidConfigurationException], () =>
validator.validate(
- new ConfigResource(GROUP, "group"), config, emptyMap())).getMessage)
+ new ConfigResource(GROUP, "group"), config, emptyMap(),
preGroupBrokerValidationMv)).getMessage)
+ }
+
+ @Test
+ def testInvalidGroupConfigStillRejectedBelowGatedMetadataVersion(): Unit = {
+ val config = new util.TreeMap[String, String]()
+ config.put("foobar", "abc")
+ assertEquals("Unknown group config name: foobar",
+ assertThrows(classOf[InvalidConfigurationException], () =>
validator.validate(
+ new ConfigResource(GROUP, "group"), config, emptyMap(),
MetadataVersion.IBP_4_4_IV2)).getMessage)
+ }
+
+ @Test
+ def
testInvalidGroupConfigSkippedOnceMetadataVersionGuaranteesBrokerValidation():
Unit = {
Review Comment:
```suggestion
def testInvalidGroupConfigNotValidatedAfterMetadataVersion4_5(): Unit = {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
