gabriellefu commented on code in PR #23121:
URL: https://github.com/apache/kafka/pull/23121#discussion_r3868356761
##########
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:
i have deleted the 2 tests, thanks
--
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]
