o `j@sddlZddlZddlZddlZddlZddlZddlmZGdddZGdddej Z ddZ e d krBe ZeedSdS) N) KafkaConsumerc@seZdZeddZdS)ConsumerPerformancec Cszi}|jD]1}|d\}}zt|}Wn tyYnw|dkr&d}n |dkr-d}n|dkr3d}|||<qtd|j|d<d |d <d |vrMd |d <|jd |d<|D] \}}td||qXt |j fi|}td|j td|jtd|j t }t|j|||j d}|tdtt}d} |D] } | d7} | |jkrnqt} ||td| td| |dWdStyt} tj| tdYdSw)N=NoneFalseFTrueTzInitializing Consumer...bootstrap_serversearliestauto_offset_resetconsumer_timeout_msi'imetrics_sample_window_msz ---> {0}={1}z---> topic={0}z ---> report stats every {0} secsz---> raw metrics? {0})event raw_metricsz-> OK!rzConsumed {0} recordszExecution time:secs)consumer_configsplitint ValueErrorprintrstats_intervalitemsformatrtopicr threadingEvent StatsReporterstarttime monotonic num_recordssetjoin Exceptionsysexc_info tracebackprint_exceptionexit) argspropspropkvconsumer timer_stoptimer start_timerecordsmsgend_timer%r5c/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/benchmarks/consumer_performance.pyrunsj        zConsumerPerformance.runN)__name__ __module__ __qualname__ staticmethodr7r5r5r5r6rsrcs6eZdZd fdd ZddZddZd d ZZS) rNFcs&t||_||_||_||_dSN)super__init__intervalr.r r)selfr?r.r r __class__r5r6r>Is  zStatsReporter.__init__cCs:|j}|jrt|dStdjdi|ddS)Na{records-consumed-rate:.0f} records/sec ({bytes-consumed-rate:.0f} B/sec), {fetch-latency-avg:.0f}ms avg latency, {fetch-rate:.0f} avg fetch requests/sec, {fetch-size-avg:.0f} avg fetch size, {records-lag-max:.0f} max record lag, {records-per-request-avg:.0f} avg records/reqzconsumer-fetch-manager-metricsr5)r.metricsrpprintrr)r@rCr5r5r6 print_statsPs  zStatsReporter.print_statscCs |dSr<)rEr@r5r5r6 print_final^s zStatsReporter.print_finalcCs<|jr|j|js||jr|j|jr |dSr<)r waitr?rErGrFr5r5r6r7as zStatsReporter.run)NF)r8r9r:r>rErGr7 __classcell__r5r5rAr6rHs rcCstjdd}|jddtdddd|jd d td d d |jdtddd |jddtdddd|jdtdd|jdtddd |jdddd|S)Nz5This tool is used to verify the consumer performance.) descriptionz-bz--bootstrap-servers+r5z'host:port for cluster bootstrap servers)typenargsdefaulthelpz-tz--topicz>Topic for consumer test (default: kafka-python-benchmark-test)zkafka-python-benchmark-test)rLrOrNz --num-recordsz0number of messages to consume (default: 1000000)i@Bz-cz--consumer-configzVkafka consumer related configuration properties like bootstrap_servers,client_id etc..z--fixture-compressionzBspecify a compression type for use with broker fixtures / producer)rLrOz--stats-intervalz?Interval in seconds for stats reporting to console (default: 5)z --raw-metrics store_truezs :