&`i8W ,dZddlZddlZddlmZmZddlmZmZm Z m Z m Z m Z m Z mZmZddlZerddlmZmZddlmZmZddlmZddlmZmZdd lmZdd lm Z m!Z!ej"e#Z$eGd d Z%d e e&efde e%ddfdZ'de e&de e%de e&effdZ(de e&de e%de e&effdZ)dddddee*e dfdee*e dfdee*e*ff dZ+Gdde Z,dS)aKafka datasource for bounded data reads. This module provides a Kafka datasource implementation for Ray Data that supports bounded reads with offset-based range queries. Message keys and values are returned as raw bytes to support any serialization format (JSON, Avro, Protobuf, etc.). Users can decode them using map operations. Requires: - kafka-python: https://kafka-python.readthedocs.io/ N) dataclassfields) TYPE_CHECKINGAnyDictIterableListLiteralOptionalTupleUnion KafkaConsumerTopicPartition)BlockOutputBufferOutputBlockSizeOption) _check_import)Block BlockMetadata) DataContext) DatasourceReadTaskceZdZUdZdZeeed<dZeeed<dZ eeed<dZ eeed<dZ eeed<dZ eeed<dZ eeed <dZeeed <dZeeed <dZeeed <dZeeed <dZeeed<dZeeed<dZeeed<dZeeed<dZeeed<dS)KafkaAuthConfiga` Authentication configuration for Kafka connections. Uses standard kafka-python parameter names. See kafka-python documentation for full details: https://kafka-python.readthedocs.io/ security_protocol: Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL. Default: PLAINTEXT. sasl_mechanism: Authentication mechanism when security_protocol is configured for SASL_PLAINTEXT or SASL_SSL. Valid values are: PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512. sasl_plain_username: username for sasl PLAIN and SCRAM authentication. Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms. sasl_plain_password: password for sasl PLAIN and SCRAM authentication. Required if sasl_mechanism is PLAIN or one of the SCRAM mechanisms. sasl_kerberos_name: Constructed gssapi.Name for use with sasl mechanism handshake. If provided, sasl_kerberos_service_name and sasl_kerberos_domain name are ignored. Default: None. sasl_kerberos_service_name: Service name to include in GSSAPI sasl mechanism handshake. Default: 'kafka' sasl_kerberos_domain_name: kerberos domain name to use in GSSAPI sasl mechanism handshake. Default: one of bootstrap servers sasl_oauth_token_provider: OAuthBearer token provider instance. Default: None ssl_context: Pre-configured SSLContext for wrapping socket connections. If provided, all other ssl_* configurations will be ignored. Default: None. ssl_check_hostname: Flag to configure whether ssl handshake should verify that the certificate matches the brokers hostname. Default: True. ssl_cafile: Optional filename of ca file to use in certificate verification. Default: None. ssl_certfile: Optional filename of file in pem format containing the client certificate, as well as any ca certificates needed to establish the certificate's authenticity. Default: None. ssl_keyfile: Optional filename containing the client private key. Default: None. ssl_password: Optional password to be used when loading the certificate chain. Default: None. ssl_crlfile: Optional filename containing the CRL to check for certificate expiration. By default, no CRL check is done. When providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. Default: None. ssl_ciphers: optionally set the available ciphers for ssl connections. It should be a string in the OpenSSL cipher list format. If no cipher can be selected (because compile-time options or other configuration forbids use of all the specified ciphers), an ssl.SSLError will be raised. See ssl.SSLContext.set_ciphers Nsecurity_protocolsasl_mechanismsasl_plain_usernamesasl_plain_passwordsasl_kerberos_namesasl_kerberos_service_namesasl_kerberos_domain_namesasl_oauth_token_provider ssl_contextssl_check_hostname ssl_cafile ssl_certfile ssl_keyfile ssl_password ssl_ciphers ssl_crlfile)__name__ __module__ __qualname____doc__rr str__annotations__rrrrr r!r"rr#r$boolr%r&r'r(r)r*/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/datasource/kafka_datasource.pyrr*sk11h(,x}+++%)NHSM((()-#---)-#---(, ,,,04 444/3x}333/3x}333"&K#%%%)---- $J $$$"&L(3-&&&!%K#%%%"&L(3-&&&!%K#%%%!%K#%%%%%r3rconfigkafka_auth_configreturnct|r3t|D]%}t||j}| |||j<$dSdS)zAdd authentication configuration to consumer config in-place. Args: config: Consumer config dict to modify. kafka_auth_config: Authentication configuration. N)rgetattrname)r5r6fieldvalues r4_add_authentication_to_configr=vs]+-.. + +E-uz::E %*uz" ++ + +r3bootstrap_serversc2|ddd}t|||S)zBuild minimal consumer config for partition discovery. Args: bootstrap_servers: List of Kafka broker addresses. kafka_auth_config: Authentication configuration. Returns: Consumer configuration dict for discovery. F)r>enable_auto_commitconsumer_timeout_msr=r>r6r5s r4$_build_consumer_config_for_discoveryrEs1/##F "&*;<<< Mr3c8|dddd}t|||S)zBuild full consumer config for reading messages. Args: bootstrap_servers: List of Kafka broker addresses. kafka_auth_config: Authentication configuration. Returns: Consumer configuration dict for reading. Fc|SNr2)vs r4z1_build_consumer_config_for_read..sr3c|SrHr2)ks r4rJz1_build_consumer_config_for_read..sar3)r>rAvalue_deserializerkey_deserializerrCrDs r4_build_consumer_config_for_readrOs8/#)k'K F "&*;<<< Mr3consumerrtopic_partitionr start_offsetearliest end_offsetlatestc D||g|}||g|}|}|}|dks||}|dks||}||krG||kr|n|d|d}||kr|n|d|d} td|d| d|jd |j||fS) ajResolve start and end offsets to actual integer offsets. Args: consumer: Kafka consumer instance. topic_partition: TopicPartition to resolve offsets for. start_offset: Start offset (int or "earliest"). end_offset: End offset (int or "latest"). Returns: Tuple of (resolved_start_offset, resolved_end_offset). rSNrUz (resolved to )zstart_offset (z) > end_offset (z) for partition z in topic )beginning_offsets end_offsets ValueError partitiontopic) rPrQrRrTearliest_offset latest_offsetoriginal_start original_end start_strend_strs r4_resolve_offsetsrcsD"00/1BCCOTO((/):;;OLM"NLz!!\%9& X!3" j  --  "AA,AAA z))   == ===   ZY Z Z Z Z,6 Z ZBQBW Z Z     ##r3c eZdZdZdZ ddeeeefdeeeefd eee dfd eee dfd e e d ef d Z de efdZ ddede edeefdZdS)KafkaDatasourcezBKafka datasource for reading from Kafka topics with bounded reads.r@rSrUN'topicsr>rRrTr6 timeout_msct|dd|std|std|dkrtdt|tr*t|tr||krtdt|tr|d krtd t|tr|d krtd t|tr|rd |vrtd|dnWt|t rB|std|D].}t|trd |vrtd|d/t|t r|n|g|_t|t r|n|g|_||_||_ ||_ ||_ tj j|_dS)aInitialize Kafka datasource. Args: topics: Kafka topic name(s) to read from. bootstrap_servers: Kafka broker addresses (string or list of strings). start_offset: Starting position. Can be: - int: Offset number - str: "earliest" end_offset: Ending position. Can be: - int: Offset number - str: "latest" kafka_auth_config: Authentication configuration. See KafkaAuthConfig for details. timeout_ms: Timeout in milliseconds for every read task to poll until reaching end_offset (default 10000ms). If the read task does not reach end_offset within the timeout, it will stop polling and return the messages it has read so far. Raises: ValueError: If required configuration is missing. ImportError: If kafka-python is not installed. kafkaz kafka-python)modulepackageztopics cannot be emptyz!bootstrap_servers cannot be emptyrztimeout_ms must be positivez)start_offset must be less than end_offsetrUzstart_offset cannot be 'latest'rSzend_offset cannot be 'earliest':z"Invalid bootstrap_servers format: z6. Expected 'host:port' or list of 'host:port' strings.z&bootstrap_servers cannot be empty listz. Expected 'host:port' string.N)rrZ isinstanceintr/list_topics_bootstrap_servers _start_offset _end_offset_kafka_auth_config _timeout_msr get_currenttarget_max_block_size_target_max_block_size)selfrgr>rRrTr6rhservers r4__init__zKafkaDatasource.__init__s[: d7NCCCC 7566 6  B@AA A ??:;; ; lC ( ( NZ C-H-H Nj(( !LMMM lC ( ( @\X-E-E>?? ? j# & & @:+C+C>?? ? ' - - $ 3D(D(D K9JKKK)E )4 0 0 $ K !IJJJ+  !&#..#V2C2C$7V7773D ",FD!9!9Gvvx +T22 %  #$  *%"3%&1&=&?&?&U###r3r7cdS)zBReturn an estimate of the in-memory data size, or None if unknown.Nr2)rzs r4estimate_inmemory_data_sizez+KafkaDatasource.estimate_inmemory_data_size6str3 parallelismper_task_row_limitc^ddlm}t|j|j}g}d} |di|}|jD]H}||}|std|d|D]} ||| fI |r| n#|r| wwxYw|j} |j } |j } |j } |j}|j }g}tjdtjfdtjfdtjfd tjfd tjfd tjfd tjfd tjtjtjfg}|D]\}}||| | | || |fdt*dt,dt.t*dt0t2t,t4dfdt0t2t,t4dfdt0t6dt,dt,fd}t9ddd|d|gd}|||}t;||||}|||S)awCreate read tasks for Kafka partitions. Creates one read task per partition. Each task reads from a single partition of a single topic. Args: parallelism: This argument is deprecated. per_task_row_limit: Maximum number of rows per read task. Returns: List of ReadTask objects, one per partition. r)rNzTopic z# has no partitions or doesn't existoffsetkeyr<r\r[ timestamptimestamp_typeheaders topic_name partition_idr>rRrSrTrUr6rhrxc Tdttffd }|S)aCreate a Kafka read function with captured variables. This factory function captures configuration variables as default arguments to avoid serialization issues when the read function is executed remotely by Ray. Using default arguments ensures all needed config is available in the remote task without requiring 'self' to be serialized. r7c3pKddlm}m}t}|di|} |}||gt ||\}}|||g}ttj }d} tj } dz } | stj | z } | | kr9t dddd |d t|d ny||} | |krn\t| | z d z}|t#|d }|s||g}|D]}||j|krd} n|jrt+|jni}||j|j|j|j|j|j|j|dt|t:jkrtt>j !|}|"||#r*|$V|#*g}| |r4t>j !|}|"||%|#r*|$V|#*|&dS#|&wxYw)aTRead function for a single Kafka partition using kafka-python. This function runs remotely in a Ray task. It creates a KafkaConsumer, reads messages from a single assigned partition, and yields PyArrow tables incrementally for efficient streaming processing. rr)rxFg@@z Kafka read task timed out after zms while reading partition z of topic z ; end_offset z was not reached. Returning z- messages collected in this read task so far.r@rf)rhNT)rrr<r\r[rrrr2)'rjrrrOassignrcseekrroftimeloggerwarninglenpositionropollmingetrrdictappendrr<r\r[rrreBATCH_SIZE_FOR_YIELDpaTable from_pylist add_blockhas_nextnextfinalizeclose)rrconsumer_configrPrQ start_offend_offrecords output_bufferpartition_done start_timetimeout_seconds elapsed_timecurrent_positionremaining_timeout_ms msg_batchmessagesmsg headers_dicttabler>rTr6rrRrxrhrs r4 kafka_read_fnzSKafkaDatasource.get_read_tasks..create_kafka_read_fn..kafka_read_fnsDCCCCCCC'F)+<''O -}????Ha)*8.\*R*R (9:::-=$o|Z..* 7! oyAAA"$(9146K)) */%)Y[[ *4v*="0<1+/9;;+CL+>> &%Sz%S%Snz%S%SGQ%S%S29%S%SWZ[bWcWc%S%S%S!"!"!"!&08/@/@/Q/Q,/7:: %47!0rRrTr6rhrxrs ```````` r4create_kafka_read_fnz.create_kafka_read_fn{sm$q)xq)q)q)q)q)q)q)q)q)q)q)q)q)f%$r3zkafka:///)num_rows size_bytes input_files exec_stats)read_fnmetadataschemarr2)rjrrErrrurqpartitions_for_topicrZrrrsrtrvryrrint64binarystringint32map_r/ror r r r rrr)rzrrrrtopic_partitionsdiscovery_consumerr\ partitionsr[r>rRrTrhr6rxtasksrrrrrrtasks r4get_read_taskszKafkaDatasource.get_read_tasks:sV$ ('''''?  #T%<   ! +!.!A!A!A!A  @ @/DDUKK !$KKKK",@@I$++UI,>????@  @" +"((***" +"((**** +!3) % %  3 $ ;28::& $")++&")++&bhjj)bhjj)!28::.BGBIKK==>    )9Y Y  $J #-$0/@JVFP?P",-BE %E %E %!E %$(9E %'uS'*2E-E'FG E % %U30A+A%BC E % $,O#< E % E %(+E %E %E %E %P%C CC\CCD H10\JJM%!#5 D LL     s ABB/)rSrUNrfrH)r+r,r-r.rr r/r ror r rr|r~rrr2r3r4reresALL  9C4<7;LVLVc49n%LV!d3i0LVC!445 LV #wx001 LV $O4 LVLVLVLVLV\Xc] EIZZZ4rs   ))))))))                      433333333UUUUUUUU111111////////((((((44444444  8 $ $ H&H&H&H&H&H&H& H&V+ cN+/7/H+ ++++"Cy5=o5N #s(^*Cy0 #s(^.,$,$%,$WZ001,$c78,,- ,$  38_ ,$,$,$,$^rrrrrjrrrrrr3