You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by va...@apache.org on 2019/02/11 22:39:12 UTC
[kafka] branch trunk updated: KAFKA-7236: Add --under-min-isr
option to describe topics command (KIP-351) (#6224)
This is an automated email from the ASF dual-hosted git repository.
vahid 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 3fb1d70 KAFKA-7236: Add --under-min-isr option to describe topics command (KIP-351) (#6224)
3fb1d70 is described below
commit 3fb1d70b6b2dbb78aafe60bae081237980fcd326
Author: Kevin Lu <lu...@berkeley.edu>
AuthorDate: Mon Feb 11 14:39:02 2019 -0800
KAFKA-7236: Add --under-min-isr option to describe topics command (KIP-351) (#6224)
* KAFKA-7236: Add --under-min-isr option to describe topics command (KIP-351)
* Minor changes to description and make test consistent with others
* Fix option, and add additional test with mixed partition status
* Add fully-replicated-topic to test case
* Address review nits
---
core/src/main/scala/kafka/admin/TopicCommand.scala | 33 ++++++++---
.../admin/TopicCommandWithAdminClientTest.scala | 68 +++++++++++++++++++++-
2 files changed, 91 insertions(+), 10 deletions(-)
diff --git a/core/src/main/scala/kafka/admin/TopicCommand.scala b/core/src/main/scala/kafka/admin/TopicCommand.scala
index 48fdaec..2fcbe17 100755
--- a/core/src/main/scala/kafka/admin/TopicCommand.scala
+++ b/core/src/main/scala/kafka/admin/TopicCommand.scala
@@ -30,7 +30,7 @@ import kafka.zk.{AdminZkClient, KafkaZkClient}
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin.{ListTopicsOptions, NewPartitions, NewTopic, AdminClient => JAdminClient}
import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
import org.apache.kafka.common.config.ConfigResource.Type
import org.apache.kafka.common.errors.{InvalidTopicException, TopicExistsException}
import org.apache.kafka.common.internals.Topic
@@ -99,17 +99,18 @@ object TopicCommand extends Logging {
leader: Option[Int],
assignedReplicas: Seq[Int],
isr: Seq[Int],
+ minIsrCount: Int,
markedForDeletion: Boolean,
describeConfigs: Boolean)
class DescribeOptions(opts: TopicCommandOptions, liveBrokers: Set[Int]) {
- val describeConfigs: Boolean = !opts.reportUnavailablePartitions && !opts.reportUnderReplicatedPartitions
+ val describeConfigs: Boolean = !opts.reportUnavailablePartitions && !opts.reportUnderReplicatedPartitions && !opts.reportUnderMinIsrPartitions
val describePartitions: Boolean = !opts.reportOverriddenConfigs
- private def hasUnderreplicatedPartitions(partitionDescription: PartitionDescription) = {
+ private def hasUnderReplicatedPartitions(partitionDescription: PartitionDescription) = {
partitionDescription.isr.size < partitionDescription.assignedReplicas.size
}
private def shouldPrintUnderReplicatedPartitions(partitionDescription: PartitionDescription) = {
- opts.reportUnderReplicatedPartitions && hasUnderreplicatedPartitions(partitionDescription)
+ opts.reportUnderReplicatedPartitions && hasUnderReplicatedPartitions(partitionDescription)
}
private def hasUnavailablePartitions(partitionDescription: PartitionDescription) = {
partitionDescription.leader.isEmpty || !liveBrokers.contains(partitionDescription.leader.get)
@@ -117,11 +118,18 @@ object TopicCommand extends Logging {
private def shouldPrintUnavailablePartitions(partitionDescription: PartitionDescription) = {
opts.reportUnavailablePartitions && hasUnavailablePartitions(partitionDescription)
}
+ private def hasUnderMinIsrPartitions(partitionDescription: PartitionDescription) = {
+ partitionDescription.isr.size < partitionDescription.minIsrCount
+ }
+ private def shouldPrintUnderMinIsrPartitions(partitionDescription: PartitionDescription) = {
+ opts.reportUnderMinIsrPartitions && hasUnderMinIsrPartitions(partitionDescription)
+ }
def shouldPrintTopicPartition(partitionDesc: PartitionDescription): Boolean = {
describeConfigs ||
shouldPrintUnderReplicatedPartitions(partitionDesc) ||
- shouldPrintUnavailablePartitions(partitionDesc)
+ shouldPrintUnavailablePartitions(partitionDesc) ||
+ shouldPrintUnderMinIsrPartitions(partitionDesc)
}
}
@@ -220,6 +228,8 @@ object TopicCommand extends Logging {
}
}
if (describeOptions.describePartitions) {
+ val computedMinIsrCount = if (opts.reportUnderMinIsrPartitions)
+ allConfigs.get(new ConfigResource(ConfigResource.Type.TOPIC, td.name())).get().get(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG).value().toInt else 0
for (partition <- sortedPartitions) {
val partitionDesc = PartitionDescription(
topic = td.name(),
@@ -227,6 +237,7 @@ object TopicCommand extends Logging {
leader = Option(partition.leader()).map(_.id()),
assignedReplicas = partition.replicas().asScala.map(_.id()),
isr = partition.isr().asScala.map(_.id()),
+ minIsrCount = computedMinIsrCount,
markedForDeletion = false,
describeOptions.describeConfigs
)
@@ -357,6 +368,7 @@ object TopicCommand extends Logging {
leader = if (leaderIsrEpoch.isEmpty) None else Option(leaderIsrEpoch.get.leaderAndIsr.leader),
assignedReplicas,
isr = if (leaderIsrEpoch.isEmpty) Seq.empty[Int] else leaderIsrEpoch.get.leaderAndIsr.isr,
+ minIsrCount = 0,
markedForDeletion,
describeOptions.describeConfigs
)
@@ -536,6 +548,8 @@ object TopicCommand extends Logging {
"if set when describing topics, only show under replicated partitions")
private val reportUnavailablePartitionsOpt = parser.accepts("unavailable-partitions",
"if set when describing topics, only show partitions whose leader is not available")
+ private val reportUnderMinIsrPartitionsOpt = parser.accepts("under-min-isr-partitions",
+ "if set when describing topics, only show partitions whose isr count is less than the configured minimum. Not supported with the --zookeeper option.")
private val topicsWithOverridesOpt = parser.accepts("topics-with-overrides",
"if set when describing topics, only show topics that have overridden configs")
private val ifExistsOpt = parser.accepts("if-exists",
@@ -577,6 +591,7 @@ object TopicCommand extends Logging {
def rackAwareMode: RackAwareMode = if (has(disableRackAware)) RackAwareMode.Disabled else RackAwareMode.Enforced
def reportUnderReplicatedPartitions: Boolean = has(reportUnderReplicatedPartitionsOpt)
def reportUnavailablePartitions: Boolean = has(reportUnavailablePartitionsOpt)
+ def reportUnderMinIsrPartitions: Boolean = has(reportUnderMinIsrPartitionsOpt)
def reportOverriddenConfigs: Boolean = has(topicsWithOverridesOpt)
def ifExists: Boolean = has(ifExistsOpt)
def ifNotExists: Boolean = has(ifNotExistsOpt)
@@ -620,11 +635,13 @@ object TopicCommand extends Logging {
if(options.has(createOpt))
CommandLineUtils.checkInvalidArgs(parser, options, replicaAssignmentOpt, Set(partitionsOpt, replicationFactorOpt))
CommandLineUtils.checkInvalidArgs(parser, options, reportUnderReplicatedPartitionsOpt,
- allTopicLevelOpts -- Set(describeOpt) + reportUnavailablePartitionsOpt + topicsWithOverridesOpt)
+ allTopicLevelOpts -- Set(describeOpt) + reportUnderMinIsrPartitionsOpt + reportUnavailablePartitionsOpt + topicsWithOverridesOpt)
+ CommandLineUtils.checkInvalidArgs(parser, options, reportUnderMinIsrPartitionsOpt,
+ allTopicLevelOpts -- Set(describeOpt) + reportUnderReplicatedPartitionsOpt + reportUnavailablePartitionsOpt + topicsWithOverridesOpt + zkConnectOpt)
CommandLineUtils.checkInvalidArgs(parser, options, reportUnavailablePartitionsOpt,
- allTopicLevelOpts -- Set(describeOpt) + reportUnderReplicatedPartitionsOpt + topicsWithOverridesOpt)
+ allTopicLevelOpts -- Set(describeOpt) + reportUnderReplicatedPartitionsOpt + reportUnderReplicatedPartitionsOpt + topicsWithOverridesOpt)
CommandLineUtils.checkInvalidArgs(parser, options, topicsWithOverridesOpt,
- allTopicLevelOpts -- Set(describeOpt) + reportUnderReplicatedPartitionsOpt + reportUnavailablePartitionsOpt)
+ allTopicLevelOpts -- Set(describeOpt) + reportUnderReplicatedPartitionsOpt + reportUnderMinIsrPartitionsOpt + reportUnavailablePartitionsOpt)
CommandLineUtils.checkInvalidArgs(parser, options, ifExistsOpt, allTopicLevelOpts -- Set(alterOpt, deleteOpt, describeOpt) ++ Set(bootstrapServerOpt))
CommandLineUtils.checkInvalidArgs(parser, options, ifNotExistsOpt, allTopicLevelOpts -- Set(createOpt) ++ Set(bootstrapServerOpt))
CommandLineUtils.checkInvalidArgs(parser, options, excludeInternalTopicOpt, allTopicLevelOpts -- Set(listOpt, describeOpt))
diff --git a/core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala b/core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala
index 3ebd689..238afa3 100644
--- a/core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/TopicCommandWithAdminClientTest.scala
@@ -26,7 +26,7 @@ import kafka.utils.{Exit, Logging, TestUtils}
import kafka.zk.{ConfigEntityChangeNotificationZNode, DeleteTopicsTopicZNode}
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin.{ListTopicsOptions, NewTopic, AdminClient => JAdminClient}
-import org.apache.kafka.common.config.{ConfigException, ConfigResource}
+import org.apache.kafka.common.config.{ConfigException, ConfigResource, TopicConfig}
import org.apache.kafka.common.internals.Topic
import org.junit.Assert.{assertEquals, assertFalse, assertTrue}
import org.junit.{After, Before, Rule, Test}
@@ -527,7 +527,7 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
}
@Test
- def testDescribeUnderreplicatedPartitions(): Unit = {
+ def testDescribeUnderReplicatedPartitions(): Unit = {
adminClient.createTopics(
Collections.singletonList(new NewTopic(testTopicName, 1, 6))).all().get()
waitForTopicCreated(testTopicName)
@@ -544,6 +544,70 @@ class TopicCommandWithAdminClientTest extends KafkaServerTestHarness with Loggin
}
@Test
+ def testDescribeUnderMinIsrPartitions(): Unit = {
+ val configMap = new java.util.HashMap[String, String]()
+ configMap.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "6")
+
+ adminClient.createTopics(
+ Collections.singletonList(new NewTopic(testTopicName, 1, 6).configs(configMap))).all().get()
+ waitForTopicCreated(testTopicName)
+
+ try {
+ killBroker(0)
+ val output = TestUtils.grabConsoleOutput(
+ topicService.describeTopic(new TopicCommandOptions(Array("--under-min-isr-partitions"))))
+ val rows = output.split("\n")
+ assertTrue(rows(0).startsWith(s"\tTopic: $testTopicName"))
+ } finally {
+ restartDeadBrokers()
+ }
+ }
+
+ /**
+ * Test describe --under-min-isr-partitions option with four topics:
+ * (1) topic with partition under the configured min ISR count
+ * (2) topic with under-replicated partition (but not under min ISR count)
+ * (3) topic with offline partition
+ * (4) topic with fully replicated partition
+ *
+ * Output should only display the (1) topic with partition under min ISR count and (3) topic with offline partition
+ */
+ @Test
+ def testDescribeUnderMinIsrPartitionsMixed(): Unit = {
+ val underMinIsrTopic = "under-min-isr-topic"
+ val notUnderMinIsrTopic = "not-under-min-isr-topic"
+ val offlineTopic = "offline-topic"
+ val fullyReplicatedTopic = "fully-replicated-topic"
+
+ val configMap = new java.util.HashMap[String, String]()
+ configMap.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "6")
+
+ adminClient.createTopics(
+ java.util.Arrays.asList(
+ new NewTopic(underMinIsrTopic, 1, 6).configs(configMap),
+ new NewTopic(notUnderMinIsrTopic, 1, 6),
+ new NewTopic(offlineTopic, Collections.singletonMap(0, Collections.singletonList(0))),
+ new NewTopic(fullyReplicatedTopic, Collections.singletonMap(0, java.util.Arrays.asList(1, 2, 3))))).all().get()
+
+ waitForTopicCreated(underMinIsrTopic)
+ waitForTopicCreated(notUnderMinIsrTopic)
+ waitForTopicCreated(offlineTopic)
+ waitForTopicCreated(fullyReplicatedTopic)
+
+ try {
+ killBroker(0)
+ val output = TestUtils.grabConsoleOutput(
+ topicService.describeTopic(new TopicCommandOptions(Array("--under-min-isr-partitions"))))
+ val rows = output.split("\n")
+ assertTrue(rows(0).startsWith(s"\tTopic: $underMinIsrTopic"))
+ assertTrue(rows(1).startsWith(s"\tTopic: $offlineTopic"))
+ assertEquals(2, rows.length);
+ } finally {
+ restartDeadBrokers()
+ }
+ }
+
+ @Test
def testDescribeReportOverriddenConfigs(): Unit = {
val config = "file.delete.delay.ms=1000"
createAndWaitTopic(new TopicCommandOptions(