From be43bf3a1257ca5f058e38af3e185ba775749614 Mon Sep 17 00:00:00 2001 From: amethystic Date: Thu, 1 Jun 2017 17:15:15 +0800 Subject: [PATCH 01/14] KAFKA-5358: Consumer perf tool should count rebalance time. Added 'join.group.ms' for new consumer to calculate the time of joining group. --- .../kafka/tools/ConsumerPerformance.scala | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index a5d4d5d73771d..a833c4b091822 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -54,10 +54,11 @@ 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) { if (!config.showDetailedStats) - println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + println(s"start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec${if (!config.useOldConsumer) ", join.group.ms" else ""}") else println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") } @@ -67,7 +68,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) endMs = System.currentTimeMillis if (config.printMetrics) { @@ -100,8 +101,14 @@ object ConsumerPerformance { val elapsedSecs = (endMs - startMs) / 1000.0 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)) + println(s"%s, %s, %.4f, %.4f, %d, %.4f${if (!config.useOldConsumer) ", %d" else "%s"}".format( + config.dateFormat.format(startMs), + config.dateFormat.format(endMs), + totalMBRead, + totalMBRead / elapsedSecs, + totalMessagesRead.get, + totalMessagesRead.get / elapsedSecs, + if (!config.useOldConsumer) joinGroupTimeInMs.get else "")) } if (metrics != null) { @@ -110,11 +117,12 @@ object ConsumerPerformance { } - def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], topics: List[String], count: Long, timeout: Long, config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong) { + def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], topics: List[String], count: Long, timeout: Long, config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong, joinTime: AtomicLong) { var bytesRead = 0L var messagesRead = 0L var lastBytesRead = 0L var lastMessagesRead = 0L + var joinEnd = 0L // Wait for group join, metadata fetch, etc val joinTimeout = 10000 @@ -122,6 +130,7 @@ object ConsumerPerformance { consumer.subscribe(topics.asJava, new ConsumerRebalanceListener { def onPartitionsAssigned(partitions: util.Collection[TopicPartition]) { isAssigned.set(true) + joinEnd = System.currentTimeMillis } def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) { isAssigned.set(false) @@ -165,6 +174,7 @@ object ConsumerPerformance { totalMessagesRead.set(messagesRead) totalBytesRead.set(bytesRead) + joinTime.set(joinEnd - joinStart) } def printProgressMessage(id: Int, bytesRead: Long, lastBytesRead: Long, messagesRead: Long, lastMessagesRead: Long, From cbdf6c10d12e1b0528dc45c078e61bb6ee1b0d2f Mon Sep 17 00:00:00 2001 From: amethystic Date: Fri, 2 Jun 2017 15:11:03 +0800 Subject: [PATCH 02/14] 1. Refined the name to `total.rebalance.time` 2. Added `total.fetch.time` 3. Add support to count total time for multiple rebalances --- .../kafka/tools/ConsumerPerformance.scala | 31 +++++++++++++------ 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index a833c4b091822..b902d3b4804a3 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -55,10 +55,11 @@ object ConsumerPerformance { val consumerTimeout = new AtomicBoolean(false) var metrics: mutable.Map[MetricName, _ <: Metric] = null val joinGroupTimeInMs = new AtomicLong(0) + val fetchTimeInMs = new AtomicLong(0) if (!config.hideHeader) { if (!config.showDetailedStats) - println(s"start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec${if (!config.useOldConsumer) ", join.group.ms" else ""}") + println(s"start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec${if (!config.useOldConsumer) ", total.rebalance.time.ms, fetch.time.ms" else ""}") else println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") } @@ -68,7 +69,8 @@ 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, joinGroupTimeInMs) + consume(consumer, List(config.topic), config.numMessages, 1000, + config, totalMessagesRead, totalBytesRead, joinGroupTimeInMs, fetchTimeInMs) endMs = System.currentTimeMillis if (config.printMetrics) { @@ -101,14 +103,14 @@ object ConsumerPerformance { val elapsedSecs = (endMs - startMs) / 1000.0 if (!config.showDetailedStats) { val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) - println(s"%s, %s, %.4f, %.4f, %d, %.4f${if (!config.useOldConsumer) ", %d" else "%s"}".format( + println(s"%s, %s, %.4f, %.4f, %d, %.4f${if (!config.useOldConsumer) ", %s" else ""}".format( config.dateFormat.format(startMs), config.dateFormat.format(endMs), totalMBRead, totalMBRead / elapsedSecs, totalMessagesRead.get, totalMessagesRead.get / elapsedSecs, - if (!config.useOldConsumer) joinGroupTimeInMs.get else "")) + if (!config.useOldConsumer) joinGroupTimeInMs.get.toString + ", " + fetchTimeInMs.get.toString else "")) } if (metrics != null) { @@ -117,12 +119,14 @@ object ConsumerPerformance { } - def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], topics: List[String], count: Long, timeout: Long, config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong, joinTime: AtomicLong) { + def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], topics: List[String], count: Long, timeout: Long, config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong, joinTime: AtomicLong, fetchTime: AtomicLong) { var bytesRead = 0L var messagesRead = 0L var lastBytesRead = 0L var lastMessagesRead = 0L - var joinEnd = 0L + var joinStart = 0L + var totalRebalanceTimeExceptTheFirst = 0L + var isFirstRebalance = true // Wait for group join, metadata fetch, etc val joinTimeout = 10000 @@ -130,14 +134,21 @@ object ConsumerPerformance { consumer.subscribe(topics.asJava, new ConsumerRebalanceListener { def onPartitionsAssigned(partitions: util.Collection[TopicPartition]) { isAssigned.set(true) - joinEnd = System.currentTimeMillis + val time = System.currentTimeMillis - joinStart + joinTime.addAndGet(time) + if (isFirstRebalance) { + isFirstRebalance = false + } else { + totalRebalanceTimeExceptTheFirst += time + } } def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) { isAssigned.set(false) + joinStart = System.currentTimeMillis }}) - val joinStart = System.currentTimeMillis() + val joinStartForFirstTime = System.currentTimeMillis() while (!isAssigned.get()) { - if (System.currentTimeMillis() - joinStart >= joinTimeout) { + if (System.currentTimeMillis() - joinStartForFirstTime >= joinTimeout) { throw new Exception("Timed out waiting for initial group join.") } consumer.poll(100) @@ -174,7 +185,7 @@ object ConsumerPerformance { totalMessagesRead.set(messagesRead) totalBytesRead.set(bytesRead) - joinTime.set(joinEnd - joinStart) + fetchTime.set(System.currentTimeMillis - startMs - totalRebalanceTimeExceptTheFirst) } def printProgressMessage(id: Int, bytesRead: Long, lastBytesRead: Long, messagesRead: Long, lastMessagesRead: Long, From 7600e6b994c251422444c905f5d4c718bf6f9935 Mon Sep 17 00:00:00 2001 From: huxihx Date: Tue, 6 Jun 2017 08:42:29 +0800 Subject: [PATCH 03/14] Refined time counting for both fetcher and rebalance as per hachikuji's comments. --- .../scala/kafka/tools/ConsumerPerformance.scala | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index b902d3b4804a3..6d4be8e6d3151 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -70,7 +70,7 @@ object ConsumerPerformance { consumer.subscribe(Collections.singletonList(config.topic)) startMs = System.currentTimeMillis consume(consumer, List(config.topic), config.numMessages, 1000, - config, totalMessagesRead, totalBytesRead, joinGroupTimeInMs, fetchTimeInMs) + config, totalMessagesRead, totalBytesRead, joinGroupTimeInMs) endMs = System.currentTimeMillis if (config.printMetrics) { @@ -101,6 +101,7 @@ object ConsumerPerformance { consumerConnector.shutdown() } val elapsedSecs = (endMs - startMs) / 1000.0 + fetchTimeInMs.set((endMs - startMs) - fetchTimeInMs.get) if (!config.showDetailedStats) { val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) println(s"%s, %s, %.4f, %.4f, %d, %.4f${if (!config.useOldConsumer) ", %s" else ""}".format( @@ -119,14 +120,12 @@ object ConsumerPerformance { } - def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], topics: List[String], count: Long, timeout: Long, config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong, joinTime: AtomicLong, fetchTime: AtomicLong) { + def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], topics: List[String], count: Long, timeout: Long, config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong, joinTime: AtomicLong) { var bytesRead = 0L var messagesRead = 0L var lastBytesRead = 0L var lastMessagesRead = 0L var joinStart = 0L - var totalRebalanceTimeExceptTheFirst = 0L - var isFirstRebalance = true // Wait for group join, metadata fetch, etc val joinTimeout = 10000 @@ -134,13 +133,7 @@ object ConsumerPerformance { consumer.subscribe(topics.asJava, new ConsumerRebalanceListener { def onPartitionsAssigned(partitions: util.Collection[TopicPartition]) { isAssigned.set(true) - val time = System.currentTimeMillis - joinStart - joinTime.addAndGet(time) - if (isFirstRebalance) { - isFirstRebalance = false - } else { - totalRebalanceTimeExceptTheFirst += time - } + joinTime.addAndGet(System.currentTimeMillis - joinStart) } def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) { isAssigned.set(false) @@ -185,7 +178,6 @@ object ConsumerPerformance { totalMessagesRead.set(messagesRead) totalBytesRead.set(bytesRead) - fetchTime.set(System.currentTimeMillis - startMs - totalRebalanceTimeExceptTheFirst) } def printProgressMessage(id: Int, bytesRead: Long, lastBytesRead: Long, messagesRead: Long, lastMessagesRead: Long, From 08ff452fcb41136b54cfc2850540b18d737e909d Mon Sep 17 00:00:00 2001 From: huxihx Date: Wed, 7 Jun 2017 14:27:00 +0800 Subject: [PATCH 04/14] Correct the counting for total fetch time. --- core/src/main/scala/kafka/tools/ConsumerPerformance.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 6d4be8e6d3151..d4a25a98c61c6 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -101,7 +101,7 @@ object ConsumerPerformance { consumerConnector.shutdown() } val elapsedSecs = (endMs - startMs) / 1000.0 - fetchTimeInMs.set((endMs - startMs) - fetchTimeInMs.get) + fetchTimeInMs.set((endMs - startMs) - joinGroupTimeInMs.get) if (!config.showDetailedStats) { val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) println(s"%s, %s, %.4f, %.4f, %d, %.4f${if (!config.useOldConsumer) ", %s" else ""}".format( From 8c80f1376dba7ac715fa0887e128293690a7014c Mon Sep 17 00:00:00 2001 From: huxihx Date: Thu, 8 Jun 2017 09:31:49 +0800 Subject: [PATCH 05/14] 1. Split `MB.sec` into two parts: `total.MB.sec` and `fetch.MB.sec` 2. Ditto for `nMsg.sec` 3. Refined output format --- .../kafka/tools/ConsumerPerformance.scala | 32 ++++++++++++------- 1 file changed, 21 insertions(+), 11 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index d4a25a98c61c6..74cae50f65d19 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -58,9 +58,10 @@ object ConsumerPerformance { val fetchTimeInMs = new AtomicLong(0) if (!config.hideHeader) { - if (!config.showDetailedStats) - println(s"start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec${if (!config.useOldConsumer) ", total.rebalance.time.ms, fetch.time.ms" else ""}") - else + if (!config.showDetailedStats) { + if (config.useOldConsumer) + println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + } else println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") } @@ -104,14 +105,23 @@ object ConsumerPerformance { fetchTimeInMs.set((endMs - startMs) - joinGroupTimeInMs.get) if (!config.showDetailedStats) { val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) - println(s"%s, %s, %.4f, %.4f, %d, %.4f${if (!config.useOldConsumer) ", %s" else ""}".format( - config.dateFormat.format(startMs), - config.dateFormat.format(endMs), - totalMBRead, - totalMBRead / elapsedSecs, - totalMessagesRead.get, - totalMessagesRead.get / elapsedSecs, - if (!config.useOldConsumer) joinGroupTimeInMs.get.toString + ", " + fetchTimeInMs.get.toString else "")) + if (config.useOldConsumer) { + println(s"%s, %s, %.4f, %.4f, %d, %.4f${if (!config.useOldConsumer) ", %s" else ""}".format( + config.dateFormat.format(startMs), config.dateFormat.format(endMs), totalMBRead, + totalMBRead / elapsedSecs, totalMessagesRead.get, totalMessagesRead.get / elapsedSecs)) + } else { + println("---- performance test result ----") + println(s"Start time:${"\t" * 5}" + config.dateFormat.format(startMs)) + println(s"End time:${"\t" * 5}" + config.dateFormat.format(startMs)) + println(s"Total time:${"\t" * 5}%.4fs [%dms(rebalance time) + %dms(fetch time)]".format( + elapsedSecs, joinGroupTimeInMs.get, fetchTimeInMs.get)) + println(s"Total MB read:${"\t" * 4}%.4f".format(totalMBRead)) + println(s"Total throughput(MB/s):${"\t" * 2}%.4f".format(totalMBRead / elapsedSecs)) + println(s"Fetch throughput(MB/s):${"\t" * 2}%.4f".format(totalMBRead / (fetchTimeInMs.get / 1000.0))) + println(s"Total messages read:${"\t" * 2}%d".format(totalMessagesRead.get)) + println(s"Total message rate(msg/s):${"\t"}%.4f".format(totalMessagesRead.get / elapsedSecs)) + println(s"Fetch message rate(msg/s):\t%.4f".format(totalMessagesRead.get / (fetchTimeInMs.get / 1000.0))) + } } if (metrics != null) { From edf1d0888728bc5c39b6ee59d23d3eed108243c6 Mon Sep 17 00:00:00 2001 From: huxihx Date: Wed, 21 Jun 2017 09:24:28 +0800 Subject: [PATCH 06/14] returned back to the original output format for new consumer --- .../kafka/tools/ConsumerPerformance.scala | 36 +++++++++---------- 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 74cae50f65d19..f58bf4edb8eff 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -59,8 +59,9 @@ object ConsumerPerformance { if (!config.hideHeader) { if (!config.showDetailedStats) { - if (config.useOldConsumer) - println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + println("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec" + + s"${if (!config.useOldConsumer) ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else ""}" + ) } else println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") } @@ -105,22 +106,21 @@ object ConsumerPerformance { fetchTimeInMs.set((endMs - startMs) - joinGroupTimeInMs.get) if (!config.showDetailedStats) { val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) - if (config.useOldConsumer) { - println(s"%s, %s, %.4f, %.4f, %d, %.4f${if (!config.useOldConsumer) ", %s" else ""}".format( - config.dateFormat.format(startMs), config.dateFormat.format(endMs), totalMBRead, - totalMBRead / elapsedSecs, totalMessagesRead.get, totalMessagesRead.get / elapsedSecs)) - } else { - println("---- performance test result ----") - println(s"Start time:${"\t" * 5}" + config.dateFormat.format(startMs)) - println(s"End time:${"\t" * 5}" + config.dateFormat.format(startMs)) - println(s"Total time:${"\t" * 5}%.4fs [%dms(rebalance time) + %dms(fetch time)]".format( - elapsedSecs, joinGroupTimeInMs.get, fetchTimeInMs.get)) - println(s"Total MB read:${"\t" * 4}%.4f".format(totalMBRead)) - println(s"Total throughput(MB/s):${"\t" * 2}%.4f".format(totalMBRead / elapsedSecs)) - println(s"Fetch throughput(MB/s):${"\t" * 2}%.4f".format(totalMBRead / (fetchTimeInMs.get / 1000.0))) - println(s"Total messages read:${"\t" * 2}%d".format(totalMessagesRead.get)) - println(s"Total message rate(msg/s):${"\t"}%.4f".format(totalMessagesRead.get / elapsedSecs)) - println(s"Fetch message rate(msg/s):\t%.4f".format(totalMessagesRead.get / (fetchTimeInMs.get / 1000.0))) + 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.get, + totalMBRead / (fetchTimeInMs.get / 1000.0), + totalMessagesRead.get / (fetchTimeInMs.get / 1000.0) + )) } } From 722e16df1b96275a04bd2e6a5e3deec476b028a6 Mon Sep 17 00:00:00 2001 From: huxihx Date: Tue, 27 Jun 2017 10:03:55 +0800 Subject: [PATCH 07/14] As per hackikuji's comments, refined code to print out rebalance time even when is set. --- .../kafka/tools/ConsumerPerformance.scala | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index f58bf4edb8eff..125937e48186c 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -63,7 +63,8 @@ object ConsumerPerformance { s"${if (!config.useOldConsumer) ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else ""}" ) } else - println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") + println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec" + + s"${if (!config.useOldConsumer) ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else ""}") } var startMs, endMs = 0L @@ -178,7 +179,7 @@ object ConsumerPerformance { if (currentTimeMillis - lastReportTime >= config.reportingInterval) { if (config.showDetailedStats) - printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, currentTimeMillis, config.dateFormat) + printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, currentTimeMillis, config.dateFormat, Some(joinTime.get)) lastReportTime = currentTimeMillis lastMessagesRead = messagesRead lastBytesRead = bytesRead @@ -191,12 +192,21 @@ object ConsumerPerformance { } def printProgressMessage(id: Int, bytesRead: Long, lastBytesRead: Long, messagesRead: Long, lastMessagesRead: Long, - startMs: Long, endMs: Long, dateFormat: SimpleDateFormat) = { + startMs: Long, endMs: Long, dateFormat: SimpleDateFormat, joinTimeMs: Option[Long] = None) = { 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)) + + joinTimeMs match { + case None => + println("%s, %d, %.4f, %.4f, %d, %.4f".format(dateFormat.format(endMs), id, totalMBRead, + 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0)) + case Some(rebalanceTimeMs) => + val fetchTimeMs = elapsedMs - rebalanceTimeMs + println("%s, %d, %.4f, %.4f, %d, %.4f, %.2f, %.2f, %.4f, %.4f".format(dateFormat.format(endMs), id, totalMBRead, + 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0, + rebalanceTimeMs.toDouble, fetchTimeMs, mbRead / (fetchTimeMs / 1000.0), (messagesRead - lastMessagesRead) / (fetchTimeMs / 1000.0))) + } } class ConsumerPerfConfig(args: Array[String]) extends PerfConfig(args) { From 57bd0e44723b681a191a12135e3cc60188c18e9d Mon Sep 17 00:00:00 2001 From: huxihx Date: Fri, 4 Aug 2017 10:27:51 +0800 Subject: [PATCH 08/14] resovled conflicts with trunk --- .../kafka/tools/ConsumerPerformance.scala | 36 ++++++++++++++++--- 1 file changed, 31 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 125937e48186c..8c53e4faf2369 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -131,7 +131,21 @@ object ConsumerPerformance { } - def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], topics: List[String], count: Long, timeout: Long, config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong, joinTime: AtomicLong) { + 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") + } + + def consume(consumer: KafkaConsumer[Array[Byte], Array[Byte]], + topics: List[String], + count: Long, + timeout: Long, + config: ConsumerPerfConfig, + totalMessagesRead: AtomicLong, + totalBytesRead: AtomicLong, + joinTime: AtomicLong) { var bytesRead = 0L var messagesRead = 0L var lastBytesRead = 0L @@ -191,8 +205,15 @@ object ConsumerPerformance { totalBytesRead.set(bytesRead) } - def printProgressMessage(id: Int, bytesRead: Long, lastBytesRead: Long, messagesRead: Long, lastMessagesRead: Long, - startMs: Long, endMs: Long, dateFormat: SimpleDateFormat, joinTimeMs: Option[Long] = None) = { + def printProgressMessage(id: Int, + bytesRead: Long, + lastBytesRead: Long, + messagesRead: Long, + lastMessagesRead: Long, + startMs: Long, + endMs: Long, + dateFormat: SimpleDateFormat, + joinTimeMs: Option[Long] = None) = { val elapsedMs: Double = endMs - startMs val totalMBRead = (bytesRead * 1.0) / (1024 * 1024) val mbRead = ((bytesRead - lastBytesRead) * 1.0) / (1024 * 1024) @@ -304,8 +325,13 @@ object ConsumerPerformance { val hideHeader = options.has(hideHeaderOpt) } - class ConsumerPerfThread(threadId: Int, name: String, stream: KafkaStream[Array[Byte], Array[Byte]], - config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong, consumerTimeout: AtomicBoolean) + class ConsumerPerfThread(threadId: Int, + name: String, + stream: KafkaStream[Array[Byte], Array[Byte]], + config: ConsumerPerfConfig, + totalMessagesRead: AtomicLong, + totalBytesRead: AtomicLong, + consumerTimeout: AtomicBoolean) extends Thread(name) { override def run() { From 37982726a6fd09ef202eb428a2dd364aded1b929 Mon Sep 17 00:00:00 2001 From: huxihx Date: Fri, 4 Aug 2017 12:06:25 +0800 Subject: [PATCH 09/14] KAFKA-5358: Did not show newly-created headers if `--show-detailed-stats` is set since rebalance time does not change during most of the consuming rounds. --- .../kafka/tools/ConsumerPerformance.scala | 21 +++++-------------- 1 file changed, 5 insertions(+), 16 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 3251a75be1ab0..c852262b79d85 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -131,8 +131,7 @@ object ConsumerPerformance { s"${if (!useOldConsumer) ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else ""}" ) } else { - println("time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec" + - s"${if (!useOldConsumer) ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else ""}") + println("time, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec") } } @@ -191,7 +190,7 @@ object ConsumerPerformance { if (currentTimeMillis - lastReportTime >= config.reportingInterval) { if (config.showDetailedStats) - printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, currentTimeMillis, config.dateFormat, Some(joinTime.get)) + printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, currentTimeMillis, config.dateFormat) lastReportTime = currentTimeMillis lastMessagesRead = messagesRead lastBytesRead = bytesRead @@ -210,22 +209,12 @@ object ConsumerPerformance { lastMessagesRead: Long, startMs: Long, endMs: Long, - dateFormat: SimpleDateFormat, - joinTimeMs: Option[Long] = None) = { + dateFormat: SimpleDateFormat) = { val elapsedMs: Double = endMs - startMs val totalMBRead = (bytesRead * 1.0) / (1024 * 1024) val mbRead = ((bytesRead - lastBytesRead) * 1.0) / (1024 * 1024) - - joinTimeMs match { - case None => - println("%s, %d, %.4f, %.4f, %d, %.4f".format(dateFormat.format(endMs), id, totalMBRead, - 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0)) - case Some(rebalanceTimeMs) => - val fetchTimeMs = elapsedMs - rebalanceTimeMs - println("%s, %d, %.4f, %.4f, %d, %.4f, %.2f, %.2f, %.4f, %.4f".format(dateFormat.format(endMs), id, totalMBRead, - 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0, - rebalanceTimeMs.toDouble, fetchTimeMs, mbRead / (fetchTimeMs / 1000.0), (messagesRead - lastMessagesRead) / (fetchTimeMs / 1000.0))) - } + println("%s, %d, %.4f, %.4f, %d, %.4f".format(dateFormat.format(endMs), id, totalMBRead, + 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0)) } class ConsumerPerfConfig(args: Array[String]) extends PerfConfig(args) { From bbba9b00a6538e69bc4261ed456a31993e42650e Mon Sep 17 00:00:00 2001 From: huxihx Date: Wed, 9 Aug 2017 10:15:56 +0800 Subject: [PATCH 10/14] correct printHeader invoking to have testHeaderMatchBody passed --- .../test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala index 2fa774c6f1f96..e60812135b5aa 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -30,7 +30,7 @@ class ConsumerPerformanceTest { @Test def testHeaderMatchBody(): Unit = { Console.withOut(outContent) { - ConsumerPerformance.printHeader(true) + ConsumerPerformance.printHeader(true, false) ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS") ) From 3a1ce0f8897577608182ffe1a8ce2f9bfbe41f55 Mon Sep 17 00:00:00 2001 From: huxihx Date: Thu, 17 Aug 2017 10:44:23 +0800 Subject: [PATCH 11/14] Added newly-created fields for detailed views. --- .../kafka/tools/ConsumerPerformance.scala | 38 ++++++++++++------- .../kafka/tools/ConsumerPerformanceTest.scala | 36 ++++++++++++------ 2 files changed, 49 insertions(+), 25 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index c852262b79d85..3d52a651b6d6d 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -55,7 +55,7 @@ object ConsumerPerformance { val consumerTimeout = new AtomicBoolean(false) var metrics: mutable.Map[MetricName, _ <: Metric] = null val joinGroupTimeInMs = new AtomicLong(0) - val fetchTimeInMs = new AtomicLong(0) + var fetchTimeInMs = 0L if (!config.hideHeader) { printHeader(config.showDetailedStats, config.useOldConsumer) @@ -67,7 +67,7 @@ object ConsumerPerformance { consumer.subscribe(Collections.singletonList(config.topic)) startMs = System.currentTimeMillis consume(consumer, List(config.topic), config.numMessages, 1000, - config, totalMessagesRead, totalBytesRead, joinGroupTimeInMs) + config, totalMessagesRead, totalBytesRead, joinGroupTimeInMs, startMs) endMs = System.currentTimeMillis if (config.printMetrics) { @@ -98,7 +98,7 @@ object ConsumerPerformance { consumerConnector.shutdown() } val elapsedSecs = (endMs - startMs) / 1000.0 - fetchTimeInMs.set((endMs - startMs) - joinGroupTimeInMs.get) + fetchTimeInMs = (endMs - startMs) - joinGroupTimeInMs.get if (!config.showDetailedStats) { val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) print("%s, %s, %.4f, %.4f, %d, %.4f".format( @@ -112,9 +112,9 @@ object ConsumerPerformance { if (!config.useOldConsumer) { print(", %d, %d, %.4f, %.4f".format( joinGroupTimeInMs.get, - fetchTimeInMs.get, - totalMBRead / (fetchTimeInMs.get / 1000.0), - totalMessagesRead.get / (fetchTimeInMs.get / 1000.0) + fetchTimeInMs, + totalMBRead / (fetchTimeInMs / 1000.0), + totalMessagesRead.get / (fetchTimeInMs / 1000.0) )) } } @@ -126,12 +126,11 @@ object ConsumerPerformance { } private[tools] def printHeader(showDetailedStats: Boolean, useOldConsumer: Boolean): Unit = { + val newFieldsInHeader = s"${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" + - s"${if (!useOldConsumer) ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else ""}" - ) + 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") + println("time, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec" + newFieldsInHeader) } } @@ -142,7 +141,8 @@ object ConsumerPerformance { config: ConsumerPerfConfig, totalMessagesRead: AtomicLong, totalBytesRead: AtomicLong, - joinTime: AtomicLong) { + joinTime: AtomicLong, + testStartTime: Long) { var bytesRead = 0L var messagesRead = 0L var lastBytesRead = 0L @@ -190,7 +190,8 @@ object ConsumerPerformance { if (currentTimeMillis - lastReportTime >= config.reportingInterval) { if (config.showDetailedStats) - printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, lastReportTime, currentTimeMillis, config.dateFormat) + printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, + lastReportTime, currentTimeMillis, config.dateFormat, Some(testStartTime), Some(joinTime.get)) lastReportTime = currentTimeMillis lastMessagesRead = messagesRead lastBytesRead = bytesRead @@ -209,12 +210,21 @@ object ConsumerPerformance { lastMessagesRead: Long, startMs: Long, endMs: Long, - dateFormat: SimpleDateFormat) = { + dateFormat: SimpleDateFormat, + beginTime: Option[Long] = None, + joinTimeOpt: Option[Long] = None) = { 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, + print("%s, %d, %.4f, %.4f, %d, %.4f".format(dateFormat.format(endMs), id, totalMBRead, 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0)) + joinTimeOpt match { + case Some(joinTime) => + val fetchTimeMs = endMs - beginTime.getOrElse(startMs) - joinTime + println(", %d, %d, %.4f, %.4f".format(joinTime, fetchTimeMs, (bytesRead * 1.0 / (1024 * 1024) / (fetchTimeMs / 1000.0)), + messagesRead / (fetchTimeMs / 1000.0))) + case None => println() + } } class ConsumerPerfConfig(args: Array[String]) extends PerfConfig(args) { diff --git a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala index e60812135b5aa..f1679c8b7fe29 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -26,21 +26,35 @@ 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(true, false, 2, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, + dateFormat, Some(1L), Some(1L))) + testHeaderMatchContent(true, true, 4, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, + dateFormat)) + } + + @Test + def testNonDetailedHeaderMatchBody(): Unit = { + testHeaderMatchContent(false, 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(false, 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, false) - 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) + } } } From c51781586646664bb9514d8ab03aa974f9e6a942 Mon Sep 17 00:00:00 2001 From: huxihx Date: Mon, 21 Aug 2017 09:18:24 +0800 Subject: [PATCH 12/14] 1. Added parameter name when invoking testDetailedHeaderMatchBody; 2. Removed useless initialization for --- core/src/main/scala/kafka/tools/ConsumerPerformance.scala | 3 +-- .../scala/unit/kafka/tools/ConsumerPerformanceTest.scala | 8 ++++---- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 3d52a651b6d6d..1dc827673085e 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -55,7 +55,6 @@ object ConsumerPerformance { val consumerTimeout = new AtomicBoolean(false) var metrics: mutable.Map[MetricName, _ <: Metric] = null val joinGroupTimeInMs = new AtomicLong(0) - var fetchTimeInMs = 0L if (!config.hideHeader) { printHeader(config.showDetailedStats, config.useOldConsumer) @@ -98,7 +97,7 @@ object ConsumerPerformance { consumerConnector.shutdown() } val elapsedSecs = (endMs - startMs) / 1000.0 - fetchTimeInMs = (endMs - startMs) - joinGroupTimeInMs.get + val fetchTimeInMs = (endMs - startMs) - joinGroupTimeInMs.get if (!config.showDetailedStats) { val totalMBRead = (totalBytesRead.get * 1.0) / (1024 * 1024) print("%s, %s, %.4f, %.4f, %d, %.4f".format( diff --git a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala index f1679c8b7fe29..5da45d60c3a5d 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -30,17 +30,17 @@ class ConsumerPerformanceTest { @Test def testDetailedHeaderMatchBody(): Unit = { - testHeaderMatchContent(true, false, 2, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, + testHeaderMatchContent(detailed = true, useOldConsumer = false, 2, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, dateFormat, Some(1L), Some(1L))) - testHeaderMatchContent(true, true, 4, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, + testHeaderMatchContent(detailed = true, useOldConsumer = true, 4, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, dateFormat)) } @Test def testNonDetailedHeaderMatchBody(): Unit = { - testHeaderMatchContent(false, false, 2, () => println(s"${dateFormat.format(System.currentTimeMillis)}, " + + 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(false, true, 4, () => println(s"${dateFormat.format(System.currentTimeMillis)}, " + + testHeaderMatchContent(detailed = false, useOldConsumer = true, 4, () => println(s"${dateFormat.format(System.currentTimeMillis)}, " + s"${dateFormat.format(System.currentTimeMillis)}, 1.0, 1.0, 1, 1.0")) } From 9adda74004c531894a31b7653529b32336682316 Mon Sep 17 00:00:00 2001 From: huxihx Date: Wed, 6 Sep 2017 10:41:42 +0800 Subject: [PATCH 13/14] added a field that tracks periodic join time. --- .../kafka/tools/ConsumerPerformance.scala | 19 ++++++++++++------- .../kafka/tools/ConsumerPerformanceTest.scala | 2 +- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index 1dc827673085e..ea9e976ecd64c 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -65,8 +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, joinGroupTimeInMs, startMs) + consume(consumer, List(config.topic), config.numMessages, 1000, config, totalMessagesRead, totalBytesRead, joinGroupTimeInMs, startMs) endMs = System.currentTimeMillis if (config.printMetrics) { @@ -125,11 +124,12 @@ object ConsumerPerformance { } private[tools] def printHeader(showDetailedStats: Boolean, useOldConsumer: Boolean): Unit = { - val newFieldsInHeader = s"${if (!useOldConsumer) ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else ""}" + val newFieldsInHeader = s"${if (!useOldConsumer) ", total.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) + println("time, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec" + + s"${if (useOldConsumer) "" else ", rebalance.time.ms" + newFieldsInHeader}") } } @@ -147,6 +147,7 @@ object ConsumerPerformance { var lastBytesRead = 0L var lastMessagesRead = 0L var joinStart = 0L + var joinTimeMsInSingleRound = 0L // Wait for group join, metadata fetch, etc val joinTimeout = 10000 @@ -155,6 +156,7 @@ object ConsumerPerformance { 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) @@ -190,7 +192,8 @@ object ConsumerPerformance { if (currentTimeMillis - lastReportTime >= config.reportingInterval) { if (config.showDetailedStats) printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, - lastReportTime, currentTimeMillis, config.dateFormat, Some(testStartTime), Some(joinTime.get)) + lastReportTime, currentTimeMillis, config.dateFormat, Some(testStartTime), Some(joinTime.get), Some(joinTimeMsInSingleRound)) + joinTimeMsInSingleRound = 0L lastReportTime = currentTimeMillis lastMessagesRead = messagesRead lastBytesRead = bytesRead @@ -211,7 +214,8 @@ object ConsumerPerformance { endMs: Long, dateFormat: SimpleDateFormat, beginTime: Option[Long] = None, - joinTimeOpt: Option[Long] = None) = { + joinTimeOpt: Option[Long] = None, + periodicJoinTimeInMs: Option[Long] = None) = { val elapsedMs: Double = endMs - startMs val totalMBRead = (bytesRead * 1.0) / (1024 * 1024) val mbRead = ((bytesRead - lastBytesRead) * 1.0) / (1024 * 1024) @@ -220,7 +224,8 @@ object ConsumerPerformance { joinTimeOpt match { case Some(joinTime) => val fetchTimeMs = endMs - beginTime.getOrElse(startMs) - joinTime - println(", %d, %d, %.4f, %.4f".format(joinTime, fetchTimeMs, (bytesRead * 1.0 / (1024 * 1024) / (fetchTimeMs / 1000.0)), + println(", %d, %d, %d, %.4f, %.4f".format( + periodicJoinTimeInMs.getOrElse(0L), joinTime, fetchTimeMs, (bytesRead * 1.0 / (1024 * 1024) / (fetchTimeMs / 1000.0)), messagesRead / (fetchTimeMs / 1000.0))) case None => println() } diff --git a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala index 5da45d60c3a5d..f3f7e0bf4155d 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -31,7 +31,7 @@ class ConsumerPerformanceTest { @Test def testDetailedHeaderMatchBody(): Unit = { testHeaderMatchContent(detailed = true, useOldConsumer = false, 2, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, - dateFormat, Some(1L), Some(1L))) + dateFormat, Some(1L), Some(1L), Some(1L))) testHeaderMatchContent(detailed = true, useOldConsumer = true, 4, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, dateFormat)) } From dfa9422cc9d842db411435745aa6fac0961f272a Mon Sep 17 00:00:00 2001 From: huxihx Date: Mon, 18 Sep 2017 09:57:28 +0800 Subject: [PATCH 14/14] addressed Jason's last comments 1. refactoring printing to common functions 2. removed redundant codes --- .../kafka/tools/ConsumerPerformance.scala | 94 +++++++++++-------- .../kafka/tools/ConsumerPerformanceTest.scala | 7 +- 2 files changed, 59 insertions(+), 42 deletions(-) diff --git a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala index ea9e976ecd64c..57b7c35ed7688 100644 --- a/core/src/main/scala/kafka/tools/ConsumerPerformance.scala +++ b/core/src/main/scala/kafka/tools/ConsumerPerformance.scala @@ -124,12 +124,11 @@ object ConsumerPerformance { } private[tools] def printHeader(showDetailedStats: Boolean, useOldConsumer: Boolean): Unit = { - val newFieldsInHeader = s"${if (!useOldConsumer) ", total.rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec" else ""}" + 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" + - s"${if (useOldConsumer) "" else ", rebalance.time.ms" + newFieldsInHeader}") + println("time, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec" + newFieldsInHeader) } } @@ -149,26 +148,15 @@ object ConsumerPerformance { 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 + joinTimeMsInSingleRound += System.currentTimeMillis - joinStart } def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) { - isAssigned.set(false) joinStart = System.currentTimeMillis }}) - val joinStartForFirstTime = System.currentTimeMillis() - while (!isAssigned.get()) { - if (System.currentTimeMillis() - joinStartForFirstTime >= 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 @@ -191,8 +179,8 @@ object ConsumerPerformance { if (currentTimeMillis - lastReportTime >= config.reportingInterval) { if (config.showDetailedStats) - printProgressMessage(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, - lastReportTime, currentTimeMillis, config.dateFormat, Some(testStartTime), Some(joinTime.get), Some(joinTimeMsInSingleRound)) + printNewConsumerProgress(0, bytesRead, lastBytesRead, messagesRead, lastMessagesRead, + lastReportTime, currentTimeMillis, config.dateFormat, joinTimeMsInSingleRound) joinTimeMsInSingleRound = 0L lastReportTime = currentTimeMillis lastMessagesRead = messagesRead @@ -205,30 +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, - beginTime: Option[Long] = None, - joinTimeOpt: Option[Long] = None, - periodicJoinTimeInMs: Option[Long] = None) = { + 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) print("%s, %d, %.4f, %.4f, %d, %.4f".format(dateFormat.format(endMs), id, totalMBRead, 1000.0 * (mbRead / elapsedMs), messagesRead, ((messagesRead - lastMessagesRead) / elapsedMs) * 1000.0)) - joinTimeOpt match { - case Some(joinTime) => - val fetchTimeMs = endMs - beginTime.getOrElse(startMs) - joinTime - println(", %d, %d, %d, %.4f, %.4f".format( - periodicJoinTimeInMs.getOrElse(0L), joinTime, fetchTimeMs, (bytesRead * 1.0 / (1024 * 1024) / (fetchTimeMs / 1000.0)), - messagesRead / (fetchTimeMs / 1000.0))) - case None => println() - } + } + + 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) { @@ -353,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 @@ -369,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 f3f7e0bf4155d..bafe8ed136bb1 100644 --- a/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala +++ b/core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala @@ -30,9 +30,10 @@ class ConsumerPerformanceTest { @Test def testDetailedHeaderMatchBody(): Unit = { - testHeaderMatchContent(detailed = true, useOldConsumer = false, 2, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, - dateFormat, Some(1L), Some(1L), Some(1L))) - testHeaderMatchContent(detailed = true, useOldConsumer = true, 4, () => ConsumerPerformance.printProgressMessage(1, 1024 * 1024, 0, 1, 0, 0, 1, + 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)) }