You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2024/08/27 09:16:53 UTC
(kafka) branch trunk updated: KAFKA-17327; Add support of group in kafka-configs.sh (#16887)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 006af8b9390 KAFKA-17327; Add support of group in kafka-configs.sh (#16887)
006af8b9390 is described below
commit 006af8b939017cb0aed3bf8ce5f3f26e137faf77
Author: DL1231 <53...@users.noreply.github.com>
AuthorDate: Tue Aug 27 17:16:46 2024 +0800
KAFKA-17327; Add support of group in kafka-configs.sh (#16887)
The patch adds support of alter/describe configs for group in kafka-configs.sh.
Reviewers: Andrew Schofield <as...@confluent.io>, Chia-Ping Tsai <ch...@gmail.com>, David Jacot <dj...@confluent.io>
---
checkstyle/import-control-core.xml | 1 +
.../kafka/clients/admin/MockAdminClient.java | 57 +++++-
.../src/main/scala/kafka/admin/ConfigCommand.scala | 75 ++++----
.../main/scala/kafka/server/DynamicConfig.scala | 11 +-
.../kafka/admin/ConfigCommandIntegrationTest.java | 106 ++++++++++-
.../test/java/kafka/admin/ConfigCommandTest.java | 206 +++++++++++++++++++++
.../server/metadata/ZkConfigRepositoryTest.scala | 2 +
.../kafka/coordinator/group/GroupConfig.java | 4 +
8 files changed, 417 insertions(+), 45 deletions(-)
diff --git a/checkstyle/import-control-core.xml b/checkstyle/import-control-core.xml
index 0e9ea061374..1d9f2029c8e 100644
--- a/checkstyle/import-control-core.xml
+++ b/checkstyle/import-control-core.xml
@@ -131,6 +131,7 @@
<allow pkg="kafka.server"/>
<allow pkg="kafka.zk"/>
<allow pkg="org.apache.kafka.clients.admin"/>
+ <allow pkg="org.apache.kafka.coordinator.group"/>
<allow pkg="org.apache.kafka.metadata.authorizer"/>
<allow pkg="org.apache.kafka.security"/>
<allow pkg="org.apache.kafka.server"/>
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
index 6758f21a3f4..68f483b195e 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
@@ -91,6 +91,8 @@ public class MockAdminClient extends AdminClient {
private final List<List<String>> brokerLogDirs;
private final List<Map<String, String>> brokerConfigs;
private final Map<String, Map<String, String>> clientMetricsConfigs;
+ private final Map<String, Map<String, String>> groupConfigs;
+ private final Map<String, String> defaultGroupConfigs;
private Node controller;
private int timeoutNextRequests = 0;
@@ -121,6 +123,7 @@ public class MockAdminClient extends AdminClient {
private Map<String, Short> featureLevels = Collections.emptyMap();
private Map<String, Short> minSupportedFeatureLevels = Collections.emptyMap();
private Map<String, Short> maxSupportedFeatureLevels = Collections.emptyMap();
+ private Map<String, String> defaultGroupConfigs = Collections.emptyMap();
@SuppressWarnings("this-escape")
public Builder() {
@@ -191,6 +194,11 @@ public class MockAdminClient extends AdminClient {
return this;
}
+ public Builder defaultGroupConfigs(Map<String, String> defaultGroupConfigs) {
+ this.defaultGroupConfigs = defaultGroupConfigs;
+ return this;
+ }
+
public MockAdminClient build() {
return new MockAdminClient(brokers,
controller == null ? brokers.get(0) : controller,
@@ -201,7 +209,8 @@ public class MockAdminClient extends AdminClient {
usingRaftController,
featureLevels,
minSupportedFeatureLevels,
- maxSupportedFeatureLevels);
+ maxSupportedFeatureLevels,
+ defaultGroupConfigs);
}
}
@@ -219,6 +228,7 @@ public class MockAdminClient extends AdminClient {
false,
Collections.emptyMap(),
Collections.emptyMap(),
+ Collections.emptyMap(),
Collections.emptyMap());
}
@@ -233,7 +243,8 @@ public class MockAdminClient extends AdminClient {
boolean usingRaftController,
Map<String, Short> featureLevels,
Map<String, Short> minSupportedFeatureLevels,
- Map<String, Short> maxSupportedFeatureLevels
+ Map<String, Short> maxSupportedFeatureLevels,
+ Map<String, String> defaultGroupConfigs
) {
this.brokers = brokers;
controller(controller);
@@ -243,6 +254,8 @@ public class MockAdminClient extends AdminClient {
this.brokerLogDirs = brokerLogDirs;
this.brokerConfigs = new ArrayList<>();
this.clientMetricsConfigs = new HashMap<>();
+ this.groupConfigs = new HashMap<>();
+ this.defaultGroupConfigs = new HashMap<>(defaultGroupConfigs);
for (int i = 0; i < brokers.size(); i++) {
final Map<String, String> config = new HashMap<>();
config.put("default.replication.factor", String.valueOf(defaultReplicationFactor));
@@ -713,7 +726,9 @@ public class MockAdminClient extends AdminClient {
@Override
public synchronized ListConsumerGroupsResult listConsumerGroups(ListConsumerGroupsOptions options) {
- throw new UnsupportedOperationException("Not implemented yet");
+ KafkaFutureImpl<Collection<Object>> future = new KafkaFutureImpl<>();
+ future.complete(groupConfigs.keySet().stream().map(g -> new ConsumerGroupListing(g, false)).collect(Collectors.toList()));
+ return new ListConsumerGroupsResult(future);
}
@Override
@@ -824,6 +839,15 @@ public class MockAdminClient extends AdminClient {
}
return toConfigObject(clientMetricsConfigs.get(resourceName));
}
+ case GROUP: {
+ String resourceName = resource.name();
+ if (resourceName.isEmpty()) {
+ throw new InvalidRequestException("Empty resource name");
+ }
+ Map<String, String> groupConfig = groupConfigs.getOrDefault(resourceName, new HashMap<>());
+ defaultGroupConfigs.forEach(groupConfig::putIfAbsent);
+ return toConfigObject(groupConfig);
+ }
default:
throw new UnsupportedOperationException("Not implemented yet");
}
@@ -945,6 +969,33 @@ public class MockAdminClient extends AdminClient {
clientMetricsConfigs.put(resourceName, newMap);
return null;
}
+ case GROUP: {
+ String resourceName = resource.name();
+ if (resourceName.isEmpty()) {
+ return new InvalidRequestException("Empty resource name");
+ }
+
+ if (!groupConfigs.containsKey(resourceName)) {
+ groupConfigs.put(resourceName, new HashMap<>());
+ }
+
+ HashMap<String, String> newMap = new HashMap<>(groupConfigs.get(resourceName));
+ for (AlterConfigOp op : ops) {
+ switch (op.opType()) {
+ case SET:
+ newMap.put(op.configEntry().name(), op.configEntry().value());
+ break;
+ case DELETE:
+ newMap.remove(op.configEntry().name());
+ break;
+ default:
+ return new InvalidRequestException(
+ "Unsupported op type " + op.opType());
+ }
+ }
+ groupConfigs.put(resourceName, newMap);
+ return null;
+ }
default:
return new UnsupportedOperationException();
}
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index 8f84e309562..2c5439f05c2 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -45,7 +45,7 @@ import scala.jdk.CollectionConverters._
import scala.collection._
/**
- * This script can be used to change configs for topics/clients/users/brokers/ips/client-metrics dynamically
+ * This script can be used to change configs for topics/clients/users/brokers/ips/client-metrics/groups dynamically
* An entity described or altered by the command may be one of:
* <ul>
* <li> topic: --topic <topic> OR --entity-type topics --entity-name <topic>
@@ -57,6 +57,7 @@ import scala.collection._
* <li> broker-logger: --broker-logger <broker-id> OR --entity-type broker-loggers --entity-name <broker-id>
* <li> ip: --ip <ip> OR --entity-type ips --entity-name <ip>
* <li> client-metrics: --client-metrics <name> OR --entity-type client-metrics --entity-name <name>
+ * <li> group: --group <group> OR --entity-type groups --entity-name <group>
* </ul>
* --entity-type <users|clients|brokers|ips> --entity-default may be specified in place of --entity-type <users|clients|brokers|ips> --entity-name <entityName>
* when describing or altering default configuration for users, clients, brokers, or ips, respectively.
@@ -78,7 +79,7 @@ object ConfigCommand extends Logging {
val BrokerDefaultEntityName = ""
val BrokerLoggerConfigType = "broker-loggers"
- private val BrokerSupportedConfigTypes = ConfigType.ALL.asScala :+ BrokerLoggerConfigType :+ ConfigType.CLIENT_METRICS
+ private val BrokerSupportedConfigTypes = ConfigType.ALL.asScala :+ BrokerLoggerConfigType :+ ConfigType.CLIENT_METRICS :+ ConfigType.GROUP
private val ZkSupportedConfigTypes = Seq(ConfigType.USER, ConfigType.BROKER)
private val DefaultScramIterations = 4096
@@ -86,7 +87,8 @@ object ConfigCommand extends Logging {
try {
val opts = new ConfigCommandOptions(args)
- CommandLineUtils.maybePrintHelpOrVersion(opts, "This tool helps to manipulate and describe entity config for a topic, client, user, broker, ip or client-metrics")
+ CommandLineUtils.maybePrintHelpOrVersion(opts,
+ "This tool helps to manipulate and describe entity config for a topic, client, user, broker, ip, client-metrics or group")
opts.checkArgs()
@@ -366,20 +368,7 @@ object ConfigCommand extends Logging {
entityTypeHead match {
case ConfigType.TOPIC =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
-
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
-
- val configResource = new ConfigResource(ConfigResource.Type.TOPIC, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ alterResourceConfig(adminClient, entityTypeHead, entityNameHead, configsToBeDeleted, configsToBeAdded, ConfigResource.Type.TOPIC)
case ConfigType.BROKER =>
val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
@@ -455,20 +444,11 @@ object ConfigCommand extends Logging {
throw new IllegalArgumentException(s"Only connection quota configs can be added for '${ConfigType.IP}' using --bootstrap-server. Unexpected config names: ${unknownConfigs.mkString(",")}")
alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
case ConfigType.CLIENT_METRICS =>
- val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
- .map { entry => (entry.name, entry) }.toMap
+ alterResourceConfig(adminClient, entityTypeHead, entityNameHead, configsToBeDeleted, configsToBeAdded, ConfigResource.Type.CLIENT_METRICS)
- // fail the command if any of the configs to be deleted does not exist
- val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
- if (invalidConfigs.nonEmpty)
- throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+ case ConfigType.GROUP =>
+ alterResourceConfig(adminClient, entityTypeHead, entityNameHead, configsToBeDeleted, configsToBeAdded, ConfigResource.Type.GROUP)
- val configResource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, entityNameHead)
- val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
- val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
- ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
- ).asJavaCollection
- adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
case _ => throw new IllegalArgumentException(s"Unsupported entity type: $entityTypeHead")
}
@@ -542,7 +522,7 @@ object ConfigCommand extends Logging {
val describeAll = opts.options.has(opts.allOpt)
entityTypes.head match {
- case ConfigType.TOPIC | ConfigType.BROKER | BrokerLoggerConfigType | ConfigType.CLIENT_METRICS =>
+ case ConfigType.TOPIC | ConfigType.BROKER | BrokerLoggerConfigType | ConfigType.CLIENT_METRICS | ConfigType.GROUP =>
describeResourceConfig(adminClient, entityTypes.head, entityNames.headOption, describeAll)
case ConfigType.USER | ConfigType.CLIENT =>
describeClientQuotaAndUserScramCredentialConfigs(adminClient, entityTypes, entityNames)
@@ -562,6 +542,8 @@ object ConfigCommand extends Logging {
adminClient.describeCluster(new DescribeClusterOptions()).nodes().get().asScala.map(_.idString).toSeq :+ BrokerDefaultEntityName
case ConfigType.CLIENT_METRICS =>
adminClient.listClientMetricsResources().all().get().asScala.map(_.name).toSeq
+ case ConfigType.GROUP =>
+ adminClient.listConsumerGroups().all.get.asScala.map(_.groupId).toSeq
case entityType => throw new IllegalArgumentException(s"Invalid entity type: $entityType")
})
@@ -580,6 +562,23 @@ object ConfigCommand extends Logging {
}
}
+ private def alterResourceConfig(adminClient: Admin, entityTypeHead: String, entityNameHead: String, configsToBeDeleted: Seq[String], configsToBeAdded: Map[String, ConfigEntry], resourceType: ConfigResource.Type): Unit = {
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
+ .map { entry => (entry.name, entry) }.toMap
+
+ // fail the command if any of the configs to be deleted does not exist
+ val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
+ if (invalidConfigs.nonEmpty)
+ throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+
+ val configResource = new ConfigResource(resourceType, entityNameHead)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
+ ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
+ ).asJavaCollection
+ adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
+ }
+
private def getResourceConfig(adminClient: Admin, entityType: String, entityName: String, includeSynonyms: Boolean, describeAll: Boolean) = {
def validateBrokerId(): Unit = try entityName.toInt catch {
case _: NumberFormatException =>
@@ -604,6 +603,8 @@ object ConfigCommand extends Logging {
(ConfigResource.Type.BROKER_LOGGER, None)
case ConfigType.CLIENT_METRICS =>
(ConfigResource.Type.CLIENT_METRICS, Some(ConfigEntry.ConfigSource.DYNAMIC_CLIENT_METRICS_CONFIG))
+ case ConfigType.GROUP =>
+ (ConfigResource.Type.GROUP, Some(ConfigEntry.ConfigSource.DYNAMIC_GROUP_CONFIG))
case entityType => throw new IllegalArgumentException(s"Invalid entity type: $entityType")
}
@@ -700,6 +701,7 @@ object ConfigCommand extends Logging {
case ConfigType.USER => "user-principal"
case ConfigType.CLIENT => "client-id"
case ConfigType.TOPIC => "topic"
+ case ConfigType.GROUP => "group"
case t => t
}
sanitizedName match {
@@ -791,7 +793,7 @@ object ConfigCommand extends Logging {
}
val entities = entityTypes.map(t => Entity(t, if (sortedNames.hasNext) Some(sanitizeName(t, sortedNames.next())) else None))
- ConfigEntity(entities.head, if (entities.size > 1) Some(entities(1)) else None)
+ ConfigEntity(entities.head, entities.lift(1))
}
class ConfigCommandOptions(args: Array[String]) extends CommandDefaultOptions(args) {
@@ -819,10 +821,10 @@ object ConfigCommand extends Logging {
val describeOpt: OptionSpecBuilder = parser.accepts("describe", "List configs for the given entity.")
val allOpt: OptionSpecBuilder = parser.accepts("all", "List all configs for the given topic, broker, or broker-logger entity (includes static configuration when the entity type is brokers)")
- val entityType: OptionSpec[String] = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers/ips/client-metrics)")
+ val entityType: OptionSpec[String] = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers/ips/client-metrics/groups)")
.withRequiredArg
.ofType(classOf[String])
- val entityName: OptionSpec[String] = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/ip/client metrics)")
+ val entityName: OptionSpec[String] = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/ip/client metrics/group id)")
.withRequiredArg
.ofType(classOf[String])
private val entityDefault: OptionSpecBuilder = parser.accepts("entity-default", "Default entity name for clients/users/brokers/ips (applies to corresponding entity type in command line)")
@@ -835,6 +837,7 @@ object ConfigCommand extends Logging {
"For entity-type '" + ConfigType.CLIENT + "': " + DynamicConfig.Client.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
"For entity-type '" + ConfigType.IP + "': " + DynamicConfig.Ip.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
"For entity-type '" + ConfigType.CLIENT_METRICS + "': " + DynamicConfig.ClientMetrics.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
+ "For entity-type '" + ConfigType.GROUP + "': " + DynamicConfig.Group.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
s"Entity types '${ConfigType.USER}' and '${ConfigType.CLIENT}' may be specified together to update config for clients of a specific user.")
.withRequiredArg
.ofType(classOf[String])
@@ -868,6 +871,9 @@ object ConfigCommand extends Logging {
val ip: OptionSpec[String] = parser.accepts("ip", "The IP address.")
.withRequiredArg
.ofType(classOf[String])
+ val group: OptionSpec[String] = parser.accepts("group", "The group's ID.")
+ .withRequiredArg
+ .ofType(classOf[String])
val zkTlsConfigFile: OptionSpec[String] = parser.accepts("zk-tls-config-file",
"Identifies the file where ZooKeeper client TLS connectivity properties are defined. Any properties other than " +
ZkConfigs.ZK_SSL_CONFIG_TO_SYSTEM_PROPERTY_MAP.asScala.keys.toList.sorted.mkString(", ") + " are ignored.")
@@ -879,7 +885,8 @@ object ConfigCommand extends Logging {
(user, ConfigType.USER),
(broker, ConfigType.BROKER),
(brokerLogger, BrokerLoggerConfigType),
- (ip, ConfigType.IP))
+ (ip, ConfigType.IP),
+ (group, ConfigType.GROUP))
private val entityDefaultsFlags = List((clientDefaults, ConfigType.CLIENT),
(userDefaults, ConfigType.USER),
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index f17efd57435..033c41fd096 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -22,6 +22,7 @@ import kafka.server.DynamicBrokerConfig.AllDynamicConfigs
import java.net.{InetAddress, UnknownHostException}
import java.util.Properties
import org.apache.kafka.common.config.ConfigDef
+import org.apache.kafka.coordinator.group.GroupConfig
import org.apache.kafka.server.config.{QuotaConfigs, ZooKeeperInternals}
import java.util
@@ -75,7 +76,7 @@ object DynamicConfig {
}
object Ip {
- private val ipConfigs = QuotaConfigs.ipConfigs()
+ private val ipConfigs = QuotaConfigs.ipConfigs
def configKeys: util.Map[String, ConfigDef.ConfigKey] = ipConfigs.configKeys
@@ -101,9 +102,15 @@ object DynamicConfig {
def names: util.Set[String] = clientConfigs.names
}
+ object Group {
+ private val groupConfigs = GroupConfig.configDef()
+
+ def names: util.Set[String] = groupConfigs.names
+ }
+
private def validate(configDef: ConfigDef, props: Properties, customPropsAllowed: Boolean) = {
// Validate Names
- val names = configDef.names()
+ val names = configDef.names
val propKeys = props.keySet.asScala.map(_.asInstanceOf[String])
if (!customPropsAllowed) {
val unknownKeys = propKeys.filterNot(names.contains(_))
diff --git a/core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java b/core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java
index e666fb8c7ac..2270b3f1c8a 100644
--- a/core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java
+++ b/core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java
@@ -19,6 +19,7 @@ package kafka.admin;
import kafka.cluster.Broker;
import kafka.cluster.EndPoint;
import kafka.test.ClusterInstance;
+import kafka.test.annotation.ClusterConfigProperty;
import kafka.test.annotation.ClusterTest;
import kafka.test.annotation.Type;
import kafka.test.junit.ClusterTestExtensions;
@@ -62,6 +63,8 @@ import static java.util.Arrays.asList;
import static java.util.Collections.singleton;
import static java.util.Collections.singletonList;
import static java.util.Collections.singletonMap;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.NEW_GROUP_COORDINATOR_ENABLE_CONFIG;
import static org.apache.kafka.security.PasswordEncoderConfigs.PASSWORD_ENCODER_CIPHER_ALGORITHM_CONFIG;
import static org.apache.kafka.security.PasswordEncoderConfigs.PASSWORD_ENCODER_ITERATIONS_CONFIG;
import static org.apache.kafka.security.PasswordEncoderConfigs.PASSWORD_ENCODER_KEYFACTORY_ALGORITHM_CONFIG;
@@ -78,6 +81,7 @@ public class ConfigCommandIntegrationTest {
private List<String> alterOpts;
private final String defaultBrokerId = "0";
+ private final String defaultGroupName = "group";
private final ClusterInstance cluster;
private static Runnable run(Stream<String> command) {
@@ -115,6 +119,21 @@ public class ConfigCommandIntegrationTest {
errOut -> assertTrue(errOut.contains("User configuration updates using ZooKeeper are only supported for SCRAM credential updates."), errOut));
}
+ @ClusterTest(types = {Type.ZK})
+ public void testExitWithNonZeroStatusOnZkCommandAlterGroup() {
+ assertNonZeroStatusExit(Stream.concat(quorumArgs(), Stream.of(
+ "--entity-type", "groups",
+ "--entity-name", "group",
+ "--alter", "--add-config", "consumer.session.timeout.ms=50000")),
+ errOut -> assertTrue(errOut.contains("Invalid entity type groups, the entity type must be one of users, brokers with a --zookeeper argument"), errOut));
+
+ // Test for the --group alias
+ assertNonZeroStatusExit(Stream.concat(quorumArgs(), Stream.of(
+ "--group", "group",
+ "--alter", "--add-config", "consumer.session.timeout.ms=50000")),
+ errOut -> assertTrue(errOut.contains("Invalid entity type groups, the entity type must be one of users, brokers with a --zookeeper argument"), errOut));
+ }
+
@ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT})
public void testNullStatusOnKraftCommandAlterUserQuota() {
Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
@@ -126,6 +145,23 @@ public class ConfigCommandIntegrationTest {
assertTrue(StringUtils.isBlank(message), message);
}
+ @ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT})
+ public void testNullStatusOnKraftCommandAlterGroup() {
+ Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
+ "--entity-type", "groups",
+ "--entity-name", "group",
+ "--alter", "--add-config", "consumer.session.timeout.ms=50000"));
+ String message = captureStandardMsg(run(command));
+ assertTrue(StringUtils.isBlank(message), message);
+
+ // Test for the --group alias
+ command = Stream.concat(quorumArgs(), Stream.of(
+ "--group", "group",
+ "--alter", "--add-config", "consumer.session.timeout.ms=50000"));
+ message = captureStandardMsg(run(command));
+ assertTrue(StringUtils.isBlank(message), message);
+ }
+
@ClusterTest(types = Type.ZK)
public void testDynamicBrokerConfigUpdateUsingZooKeeper() throws Exception {
cluster.shutdownBroker(0);
@@ -236,7 +272,7 @@ public class ConfigCommandIntegrationTest {
singletonMap("listener.name.internal.ssl.keystore.location", "/tmp/test.jks"));
// Per-broker config configured at default cluster-level should fail
assertThrows(ExecutionException.class,
- () -> alterConfigWithKraft(client, Optional.empty(),
+ () -> alterConfigWithKraft(client, Optional.empty(),
singletonMap("listener.name.internal.ssl.keystore.location", "/tmp/test.jks")));
deleteAndVerifyConfigValue(client, defaultBrokerId,
singleton("listener.name.internal.ssl.keystore.location"), false);
@@ -256,6 +292,39 @@ public class ConfigCommandIntegrationTest {
}
}
+ @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, serverProperties = {
+ @ClusterConfigProperty(key = NEW_GROUP_COORDINATOR_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, value = "classic,consumer"),
+ })
+ public void testGroupConfigUpdateUsingKraft() throws Exception {
+ alterOpts = asList("--bootstrap-server", cluster.bootstrapServers(), "--entity-type", "groups", "--alter");
+ verifyGroupConfigUpdate();
+
+ // Test for the --group alias
+ alterOpts = asList("--bootstrap-server", cluster.bootstrapServers(), "--group", "--alter");
+ verifyGroupConfigUpdate();
+ }
+
+ private void verifyGroupConfigUpdate() throws Exception {
+ try (Admin client = cluster.createAdminClient()) {
+ // Add config
+ Map<String, String> configs = new HashMap<>();
+ configs.put("consumer.session.timeout.ms", "50000");
+ configs.put("consumer.heartbeat.interval.ms", "6000");
+ alterAndVerifyGroupConfig(client, defaultGroupName, configs);
+
+ // Delete config
+ configs.put("consumer.session.timeout.ms", "45000");
+ configs.put("consumer.heartbeat.interval.ms", "5000");
+ deleteAndVerifyGroupConfigValue(client, defaultGroupName, configs);
+
+ // Unknown config configured should fail
+ assertThrows(ExecutionException.class,
+ () -> alterConfigWithKraft(client, Optional.of(defaultGroupName),
+ singletonMap("unknown.config", "20000")));
+ }
+ }
+
@ClusterTest(types = {Type.ZK})
public void testAlterReadOnlyConfigInZookeeperThenShouldFail() {
cluster.shutdownBroker(0);
@@ -488,10 +557,15 @@ public class ConfigCommandIntegrationTest {
verifyConfig(client, brokerId, config);
}
- private void alterConfigWithKraft(Admin client, Optional<String> brokerId, Map<String, String> config) {
+ private void alterAndVerifyGroupConfig(Admin client, String groupName, Map<String, String> config) throws Exception {
+ alterConfigWithKraft(client, Optional.of(groupName), config);
+ verifyGroupConfig(client, groupName, config);
+ }
+
+ private void alterConfigWithKraft(Admin client, Optional<String> resourceName, Map<String, String> config) {
String configStr = transferConfigMapToString(config);
ConfigCommand.ConfigCommandOptions addOpts =
- new ConfigCommand.ConfigCommandOptions(toArray(alterOpts, entityOp(brokerId), asList("--add-config", configStr)));
+ new ConfigCommand.ConfigCommandOptions(toArray(alterOpts, entityOp(resourceName), asList("--add-config", configStr)));
ConfigCommand.alterConfig(client, addOpts);
}
@@ -505,6 +579,16 @@ public class ConfigCommandIntegrationTest {
}, 10000, config + " are not updated");
}
+ private void verifyGroupConfig(Admin client, String groupName, Map<String, String> config) throws Exception {
+ ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupName);
+ TestUtils.waitForCondition(() -> {
+ Map<String, String> current = getConfigEntryStream(client, configResource)
+ .filter(configEntry -> Objects.nonNull(configEntry.value()))
+ .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value));
+ return config.entrySet().stream().allMatch(e -> e.getValue().equals(current.get(e.getKey())));
+ }, 10000, config + " are not updated");
+ }
+
private Stream<ConfigEntry> getConfigEntryStream(Admin client,
ConfigResource configResource) throws InterruptedException, ExecutionException {
return client.describeConfigs(singletonList(configResource))
@@ -515,9 +599,9 @@ public class ConfigCommandIntegrationTest {
.flatMap(e -> e.entries().stream());
}
- private void deleteAndVerifyConfigValue(Admin client,
- String brokerId,
- Set<String> config,
+ private void deleteAndVerifyConfigValue(Admin client,
+ String brokerId,
+ Set<String> config,
boolean hasDefaultValue) throws Exception {
ConfigCommand.ConfigCommandOptions deleteOpts =
new ConfigCommand.ConfigCommandOptions(toArray(alterOpts, asList("--entity-name", brokerId),
@@ -526,6 +610,16 @@ public class ConfigCommandIntegrationTest {
verifyPerBrokerConfigValue(client, brokerId, config, hasDefaultValue);
}
+ private void deleteAndVerifyGroupConfigValue(Admin client,
+ String groupName,
+ Map<String, String> defaultConfigs) throws Exception {
+ ConfigCommand.ConfigCommandOptions deleteOpts =
+ new ConfigCommand.ConfigCommandOptions(toArray(alterOpts, asList("--entity-name", groupName),
+ asList("--delete-config", String.join(",", defaultConfigs.keySet()))));
+ ConfigCommand.alterConfig(client, deleteOpts);
+ verifyGroupConfig(client, groupName, defaultConfigs);
+ }
+
private void verifyPerBrokerConfigValue(Admin client,
String brokerId,
Set<String> config,
diff --git a/core/src/test/java/kafka/admin/ConfigCommandTest.java b/core/src/test/java/kafka/admin/ConfigCommandTest.java
index fccccff1f7a..82426c8939c 100644
--- a/core/src/test/java/kafka/admin/ConfigCommandTest.java
+++ b/core/src/test/java/kafka/admin/ConfigCommandTest.java
@@ -127,6 +127,13 @@ public class ConfigCommandTest {
"--describe")));
}
+ @Test
+ public void shouldExitWithNonZeroStatusOnZkCommandWithGroupsEntity() {
+ assertNonZeroStatusExit(toArray(ZOOKEEPER_BOOTSTRAP, Arrays.asList(
+ "--entity-type", "groups",
+ "--describe")));
+ }
+
@Test
public void shouldExitWithNonZeroStatusAlterUserQuotaWithoutEntityName() {
assertNonZeroStatusExit(toArray(BROKER_BOOTSTRAP, Arrays.asList(
@@ -261,6 +268,21 @@ public class ConfigCommandTest {
testArgumentParse(CONTROLLER_BOOTSTRAP, "ips");
}
+ @Test
+ public void shouldFailParseArgumentsForGroupEntityTypeUsingZookeeper() {
+ assertThrows(IllegalArgumentException.class, () -> testArgumentParse(ZOOKEEPER_BOOTSTRAP, "groups"));
+ }
+
+ @Test
+ public void shouldParseArgumentsForGroupEntityTypeWithBrokerBootstrap() {
+ testArgumentParse(BROKER_BOOTSTRAP, "groups");
+ }
+
+ @Test
+ public void shouldParseArgumentsForGroupEntityTypeWithControllerBootstrap() {
+ testArgumentParse(CONTROLLER_BOOTSTRAP, "groups");
+ }
+
public void testArgumentParse(List<String> bootstrapArguments, String entityType) {
String shortFlag = "--" + entityType.substring(0, entityType.length() - 1);
String connectOpts1 = bootstrapArguments.get(0);
@@ -437,18 +459,21 @@ public class ConfigCommandTest {
if (!zkConfig) {
testExpectedEntityTypeNames(Collections.singletonList(ConfigType.TOPIC), Collections.singletonList("A"), connectOpts, "--entity-type", "topics", "--entity-name", "A");
testExpectedEntityTypeNames(Collections.singletonList(ConfigType.IP), Collections.singletonList("1.2.3.4"), connectOpts, "--entity-name", "1.2.3.4", "--entity-type", "ips");
+ testExpectedEntityTypeNames(Collections.singletonList(ConfigType.GROUP), Collections.singletonList("A"), connectOpts, "--entity-type", "groups", "--entity-name", "A");
testExpectedEntityTypeNames(Arrays.asList(ConfigType.USER, ConfigType.CLIENT), Arrays.asList("A", ""), connectOpts,
"--entity-type", "users", "--entity-type", "clients", "--entity-name", "A", "--entity-default");
testExpectedEntityTypeNames(Arrays.asList(ConfigType.USER, ConfigType.CLIENT), Arrays.asList("", "B"), connectOpts,
"--entity-default", "--entity-name", "B", "--entity-type", "users", "--entity-type", "clients");
testExpectedEntityTypeNames(Collections.singletonList(ConfigType.TOPIC), Collections.singletonList("A"), connectOpts, "--topic", "A");
testExpectedEntityTypeNames(Collections.singletonList(ConfigType.IP), Collections.singletonList("1.2.3.4"), connectOpts, "--ip", "1.2.3.4");
+ testExpectedEntityTypeNames(Collections.singletonList(ConfigType.GROUP), Collections.singletonList("A"), connectOpts, "--group", "A");
testExpectedEntityTypeNames(Arrays.asList(ConfigType.CLIENT, ConfigType.USER), Arrays.asList("B", "A"), connectOpts, "--client", "B", "--user", "A");
testExpectedEntityTypeNames(Arrays.asList(ConfigType.CLIENT, ConfigType.USER), Arrays.asList("B", ""), connectOpts, "--client", "B", "--user-defaults");
testExpectedEntityTypeNames(Arrays.asList(ConfigType.CLIENT, ConfigType.USER), Collections.singletonList("A"), connectOpts,
"--entity-type", "clients", "--entity-type", "users", "--entity-name", "A");
testExpectedEntityTypeNames(Collections.singletonList(ConfigType.TOPIC), Collections.emptyList(), connectOpts, "--entity-type", "topics");
testExpectedEntityTypeNames(Collections.singletonList(ConfigType.IP), Collections.emptyList(), connectOpts, "--entity-type", "ips");
+ testExpectedEntityTypeNames(Collections.singletonList(ConfigType.GROUP), Collections.emptyList(), connectOpts, "--entity-type", "groups");
}
testExpectedEntityTypeNames(Collections.singletonList(ConfigType.BROKER), Collections.singletonList("0"), connectOpts, "--entity-name", "0", "--entity-type", "brokers");
@@ -1918,6 +1943,187 @@ public class ConfigCommandTest {
assertEquals("client-metrics is not a known entityType. Should be one of List(topics, clients, users, brokers, ips)", exception.getMessage());
}
+ @Test
+ public void shouldAlterGroupConfig() {
+ Node node = new Node(1, "localhost", 9092);
+ verifyAlterGroupConfig(node, "group", Arrays.asList("--entity-type", "groups", "--entity-name", "group"));
+
+ // Test for the --group alias
+ verifyAlterGroupConfig(node, "groupUsingAlias", Arrays.asList("--group", "groupUsingAlias"));
+ }
+
+ private void verifyAlterGroupConfig(Node node, String resourceName, List<String> resourceOpts) {
+ List<String> optsList = concat(Arrays.asList("--bootstrap-server", "localhost:9092",
+ "--alter",
+ "--delete-config", "consumer.session.timeout.ms",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"), resourceOpts);
+ ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray(optsList));
+
+ ConfigResource resource = new ConfigResource(ConfigResource.Type.GROUP, resourceName);
+ List<ConfigEntry> configEntries = Collections.singletonList(new ConfigEntry("consumer.session.timeout.ms", "45000",
+ ConfigEntry.ConfigSource.DYNAMIC_GROUP_CONFIG, false, false, Collections.emptyList(),
+ ConfigEntry.ConfigType.UNKNOWN, null));
+ KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
+ future.complete(Collections.singletonMap(resource, new Config(configEntries)));
+ DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
+ when(describeResult.all()).thenReturn(future);
+
+ KafkaFutureImpl<Void> alterFuture = new KafkaFutureImpl<>();
+ alterFuture.complete(null);
+ AlterConfigsResult alterResult = mock(AlterConfigsResult.class);
+ when(alterResult.all()).thenReturn(alterFuture);
+
+ MockAdminClient mockAdminClient = new MockAdminClient(Collections.singletonList(node), node) {
+ @Override
+ public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
+ assertFalse(options.includeSynonyms(), "Config synonyms requested unnecessarily");
+ assertEquals(1, resources.size());
+ ConfigResource res = resources.iterator().next();
+ assertEquals(ConfigResource.Type.GROUP, res.type());
+ assertEquals(resourceName, res.name());
+ return describeResult;
+ }
+
+ @Override
+ public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs, AlterConfigsOptions options) {
+ assertEquals(1, configs.size());
+ Map.Entry<ConfigResource, Collection<AlterConfigOp>> entry = configs.entrySet().iterator().next();
+ ConfigResource res = entry.getKey();
+ Collection<AlterConfigOp> alterConfigOps = entry.getValue();
+ assertEquals(ConfigResource.Type.GROUP, res.type());
+ assertEquals(2, alterConfigOps.size());
+
+ List<AlterConfigOp> expectedConfigOps = Arrays.asList(
+ new AlterConfigOp(new ConfigEntry("consumer.heartbeat.interval.ms", "6000"), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry("consumer.session.timeout.ms", ""), AlterConfigOp.OpType.DELETE)
+ );
+ assertEquals(expectedConfigOps.size(), alterConfigOps.size());
+ Iterator<AlterConfigOp> alterConfigOpsIter = alterConfigOps.iterator();
+ for (AlterConfigOp expectedConfigOp : expectedConfigOps) {
+ assertEquals(expectedConfigOp, alterConfigOpsIter.next());
+ }
+ return alterResult;
+ }
+ };
+ ConfigCommand.alterConfig(mockAdminClient, alterOpts);
+ verify(describeResult).all();
+ verify(alterResult).all();
+ }
+
+ @Test
+ public void shouldDescribeGroupConfigWithoutEntityName() {
+ ConfigCommand.ConfigCommandOptions describeOpts = new ConfigCommand.ConfigCommandOptions(toArray("--bootstrap-server", "localhost:9092",
+ "--entity-type", "groups",
+ "--entity-name", "group",
+ "--describe"));
+ verifyDescribeGroupConfig(describeOpts, "group");
+
+ // Test for the --group alias
+ describeOpts = new ConfigCommand.ConfigCommandOptions(toArray("--bootstrap-server", "localhost:9092", "--group", "groupUsingAlias", "--describe"));
+ verifyDescribeGroupConfig(describeOpts, "groupUsingAlias");
+ }
+
+ private void verifyDescribeGroupConfig(ConfigCommand.ConfigCommandOptions describeOpts, String resourceName) {
+ ConfigResource resourceCustom = new ConfigResource(ConfigResource.Type.GROUP, resourceName);
+ ConfigEntry configEntry = new ConfigEntry("consumer.heartbeat.interval.ms", "6000");
+ KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
+ DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
+ when(describeResult.all()).thenReturn(future);
+
+ Node node = new Node(1, "localhost", 9092);
+ MockAdminClient mockAdminClient = new MockAdminClient(Collections.singletonList(node), node) {
+ @Override
+ public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
+ assertTrue(options.includeSynonyms());
+ assertEquals(1, resources.size());
+ ConfigResource resource = resources.iterator().next();
+ assertEquals(ConfigResource.Type.GROUP, resource.type());
+ assertEquals(resourceCustom.name(), resource.name());
+ future.complete(Collections.singletonMap(resourceCustom, new Config(Collections.singletonList(configEntry))));
+ return describeResult;
+ }
+ };
+ mockAdminClient.incrementalAlterConfigs(Collections.singletonMap(resourceCustom,
+ Collections.singletonList(new AlterConfigOp(configEntry, AlterConfigOp.OpType.SET))), new AlterConfigsOptions());
+ ConfigCommand.describeConfig(mockAdminClient, describeOpts);
+ verify(describeResult).all();
+ }
+
+ @Test
+ public void shouldNotAlterGroupConfigWithoutEntityName() {
+ ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray("--bootstrap-server", "localhost:9092",
+ "--entity-type", "groups",
+ "--alter",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"));
+
+ IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, alterOpts::checkArgs);
+ assertEquals("an entity name must be specified with --alter of groups", exception.getMessage());
+ }
+
+ @Test
+ public void shouldNotSupportAlterGroupConfigWithZookeeperArg() {
+ ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--entity-name", "group",
+ "--entity-type", "groups",
+ "--alter",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"));
+
+ IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, alterOpts::checkArgs);
+ assertEquals("Invalid entity type groups, the entity type must be one of users, brokers with a --zookeeper argument", exception.getMessage());
+
+ // Test for the --group alias
+ alterOpts = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--group", "group",
+ "--alter",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"));
+
+ exception = assertThrows(IllegalArgumentException.class, alterOpts::checkArgs);
+ assertEquals("Invalid entity type groups, the entity type must be one of users, brokers with a --zookeeper argument", exception.getMessage());
+
+ }
+
+ @Test
+ public void shouldNotSupportDescribeGroupConfigWithZookeeperArg() {
+ ConfigCommand.ConfigCommandOptions describeOpts = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--entity-name", "group",
+ "--entity-type", "groups",
+ "--describe"));
+
+ IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, describeOpts::checkArgs);
+ assertEquals("Invalid entity type groups, the entity type must be one of users, brokers with a --zookeeper argument", exception.getMessage());
+
+ // Test for the --group alias
+ describeOpts = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--group", "group",
+ "--describe"));
+
+ exception = assertThrows(IllegalArgumentException.class, describeOpts::checkArgs);
+ assertEquals("Invalid entity type groups, the entity type must be one of users, brokers with a --zookeeper argument", exception.getMessage());
+
+ }
+
+ @Test
+ public void shouldNotSupportAlterGroupConfigWithZookeeper() {
+ ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--entity-name", "group",
+ "--entity-type", "groups",
+ "--alter",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"));
+
+ IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> ConfigCommand.alterConfigWithZk(null, alterOpts, DUMMY_ADMIN_ZK_CLIENT));
+ assertEquals("groups is not a known entityType. Should be one of List(topics, clients, users, brokers, ips)", exception.getMessage());
+
+ // Test for the --group alias
+ ConfigCommand.ConfigCommandOptions alterOptsUsingAlias = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--group", "group",
+ "--alter",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"));
+
+ exception = assertThrows(IllegalArgumentException.class, () -> ConfigCommand.alterConfigWithZk(null, alterOptsUsingAlias, DUMMY_ADMIN_ZK_CLIENT));
+ assertEquals("groups is not a known entityType. Should be one of List(topics, clients, users, brokers, ips)", exception.getMessage());
+
+ }
+
public static String[] toArray(String... first) {
return first;
}
diff --git a/core/src/test/scala/unit/kafka/server/metadata/ZkConfigRepositoryTest.scala b/core/src/test/scala/unit/kafka/server/metadata/ZkConfigRepositoryTest.scala
index 6ba5e954a55..33ec03038d0 100644
--- a/core/src/test/scala/unit/kafka/server/metadata/ZkConfigRepositoryTest.scala
+++ b/core/src/test/scala/unit/kafka/server/metadata/ZkConfigRepositoryTest.scala
@@ -53,5 +53,7 @@ class ZkConfigRepositoryTest {
assertThrows(classOf[IllegalArgumentException], () => zkConfigRepository.config(new ConfigResource(value, value.toString))))
// Validate exception for CLIENT_METRICS.
assertThrows(classOf[InvalidRequestException], () => zkConfigRepository.config(new ConfigResource(Type.CLIENT_METRICS, Type.CLIENT_METRICS.toString)))
+ // Validate exception for GROUP.
+ assertThrows(classOf[InvalidRequestException], () => zkConfigRepository.config(new ConfigResource(Type.GROUP, Type.GROUP.toString)))
}
}
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 5cc5ceb8df7..6c563741b19 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
@@ -66,6 +66,10 @@ public class GroupConfig extends AbstractConfig {
this.consumerHeartbeatIntervalMs = getInt(CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG);
}
+ public static ConfigDef configDef() {
+ return CONFIG;
+ }
+
public static Optional<Type> configType(String configName) {
return Optional.ofNullable(CONFIG.configKeys().get(configName)).map(c -> c.type);
}