o `j@sddlZddlZddlZddlZddlZddlZddlmZGdddZGdddej Z ddZ e d krBe ZeedSdS) N) KafkaProducerc@seZdZeddZdS)ProducerPerformancec szi}jD]1}|d\}}zt|}Wn tyYnw|dkr&d}n |dkr-d}n|dkr3d}|||<qtdj|d<ttjj d |d <t di|| D] \}}td ||qZtd j td jtd j td jt}tj |jd}|tdtfdd}t}|t} ||td| |dWdStyt} tj| tdYdSw)N=NoneFalseFTrueTzInitializing producer...bootstrap_serversimetrics_sample_window_msz ---> {0}={1}z---> topic={0}z---> send {0} byte recordsz ---> report stats every {0} secsz---> raw metrics? {0})event raw_metricsz-> OK!c sg}j}|j}j}j}t|D] }|||qtdd\}}|D]}|r9|d7}q.| rB|d7}q.t |td||fdS)NzSend complete...)rrz%d suceeded, %d failed) sendappendtopic num_recordsrangeprintflushclose succeededfailed ValueError) resultsr rrr_ count_success count_failurerargsproducerrecordc/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/benchmarks/producer_performance.py _benchmark5s$   z+ProducerPerformance.run.._benchmarkzExecution time:secsr r!)producer_configsplitintrrrbytes bytearray record_sizestats_intervalritemsformatrr threadingEvent StatsReporterstarttime monotonicsetjoin Exceptionsysexc_info tracebackprint_exceptionexit) rpropspropkv timer_stoptimerr# start_timeend_timer8r!rr"runs^       zProducerPerformance.runN)__name__ __module__ __qualname__ staticmethodrDr!r!r!r"rsrcs6eZdZd fdd ZddZddZd d ZZS) r0NFcs&t||_||_||_||_dSN)super__init__intervalrr r )selfrLrr r  __class__r!r"rK]s  zStatsReporter.__init__cCsB|j}|s dS|jrt|dStdjdi|ddS)Na+{record-send-rate:.0f} records/sec ({byte-rate:.0f} B/sec), {request-rate:.0f} avg requests/sec, {request-latency-avg:.0f}ms avg latency, {throttle-time-max:.0f}ms max throttle, {record-size-avg:.0f} avg record size, {batch-size-avg:.0f} avg batch size, {records-per-request-avg:.0f} avg records/reqzproducer-metricsr!)rmetricsr pprintrr-)rMrPr!r!r" print_statsds  zStatsReporter.print_statscCs |dSrI)rRrMr!r!r" print_finalts zStatsReporter.print_finalcCs<|jr|j|js||jr|j|jr |dSrI)r waitrLrRrTrSr!r!r"rDws zStatsReporter.run)NF)rErFrGrKrRrTrD __classcell__r!r!rNr"r0\s r0cCstjdd}|jddtdddd|jd d td d d |jdtddd |jdtddd |jddtdddd|jdtddd |jdddd|S)Nz5This tool is used to verify the producer performance.) descriptionz-bz--bootstrap-servers+r!z&host:port for cluster bootstrap server)typenargsdefaulthelpz-tz--topicz:Topic name for test (default: kafka-python-benchmark-test)zkafka-python-benchmark-test)rYr\r[z --num-recordsz0number of messages to produce (default: 1000000)i@Bz --record-sizez$message size in bytes (default: 100)dz-cz--producer-configzWkafka producer related configuaration properties like bootstrap_servers,client_id etc..z--stats-intervalz?Interval in seconds for stats reporting to console (default: 5)z --raw-metrics store_truezs N"!