From e536013bf58ad730c11a1d7b9f28b6729ef2bf0f Mon Sep 17 00:00:00 2001 From: Paolo Patierno Date: Wed, 30 May 2018 12:36:43 +0200 Subject: [PATCH 1/8] Removed the deprecated "--new-consumer" option for all consumer based tools --- .../kafka/admin/ConsumerGroupCommand.scala | 10 ---- .../scala/kafka/tools/ConsoleConsumer.scala | 9 ---- .../kafka/tools/ConsumerPerformance.scala | 10 +--- .../kafka/tools/ConsoleConsumerTest.scala | 22 ++++++-- .../kafka/tools/ConsumerPerformanceTest.scala | 53 ++++++++++++++++++- 5 files changed, 70 insertions(+), 34 deletions(-) diff --git a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala index 689a63c2e4a42..44d09fda5fa6e 100755 --- a/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala +++ b/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala @@ -889,8 +889,6 @@ object ConsumerGroupCommand extends Logging { "Pass in just a topic to delete the given topic's partition offsets and ownership information " + "for every consumer group. For instance --topic t1" + nl + "WARNING: Group deletion only works for old ZK-based consumer groups, and one has to use it carefully to only delete groups that are not active." - val NewConsumerDoc = "Use the new consumer implementation. This is the default, so this option is deprecated and " + - "will be removed in a future release." val TimeoutMsDoc = "The timeout that can be set for some use cases. For example, it can be used when describing the group " + "to specify the maximum amount of time in milliseconds to wait before the group stabilizes (when the group is just created, " + "or is going through some changes)." @@ -943,7 +941,6 @@ object ConsumerGroupCommand extends Logging { val listOpt = parser.accepts("list", ListDoc) val describeOpt = parser.accepts("describe", DescribeDoc) val deleteOpt = parser.accepts("delete", DeleteDoc) - val newConsumerOpt = parser.accepts("new-consumer", NewConsumerDoc) val timeoutMsOpt = parser.accepts("timeout", TimeoutMsDoc) .withRequiredArg .describedAs("timeout (ms)") @@ -1011,16 +1008,9 @@ object ConsumerGroupCommand extends Logging { if (useOldConsumer) { if (options.has(bootstrapServerOpt)) CommandLineUtils.printUsageAndDie(parser, s"Option $bootstrapServerOpt is not valid with $zkConnectOpt.") - else if (options.has(newConsumerOpt)) - CommandLineUtils.printUsageAndDie(parser, s"Option $newConsumerOpt is not valid with $zkConnectOpt.") } else { CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServerOpt) - if (options.has(newConsumerOpt)) { - Console.err.println(s"The --new-consumer option is deprecated and will be removed in a future major release. " + - s"The new consumer is used by default if the --bootstrap-server option is provided.") - } - if (options.has(deleteOpt) && options.has(topicOpt)) CommandLineUtils.printUsageAndDie(parser, s"When deleting a consumer group the option $topicOpt is only " + s"valid with $zkConnectOpt. The new consumer does not support topic-specific offset deletion from a consumer group.") diff --git a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala index c1f8b81a06697..b3103ebfa8851 100755 --- a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala +++ b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala @@ -324,8 +324,6 @@ object ConsoleConsumer extends Logging { .withRequiredArg .describedAs("metrics directory") .ofType(classOf[java.lang.String]) - val newConsumerOpt = parser.accepts("new-consumer", "Use the new consumer implementation. This is the default, so " + - "this option is deprecated and will be removed in a future release.") val bootstrapServerOpt = parser.accepts("bootstrap-server", "REQUIRED (unless old consumer is used): The server to connect to.") .withRequiredArg .describedAs("server to connect to") @@ -397,8 +395,6 @@ object ConsoleConsumer extends Logging { if (useOldConsumer) { if (options.has(bootstrapServerOpt)) CommandLineUtils.printUsageAndDie(parser, s"Option $bootstrapServerOpt is not valid with $zkConnectOpt.") - else if (options.has(newConsumerOpt)) - CommandLineUtils.printUsageAndDie(parser, s"Option $newConsumerOpt is not valid with $zkConnectOpt.") val topicOrFilterOpt = List(topicIdOpt, whitelistOpt, blacklistOpt).filter(options.has) if (topicOrFilterOpt.size != 1) CommandLineUtils.printUsageAndDie(parser, "Exactly one of whitelist/blacklist/topic is required.") @@ -449,11 +445,6 @@ object ConsoleConsumer extends Logging { if (!useOldConsumer) { CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServerOpt) - - if (options.has(newConsumerOpt)) { - Console.err.println("The --new-consumer option is deprecated and will be removed in a future major release. " + - "The new consumer is used by default if the --bootstrap-server option is provided.") - } } if (options.has(csvMetricsReporterEnabledOpt)) { diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 4fff87745e787..f4221fe294641 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -296,8 +296,6 @@ object ConsumerPerformance extends LazyLogging { .describedAs("count") .ofType(classOf[java.lang.Integer]) .defaultsTo(1) - val newConsumerOpt = parser.accepts("new-consumer", "Use the new consumer implementation. This is the default, so " + - "this option is deprecated and will be removed in a future release.") val consumerConfigOpt = parser.accepts("consumer.config", "Consumer config properties file.") .withRequiredArg .describedAs("config file") @@ -325,11 +323,6 @@ object ConsumerPerformance extends LazyLogging { if (!useOldConsumer) { CommandLineUtils.checkRequiredArgs(parser, options, bootstrapServersOpt) - if (options.has(newConsumerOpt)) { - Console.err.println("The --new-consumer option is deprecated and will be removed in a future major release. " + - "The new consumer is used by default if the --broker-list option is provided.") - } - import org.apache.kafka.clients.consumer.ConsumerConfig props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, options.valueOf(bootstrapServersOpt)) props.put(ConsumerConfig.GROUP_ID_CONFIG, options.valueOf(groupIdOpt)) @@ -342,8 +335,7 @@ object ConsumerPerformance extends LazyLogging { } else { if (options.has(bootstrapServersOpt)) CommandLineUtils.printUsageAndDie(parser, s"Option $bootstrapServersOpt is not valid with $zkConnectOpt.") - else if (options.has(newConsumerOpt)) - CommandLineUtils.printUsageAndDie(parser, s"Option $newConsumerOpt is not valid with $zkConnectOpt.") + CommandLineUtils.checkRequiredArgs(parser, options, zkConnectOpt, numMessagesOpt) props.put("group.id", options.valueOf(groupIdOpt)) props.put("socket.receive.buffer.bytes", options.valueOf(socketBufferSizeOpt).toString) diff --git a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala index 6f465557d7b5f..8bf1b58f12da4 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala @@ -149,8 +149,7 @@ class ConsoleConsumerTest { val args: Array[String] = Array( "--bootstrap-server", "localhost:9092", "--topic", "test", - "--from-beginning", - "--new-consumer") //new + "--from-beginning") //When val config = new ConsoleConsumer.ConsumerConfig(args) @@ -169,8 +168,7 @@ class ConsoleConsumerTest { "--bootstrap-server", "localhost:9092", "--topic", "test", "--partition", "0", - "--offset", "3", - "--new-consumer") //new + "--offset", "3") //When val config = new ConsoleConsumer.ConsumerConfig(args) @@ -200,6 +198,21 @@ class ConsoleConsumerTest { assertFalse(config.useOldConsumer) } + @Test + def testNewConsumerRemovedOption() { + //Given + val args: Array[String] = Array( + "--bootstrap-server", "localhost:9092", + "--topic", "test", + "--from-beginning") + + //When + val config = new ConsoleConsumer.ConsumerConfig(args) + + //Then + assertFalse(config.useOldConsumer) + } + @Test def shouldParseValidNewSimpleConsumerValidConfigWithStringOffset() { //Given @@ -208,7 +221,6 @@ class ConsoleConsumerTest { "--topic", "test", "--partition", "0", "--offset", "LatEst", - "--new-consumer", //new "--property", "print.value=false") //When diff --git a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala index bafe8ed136bb1..80a00bef44f58 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -20,7 +20,7 @@ package kafka.tools import java.io.ByteArrayOutputStream import java.text.SimpleDateFormat -import org.junit.Assert.assertEquals +import org.junit.Assert.{assertEquals, assertFalse, assertTrue} import org.junit.Test class ConsumerPerformanceTest { @@ -45,6 +45,57 @@ class ConsumerPerformanceTest { s"${dateFormat.format(System.currentTimeMillis)}, 1.0, 1.0, 1, 1.0")) } + @Test + def testConfigUsingNewConsumer(): Unit = { + //Given + val args: Array[String] = Array( + "--broker-list", "localhost:9092", + "--topic", "test", + "--messages", "10" + ) + + //When + val config = new ConsumerPerformance.ConsumerPerfConfig(args) + + //Then + assertFalse(config.useOldConsumer) + assertEquals("localhost:9092", config.bootstrapServers) + assertEquals("test", config.topic) + assertEquals(10, config.numMessages) + } + + @Test + def testConfigUsingOldConsumer() { + //Given + val args: Array[String] = Array( + "--zookeeper", "localhost:2181", + "--topic", "test", + "--messages", "10") + + //When + val config = new ConsumerPerformance.ConsumerPerfConfig(args) + + //Then + assertTrue(config.useOldConsumer) + assertEquals("localhost:2181", config.zkConnectionStr) + assertEquals("test", config.topic) + assertEquals(10, config.numMessages) + } + + @Test(expected = classOf[joptsimple.OptionException]) + def testConfigUsingNewConsumerUnrecognizedOption(): Unit = { + //Given + val args: Array[String] = Array( + "--broker-list", "localhost:9092", + "--topic", "test", + "--messages", "10", + "--new-consumer" + ) + + //When + new ConsumerPerformance.ConsumerPerfConfig(args) + } + private def testHeaderMatchContent(detailed: Boolean, useOldConsumer: Boolean, expectedOutputLineCount: Int, fun: () => Unit): Unit = { Console.withOut(outContent) { ConsumerPerformance.printHeader(detailed, useOldConsumer) From 179e80e118814088b369fbe3e88ec5723b749bbc Mon Sep 17 00:00:00 2001 From: Paolo Patierno Date: Thu, 31 May 2018 09:15:36 +0200 Subject: [PATCH 2/8] Updated upgrade doc with description of KIP-176 --- docs/upgrade.html | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/docs/upgrade.html b/docs/upgrade.html index 056fb8366e44f..ceebc7b0ba377 100644 --- a/docs/upgrade.html +++ b/docs/upgrade.html @@ -97,6 +97,11 @@
Notable changes in 2 will be removed in a future version.
  • The internal method kafka.admin.AdminClient.deleteRecordsBefore has been removed. Users are encouraged to migrate to org.apache.kafka.clients.admin.AdminClient.deleteRecords.
  • The tool kafka.tools.ReplayLogProducer has been removed.
    • +
  • KIP-176 removes + the --new-consumer option for all consumer based tools as kafka-console-consumer, kafka-consumer-perf-test + and kafka-consumer-groups. The new consumer is automatically used if the bootstrap servers list is provided on the command line + otherwise, providing the zookeeper connection instead, the old one is used. +
    • New Protocol Versions
    From ec775733a8eb3faa2b326c2d317de66b18ad4239 Mon Sep 17 00:00:00 2001 From: Paolo Patierno Date: Fri, 1 Jun 2018 11:16:28 +0200 Subject: [PATCH 3/8] Added check on using --new-consumer based on the node version --- tests/kafkatest/services/kafka/kafka.py | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/tests/kafkatest/services/kafka/kafka.py b/tests/kafkatest/services/kafka/kafka.py index ba5abc719f617..7e919b37249a5 100644 --- a/tests/kafkatest/services/kafka/kafka.py +++ b/tests/kafkatest/services/kafka/kafka.py @@ -30,7 +30,7 @@ from kafkatest.services.monitor.jmx import JmxMixin from kafkatest.services.security.minikdc import MiniKdc from kafkatest.services.security.security_config import SecurityConfig -from kafkatest.version import DEV_BRANCH +from kafkatest.version import DEV_BRANCH, LATEST_0_10_0 Port = collections.namedtuple('Port', ['name', 'number', 'open']) @@ -584,8 +584,12 @@ def list_consumer_groups(self, node=None, new_consumer=True, command_config=None command_config = "--command-config " + command_config if new_consumer: - cmd = "%s --new-consumer --bootstrap-server %s %s --list" % \ + new_consumer_opt = "" + if node.version <= LATEST_0_10_0: + new_consumer_opt = "--new-consumer" + cmd = "%s %s --bootstrap-server %s %s --list" % \ (consumer_group_script, + new_consumer_opt, self.bootstrap_servers(self.security_protocol), command_config) else: @@ -611,8 +615,14 @@ def describe_consumer_group(self, group, node=None, new_consumer=True, command_c command_config = "--command-config " + command_config if new_consumer: - cmd = "%s --new-consumer --bootstrap-server %s %s --group %s --describe" % \ - (consumer_group_script, self.bootstrap_servers(self.security_protocol), command_config, group) + new_consumer_opt = "" + if node.version <= LATEST_0_10_0: + new_consumer_opt = "--new-consumer" + cmd = "%s %s --bootstrap-server %s %s --group %s --describe" % \ + (consumer_group_script, + new_consumer_opt, + self.bootstrap_servers(self.security_protocol), + command_config, group) else: cmd = "%s --zookeeper %s %s --group %s --describe" % \ (consumer_group_script, self.zk_connect_setting(), command_config, group) From 9f8d327c13fda7bd07fa26474b0c2e9f91202033 Mon Sep 17 00:00:00 2001 From: Paolo Patierno Date: Fri, 1 Jun 2018 12:14:53 +0200 Subject: [PATCH 4/8] Added unit tests for checking unrecognized --new-consumer option --- .../admin/DeleteConsumerGroupsTest.scala | 7 ++++++ .../admin/DescribeConsumerGroupTest.scala | 6 +++++ .../kafka/admin/ListConsumerGroupTest.scala | 6 +++++ .../admin/ResetConsumerGroupOffsetTest.scala | 8 +++++++ .../kafka/tools/ConsoleConsumerTest.scala | 24 +++++++++++++++++++ 5 files changed, 51 insertions(+) diff --git a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala index 4cc28372449b1..071640dc93b84 100644 --- a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala +++ b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala @@ -222,4 +222,11 @@ class DeleteConsumerGroupTest extends ConsumerGroupCommandTest { result.size == 1 && result.keySet.contains(group) && result.get(group).contains(Errors.COORDINATOR_NOT_AVAILABLE)) } + + @Test(expected = classOf[joptsimple.OptionException]) + def testDeleteWithUnrecognizedNewConsumerOption() { + val cgcArgs = Array("--new-consumer", "--bootstrap-server", brokerList, "--delete", "--group", group) + getConsumerGroupService(cgcArgs) + fail("Expected an error due to presence of unrecognized --new-consumer option") + } } diff --git a/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala b/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala index 5725568e10ed4..d085eeada09b4 100644 --- a/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala +++ b/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala @@ -662,6 +662,12 @@ class DescribeConsumerGroupTest extends ConsumerGroupCommandTest { } } + @Test(expected = classOf[joptsimple.OptionException]) + def testDescribeWithUnrecognizedNewConsumerOption() { + val cgcArgs = Array("--new-consumer", "--bootstrap-server", brokerList, "--describe", "--group", group) + getConsumerGroupService(cgcArgs) + fail("Expected an error due to presence of unrecognized --new-consumer option") + } } diff --git a/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala b/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala index 13dccbe113e8f..1c756d8b0d3e8 100644 --- a/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala @@ -86,4 +86,10 @@ class ListConsumerGroupTest extends ConsumerGroupCommandTest { }, s"Expected --list to show groups $expectedGroups, but found $foundGroups.") } + @Test(expected = classOf[joptsimple.OptionException]) + def testListWithUnrecognizedNewConsumerOption() { + val cgcArgs = Array("--new-consumer", "--bootstrap-server", brokerList, "--list") + getConsumerGroupService(cgcArgs) + fail("Expected an error due to presence of unrecognized --new-consumer option") + } } diff --git a/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala b/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala index 04fc428514bfc..c5f83063de503 100644 --- a/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala @@ -335,6 +335,14 @@ class ResetConsumerGroupOffsetTest extends ConsumerGroupCommandTest { adminZkClient.deleteTopic(topic) } + @Test(expected = classOf[joptsimple.OptionException]) + def testResetWithUnrecognizedNewConsumerOption() { + val cgcArgs = Array("--new-consumer", "--bootstrap-server", brokerList, "--reset-offsets", "--group", group, "--all-topics", + "--to-offset", "2", "--export") + getConsumerGroupService(cgcArgs) + fail("Expected an error due to presence of unrecognized --new-consumer option") + } + private def produceMessages(topic: String, numMessages: Int): Unit = { val records = (0 until numMessages).map(_ => new ProducerRecord[Array[Byte], Array[Byte]](topic, new Array[Byte](100 * 1000))) diff --git a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala index 8bf1b58f12da4..caa2b8c5eb9c1 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala @@ -183,6 +183,30 @@ class ConsoleConsumerTest { } + @Test(expected = classOf[IllegalArgumentException]) + def shouldExitOnUnrecognizedNewConsumerOption(): Unit = { + + // Override exit procedure to throw an exception instead of exiting, so we can catch the exit + // properly for this test case + Exit.setExitProcedure((_, message) => throw new IllegalArgumentException(message.orNull)) + + //Given + val args: Array[String] = Array( + "--new-consumer", + "--bootstrap-server", "localhost:9092", + "--topic", "test", + "--from-beginning") + + //When + try { + new ConsoleConsumer.ConsumerConfig(args) + } finally { + Exit.resetExitProcedure() + } + + fail("Expected consumer config construction to fail due to unrecognized --new-consumer option") + } + @Test def testDefaultConsumer() { //Given From 56b220496650aacb68c9f36c523677376e38734b Mon Sep 17 00:00:00 2001 From: Paolo Patierno Date: Tue, 5 Jun 2018 18:09:06 +0200 Subject: [PATCH 5/8] Addressed some reequested changes on review --- .../unit/kafka/admin/DeleteConsumerGroupsTest.scala | 4 ++-- .../scala/unit/kafka/tools/ConsoleConsumerTest.scala | 6 ------ docs/upgrade.html | 9 ++++++--- 3 files changed, 8 insertions(+), 11 deletions(-) diff --git a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala index 071640dc93b84..f0a6d8fd5da36 100644 --- a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala +++ b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala @@ -16,6 +16,7 @@ */ package unit.kafka.admin +import joptsimple.OptionException import kafka.admin.ConsumerGroupCommandTest import kafka.utils.TestUtils import org.apache.kafka.common.protocol.Errors @@ -223,10 +224,9 @@ class DeleteConsumerGroupTest extends ConsumerGroupCommandTest { result.keySet.contains(group) && result.get(group).contains(Errors.COORDINATOR_NOT_AVAILABLE)) } - @Test(expected = classOf[joptsimple.OptionException]) + @Test(expected = classOf[OptionException]) def testDeleteWithUnrecognizedNewConsumerOption() { val cgcArgs = Array("--new-consumer", "--bootstrap-server", brokerList, "--delete", "--group", group) getConsumerGroupService(cgcArgs) - fail("Expected an error due to presence of unrecognized --new-consumer option") } } diff --git a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala index caa2b8c5eb9c1..dcbfd5696bc62 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala @@ -186,8 +186,6 @@ class ConsoleConsumerTest { @Test(expected = classOf[IllegalArgumentException]) def shouldExitOnUnrecognizedNewConsumerOption(): Unit = { - // Override exit procedure to throw an exception instead of exiting, so we can catch the exit - // properly for this test case Exit.setExitProcedure((_, message) => throw new IllegalArgumentException(message.orNull)) //Given @@ -403,8 +401,6 @@ class ConsoleConsumerTest { @Test(expected = classOf[IllegalArgumentException]) def shouldExitOnInvalidConfigWithAutoOffsetResetAndConflictingFromBeginningNewConsumer() { - // Override exit procedure to throw an exception instead of exiting, so we can catch the exit - // properly for this test case Exit.setExitProcedure((_, message) => throw new IllegalArgumentException(message.orNull)) //Given @@ -427,8 +423,6 @@ class ConsoleConsumerTest { @Test(expected = classOf[IllegalArgumentException]) def shouldExitOnInvalidConfigWithAutoOffsetResetAndConflictingFromBeginningOldConsumer() { - // Override exit procedure to throw an exception instead of exiting, so we can catch the exit - // properly for this test case Exit.setExitProcedure((_, message) => throw new IllegalArgumentException(message.orNull)) //Given diff --git a/docs/upgrade.html b/docs/upgrade.html index ceebc7b0ba377..cfc3c833ffe85 100644 --- a/docs/upgrade.html +++ b/docs/upgrade.html @@ -97,10 +97,13 @@
    Notable changes in 2 will be removed in a future version.
  • The internal method kafka.admin.AdminClient.deleteRecordsBefore has been removed. Users are encouraged to migrate to org.apache.kafka.clients.admin.AdminClient.deleteRecords.
  • The tool kafka.tools.ReplayLogProducer has been removed.
    • -
  • KIP-176 removes +
  • KIP-176 finally removes the --new-consumer option for all consumer based tools as kafka-console-consumer, kafka-consumer-perf-test - and kafka-consumer-groups. The new consumer is automatically used if the bootstrap servers list is provided on the command line - otherwise, providing the zookeeper connection instead, the old one is used. + and kafka-consumer-groups. + The new consumer is automatically used if the bootstrap servers list is provided on the command line + otherwise, when the zookeeper connection is provided, the old consumer is used. + The --new-consumer option had already been ignored as the way of selecting the consumer since Kafka 1.0.0, + this KIP just removes the option.
    • From 136f2eb38755677afbd9f0a9161c58fb5b85c674 Mon Sep 17 00:00:00 2001 From: Paolo Patierno Date: Tue, 5 Jun 2018 18:47:18 +0200 Subject: [PATCH 6/8] Added missing field in the ConsumerPerformanceConfig (lost after a rebase) --- core/src/main/scala/kafka/tools/ConsumerPerformance.scala | 2 ++ 1 file changed, 2 insertions(+) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index f4221fe294641..5afe56f55177b 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -355,6 +355,8 @@ object ConsumerPerformance extends LazyLogging { val dateFormat = new SimpleDateFormat(options.valueOf(dateFormatOpt)) val hideHeader = options.has(hideHeaderOpt) val recordFetchTimeoutMs = options.valueOf(recordFetchTimeoutOpt).longValue() + val bootstrapServers = options.valueOf(bootstrapServersOpt) + val zkConnectionStr = options.valueOf(zkConnectOpt) } class ConsumerPerfThread(threadId: Int, From cf68494668955f500c424b3e72551f599f6c4c7c Mon Sep 17 00:00:00 2001 From: Paolo Patierno Date: Tue, 5 Jun 2018 22:47:14 +0200 Subject: [PATCH 7/8] Added import package and removed useless fail --- .../test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala | 3 +-- .../test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala | 3 ++- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala index f0a6d8fd5da36..ef3b17c393850 100644 --- a/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala +++ b/core/src/test/scala/unit/kafka/admin/DeleteConsumerGroupsTest.scala @@ -25,12 +25,11 @@ import org.junit.Test class DeleteConsumerGroupTest extends ConsumerGroupCommandTest { - @Test(expected = classOf[joptsimple.OptionException]) + @Test(expected = classOf[OptionException]) def testDeleteWithTopicOption() { TestUtils.createOffsetsTopic(zkClient, servers) val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group, "--topic") getConsumerGroupService(cgcArgs) - fail("Expected an error due to presence of mutually exclusive options") } @Test diff --git a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala index 80a00bef44f58..640d44fc5c2b0 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -20,6 +20,7 @@ package kafka.tools import java.io.ByteArrayOutputStream import java.text.SimpleDateFormat +import joptsimple.OptionException import org.junit.Assert.{assertEquals, assertFalse, assertTrue} import org.junit.Test @@ -82,7 +83,7 @@ class ConsumerPerformanceTest { assertEquals(10, config.numMessages) } - @Test(expected = classOf[joptsimple.OptionException]) + @Test(expected = classOf[OptionException]) def testConfigUsingNewConsumerUnrecognizedOption(): Unit = { //Given val args: Array[String] = Array( From 80e035a255981933810142389fdaf4fc912e814b Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Tue, 5 Jun 2018 17:32:30 -0700 Subject: [PATCH 8/8] Minor tweaks --- .../src/main/scala/kafka/tools/ConsumerPerformance.scala | 2 -- .../unit/kafka/admin/DescribeConsumerGroupTest.scala | 4 ++-- .../scala/unit/kafka/admin/ListConsumerGroupTest.scala | 4 ++-- .../unit/kafka/admin/ResetConsumerGroupOffsetTest.scala | 4 ++-- .../scala/unit/kafka/tools/ConsoleConsumerTest.scala | 9 --------- .../scala/unit/kafka/tools/ConsumerPerformanceTest.scala | 4 ++-- 6 files changed, 8 insertions(+), 19 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 5afe56f55177b..f4221fe294641 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -355,8 +355,6 @@ object ConsumerPerformance extends LazyLogging { val dateFormat = new SimpleDateFormat(options.valueOf(dateFormatOpt)) val hideHeader = options.has(hideHeaderOpt) val recordFetchTimeoutMs = options.valueOf(recordFetchTimeoutOpt).longValue() - val bootstrapServers = options.valueOf(bootstrapServersOpt) - val zkConnectionStr = options.valueOf(zkConnectOpt) } class ConsumerPerfThread(threadId: Int, diff --git a/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala b/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala index d085eeada09b4..a2361b706a3bf 100644 --- a/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala +++ b/core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala @@ -16,6 +16,7 @@ */ package kafka.admin +import joptsimple.OptionException import kafka.utils.TestUtils import org.apache.kafka.clients.consumer.RoundRobinAssignor import org.apache.kafka.common.TopicPartition @@ -112,12 +113,11 @@ class DescribeConsumerGroupTest extends ConsumerGroupCommandTest { } } - @Test(expected = classOf[joptsimple.OptionException]) + @Test(expected = classOf[OptionException]) def testDescribeWithMultipleSubActions() { TestUtils.createOffsetsTopic(zkClient, servers) val cgcArgs = Array("--bootstrap-server", brokerList, "--describe", "--group", group, "--members", "--state") getConsumerGroupService(cgcArgs) - fail("Expected an error due to presence of mutually exclusive options") } @Test diff --git a/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala b/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala index 1c756d8b0d3e8..c83e00287970d 100644 --- a/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ListConsumerGroupTest.scala @@ -18,6 +18,7 @@ package kafka.admin import java.util.Properties +import joptsimple.OptionException import org.junit.Test import kafka.admin.ConsumerGroupCommand.ConsumerGroupCommandOptions import kafka.admin.ConsumerGroupCommand.ZkConsumerGroupService @@ -86,10 +87,9 @@ class ListConsumerGroupTest extends ConsumerGroupCommandTest { }, s"Expected --list to show groups $expectedGroups, but found $foundGroups.") } - @Test(expected = classOf[joptsimple.OptionException]) + @Test(expected = classOf[OptionException]) def testListWithUnrecognizedNewConsumerOption() { val cgcArgs = Array("--new-consumer", "--bootstrap-server", brokerList, "--list") getConsumerGroupService(cgcArgs) - fail("Expected an error due to presence of unrecognized --new-consumer option") } } diff --git a/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala b/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala index c5f83063de503..116b455cb87cf 100644 --- a/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala +++ b/core/src/test/scala/unit/kafka/admin/ResetConsumerGroupOffsetTest.scala @@ -16,6 +16,7 @@ import java.io.{BufferedWriter, File, FileWriter} import java.text.{ParseException, SimpleDateFormat} import java.util.{Calendar, Date, Properties} +import joptsimple.OptionException import kafka.admin.ConsumerGroupCommand.ConsumerGroupService import kafka.server.KafkaConfig import kafka.utils.TestUtils @@ -335,12 +336,11 @@ class ResetConsumerGroupOffsetTest extends ConsumerGroupCommandTest { adminZkClient.deleteTopic(topic) } - @Test(expected = classOf[joptsimple.OptionException]) + @Test(expected = classOf[OptionException]) def testResetWithUnrecognizedNewConsumerOption() { val cgcArgs = Array("--new-consumer", "--bootstrap-server", brokerList, "--reset-offsets", "--group", group, "--all-topics", "--to-offset", "2", "--export") getConsumerGroupService(cgcArgs) - fail("Expected an error due to presence of unrecognized --new-consumer option") } private def produceMessages(topic: String, numMessages: Int): Unit = { diff --git a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala index dcbfd5696bc62..1a32bf4207211 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsoleConsumerTest.scala @@ -185,7 +185,6 @@ class ConsoleConsumerTest { @Test(expected = classOf[IllegalArgumentException]) def shouldExitOnUnrecognizedNewConsumerOption(): Unit = { - Exit.setExitProcedure((_, message) => throw new IllegalArgumentException(message.orNull)) //Given @@ -201,8 +200,6 @@ class ConsoleConsumerTest { } finally { Exit.resetExitProcedure() } - - fail("Expected consumer config construction to fail due to unrecognized --new-consumer option") } @Test @@ -400,7 +397,6 @@ class ConsoleConsumerTest { @Test(expected = classOf[IllegalArgumentException]) def shouldExitOnInvalidConfigWithAutoOffsetResetAndConflictingFromBeginningNewConsumer() { - Exit.setExitProcedure((_, message) => throw new IllegalArgumentException(message.orNull)) //Given @@ -416,13 +412,10 @@ class ConsoleConsumerTest { } finally { Exit.resetExitProcedure() } - - fail("Expected consumer property construction to fail due to inconsistent reset options") } @Test(expected = classOf[IllegalArgumentException]) def shouldExitOnInvalidConfigWithAutoOffsetResetAndConflictingFromBeginningOldConsumer() { - Exit.setExitProcedure((_, message) => throw new IllegalArgumentException(message.orNull)) //Given @@ -438,8 +431,6 @@ class ConsoleConsumerTest { } finally { Exit.resetExitProcedure() } - - fail("Expected consumer property construction to fail due to inconsistent reset options") } @Test diff --git a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala index 640d44fc5c2b0..bc199f658abed 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -60,7 +60,7 @@ class ConsumerPerformanceTest { //Then assertFalse(config.useOldConsumer) - assertEquals("localhost:9092", config.bootstrapServers) + assertEquals("localhost:9092", config.options.valueOf(config.bootstrapServersOpt)) assertEquals("test", config.topic) assertEquals(10, config.numMessages) } @@ -78,7 +78,7 @@ class ConsumerPerformanceTest { //Then assertTrue(config.useOldConsumer) - assertEquals("localhost:2181", config.zkConnectionStr) + assertEquals("localhost:2181", config.options.valueOf(config.zkConnectOpt)) assertEquals("test", config.topic) assertEquals(10, config.numMessages) }