-
Notifications
You must be signed in to change notification settings - Fork 13.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
KAFKA-12454: Add ERROR logging on kafka-log-dirs when given brokerIds do not exist in current kafka cluster #10304
Changes from 6 commits
e71e5a4
91fd305
b473a44
a4d6ae7
2b623e2
fefdef9
c88b348
848822f
587800d
1e26832
5641815
b245e5f
3565cca
24f520f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -39,19 +39,29 @@ object LogDirsCommand { | |
def describe(args: Array[String], out: PrintStream): Unit = { | ||
val opts = new LogDirsCommandOptions(args) | ||
val adminClient = createAdminClient(opts) | ||
val topicList = opts.options.valueOf(opts.topicListOpt).split(",").filter(!_.isEmpty) | ||
val brokerList = Option(opts.options.valueOf(opts.brokerListOpt)) match { | ||
case Some(brokerListStr) => brokerListStr.split(',').filter(!_.isEmpty).map(_.toInt) | ||
case None => adminClient.describeCluster().nodes().get().asScala.map(_.id()).toArray | ||
} | ||
val topicList = opts.options.valueOf(opts.topicListOpt).split(",").filter(_.nonEmpty) | ||
try { | ||
val clusterBrokers = adminClient.describeCluster().nodes().get().asScala.map(_.id()).toSet | ||
val (existingBrokers, nonExistingBrokers) = Option(opts.options.valueOf(opts.brokerListOpt)) match { | ||
case Some(brokerListStr) => | ||
val inputBrokers = brokerListStr.split(',').filter(_.nonEmpty).map(_.toInt).toSet | ||
(inputBrokers, inputBrokers.diff(clusterBrokers)) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. As the variable is called |
||
case None => (clusterBrokers, Set.empty) | ||
} | ||
|
||
out.println("Querying brokers for log directories information") | ||
val describeLogDirsResult: DescribeLogDirsResult = adminClient.describeLogDirs(brokerList.map(Integer.valueOf).toSeq.asJava) | ||
val logDirInfosByBroker = describeLogDirsResult.allDescriptions.get().asScala.map { case (k, v) => k -> v.asScala } | ||
if (nonExistingBrokers.nonEmpty) { | ||
out.println(s"ERROR: The given node(s) does not exist from broker-list: ${nonExistingBrokers.mkString(",")}. Current cluster exist node(s): ${clusterBrokers.mkString(",")}") | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: Should we say There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Good idea.I will act right away. |
||
} else { | ||
out.println("Querying brokers for log directories information") | ||
val describeLogDirsResult: DescribeLogDirsResult = adminClient.describeLogDirs(existingBrokers.map(Integer.valueOf).toSeq.asJava) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sorry, I forgot this, I will change it right away |
||
val logDirInfosByBroker = describeLogDirsResult.allDescriptions.get().asScala.map { case (k, v) => k -> v.asScala } | ||
|
||
out.println(s"Received log directory information from brokers ${brokerList.mkString(",")}") | ||
out.println(formatAsJson(logDirInfosByBroker, topicList.toSet)) | ||
adminClient.close() | ||
out.println(s"Received log directory information from brokers ${existingBrokers.mkString(",")}") | ||
out.println(formatAsJson(logDirInfosByBroker, topicList.toSet)) | ||
} | ||
} finally { | ||
adminClient.close() | ||
} | ||
} | ||
|
||
private def formatAsJson(logDirInfosByBroker: Map[Integer, Map[String, LogDirDescription]], topicSet: Set[String]): String = { | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,52 @@ | ||
package unit.kafka.admin | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We must add the licence header here. You can copy it from another file. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sorry, after checking the compilation report, I have realized this problem and I have made changes. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. the package name should be |
||
|
||
import java.io.{ByteArrayOutputStream, PrintStream} | ||
import java.nio.charset.StandardCharsets | ||
|
||
import kafka.admin.LogDirsCommand | ||
import kafka.integration.KafkaServerTestHarness | ||
import kafka.server.KafkaConfig | ||
import kafka.utils.TestUtils | ||
import org.junit.jupiter.api.Assertions.assertTrue | ||
import org.junit.jupiter.api.Test | ||
|
||
import scala.collection.Seq | ||
|
||
class LogDirsCommandTest extends KafkaServerTestHarness { | ||
|
||
def generateConfigs: Seq[KafkaConfig] = { | ||
TestUtils.createBrokerConfigs(1, zkConnect) | ||
.map(KafkaConfig.fromProps) | ||
} | ||
|
||
@Test | ||
def checkLogDirsCommandOutput(): Unit = { | ||
val byteArrayOutputStream = new ByteArrayOutputStream | ||
val printStream = new PrintStream(byteArrayOutputStream, false, StandardCharsets.UTF_8.name()) | ||
//input exist brokerList | ||
LogDirsCommand.describe(Array("--bootstrap-server", brokerList, "--broker-list", "0", "--describe"), printStream) | ||
val existBrokersContent = new String(byteArrayOutputStream.toByteArray, StandardCharsets.UTF_8) | ||
val existBrokersLineIter = existBrokersContent.split("\n").iterator | ||
|
||
assertTrue(existBrokersLineIter.hasNext) | ||
assertTrue(existBrokersLineIter.next().contains(s"Querying brokers for log directories information")) | ||
|
||
//input nonExist brokerList | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. how about using There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. It has been modified. |
||
byteArrayOutputStream.reset() | ||
LogDirsCommand.describe(Array("--bootstrap-server", brokerList, "--broker-list", "0,1,2", "--describe"), printStream) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could you input duplicate ids and the check the output does not include duplicates? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @chia7712 I have added the duplicate ids test and it passed. |
||
val nonExistBrokersContent = new String(byteArrayOutputStream.toByteArray, StandardCharsets.UTF_8) | ||
val nonExistBrokersLineIter = nonExistBrokersContent.split("\n").iterator | ||
|
||
assertTrue(nonExistBrokersLineIter.hasNext) | ||
assertTrue(nonExistBrokersLineIter.next().contains(s"ERROR: The given node(s) does not exist from broker-list: 1,2. Current cluster exist node(s): 0")) | ||
|
||
//use all brokerList for current cluster | ||
byteArrayOutputStream.reset() | ||
LogDirsCommand.describe(Array("--bootstrap-server", brokerList, "--describe"), printStream) | ||
val allBrokersContent = new String(byteArrayOutputStream.toByteArray, StandardCharsets.UTF_8) | ||
val allBrokersLineIter = allBrokersContent.split("\n").iterator | ||
|
||
assertTrue(allBrokersLineIter.hasNext) | ||
assertTrue(allBrokersLineIter.next().contains(s"Querying brokers for log directories information")) | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
the resource declaration should be followed by
try
block.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good.It has been modified in the latest submission.