KAFKA-6743 ConsumerPerformance fails to consume all messages by rootex- · Pull Request #4818 · apache/kafka

rootex- · 2018-04-03T14:45:06Z

ConsumerPerformance fails to consume all messages

on topics with large number of partitions due to a relatively short default polling loop timeout (1000 ms) that is not reachable and modifiable by the end user.

Demo

Create a topic of 10 000 partitions, send a 50 000 000 of 100 byte records using kafka-producer-perf-test and consume them using kafka-consumer-perf-test (ConsumerPerformance). You will likely notice that the number of records returned by the kafka-consumer-perf-test is many times less than expected 50 000 000.

This happens due to specific ConsumerPerformance implementation. As the result, in some rough cases it may take a long enough time to process/iterate through the records polled in batches, thus, the time may exceed the default hardcoded polling loop timeout and this is probably not what we want from this utility.

Possible options

Increasing polling loop timeout in ConsumerPerformance implementation. It defaults to 1000 ms and is hardcoded, thus cannot be changed but we could export it as an OPTIONAL kafka-consumer-perf-test parameter to enable it on a script level configuration and available to the end user.
Decreasing max.poll.records on a Consumer config level. This is not a fine option though since we do not want to touch the default settings.

KAFKA-6743: ConsumerPerformance fails to consume all messages on topics with large number of partitions
KIP-281: ConsumerPerformance: Increase Polling Loop Timeout and Make It Reachable by the End User

Result

bin/kafka-consumer-perf-test.sh... --polling-loop-timeout <TIME_IN_MS>

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

Remove consumer offset reset on startup

hachikuji

Thanks for the patch. Left one comment. Also, maybe we can change the title? Does this patch have anything to do with offset reset behavior?

hachikuji · 2018-04-03T16:36:55Z

    val printMetricsOpt = parser.accepts("print-metrics", "Print out the metrics. This only applies to new consumer.")
    val showDetailedStatsOpt = parser.accepts("show-detailed-stats", "If set, stats are reported for each reporting " +
      "interval as configured by reporting-interval")
+    val pollLoopTimeoutOpt = parser.accepts("polling-loop-timeout", "Consumer polling loop timeout.")


The performance tool is part of the public API, so adding new options requires a short KIP.

To be honest, the default timeout of 1 second doesn't make much sense to me. I would probably consider either removing the timeout entirely or at least increasing it to something more substantial (say one minute). If we decide to keep it, we should add a message to indicate that the tool exited before consuming all messages.

Jason, thanks a lot for your remark. I'll be back in a while with new KIP.

Please review KIP-278

Thanks! I left a few comments on the dev list.

I've updated the KIP number to show the correct number KIP-281

@hachikuji please, review this one

@MosheBlumbergX thank you!

Waiting for some more feedback on KIP-281, friends.

Thanks

@rootex- Sorry, sent my initial response to the wrong thread. I've replied to the right one now.

@hachikuji No problem, appreciate for your feedback!

…rning message

…imeout to 10 seconds, updated the warning message

hachikuji · 2018-04-16T20:25:55Z

@rootex- Thanks for the updates. Can you reply to the KIP thread? Since there haven't been many comments, you can probably go ahead and start a vote.

rootex- · 2018-04-16T21:28:05Z

@hachikuji Thank you for the quick response, just replied to the KIP [DISCUSS] thread. And started the KIP [VOTE] thread

dhruvilshah3 · 2018-04-21T02:34:18Z

      }
+      if (messagesRead < count)
+        println(s"WARNING: Exiting before consuming the expected number of messages: timeout ($timeout ms) exceeded. " +
+          s"Probably too much time spent on processing a polled list of records. Be sure to increase the --timeout option")


Nit: Missing "." at the end.

I also wonder if this is the right place to print this message. Perhaps this should be outside the while loop?

@dhruvilshah3, my bad, thank you for pointing this out. Updated

dhruvilshah3 · 2018-04-21T02:42:06Z

Thanks for the PR! Left a minor comment, otherwise LGTM.

astubbs · 2018-05-29T09:19:27Z

Relates to #4082

hachikuji

@rootex- I think you have enough binding votes for this KIP, so you can close the vote. This is small enough that I think we can still get this patch into 2.0.0. I had just one small comment.

hachikuji · 2018-05-31T06:08:30Z

    val printMetricsOpt = parser.accepts("print-metrics", "Print out the metrics. This only applies to new consumer.")
    val showDetailedStatsOpt = parser.accepts("show-detailed-stats", "If set, stats are reported for each reporting " +
      "interval as configured by reporting-interval")
+    val pollLoopTimeoutOpt = parser.accepts("timeout", "Consumer polling loop timeout.")


Can we make this more descriptive? How about this?

The maximum allowed time between returned records

Also, maybe we can change the name to recordFetchTimeout or something like that?

Could we also mention that the timeout is in milliseconds? And perhaps also name the variable recordFetchTimeoutMs.

@dhruvilshah3
@hachikuji

Hello, guys. Sure, I'll take care of that today.

Would you like to apply changes for both the variable name and the UI parameter? If yes, then KIP should be updated too.

