Skip to content

KAFKA-6743 ConsumerPerformance fails to consume all messages #4818

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

Merged
hachikuji merged 9 commits into from
Jun 2, 2018
Merged

KAFKA-6743 ConsumerPerformance fails to consume all messages #4818

Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
5891628
KAFKA-6743 ConsumerPerformance resets offsets on every startup
rootex- Apr 3, 2018
a2826dd
KAFKA-6724 Increase polling loop timeout up to 60 000 ms and add a wa…
rootex- Apr 4, 2018
6ddd09c
KAFKA-6743 Renamed --polling-loop-timeout -> --timeout, decreased --t…
rootex- Apr 16, 2018
fccf3b1
Fixed warning message showup
rootex- Apr 30, 2018
2cd77bb
Renamed pollLoopTimeout -> recordFetchTimeoutMs, improved recordFetch…
rootex- Jun 2, 2018
d3314e0
Fix deprecation: ConsumerTimeoutException -> org.apache.kafka.common.…
rootex- Jun 2, 2018
c84c42a
Remove unused import statement
rootex- Jun 2, 2018
883e1f9
Rollback to ConsumerTimeoutException due to issues
rootex- Jun 2, 2018
9124458
Minor tweak to warning message
hachikuji Jun 2, 2018
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
11 changes: 10 additions & 1 deletion 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 @@ -65,7 +65,7 @@ object ConsumerPerformance extends LazyLogging {
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, config.recordFetchTimeoutMs, config, totalMessagesRead, totalBytesRead, joinGroupTimeInMs, startMs)
endMs = System.currentTimeMillis

if (config.printMetrics) {
Expand Down Expand Up @@ -188,6 +188,9 @@ object ConsumerPerformance extends LazyLogging {
}
}

if (messagesRead < count)
println(s"WARNING: Exiting before consuming the expected number of messages: timeout ($timeout ms) exceeded. " +
"You can use the --timeout option to increase the timeout."))
totalMessagesRead.set(messagesRead)
totalBytesRead.set(bytesRead)
}
Expand Down Expand Up @@ -302,6 +305,11 @@ object ConsumerPerformance extends LazyLogging {
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 recordFetchTimeoutOpt = parser.accepts("timeout", "The maximum allowed time in milliseconds between returned records.")
.withOptionalArg()
.describedAs("milliseconds")
.ofType(classOf[Long])
.defaultsTo(10000)

val options = parser.parse(args: _*)

Expand Down Expand Up @@ -354,6 +362,7 @@ object ConsumerPerformance extends LazyLogging {
val showDetailedStats = options.has(showDetailedStatsOpt)
val dateFormat = new SimpleDateFormat(options.valueOf(dateFormatOpt))
val hideHeader = options.has(hideHeaderOpt)
val recordFetchTimeoutMs = options.valueOf(recordFetchTimeoutOpt).longValue()
}

class ConsumerPerfThread(threadId: Int,
Expand Down