diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index c81866140c74e..57b7c35ed7688 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -54,9 +54,10 @@ object ConsumerPerformance { val totalBytesRead = new AtomicLong(0) val consumerTimeout = new AtomicBoolean(false) var metrics: mutable.Map[MetricName, _ <: Metric] = null + val joinGroupTimeInMs = new AtomicLong(0) if (!config.hideHeader) { - printHeader(!config.showDetailedStats) + printHeader(config.showDetailedStats, config.useOldConsumer) } var startMs, endMs = 0L @@ -64,7 +65,7 @@ object ConsumerPerformance { val consumer = new KafkaConsumer[Array[Byte], Array[Byte]](config.props) consumer.subscribe(Collections.singletonList(config.topic)) startMs = System.currentTimeMillis - consume(consumer, List(config.topic), config.numMessages, 1000, config, totalMessagesRead, totalBytesRead) + consume(consumer, List(config.topic), config.numMessages, 1000, config, totalMessagesRead, totalBytesRead, joinGroupTimeInMs, startMs) endMs = System.currentTimeMillis if (config.printMetrics) { @@ -95,10 +96,25 @@ object ConsumerPerformance { consumerConnector.shutdown() } val elapsedSecs = (endMs - startMs) / 1000.0 + val fetchTimeInMs = (endMs - startMs) - joinGroupTimeInMs.get if (!config.showDetailedStats) { val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) - println("%s, %s, %.4f, %.4f, %d, %.4f".format(config.dateFormat.format(startMs), config.dateFormat.format(endMs), - totalMBRead, totalMBRead / elapsedSecs, totalMessagesRead.get, totalMessagesRead.get / elapsedSecs)) + print("%s, %s, %.4f, %.4f, %d, %.4f".format( + config.dateFormat.format(startMs), + config.dateFormat.format(endMs), + totalMBRead, + totalMBRead / elapsedSecs, + totalMessagesRead.get, + totalMessagesRead.get / elapsedSecs + )) + if (!config.useOldConsumer) { + print(", %d, %d, %.4f, %.4f".format( + joinGroupTimeInMs.get, + fetchTimeInMs, + totalMBRead / (fetchTimeInMs / 1000.0), + totalMessagesRead.get / (fetchTimeInMs / 1000.0) + )) + } } if (metrics != null) { @@ -107,11 +123,13 @@ object ConsumerPerformance { } - private[tools] def printHeader(showDetailedStats: Boolean): Unit = { - if (showDetailedStats) - println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") - else - println("time, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + private[tools] def printHeader(showDetailedStats: Boolean, useOldConsumer: Boolean): Unit = { + val newFieldsInHeader = if (!useOldConsumer) ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else "" + if (!showDetailedStats) { + println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec" + newFieldsInHeader) + } else { + println("time, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec" + newFieldsInHeader) + } } def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], @@ -120,29 +138,25 @@ object ConsumerPerformance { timeout: Long, config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, - totalBytesRead: AtomicLong) { + totalBytesRead: AtomicLong, + joinTime: AtomicLong, + testStartTime: Long) { var bytesRead = 0L var messagesRead = 0L var lastBytesRead = 0L var lastMessagesRead = 0L + var joinStart = 0L + var joinTimeMsInSingleRound = 0L - // Wait for group join, metadata fetch, etc - val joinTimeout = 10000 - val isAssigned = new AtomicBoolean(false) consumer.subscribe(topics.asJava, new ConsumerRebalanceListener { def onPartitionsAssigned(partitions: util.Collection[TopicPartition]) { - isAssigned.set(true) + joinTime.addAndGet(System.currentTimeMillis - joinStart) + joinTimeMsInSingleRound += System.currentTimeMillis - joinStart } def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) { - isAssigned.set(false) + joinStart = System.currentTimeMillis }}) - val joinStart = System.currentTimeMillis() - while (!isAssigned.get()) { - if (System.currentTimeMillis() - joinStart >= joinTimeout) { - throw new Exception("Timed out waiting for initial group join.") - } - consumer.poll(100) - } + consumer.poll(0) consumer.seekToBeginning(Collections.emptyList()) // Now start the benchmark @@ -165,7 +179,9 @@ object ConsumerPerformance { if (currentTimeMillis - lastReportTime >= config.reportingInterval) { if (config.showDetailedStats) - printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, currentTimeMillis, config.dateFormat) + printNewConsumerProgress(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, + lastReportTime, currentTimeMillis, config.dateFormat, joinTimeMsInSingleRound) + joinTimeMsInSingleRound = 0L lastReportTime = currentTimeMillis lastMessagesRead = messagesRead lastBytesRead = bytesRead @@ -177,19 +193,57 @@ object ConsumerPerformance { totalBytesRead.set(bytesRead) } - def printProgressMessage(id: Int, - bytesRead: Long, - lastBytesRead: Long, - messagesRead: Long, - lastMessagesRead: Long, - startMs: Long, - endMs: Long, - dateFormat: SimpleDateFormat) = { + def printOldConsumerProgress(id: Int, + bytesRead: Long, + lastBytesRead: Long, + messagesRead: Long, + lastMessagesRead: Long, + startMs: Long, + endMs: Long, + dateFormat: SimpleDateFormat): Unit = { + printBasicProgress(id, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, startMs, endMs, dateFormat) + println() + } + + def printNewConsumerProgress(id: Int, + bytesRead: Long, + lastBytesRead: Long, + messagesRead: Long, + lastMessagesRead: Long, + startMs: Long, + endMs: Long, + dateFormat: SimpleDateFormat, + periodicJoinTimeInMs: Long): Unit = { + printBasicProgress(id, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, startMs, endMs, dateFormat) + printExtendedProgress(bytesRead, messagesRead, startMs, endMs, periodicJoinTimeInMs) + } + + private def printBasicProgress(id: Int, + bytesRead: Long, + lastBytesRead: Long, + messagesRead: Long, + lastMessagesRead: Long, + startMs: Long, + endMs: Long, + dateFormat: SimpleDateFormat): Unit = { val elapsedMs: Double = endMs - startMs val totalMBRead = (bytesRead * 1.0) / (1024 * 1024) val mbRead = ((bytesRead - lastBytesRead) * 1.0) / (1024 * 1024) - println("%s, %d, %.4f, %.4f, %d, %.4f".format(dateFormat.format(endMs), id, totalMBRead, - 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0)) + print("%s, %d, %.4f, %.4f, %d, %.4f".format(dateFormat.format(endMs), id, totalMBRead, + 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0)) + } + + private def printExtendedProgress(bytesRead: Long, + messagesRead: Long, + startMs: Long, + endMs: Long, + periodicJoinTimeInMs: Long): Unit = { + val fetchTimeMs = if (endMs - startMs - periodicJoinTimeInMs <= 0) 0 else endMs - startMs - periodicJoinTimeInMs + if (fetchTimeMs > 0) + println(", %d, %d, %.4f, %.4f".format(periodicJoinTimeInMs, fetchTimeMs, + (bytesRead * 1.0 / (1024 * 1024) / (fetchTimeMs / 1000.0)), messagesRead / (fetchTimeMs / 1000.0))) + else + println(", %d, %d, %.4f, %.4f".format(periodicJoinTimeInMs, fetchTimeMs, 0.0, 0.0)) } class ConsumerPerfConfig(args: Array[String]) extends PerfConfig(args) { @@ -314,7 +368,7 @@ object ConsumerPerformance { if (currentTimeMillis - lastReportTime >= config.reportingInterval) { if (config.showDetailedStats) - printProgressMessage(threadId, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, currentTimeMillis, config.dateFormat) + printOldConsumerProgress(threadId, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, currentTimeMillis, config.dateFormat) lastReportTime = currentTimeMillis lastMessagesRead = messagesRead lastBytesRead = bytesRead @@ -330,7 +384,8 @@ object ConsumerPerformance { totalMessagesRead.addAndGet(messagesRead) totalBytesRead.addAndGet(bytesRead) if (config.showDetailedStats) - printProgressMessage(threadId, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, startMs, System.currentTimeMillis, config.dateFormat) + printOldConsumerProgress(threadId, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, startMs, System.currentTimeMillis, config.dateFormat) + } } diff --git a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala index 2fa774c6f1f96..bafe8ed136bb1 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -26,21 +26,36 @@ import org.junit.Test class ConsumerPerformanceTest { private val outContent = new ByteArrayOutputStream() + private val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS") @Test - def testHeaderMatchBody(): Unit = { + def testDetailedHeaderMatchBody(): Unit = { + testHeaderMatchContent(detailed = true, useOldConsumer = false, 2, + () => ConsumerPerformance.printNewConsumerProgress(1, 1024 * 1024, 0, 1, 0, 0, 1, dateFormat, 1L)) + testHeaderMatchContent(detailed = true, useOldConsumer = true, 4, + () => ConsumerPerformance.printOldConsumerProgress(1, 1024 * 1024, 0, 1, 0, 0, 1, + dateFormat)) + } + + @Test + def testNonDetailedHeaderMatchBody(): Unit = { + testHeaderMatchContent(detailed = false, useOldConsumer = false, 2, () => println(s"${dateFormat.format(System.currentTimeMillis)}, " + + s"${dateFormat.format(System.currentTimeMillis)}, 1.0, 1.0, 1, 1.0, 1, 1, 1.1, 1.1")) + testHeaderMatchContent(detailed = false, useOldConsumer = true, 4, () => println(s"${dateFormat.format(System.currentTimeMillis)}, " + + s"${dateFormat.format(System.currentTimeMillis)}, 1.0, 1.0, 1, 1.0")) + } + + private def testHeaderMatchContent(detailed: Boolean, useOldConsumer: Boolean, expectedOutputLineCount: Int, fun: () => Unit): Unit = { Console.withOut(outContent) { - ConsumerPerformance.printHeader(true) - ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, - new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS") - ) - } + ConsumerPerformance.printHeader(detailed, useOldConsumer) + fun() - val contents = outContent.toString.split("\n") - assertEquals(2, contents.length) - val header = contents(0) - val body = contents(1) + val contents = outContent.toString.split("\n") + assertEquals(expectedOutputLineCount, contents.length) + val header = contents(0) + val body = contents(1) - assertEquals(header.split(",").length, body.split(",").length) + assertEquals(header.split(",").length, body.split(",").length) + } } }