chia7712 commented on code in PR #15840: URL: https://github.com/apache/kafka/pull/15840#discussion_r1606087930
########## core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java: ########## @@ -65,219 +74,479 @@ @ExtendWith(value = ClusterTestExtensions.class) @Tag("integration") public class ConfigCommandIntegrationTest { - AdminZkClient adminZkClient; - List<String> alterOpts; + private List<String> alterOpts; + private final String defaultBrokerId = "0"; private final ClusterInstance cluster; + private static Runnable run(Stream<String> command) { + return () -> { + try { + ConfigCommand.main(command.toArray(String[]::new)); + } catch (RuntimeException e) { + // do nothing. + } finally { + Exit.resetExitProcedure(); + } + }; + } + public ConfigCommandIntegrationTest(ClusterInstance cluster) { this.cluster = cluster; } - @ClusterTest(types = {Type.ZK, Type.KRAFT}) + @ClusterTest(types = {Type.ZK, Type.KRAFT, Type.CO_KRAFT}) public void testExitWithNonZeroStatusOnUpdatingUnallowedConfig() { assertNonZeroStatusExit(Stream.concat(quorumArgs(), Stream.of( "--entity-name", cluster.isKRaftTest() ? "0" : "1", "--entity-type", "brokers", "--alter", "--add-config", "security.inter.broker.protocol=PLAINTEXT")), - errOut -> - assertTrue(errOut.contains("Cannot update these configs dynamically: Set(security.inter.broker.protocol)"), errOut)); + errOut -> assertTrue(errOut.contains("Cannot update these configs dynamically: Set(security.inter.broker.protocol)"), errOut)); } - @ClusterTest(types = {Type.ZK}) public void testExitWithNonZeroStatusOnZkCommandAlterUserQuota() { assertNonZeroStatusExit(Stream.concat(quorumArgs(), Stream.of( "--entity-type", "users", "--entity-name", "admin", "--alter", "--add-config", "consumer_byte_rate=20000")), - errOut -> - assertTrue(errOut.contains("User configuration updates using ZooKeeper are only supported for SCRAM credential updates."), errOut)); - } - - public static void assertNonZeroStatusExit(Stream<String> args, Consumer<String> checkErrOut) { - AtomicReference<Integer> exitStatus = new AtomicReference<>(); - Exit.setExitProcedure((status, __) -> { - exitStatus.set(status); - throw new RuntimeException(); - }); - - String errOut = captureStandardErr(() -> { - try { - ConfigCommand.main(args.toArray(String[]::new)); - } catch (RuntimeException e) { - // do nothing. - } finally { - Exit.resetExitProcedure(); - } - }); - - checkErrOut.accept(errOut); - assertNotNull(exitStatus.get()); - assertEquals(1, exitStatus.get()); - } - - private Stream<String> quorumArgs() { - return cluster.isKRaftTest() - ? Stream.of("--bootstrap-server", cluster.bootstrapServers()) - : Stream.of("--zookeeper", ((ZkClusterInvocationContext.ZkClusterInstance) cluster).getUnderlying().zkConnect()); - } - - public List<String> entityOp(Optional<String> brokerId) { - return brokerId.map(id -> Arrays.asList("--entity-name", id)).orElse(Collections.singletonList("--entity-default")); - } - - public void alterConfigWithZk(KafkaZkClient zkClient, Map<String, String> configs, Optional<String> brokerId) throws Exception { - alterConfigWithZk(zkClient, configs, brokerId, Collections.emptyMap()); + errOut -> assertTrue(errOut.contains("User configuration updates using ZooKeeper are only supported for SCRAM credential updates."), errOut)); } - public void alterConfigWithZk(KafkaZkClient zkClient, Map<String, String> configs, Optional<String> brokerId, Map<String, String> encoderConfigs) { - String configStr = Stream.of(configs.entrySet(), encoderConfigs.entrySet()) - .flatMap(Set::stream) - .map(e -> e.getKey() + "=" + e.getValue()) - .collect(Collectors.joining(",")); - ConfigCommand.ConfigCommandOptions addOpts = new ConfigCommand.ConfigCommandOptions(toArray(alterOpts, entityOp(brokerId), Arrays.asList("--add-config", configStr))); - ConfigCommand.alterConfigWithZk(zkClient, addOpts, adminZkClient); - } - - void verifyConfig(KafkaZkClient zkClient, Map<String, String> configs, Optional<String> brokerId) { - Properties entityConfigs = zkClient.getEntityConfigs("brokers", brokerId.orElse(ZooKeeperInternals.DEFAULT_STRING)); - assertEquals(configs, entityConfigs); - } - - void alterAndVerifyConfig(KafkaZkClient zkClient, Map<String, String> configs, Optional<String> brokerId) throws Exception { - alterConfigWithZk(zkClient, configs, brokerId); - verifyConfig(zkClient, configs, brokerId); - } + @ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT}) + public void testNullStatusOnKraftCommandAlterUserQuota() { + Stream<String> command = Stream.concat(quorumArgs(), Stream.of( + "--entity-type", "users", + "--entity-name", "admin", + "--alter", "--add-config", "consumer_byte_rate=20000")); + String message = captureStandardMsg(run(command)); - void deleteAndVerifyConfig(KafkaZkClient zkClient, Set<String> configNames, Optional<String> brokerId) { - ConfigCommand.ConfigCommandOptions deleteOpts = new ConfigCommand.ConfigCommandOptions(toArray(alterOpts, entityOp(brokerId), Arrays.asList("--delete-config", String.join(",", configNames)))); - ConfigCommand.alterConfigWithZk(zkClient, deleteOpts, adminZkClient); - verifyConfig(zkClient, Collections.emptyMap(), brokerId); + assertTrue(StringUtils.isBlank(message), message); } - @ClusterTest(types = {Type.ZK}) + @ClusterTest(types = Type.ZK) public void testDynamicBrokerConfigUpdateUsingZooKeeper() throws Exception { cluster.shutdownBroker(0); String zkConnect = ((ZkClusterInvocationContext.ZkClusterInstance) cluster).getUnderlying().zkConnect(); KafkaZkClient zkClient = ((ZkClusterInvocationContext.ZkClusterInstance) cluster).getUnderlying().zkClient(); String brokerId = "1"; - adminZkClient = new AdminZkClient(zkClient, scala.None$.empty()); - alterOpts = Arrays.asList("--zookeeper", zkConnect, "--entity-type", "brokers", "--alter"); + AdminZkClient adminZkClient = new AdminZkClient(zkClient, scala.None$.empty()); + alterOpts = asList("--zookeeper", zkConnect, "--entity-type", "brokers", "--alter"); // Add config - alterAndVerifyConfig(zkClient, Collections.singletonMap("message.max.size", "110000"), Optional.of(brokerId)); - alterAndVerifyConfig(zkClient, Collections.singletonMap("message.max.size", "120000"), Optional.empty()); + alterAndVerifyConfig(zkClient, adminZkClient, Optional.of(brokerId), Review Comment: `message.max.size` is not a correct config in kafka. Maybe we should `message.max.bytes` instead -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org