Thanks for closing the vote session. Been away for a while, I have more time now.

hachikuji · 2018-06-01T00:06:21Z

@rootex- I went ahead and called the vote. I'm going to go ahead and push a change to this branch to move things along.

rootex- · 2018-06-01T07:20:48Z

@hachikuji Jason, thanks for closing the vote session. I'll push the changes today.

hachikuji · 2018-06-01T18:18:25Z

@rootex- Thanks, I can merge once the changes are ready.

rootex- · 2018-06-01T18:30:44Z

@hachikuji Jason, would you like to apply changes for both the variable name recordFetchTimeoutMs and the UI parameter --record-fetch-timeout? If yes, then KIP should be updated too.

hachikuji · 2018-06-01T18:41:36Z

@rootex- Oh, I was actually referring to the variable pollLoopTimeoutOpt, not the command argument. I think --timeout is fine, but we can mention in the description that it is specified in milliseconds as @dhruvilshah3 suggests.

…Timeout description

…errors.TimeoutException. Supported versions: kafka 0.11 and newer

rootex- · 2018-06-02T12:49:42Z

@hachikuji Jason, thanks for explaining. It's all done now 👌 If everything's fine, let's merge it right away 👍

hachikuji

LGTM, thanks for the patch! Note that i made a minor tweak to the warning message.

…1] (apache#4818) This patch implements KIP-281, which adds a configurable timeout to the consumer performance tool with a default value of 10 seconds. The old timeout was hard-coded as 1 second. Additionally, this patch adds a warning message when the tool exits after a timeout rather than returning silently. Reviewers: Dhruvil Shah <dhruvil@confluent.io>, Jason Gustafson <jason@confluent.io>

rootex- changed the title ~~KAFKA-6724 ConsumerPerformance resets offsets on every startup~~ KAFKA-6743 ConsumerPerformance resets offsets on every startup Apr 3, 2018

KAFKA-6743 ConsumerPerformance resets offsets on every startup

5891628

Remove consumer offset reset on startup

rootex- force-pushed the KAFKA-6743_ConsumerPerformance_export_polling_loop_timeout_option_to_the_user branch from f14792b to 5891628 Compare April 3, 2018 14:57

hachikuji reviewed Apr 3, 2018

View reviewed changes

rootex- changed the title ~~KAFKA-6743 ConsumerPerformance resets offsets on every startup~~ KAFKA-6743 ConsumerPerformance fails to consume all messages Apr 3, 2018

rootex- added 2 commits April 4, 2018 15:58

KAFKA-6724 Increase polling loop timeout up to 60 000 ms and add a wa…

a2826dd

…rning message

KAFKA-6743 Renamed --polling-loop-timeout -> --timeout, decreased --t…

6ddd09c

…imeout to 10 seconds, updated the warning message

dhruvilshah3 reviewed Apr 21, 2018

View reviewed changes

Fixed warning message showup

fccf3b1

ijuma added this to the 2.0.0 milestone May 21, 2018

ijuma assigned hachikuji May 21, 2018

rootex- mentioned this pull request May 29, 2018

KAFKA-6225: Adds an option to consume continuously #4082

Closed

hachikuji reviewed May 31, 2018

View reviewed changes

rootex- added 2 commits June 2, 2018 15:26

Renamed pollLoopTimeout -> recordFetchTimeoutMs, improved recordFetch…

2cd77bb

…Timeout description

Fix deprecation: ConsumerTimeoutException -> org.apache.kafka.common.…

d3314e0

…errors.TimeoutException. Supported versions: kafka 0.11 and newer

rootex- and others added 3 commits June 2, 2018 21:25

Remove unused import statement

c84c42a

Rollback to ConsumerTimeoutException due to issues

883e1f9

Minor tweak to warning message

9124458

hachikuji approved these changes Jun 2, 2018

View reviewed changes

hachikuji merged commit 1d5f864 into apache:trunk Jun 2, 2018

Uh oh!

Conversation

rootex- commented Apr 3, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

ConsumerPerformance fails to consume all messages

Demo

Possible options

Result

Committer Checklist (excluded from commit message)

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

hachikuji Apr 3, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rootex- Apr 4, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

MosheBlumbergX Apr 5, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Apr 16, 2018

Uh oh!

rootex- commented Apr 16, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dhruvilshah3 commented Apr 21, 2018

Uh oh!

astubbs commented May 29, 2018

Uh oh!

hachikuji left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Jun 1, 2018

Uh oh!

rootex- commented Jun 1, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hachikuji commented Jun 1, 2018

Uh oh!

rootex- commented Jun 1, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hachikuji commented Jun 1, 2018

Uh oh!

rootex- commented Jun 2, 2018

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

rootex- commented Apr 3, 2018 •

edited

Loading

hachikuji Apr 3, 2018 •

edited

Loading

rootex- Apr 4, 2018 •

edited

Loading

MosheBlumbergX Apr 5, 2018 •

edited

Loading

hachikuji left a comment •

edited

Loading

rootex- commented Jun 1, 2018 •

edited

Loading

rootex- commented Jun 1, 2018 •

edited

Loading