From 52ede8d05b6452ea67cdab4f6461dcc18c3f0715 Mon Sep 17 00:00:00 2001 From: Aneel Nazareth Date: Wed, 26 Feb 2020 14:05:24 -0600 Subject: [PATCH 1/6] KAKFA-9612 CLI Dynamic Configuration with file input --- .../scala/kafka/admin/ConfigCommand.scala | 34 +++++++--- .../unit/kafka/admin/ConfigCommandTest.scala | 66 ++++++++++++++++++- 2 files changed, 89 insertions(+), 11 deletions(-) diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala index a43fe9e539cd9..2d68d74a85652 100644 --- a/core/src/main/scala/kafka/admin/ConfigCommand.scala +++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala @@ -17,6 +17,7 @@ package kafka.admin +import java.io.{File, FileInputStream} import java.util.concurrent.TimeUnit import java.util.{Collections, Properties} @@ -247,6 +248,15 @@ object ConfigCommand extends Config { private[admin] def parseConfigsToBeAdded(opts: ConfigCommandOptions): Properties = { val props = new Properties + if (opts.options.has(opts.addConfigFile)) { + val file = opts.options.valueOf(opts.addConfigFile) + val inputStream = new FileInputStream(file) + try { + props.load(inputStream) + } finally { + inputStream.close() + } + } if (opts.options.has(opts.addConfig)) { // Split list by commas, but avoid those in [], then into KV pairs // Each KV pair is of format key=value, split them into key and value, using -1 as the limit for split() to @@ -258,10 +268,10 @@ object ConfigCommand extends Config { require(configsToBeAdded.forall(config => config.length == 2), "Invalid entity config: all configs to be added must be in the format \"key=val\".") //Create properties, parsing square brackets from values if necessary configsToBeAdded.foreach(pair => props.setProperty(pair(0).trim, pair(1).replaceAll("\\[?\\]?", "").trim)) - if (props.containsKey(LogConfig.MessageFormatVersionProp)) { - println(s"WARNING: The configuration ${LogConfig.MessageFormatVersionProp}=${props.getProperty(LogConfig.MessageFormatVersionProp)} is specified. " + - s"This configuration will be ignored if the version is newer than the inter.broker.protocol.version specified in the broker.") - } + } + if (props.containsKey(LogConfig.MessageFormatVersionProp)) { + println(s"WARNING: The configuration ${LogConfig.MessageFormatVersionProp}=${props.getProperty(LogConfig.MessageFormatVersionProp)} is specified. " + + s"This configuration will be ignored if the version is newer than the inter.broker.protocol.version specified in the broker.") } props } @@ -645,6 +655,9 @@ object ConfigCommand extends Config { 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]) + val addConfigFile = parser.accepts("add-config-file", "Path to a properties file with configs to add. See add-config for a list of valid configurations.") + .withRequiredArg + .ofType(classOf[File]) val deleteConfig = parser.accepts("delete-config", "config keys to remove 'k1,k2'") .withRequiredArg .ofType(classOf[String]) @@ -764,10 +777,15 @@ object ConfigCommand extends Config { } else if (!hasEntityName) throw new IllegalArgumentException(s"an entity name must be specified with --alter of ${entityTypeVals.mkString(",")}") - val isAddConfigPresent: Boolean = options.has(addConfig) - val isDeleteConfigPresent: Boolean = options.has(deleteConfig) - if (!isAddConfigPresent && !isDeleteConfigPresent) - throw new IllegalArgumentException("At least one of --add-config or --delete-config must be specified with --alter") + val isAddConfigPresent = options.has(addConfig) + val isAddConfigFilePresent = options.has(addConfigFile) + val isDeleteConfigPresent = options.has(deleteConfig) + + if(isAddConfigPresent && isAddConfigFilePresent) + throw new IllegalArgumentException("Only one of --add-config or --add-config-file must be specified") + + if(!isAddConfigPresent && !isAddConfigFilePresent && ! isDeleteConfigPresent) + throw new IllegalArgumentException("At least one of --add-config, --add-config-file, or --delete-config must be specified with --alter") } } } diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala index a963c74dd62bc..08cd146578a2f 100644 --- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala @@ -16,6 +16,7 @@ */ package kafka.admin +import java.io.{File, FileWriter} import java.util import java.util.Properties @@ -26,10 +27,10 @@ import kafka.server.{ConfigEntityName, ConfigType, KafkaConfig} import kafka.utils.{Exit, Logging} import kafka.zk.{AdminZkClient, BrokerInfo, KafkaZkClient, ZooKeeperTestHarness} import org.apache.kafka.clients.admin._ -import org.apache.kafka.common.config.{ConfigException, ConfigResource} -import org.apache.kafka.common.internals.KafkaFutureImpl import org.apache.kafka.common.Node +import org.apache.kafka.common.config.{ConfigException, ConfigResource} import org.apache.kafka.common.errors.InvalidConfigurationException +import org.apache.kafka.common.internals.KafkaFutureImpl import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity, ClientQuotaFilter} import org.apache.kafka.common.security.auth.SecurityProtocol @@ -40,8 +41,8 @@ import org.junit.Assert._ import org.junit.Test import org.scalatest.Assertions.intercept +import scala.collection.JavaConverters._ import scala.collection.{Seq, mutable} -import scala.jdk.CollectionConverters._ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { @@ -160,12 +161,25 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { "--add-config", "a=b,c=d")) createOpts.checkArgs() + createOpts = new ConfigCommandOptions(Array(connectOpts._1, connectOpts._2, + "--entity-name", "1", + "--entity-type", entityType, + "--alter", + "--add-config-file", "/tmp/new.properties")) + createOpts.checkArgs() + createOpts = new ConfigCommandOptions(Array(connectOpts._1, connectOpts._2, shortFlag, "1", "--alter", "--add-config", "a=b,c=d")) createOpts.checkArgs() + createOpts = new ConfigCommandOptions(Array(connectOpts._1, connectOpts._2, + shortFlag, "1", + "--alter", + "--add-config-file", "/tmp/new.properties")) + createOpts.checkArgs() + // For alter and deleted config createOpts = new ConfigCommandOptions(Array(connectOpts._1, connectOpts._2, "--entity-name", "1", @@ -226,6 +240,52 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { assertTrue(addedProps2.getProperty("f").isEmpty) } + @Test(expected = classOf[IllegalArgumentException]) + def shouldFailIfAddAndAddFile(): Unit = { + + // Should not parse correctly + val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", + "--entity-name", "1", + "--entity-type", "brokers", + "--alter", + "--add-config", "a=b,c=d", + "--add-config-file", "/tmp/new.properties" + )) + createOpts.checkArgs() + } + + @Test + def testParseConfigsToBeAddedForAddConfigFile(): Unit = { + val fileContents = + """a=b + |c = d + |json = {"key": "val"} + |nested = [[1, 2], [3, 4]] + |""".stripMargin + + val file = File.createTempFile("testParseConfigsToBeAddedForAddConfigFile", ".properties") + file.deleteOnExit() + val writer = new FileWriter(file) + writer.write(fileContents) + writer.close() + + val addConfigFileArgs = Array("--add-config-file", file.getPath) + + val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", + "--entity-name", "1", + "--entity-type", "brokers", + "--alter") + ++ addConfigFileArgs) + createOpts.checkArgs() + + val addedProps = ConfigCommand.parseConfigsToBeAdded(createOpts) + assertEquals(4, addedProps.size()) + assertEquals("b", addedProps.getProperty("a")) + assertEquals("d", addedProps.getProperty("c")) + assertEquals("{\"key\": \"val\"}", addedProps.getProperty("json")) + assertEquals("[[1, 2], [3, 4]]", addedProps.getProperty("nested")) + } + def doTestOptionEntityTypeNames(zkConfig: Boolean): Unit = { val connectOpts = if (zkConfig) ("--zookeeper", zkConnect) From d2957fb2c2901c1bb31a961cc981e5f932c6fcd9 Mon Sep 17 00:00:00 2001 From: Aneel Nazareth Date: Wed, 1 Apr 2020 14:23:37 -0500 Subject: [PATCH 2/6] PR feedback: whitespace, no stack trace for file not found --- core/src/main/scala/kafka/admin/ConfigCommand.scala | 10 +++++++--- .../scala/unit/kafka/admin/ConfigCommandTest.scala | 1 - 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala index 2d68d74a85652..b6afacc0bb364 100644 --- a/core/src/main/scala/kafka/admin/ConfigCommand.scala +++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala @@ -17,7 +17,7 @@ package kafka.admin -import java.io.{File, FileInputStream} +import java.io.{File, FileInputStream, FileNotFoundException} import java.util.concurrent.TimeUnit import java.util.{Collections, Properties} @@ -27,6 +27,7 @@ import kafka.log.LogConfig import kafka.server.{ConfigEntityName, ConfigType, Defaults, DynamicBrokerConfig, DynamicConfig, KafkaConfig} import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit, PasswordEncoder} import kafka.utils.Implicits._ +import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit, PasswordEncoder} import kafka.zk.{AdminZkClient, KafkaZkClient} import org.apache.kafka.clients.CommonClientConfigs import org.apache.kafka.clients.admin.{Admin, AlterClientQuotasOptions, AlterConfigOp, AlterConfigsOptions, ConfigEntry, DescribeClusterOptions, DescribeConfigsOptions, ListTopicsOptions, Config => JConfig} @@ -250,7 +251,10 @@ object ConfigCommand extends Config { val props = new Properties if (opts.options.has(opts.addConfigFile)) { val file = opts.options.valueOf(opts.addConfigFile) - val inputStream = new FileInputStream(file) + val inputStream = try new FileInputStream(file) catch { + case _: FileNotFoundException => + throw new IllegalArgumentException(s"No such file or directory: $file") + } try { props.load(inputStream) } finally { @@ -784,7 +788,7 @@ object ConfigCommand extends Config { if(isAddConfigPresent && isAddConfigFilePresent) throw new IllegalArgumentException("Only one of --add-config or --add-config-file must be specified") - if(!isAddConfigPresent && !isAddConfigFilePresent && ! isDeleteConfigPresent) + if(!isAddConfigPresent && !isAddConfigFilePresent && !isDeleteConfigPresent) throw new IllegalArgumentException("At least one of --add-config, --add-config-file, or --delete-config must be specified with --alter") } } diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala index 08cd146578a2f..81b04e4f78341 100644 --- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala @@ -242,7 +242,6 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { @Test(expected = classOf[IllegalArgumentException]) def shouldFailIfAddAndAddFile(): Unit = { - // Should not parse correctly val createOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", "--entity-name", "1", From 7b44920e823dbec4e06bfa005caee727d85909db Mon Sep 17 00:00:00 2001 From: Aneel Nazareth Date: Fri, 3 Apr 2020 09:54:14 -0500 Subject: [PATCH 3/6] added test --- .../unit/kafka/admin/ConfigCommandTest.scala | 23 ++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala index 81b04e4f78341..de933d9cff1d5 100644 --- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala @@ -483,12 +483,33 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { @Test def shouldAlterTopicConfig(): Unit = { + doShouldAlterTopicConfig(false) + } + + @Test + def shouldAlterTopicConfigFile(): Unit = { + doShouldAlterTopicConfig(true) + } + + def doShouldAlterTopicConfig(file: Boolean): Unit = { + var filePath = "" + val addedConfigs = Seq("delete.retention.ms=1000000", "min.insync.replicas=2") + if (file) { + val file = File.createTempFile("testParseConfigsToBeAddedForAddConfigFile", ".properties") + file.deleteOnExit() + val writer = new FileWriter(file) + writer.write(addedConfigs.mkString("\n")) + writer.close() + filePath = file.getPath + } + val resourceName = "my-topic" val alterOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092", "--entity-name", resourceName, "--entity-type", "topics", "--alter", - "--add-config", "delete.retention.ms=1000000,min.insync.replicas=2", + if (file) "--add-config-file" else "--add-config", + if (file) filePath else addedConfigs.mkString(","), "--delete-config", "unclean.leader.election.enable")) var alteredConfigs = false From 94d05f705e3944ce0794c902a2cf9ffeedff86f3 Mon Sep 17 00:00:00 2001 From: Aneel Nazareth Date: Wed, 8 Apr 2020 09:15:24 -0500 Subject: [PATCH 4/6] remove unnecessary imports changes --- core/src/main/scala/kafka/admin/ConfigCommand.scala | 1 - 1 file changed, 1 deletion(-) diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala index b6afacc0bb364..8e0d2447b8754 100644 --- a/core/src/main/scala/kafka/admin/ConfigCommand.scala +++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala @@ -27,7 +27,6 @@ import kafka.log.LogConfig import kafka.server.{ConfigEntityName, ConfigType, Defaults, DynamicBrokerConfig, DynamicConfig, KafkaConfig} import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit, PasswordEncoder} import kafka.utils.Implicits._ -import kafka.utils.{CommandDefaultOptions, CommandLineUtils, Exit, PasswordEncoder} import kafka.zk.{AdminZkClient, KafkaZkClient} import org.apache.kafka.clients.CommonClientConfigs import org.apache.kafka.clients.admin.{Admin, AlterClientQuotasOptions, AlterConfigOp, AlterConfigsOptions, ConfigEntry, DescribeClusterOptions, DescribeConfigsOptions, ListTopicsOptions, Config => JConfig} From cdff0b4dfe752eabe61cd9cf83de13fc4fec7950 Mon Sep 17 00:00:00 2001 From: Aneel Nazareth Date: Mon, 13 Apr 2020 09:23:19 -0500 Subject: [PATCH 5/6] remove "or directory" from error message about file --- core/src/main/scala/kafka/admin/ConfigCommand.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala index 8e0d2447b8754..9ded7b4b0ce56 100644 --- a/core/src/main/scala/kafka/admin/ConfigCommand.scala +++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala @@ -252,7 +252,7 @@ object ConfigCommand extends Config { val file = opts.options.valueOf(opts.addConfigFile) val inputStream = try new FileInputStream(file) catch { case _: FileNotFoundException => - throw new IllegalArgumentException(s"No such file or directory: $file") + throw new IllegalArgumentException(s"No such file: $file") } try { props.load(inputStream) From f8ef1ec1bfc9fe8f17a1f71931d566601b107414 Mon Sep 17 00:00:00 2001 From: Aneel Nazareth Date: Mon, 27 Apr 2020 09:54:23 -0500 Subject: [PATCH 6/6] PR feedback: use Utils --- .../java/org/apache/kafka/test/TestUtils.java | 14 ++++++++++++++ .../main/scala/kafka/admin/ConfigCommand.scala | 13 ++----------- .../unit/kafka/admin/ConfigCommandTest.scala | 16 ++++------------ 3 files changed, 20 insertions(+), 23 deletions(-) diff --git a/clients/src/test/java/org/apache/kafka/test/TestUtils.java b/clients/src/test/java/org/apache/kafka/test/TestUtils.java index fe0b4a0cdc321..f9be363c1a46f 100644 --- a/clients/src/test/java/org/apache/kafka/test/TestUtils.java +++ b/clients/src/test/java/org/apache/kafka/test/TestUtils.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.test; +import java.io.FileWriter; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.Cluster; @@ -228,6 +229,19 @@ public static File tempFile() throws IOException { return file; } + /** + * Create a file with the given contents in the default temporary-file directory, + * using `kafka` as the prefix and `tmp` as the suffix to generate its name. + */ + public static File tempFile(final String contents) throws IOException { + final File file = tempFile(); + final FileWriter writer = new FileWriter(file); + writer.write(contents); + writer.close(); + + return file; + } + /** * Create a temporary relative directory in the default temporary-file directory with the given prefix. * diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala index 9ded7b4b0ce56..5bac33d8489e5 100644 --- a/core/src/main/scala/kafka/admin/ConfigCommand.scala +++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala @@ -17,7 +17,6 @@ package kafka.admin -import java.io.{File, FileInputStream, FileNotFoundException} import java.util.concurrent.TimeUnit import java.util.{Collections, Properties} @@ -250,15 +249,7 @@ object ConfigCommand extends Config { val props = new Properties if (opts.options.has(opts.addConfigFile)) { val file = opts.options.valueOf(opts.addConfigFile) - val inputStream = try new FileInputStream(file) catch { - case _: FileNotFoundException => - throw new IllegalArgumentException(s"No such file: $file") - } - try { - props.load(inputStream) - } finally { - inputStream.close() - } + props.putAll(Utils.loadProps(file)) } if (opts.options.has(opts.addConfig)) { // Split list by commas, but avoid those in [], then into KV pairs @@ -660,7 +651,7 @@ object ConfigCommand extends Config { .ofType(classOf[String]) val addConfigFile = parser.accepts("add-config-file", "Path to a properties file with configs to add. See add-config for a list of valid configurations.") .withRequiredArg - .ofType(classOf[File]) + .ofType(classOf[String]) val deleteConfig = parser.accepts("delete-config", "config keys to remove 'k1,k2'") .withRequiredArg .ofType(classOf[String]) diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala index de933d9cff1d5..1a7d43e7b5100 100644 --- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala @@ -16,7 +16,6 @@ */ package kafka.admin -import java.io.{File, FileWriter} import java.util import java.util.Properties @@ -36,13 +35,14 @@ import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity, import org.apache.kafka.common.security.auth.SecurityProtocol import org.apache.kafka.common.security.scram.internals.ScramCredentialUtils import org.apache.kafka.common.utils.Sanitizer +import org.apache.kafka.test.TestUtils import org.easymock.EasyMock import org.junit.Assert._ import org.junit.Test import org.scalatest.Assertions.intercept -import scala.collection.JavaConverters._ import scala.collection.{Seq, mutable} +import scala.jdk.CollectionConverters._ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { @@ -262,11 +262,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { |nested = [[1, 2], [3, 4]] |""".stripMargin - val file = File.createTempFile("testParseConfigsToBeAddedForAddConfigFile", ".properties") - file.deleteOnExit() - val writer = new FileWriter(file) - writer.write(fileContents) - writer.close() + val file = TestUtils.tempFile(fileContents) val addConfigFileArgs = Array("--add-config-file", file.getPath) @@ -495,11 +491,7 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging { var filePath = "" val addedConfigs = Seq("delete.retention.ms=1000000", "min.insync.replicas=2") if (file) { - val file = File.createTempFile("testParseConfigsToBeAddedForAddConfigFile", ".properties") - file.deleteOnExit() - val writer = new FileWriter(file) - writer.write(addedConfigs.mkString("\n")) - writer.close() + val file = TestUtils.tempFile(addedConfigs.mkString("\n")) filePath = file.getPath }