Skip to content

KAFKA-5358: Consumer perf tool should count rebalance time. #3188

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.

Sign up for GitHub

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

Closed
huxihx wants to merge 15 commits into from
Closed

KAFKA-5358: Consumer perf tool should count rebalance time. #3188

Show file tree
Hide file tree
Changes from 12 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
be43bf3
KAFKA-5358: Consumer perf tool should count rebalance time.
huxi-2b Jun 1, 2017
cbdf6c1
1. Refined the name to `total.rebalance.time`
huxi-2b Jun 2, 2017
7600e6b
Refined time counting for both fetcher and rebalance as per hachikuji…
huxi-2b Jun 6, 2017
08ff452
Correct the counting for total fetch time.
huxi-2b Jun 7, 2017
8c80f13
1. Split `MB.sec` into two parts: `total.MB.sec` and `fetch.MB.sec`
huxi-2b Jun 8, 2017
edf1d08
returned back to the original output format for new consumer
huxi-2b Jun 21, 2017
722e16d
As per hackikuji's comments, refined code to print out rebalance time…
huxi-2b Jun 27, 2017
57bd0e4
resovled conflicts with trunk
huxi-2b Aug 4, 2017
16cc2ca
Merge branch 'trunk' into KAKFA-5358
huxi-2b Aug 4, 2017
3798272
KAFKA-5358: Did not show newly-created headers if `--show-detailed-st…
huxi-2b Aug 4, 2017
bbba9b0
correct printHeader invoking to have testHeaderMatchBody passed
huxi-2b Aug 9, 2017
3a1ce0f
Added newly-created fields for detailed views.
huxi-2b Aug 17, 2017
c517815
1. Added parameter name when invoking testDetailedHeaderMatchBody; 2.…
huxi-2b Aug 21, 2017
9adda74
added a field that tracks periodic join time.
huxi-2b Sep 6, 2017
dfa9422
addressed Jason's last comments
huxi-2b Sep 18, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 51 additions & 16 deletions core/src/main/scala/kafka/tools/ConsumerPerformance.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,17 +54,20 @@ object ConsumerPerformance {
val totalBytesRead = new AtomicLong(0)
val consumerTimeout = new AtomicBoolean(false)
var metrics: mutable.Map[MetricName, _ <: Metric] = null
val joinGroupTimeInMs = new AtomicLong(0)
var fetchTimeInMs = 0L

@hachikuji hachikuji Aug 19, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't need to be a var. We can just initialize it below, right?


if (!config.hideHeader) {
printHeader(!config.showDetailedStats)
printHeader(config.showDetailedStats, config.useOldConsumer)
}

var startMs, endMs = 0L
if (!config.useOldConsumer) {
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) {
Expand Down Expand Up @@ -95,10 +98,25 @@ object ConsumerPerformance {
consumerConnector.shutdown()
}
val elapsedSecs = (endMs - startMs) / 1000.0

@hachikuji hachikuji Jun 7, 2017 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the main point of this patch is to compute the throughput statistics based only on the time spent fetching, so maybe we can move this below and just use fetchTimeInMs.get / 1000.0? We could also report MB.sec as two separate values: total.MB.sec which uses the total time, and fetch.MB.sec which uses only the fetch time. Similarly for nMsg.sec.

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) {
Expand All @@ -107,11 +125,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 = 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" + 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]],
Expand All @@ -120,25 +140,30 @@ 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

// 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)
}
def onPartitionsRevoked(partitions: util.Collection[TopicPartition]) {
isAssigned.set(false)
joinStart = System.currentTimeMillis
}})
val joinStart = System.currentTimeMillis()
val joinStartForFirstTime = System.currentTimeMillis()

@hachikuji hachikuji Sep 8, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since we are now reporting the join time separately, I don't think we need the loop below anymore. It was initially intended to discount the rebalance overhead, but now we account for that transparently.

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)
Expand All @@ -165,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))

@hachikuji hachikuji Aug 22, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  1. There seems to be no reason for testStartTime to be an Option.
  2. Why do we call get on joinTime and rewrap it instead of just passing it through?

@huxihx huxihx Aug 23, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since joinTime is an AtomicLong, and printProgressMessage expects an Option, that's why I used Some(joinTime.get) here.
For old consumers, there is no join time, so printProgressMessage exposes an Option here. Does it make any senses?

lastReportTime = currentTimeMillis
lastMessagesRead = messagesRead
lastBytesRead = bytesRead
Expand All @@ -184,12 +210,21 @@ object ConsumerPerformance {
lastMessagesRead: Long,
startMs: Long,
endMs: Long,
dateFormat: SimpleDateFormat) = {
dateFormat: SimpleDateFormat,
beginTime: Option[Long] = None,

@hachikuji hachikuji Aug 22, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's a little weird that we have both a startTime and a beginTime. I think the reason is that beginTime is computed from the start of the test, but startTime is computed after the first rebalance. However, now that we are calculating the rebalance time separately, I wonder if we still need to maintain this distinction. Maybe we can just use the test start time for both cases.

@huxihx huxihx Aug 23, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

startTime changes in each round of printProgressMessage but beginTime does not change after a test begins. printProgressMessage must use startTime to calculate TPS in a single call.

However, we also need to pass through beginTime which does not change after perf test gets started. It is used to calculate the total fetch time and fetch-time-related measures.

What do you think?

@hachikuji hachikuji Aug 23, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, that's fair. In that case, perhaps we should do something similar for the join time. Maybe we can have one field which tracks periodic join time and is reset after each reporting interval, and another field which tracks the total join time.

@huxihx huxihx Sep 6, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

added periodicJoinTimeInMs to track join time for a single interval.

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,
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))
joinTimeOpt match {

@hachikuji hachikuji Sep 8, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe we should just move this block to a separate function so that we don't need all of the Option parameters?

@huxihx huxihx Sep 11, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, I am not sure if I fully understand this comment. Do you mean create a new function to handle all the printing things whereas printProgressMessage is that type of function?

@hachikuji hachikuji Sep 15, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it's a minor issue, but it's kind of annoying that we have three optional parameters which are all expected in the case of the new consumer. I think a cleaner approach is to use a separate method. Something approximately like this:

def printBasicProgress ...
def printExtraProgress ...

def printNewConsumerProgress {
  printBasicProgress()
  printExtraProgress()
  println()
}

def printOldConsumerProgress() {
  printBasicProgress()
  println()
}

Then we shouldn't need the optional parameters.

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)),

@hachikuji hachikuji Aug 19, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now we're printing two lines in every call to printProgressMessage? Shouldn't we just include the additional information into one message?

@huxihx huxihx Aug 21, 2017

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@hachikuji The new-added fields are within the same line since I changed the first println to print. Am I doing right?

messagesRead / (fetchTimeMs / 1000.0)))
case None => println()
}
}

class ConsumerPerfConfig(args: Array[String]) extends PerfConfig(args) {
Expand Down
36 changes: 25 additions & 11 deletions core/src/test/scala/unit/kafka/tools/ConsumerPerformanceTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,

@hachikuji hachikuji Aug 19, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you add parameter names for the two booleans?

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)
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)
}
}
}