$ ./kafka-producer-perf-test.sh
usage: producer-performance [-h] --topic TOPIC --num-records NUM-RECORDS [--payload-delimiter PAYLOAD-DELIMITER]
--throughput THROUGHPUT [--producer-props PROP-NAME=PROP-VALUE [PROP-NAME=PROP-VALUE ...]]
[--producer.config CONFIG-FILE] [--print-metrics] [--transactional-id TRANSACTIONAL-ID]
[--transaction-duration-ms TRANSACTION-DURATION] (--record-size RECORD-SIZE |
--payload-file PAYLOAD-FILE)
This tool is used to verify the producer performance.
optional arguments:
-h, --help show this help message and exit
--topic TOPIC produce messages to this topic
--num-records NUM-RECORDS
number of messages to produce
--payload-delimiter PAYLOAD-DELIMITER
provides delimiter to be used when --payload-file is provided. Defaults to new line. Note that this
parameter will be ignored if --payload-file is not provided. (default: \n)
--throughput THROUGHPUT
throttle maximum message throughput to *approximately* THROUGHPUT messages/sec. Set this to -1 to disable
throttling.
--producer-props PROP-NAME=PROP-VALUE [PROP-NAME=PROP-VALUE ...]
kafka producer related configuration properties like bootstrap.servers,client.id etc. These configs take
precedence over those passed via --producer.config.
--producer.config CONFIG-FILE
producer config properties file.
--print-metrics print out metrics at the end of the test. (default: false)
--transactional-id TRANSACTIONAL-ID
The transactionalId to use if transaction-duration-ms is > 0. Useful when testing the performance of
concurrent transactions. (default: performance-producer-default-transactional-id)
--transaction-duration-ms TRANSACTION-DURATION
The max age of each transaction. The commitTransaction will be called after this time has elapsed.
Transactions are only enabled if this value is positive. (default: 0)
either --record-size or --payload-file must be specified but not both.
--record-size RECORD-SIZE
message size in bytes. Note that you must provide exactly one of --record-size or --payload-file.
--payload-file PAYLOAD-FILE
file to read the message payloads from. This works only for UTF-8 encoded text files. Payloads will be
read from this file and a payload will be randomly selected when sending messages. Note that you must
provide exactly one of --record-size or --payload-file.
주요 argument만 살펴보도록 하겠습니다.
Argument
설명
topic
프로듀싱할 topic명
num-record
프로듀싱할 message 개수
record-size
message 사이즈
throughput
메시지 처리량, -1이면 제한 두지 않고 처리
producer-props
producer 설정(ex. bootstrap-servers...)
print-metrics
Test 종료 후 metric 출력
그럼 한번 shell script을 통해 프로듀싱으로 테스트를 진행해 보겠습니다.
$ ./kafka-producer-perf-test.sh ₩
--topic test_topic ₩
--num-records 1000 ₩
--throughput -1 ₩
--producer-props bootstrap.servers=kafka1:9092 max.request.size=10485760 ₩
--record-size 1048576 ₩
--print-metric
263 records sent, 52.5 records/sec (52.54 MB/sec), 405.0 ms avg latency, 577.0 ms max latency.
285 records sent, 56.9 records/sec (56.90 MB/sec), 551.1 ms avg latency, 582.0 ms max latency.
339 records sent, 67.7 records/sec (67.68 MB/sec), 457.4 ms avg latency, 541.0 ms max latency.
1000 records sent, 59.837243 records/sec (59.84 MB/sec), 470.39 ms avg latency, 582.00 ms max latency, 472 ms 50th, 572 ms 95th, 578 ms 99th, 582 ms 99.9th.
Metric Name Value
app-info:commit-id:{client-id=perf-producer-client} : 50029d3ed8ba576f
app-info:start-time-ms:{client-id=perf-producer-client} : 1687739710990
app-info:version:{client-id=perf-producer-client} : 3.2.3
...
...
...
print-metric 옵션을 넣으면 Metric 정보가 많이 나오는데 생략하였습니다.
1MB 사이즈를 가진 record를 보낼 때 대략 1초에 60개의 record를 보내는 것으로 결과가 나옵니다.
1000개의 99.9%면 999개까지 보낼 때 582ms정도 시간의 걸리는 것으로 나옵니다.
$ ./kafka-consumer-perf-test.sh --help
This tool helps in performance test for the full zookeeper consumer
Option Description
------ -----------
--bootstrap-server <String: server to REQUIRED unless --broker-list
connect to> (deprecated) is specified. The server
(s) to connect to.
--broker-list <String: broker-list> DEPRECATED, use --bootstrap-server
instead; ignored if --bootstrap-
server is specified. The broker
list string in the form HOST1:PORT1,
HOST2:PORT2.
--consumer.config <String: config file> Consumer config properties file.
--date-format <String: date format> The date format to use for formatting
the time field. See java.text.
SimpleDateFormat for options.
(default: yyyy-MM-dd HH:mm:ss:SSS)
--fetch-size <Integer: size> The amount of data to fetch in a
single request. (default: 1048576)
--from-latest If the consumer does not already have
an established offset to consume
from, start with the latest message
present in the log rather than the
earliest message.
--group <String: gid> The group id to consume on. (default:
perf-consumer-84403)
--help Print usage information.
--hide-header If set, skips printing the header for
the stats
--messages <Long: count> REQUIRED: The number of messages to
send or consume
--num-fetch-threads <Integer: count> DEPRECATED AND IGNORED: Number of
fetcher threads. (default: 1)
--print-metrics Print out the metrics.
--reporting-interval <Integer: Interval in milliseconds at which to
interval_ms> print progress info. (default: 5000)
--show-detailed-stats If set, stats are reported for each
reporting interval as configured by
reporting-interval
--socket-buffer-size <Integer: size> The size of the tcp RECV size.
(default: 2097152)
--threads <Integer: count> DEPRECATED AND IGNORED: Number of
processing threads. (default: 10)
--timeout [Long: milliseconds] The maximum allowed time in
milliseconds between returned
records. (default: 10000)
--topic <String: topic> REQUIRED: The topic to consume from.
--version Display Kafka version.
혹시나 테스트할 때 주의할 점이 테스트할 Topic에 메시지가 남아있어야 합니다.
아니면 아래처럼 timeout으로 테스트가 불가능합니다.
$ ./kafka-consumer-perf-test.sh --bootstrap-server kafka1:9092 --topic test-topic --messages 1000
start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
WARNING: Exiting before consuming the expected number of messages: timeout (10000 ms) exceeded. You can use the --timeout option to increase the timeout.
2023-06-25 17:48:54:340, 2023-06-25 17:49:04:370, 0.0000, 0.0000, 0, 0.0000, 405, 9625, 0.0000, 0.0000
참고로 테스트하는 Topic은 아래 command에서 알 수 있듯이 partition 2, replication 2인 Topic입니다.
for (i <- 0 until numMessages) {
val message = randomBytesOfLen(random, messageLen)
val begin = System.nanoTime
//Send message (of random bytes) synchronously then immediately poll for it
producer.send(new ProducerRecord[Array[Byte], Array[Byte]](topic, message)).get()
val recordIter = consumer.poll(Duration.ofMillis(timeout)).iterator
val elapsed = System.nanoTime - begin
//Check we got results
if (!recordIter.hasNext) {
finalise()
throw new RuntimeException(s"poll() timed out before finding a result (timeout:[$timeout])")
}
//Check result matches the original record
val sent = new String(message, StandardCharsets.UTF_8)
val read = new String(recordIter.next().value(), StandardCharsets.UTF_8)
if (!read.equals(sent)) {
finalise()
throw new RuntimeException(s"The message read [$read] did not match the message sent [$sent]")
}
//Check we only got the one message
if (recordIter.hasNext) {
val count = 1 + recordIter.asScala.size
throw new RuntimeException(s"Only one result was expected during this test. We found [$count]")
}
//Report progress
if (i % 1000 == 0)
println(i.toString + "\t" + elapsed / 1000.0 / 1000.0)
totalTime += elapsed
latencies(i) = elapsed / 1000 / 1000
}
