o `jf@sddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZddlmZddlmZmZmZddlmZddlmZdd lmZeeZGd d d ZGd d d ZdS)N)Future)KafkaConnectionMetrics)get_sasl_mechanism)ApiVersionsRequest)SaslAuthenticateRequestSaslHandshakeRequestSaslBytesRequest)BrokerVersionData) KafkaProtocol) __version__c@seZdZiddedddedddd d d d d ddddddddddddddddddZdTddZeddZeddZd d!Z ed"d#Z d$d%Z ed&d'Z dTd(d)Z dUd*d+ZdTd,d-Zd.d/Zd0d1Zd2d3Zd4d5Zd6d7Zd8d9Zd:d;Zdd?Zd@dAZdBdCZdUdDdEZdFdGZdHdIZdUdJdKZdUdLdMZedNdOZ dUdPdQZ!dRdSZ"dS)VKafkaConnection client_idz kafka-python-client_software_namez kafka-pythonclient_software_version%max_in_flight_requests_per_connectionreceive_message_max_bytesi@Brequest_timeout_msi0usecurity_protocol PLAINTEXTsasl_mechanismNsasl_plain_usernamesasl_plain_passwordsasl_kerberos_namesasl_kerberos_service_namekafkasasl_kerberos_domain_namesasl_oauth_token_providermetricsmetric_group_prefixcKst|j|_|jD] }||vr|||j|<q ||_||_d|_d|_t|_ t |_ d|_ d|_ t|_t|_t|_||_tj|_d|_t||_|jdrbt|jd|jd||_nd|_|j|j|j|jdS)NFTrrr)copyDEFAULT_CONFIGconfignode_idnet transportparser collectionsdeque_request_buffersetpaused connected initializingr _init_future _close_futurein_flight_requestsbroker_version_datar max_version_api_versions_idx_throttle_timeSaslReauthenticator_reauthr_sensors add_errbackfail_in_flight_requestsadd_both)selfr%r$r2configskeyr?R/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/net/connection.py__init__*s6     zKafkaConnection.__init__cCs|jdurdS|jjSN)r2broker_versionr<r?r?r@rCGs zKafkaConnection.broker_versioncCs|j o|j SrB)r-r.rDr?r?r@closedMszKafkaConnection.closedcCsr|jrd}n|js d}n|jrd}nd}|jrd|jnd}|jdur(|jnd}d|j|d |d |d S) Nr. disconnectedr,r-z host=[%s]r unknownz)r.r-r,r& host_portrCr$)r<staterHrCr?r?r@__str__QszKafkaConnection.__str__cC|jSrB)r/rDr?r?r@ init_future^zKafkaConnection.init_futureccs|jEdH|SrB)rL __await__rDr?r?r@rNbszKafkaConnection.__await__cCrKrB)r0rDr?r?r@ close_futurefrMzKafkaConnection.close_futurecCs^|durt}|dur||dSz||jWSty.|jdd|_||jYSw)Nr)time monotonic _timeout_secsAttributeErrorr#)r<now timeout_msr?r?r@ _timeout_atjs   zKafkaConnection._timeout_atcCs~t}|j|d}|js|jjr|j|||f|S|jr*|t d|jS|j s5|t dS|j |||d|S)N)rVz Node paused: zNode not connectedfuture timeout_at)rrWr.r7is_reauthenticatingr*appendr,failureErrorsNodeNotReadyErrorr-KafkaConnectionError _send_request)r<requestrrYrZr?r?r@ send_requestvs zKafkaConnection.send_requestc sHdurtjrtdS|jdur;z j||_Wntjy:}z |WYd}~Sd}~wwt durIj dkrVt Sj |}td|||rjfdd}j||fndjsjj tjjdkrdS)NrE)rUz%s Request %d: %scsSrB)_request_timed_outr?rYr< sent_timerZr?r@sz/KafkaConnection._send_request..r max_in_flight)rrEr]r^r` API_VERSIONr2 api_versionIncompatibleBrokerVersionrQrRrWKafkaTimeoutErrorr'rclogdebugexpect_responser%call_atr1r\successr,r&write send_byteslenr#pause)r<rbrYrZexccorrelation_id timeout_taskr?rer@rasD       zKafkaConnection._send_requestcCs4|jr|j\}}}|j|||d|jsdSdS)NrX)r*popleftra)r<rbrYrZr?r?r@ send_bufferedszKafkaConnection.send_bufferedcCsB|js|jrdS||d}td|||td|dS)NrPz6%s: Request timed out after %d ms. Closing connection.zRequest timed out after %d ms)rEis_donermwarningcloser^RequestTimedOutError)r<rYsent_atrZrVr?r?r@rds  z"KafkaConnection._request_timed_outc Cs&|jrtd|t|dS|j|}t|D]_\}\}}z |j\}}}} } Wnt y=| t dYSw||krL| t dS|j | t|d} |jrd|jj| td||| |||||qd|jvrt|j|jdkr|d|jdS) z# Called when some data is received.z3%s: Ignoring %d bytes received by closed connectionNz-Received response with no in-flight-requests!z$Received unrecognized correlation idrPz%s: Response %d (%s ms): %srhr)rErmrnrtr' receive_bytes enumerater1ry IndexErrorr}r^r`r%cancelrQrRr8 request_timerecord_maybe_throttlerqr,r#unpauser7on_response_processed) r<data responsesiresp_correlation_idresponsereq_correlation_idrYrfrWrx latency_msr?r?r@ data_receiveds,      zKafkaConnection.data_receivedcCsdS)z Called when the other end calls write_eof() or equivalent. If this returns a false value (including None), the transport will close itself. If it returns a true value, closing the transport is up to the protocol. Fr?rDr?r?r@ eof_receivedszKafkaConnection.eof_receivedcCsd|_|_d|_|j|pt}|jjs|j ||j jsC|dur4t d||j ddSt d|||j |dSdS)z Called when the connection is lost or closed. The argument is an exception object or None (the latter meaning a regular EOF is received or the connection was aborted or closed). FNz%s: Connection closedz%s: Connection lost: %s)r-r.r&r7rr^r`r/r{r]r0rminforqerror)r<rvrr?r?r@connection_losts     zKafkaConnection.connection_lostcCs~|jstd|p t}|jr |j\}}}|||js|jr=|j\}}}}}|j ||||js#dSdS)Nz4Connection must be closed to fail in flight requests) rE RuntimeErrorr^ Cancelledr*ryr]r1r%r)r<r_rYrxr?r?r@r:s    z'KafkaConnection.fail_in_flight_requestscCs|jrtd||_|j|kr|j|d|_|jzd|jg|j ddR}Wnt yBt dd}Ynwt |jd|jd |d |_t d |d S) z Called when a connection is made. The argument is the transport representing the pipe connection. To receive data, wait for data_received() calls. When the connection is closed, connection_lost() is called. z Connection closed during connectTznode=%s[%s:%s]rz%Failed to build connection log_prefixr r r)r ridentz%s: Connection madeN)rEr^r`r& get_protocol set_protocolr.resume_readingr$getPeer Exceptionrm exceptionr r#r'rn)r<r& log_prefixr?r?r@connection_mades&   $  zKafkaConnection.connection_madecCs|j|dSrB)r,add)r<vr?r?r@ru"szKafkaConnection.pausecCsfz|j|Wn tyYdSw|js+|jr-|jr/|j}|r1|j|dSdSdSdSdSrB)r,removeKeyErrorr'r&rsrr)r<rto_sendr?r?r@r%s   zKafkaConnection.unpausecC|ddS)a Called when the transport's buffer goes over the high-water mark. Pause and resume calls are paired -- pause_writing() is called once when the buffer goes strictly over the high-water mark (even if subsequent writes increases the buffer size even more), and eventually resume_writing() is called once when the buffer size reaches the low-water mark. Note that if the buffer size equals the high-water mark, pause_writing() is not called -- it must go strictly over. Conversely, resume_writing() is called when the buffer size is equal or lower than the low-water mark. These end conditions are important to ensure that things go as expected when either mark is zero. NOTE: This is the only Protocol callback that is not called through EventLoop.call_soon() -- if it were, it would have no effect when it's most needed (when the app keeps writing without yielding until pause_writing() is called). bufferN)rurDr?r?r@ pause_writing0szKafkaConnection.pause_writingcCr)zD Called when the transport's buffer drains below the low-water mark.rN)rrDr?r?r@resume_writingGzKafkaConnection.resume_writingcCsN|dur |jjs t}|js||dS|r |j|dS|jdSrB)r/r{r^r`r&rabortr})r<rr?r?r@r}Ks zKafkaConnection.closecCst|dd}|jr|jj||sdS|jdur;|jdkr;t|d}||jkr;||_|j ||j | dt d||jj|dS)Nthrottle_time_msr)rrrPthrottlez"%s: %s throttled by broker (%d ms))getattrr8 throttle_timerrCrQrRr5r%rp_maybe_unthrottlerurmr| __class____name__)r<rrrr?r?r@rVs   zKafkaConnection._maybe_throttlecCs&t|jkrd|_|ddSdS)Nrr)rQrRr5rrDr?r?r@rgsz!KafkaConnection._maybe_unthrottlec s|dur |}z||IdH|jr||IdHWnty6}z ||WYd}~dSd}~ww|td|dS)Nz %s: Connected) rW_get_api_versions sasl_enabled_sasl_authenticaterr}_init_completermr)r<rZrr?r?r@ initializelszKafkaConnection.initializec s|dur |}|jdur+z |jt|_Wntjy*td||j YdSw|t kry|j}t||j d|j dd}|j ||dIdH}t|j}|tjurWn'|tjurv|jD]}|j|jkrqt|j|j|_nq_d|_q+|tddd |jD}t|d }|jdurtd ||j||_dS|j|krtd ||j|jj||_dS|jdur|j|krtd ||j|jjdSdSdS)Nz7%s: Using pre-configured api_version %s for ApiVersionsrr)versionrrrZrz Timeout during ApiVersions checkcSsi|] }|j|j|jfqSr?)api_key min_versionr3).0rjr?r?r@ sz5KafkaConnection._get_api_versions..) api_versionsz#%s: Broker version identified as %szA%s: Broker version identified as %s (lower than user-supplied %s)zA%s: Broker version identified as %s; clamping to user-supplied %s)rWr2rjrr4r^rkrmrnrCrQrRr#rafor_code error_codeNoErrorUnsupportedVersionErrorapi_keysrAPI_KEYminr3rlr rbroker_version_str) r<rZrrbr error_typerjrbvdr?r?r@rysX             z!KafkaConnection._get_api_versionscCs|jddvS)Nr)SASL_PLAINTEXTSASL_SSL)r#rDr?r?r@rrzKafkaConnection.sasl_enabledc s|dur |}t|jddd}|j||dIdH}t|j}|tjur2t d||j ||jd|j vrGt d|jd|j f|j }|jjrR|jjn|jd}t|jdd d|i|j}d}|s|tkr|} |dkrt| } nt| } |j| |dIdH}t|j}|tjurtd |j |jf|dkr|rn||j|s|tkstt|krtd |std |jd|dkr|j|jt d ||!dS) Nr) mechanismr3rz%s: SaslHandshake failed: %szGKafka broker does not support %s sasl mechanism. Enabled mechanisms: %srhostz%s: %szSASL Authentication timed outz"Failed to authenticate via SASL %sr?)"rWrr#rar^rrrrmrr mechanismsUnsupportedSaslMechanismErrorrir&rrrr{rQrR auth_bytesrrSaslAuthenticationFailedError error_messagereceiverlis_authenticatedr7session_updatedsession_lifetime_msr auth_details) r<rZrbrrr sasl_hostr auth_responsetoken auth_requestr?r?r@rsd         z"KafkaConnection._sasl_authenticatecCs8|jrd|_d|_||jd|jdSdS)NFT)r.r-rzr/rqr7schedulerDr?r?r@rs zKafkaConnection._init_complete)NNrB)#r __module__ __qualname__r r"rApropertyrCrErJrLrNrOrWrcrarzrdrrrr:rrurrrr}rrrrrrrr?r?r?r@r s            *        /   8r c@s`eZdZdZddZeddZeddZdd Zd d Z d d Z ddZ ddZ ddZ dS)r6aKIP-368 SASL re-authentication state and scheduling for a single KafkaConnection. Owns the per-connection re-auth lifecycle so the connection doesn't have to carry the related attributes and coroutines inline. The connection plugs this in at five points: - after each successful SASL auth -> session_updated() - after init completes -> schedule() - when send_request needs to gate the public API -> is_reauthenticating - on every response popped from in_flight_requests -> on_response_processed() - on connection_lost -> cancel() cCs(||_d|_d|_d|_d|_d|_dS)NrF)_connrauthenticated_at_task_reauthenticating _drain_future)r<connr?r?r@rAs  zSaslReauthenticator.__init__cCrKrB)rrDr?r?r@r[rMz'SaslReauthenticator.is_reauthenticatingcCrK)zEThe scheduled re-auth task, or None. Exposed for tests/observability.)rrDr?r?r@taskszSaslReauthenticator.taskcCsJ|pd|_|jdkrd|_nd|jkrdkrnnd|_t|_dS)zCapture broker-advertised session lifetime after each successful auth round (initial and subsequent re-auths). Clamp negative values to 0, and require minimum non-zero lifetime of 1sec (1000).rrPN)rrQrRr)r<rr?r?r@r s  z#SaslReauthenticator.session_updatedcCsX|jjr|js dStdd}|j|d}td|j||j|jj||j |_ dS)aSchedule the next re-auth before the lifetime elapses. Jittered to 85-95% of the lifetime to avoid synchronised re-auth storms across many connections (Apache Java semantics). No-op when SASL is disabled or the broker advertised lifetime=0. Ng333333?gffffff?rPzG%s: Scheduling SASL re-authentication in %.3fs (session_lifetime_ms=%d)) rrrrandomuniformrmrnr% call_later_runr)r<pctdelayr?r?r@rs  zSaslReauthenticator.schedulecCsR|jdur|jj|jd|_|jdur!|jjs!|jtd|_d|_ dS)zvCancel any pending re-auth and fail the drain awaiter if present. Called from KafkaConnection.connection_lost.NF) rrr%rrr{r]r^r`rrDr?r?r@r$s  zSaslReauthenticator.cancelcCs@|jr|jdur|jjs|jjs|jddSdSdSdSdS)zWake the drain awaiter once in_flight_requests clears during reauth. Called from KafkaConnection.data_received after each pop.N)rrrr1r{rqrDr?r?r@r/s  z)SaslReauthenticator.on_response_processedc sd|_|jjr dSz |IdHWdStyC}z#td|j|t|tr+|nt t |}|j |WYd}~dSd}~ww)Nz%%s: SASL re-authentication failed: %s) rrrE _do_reauth BaseExceptionrmr| isinstancerr^rstrr})r<rverrr?r?r@r8szSaslReauthenticator._runcsd|_zF|jjr$|jjs$t|_|jjsn|jIdH|jjr$|jjr d|_|jjr4Wd|_d|_dStd|j|jIdHWd|_d|_nd|_d|_w|jjrXdS|j | dS)NTFz$%s: Beginning SASL re-authentication) rrr1rErrrmrnrrzrrDr?r?r@rFs0    zSaslReauthenticator._do_reauthN)rrr__doc__rArr[rrrrrrrr?r?r?r@r6s      r6) r(r!loggingrstructrQ kafka.errorserrorsr^ kafka.futurerkafka.net.metricsrkafka.net.saslrkafka.protocol.metadatarkafka.protocol.saslrrr"kafka.protocol.broker_version_datar kafka.protocol.parserr kafka.versionr getLoggerrrmr r6r?r?r?r@s*         Z