o
    `j;O                     @   s   d dl Z d dlZd dlZd dlZd dlZd dlZddlmZ ddlm	Z	 ddl
mZ ddlmZmZ d dlmZ d dlmZ d dlmZ d d	lmZ d d
lmZ d dlmZ eeZG dd dZdS )    N   )create_connection)KafkaConnection)KafkaManagerMetrics)KafkaSSLTransportKafkaTCPTransport)ClusterMetadata)WakeupNotifier)BrokerVersionData)Future)__version__c                   @   s  e Zd Zi dddde dddedd	d
ddddddddddejejdfejejdfgddddddddddddi ddd dd!dd"dd#dd$dd%dd&dd'd(d)dd*dd+dd,dd-dd.d/d0d1d2d3Z	d4Z
d5d6 Zed7d8 Zd9d: Zd;d< Zdkd=d>Zdkd?d@ZedAdB ZdCdD ZdEdF ZedGdH ZdldIdJZdmdKdLZ				dndMdNZdodOdPZdQdR ZdSdT ZdUdV ZdWdX ZdYdZ Zd[d\ Zd]d^ Z d_d` Z!dadb Z"dodcddZ#dedf Z$dgdh Z%didj Z&dS )pKafkaConnectionManagerbootstrap_serverszlocalhost:9092	client_idzkafka-python-client_software_namezkafka-pythonclient_software_versionreceive_message_max_bytesi@B reconnect_backoff_ms2   reconnect_backoff_max_msi0u  request_timeout_ms"socket_connection_setup_timeout_msi'  &socket_connection_setup_timeout_max_mssocket_optionsr   %max_in_flight_requests_per_connection   connections_max_idle_msi`= security_protocol	PLAINTEXTssl_contextNssl_check_hostnameT
ssl_cafilessl_certfilessl_keyfilessl_passwordssl_crlfilesasl_mechanismsasl_plain_usernamesasl_plain_passwordsasl_kerberos_namesasl_kerberos_service_namekafkasasl_kerberos_domain_namesasl_oauth_token_provider	proxy_urlapi_versionmetricsmetric_group_prefix metadata_max_age_msi client_dns_lookupuse_all_dns_ips)r5   (resolve_canonical_bootstrap_servers_onlyc                 K   sZ  t  | j| _| jD ]}||v r|| | j|< q
| jd | jvr,td| j| jd f |dd urF| jd d u rFtd |d | jd< || _t	| jd | jd | jd d| _
| j
|  i | _t | _i | _| jd	 d
 | _|   d | _d | _t| j| _| jd rt| jd | jd | j| _nd | _| jd d urt| jd | _d| _d S )Nr4   z+client_dns_lookup must be one of %s; got %rsocks5_proxyr.   z1socks5_proxy is deprecated, use proxy_url insteadr   r3   )r   r3   r4   r     r0   r1   r/   F)copyDEFAULT_CONFIGconfig_VALID_DNS_LOOKUP_MODES
ValueErrorgetlogwarning_netr   clusterattach_connsdict_backoff_auth_failures_idle_check_delayclose_idle_connectionsbroker_version_data_bootstrap_futurer	   _bootstrap_wakeupr   _sensorsr
   closed)selfnetconfigskey rS   O/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/net/manager.py__init__@   sL   



zKafkaConnectionManager.__init__c                 C   s   | j d u rd S | j jS N)rJ   broker_versionrO   rS   rS   rT   rW   k   s   
z%KafkaConnectionManager.broker_versionc                 C   s    t tdd | j dd dS )Nc                 S   s   | j S rV   )	connectedconnrS   rS   rT   <lambda>r   s    z?KafkaConnectionManager.least_used_connections.<locals>.<lambda>c                 S   s   | j jS rV   )	transportlast_activityrZ   rS   rS   rT   r\   r   s    )rR   )sortedfilterrD   valuesrX   rS   rS   rT   least_used_connectionsq   s    z-KafkaConnectionManager.least_used_connectionsc                    s  | j s_|d u st |k r_t| j }td|j	|j
|j|j z|d ur2|t  d nd }| j|j	|dddd}W n0 tjyp   | |j	}|d ur^t|td|t  }td|j	| | |I d H  Y qw z|I d H  W n; tjy   td   ty } z"| j|j	|| | |j	}td|j	||d  W Y d }~qd }~ww zz3| j|j	I d H  | j s| |j	}td	|d  W W td
|j	 | j|j	|  qW n> ty& } z1| |j	}td|j	 d| d|d dd W Y d }~W td
|j	 | j|j	|  qd }~ww | |j	 | j  td| j W td
|j	 | j|j	|  dS td
|j	 | j|j	|  w t d| jj!d f )Nz-Attempting bootstrap to %s at %s:%s (rack %s)r8   F)
timeout_mspop_on_closerefresh_metadata_on_errreset_backoff_on_connectr   z)Bootstrap %s not ready; waiting %.2f secszGDid you attempt to connect to a kafka controller (no metadata support)?z9Bootstrap connection to %s failed: %s (backoff %.2f secs)zBBootstrap metadata response has no brokers. Retrying in %.2f secs.zClosing bootstrap connection %szBootstrap attempt to z	 failed: z
 (backoff z.2fz secs)zBootstrap complete: %sTzUnable to bootstrap from %sr   )"rN   time	monotonicrandomchoicerB   bootstrap_brokersr?   infonode_idhostportrackget_connectionErrorsNodeNotReadyErrorconnection_delayminmaxr@   rL   IncompatibleBrokerVersionerror	ExceptionrD   popcloseupdate_backoffrefresh_metadatabrokersreset_backoffstart_refresh_loopKafkaTimeoutErrorr;   )rO   deadlinebootstrap_brokerrc   r[   delayexc
backoff_msrS   rS   rT   _do_bootstrapt   s   


&
z$KafkaConnectionManager._do_bootstrapc                 C   sh   | j d ur|r| j js| j S |d u rd nt |d  }td | | j|| _ | j dd  | j S )Nr8   zStarting new bootstrapc                 S   s   t d| S )NzBootstrap failed: %s)r?   rx   r   rS   rS   rT   r\          z8KafkaConnectionManager.bootstrap_async.<locals>.<lambda>)	rK   is_donerg   rh   r?   debug	call_soonr   add_errback)rO   rc   refreshr   rS   rS   rT   bootstrap_async   s   
z&KafkaConnectionManager.bootstrap_asyncc                 C   s   | j | j|| d S rV   )rA   runr   )rO   rc   r   rS   rS   rT   	bootstrap   s   z KafkaConnectionManager.bootstrapc                 C   s   | j d uo	| j  S rV   )rK   	succeededrX   rS   rS   rT   bootstrapped   s   z#KafkaConnectionManager.bootstrappedc                 C   s   |j j| j S rV   )r]   r^   rH   )rO   r[   rS   rS   rT   _connection_idle_at   s   z*KafkaConnectionManager._connection_idle_atc                 C   st   |   D ]}| |}t |krtd|j |  q nt | j }t	d|t   | j
|| j d S )Nz"Closing idle connection to node %sz&Next idle connections check in %d secs)rb   r   rg   rh   r?   rl   rm   r{   rH   r   rA   call_atrI   )rO   r[   next_idle_atrS   rS   rT   rI      s   

z-KafkaConnectionManager.close_idle_connectionsc                 C   s   | j d dv S )Nr   )SSLSASL_SSL)r;   rX   rS   rS   rT   ssl_enabled   s   z"KafkaConnectionManager.ssl_enabledc              
      s   t | j|j|j| jd | jd |dI d H }| jr2dd | j D }t| j|fd|ji|}n	t| j||jd}z
|	 I d H  W |S  t
yX } ztd| d }~ww )	Nr   r.   )r.   
timeout_atc                 S   s    i | ]\}}| d r||qS )ssl_)
startswith).0rR   valuerS   rS   rT   
<dictcomp>   s    z;KafkaConnectionManager._build_transport.<locals>.<dictcomp>rn   )rn   zHandshake failed: %s)r   rA   rn   ro   r;   r   itemsr   r   	handshakery   rr   KafkaConnectionError)rO   noder   sockssl_configsr]   erS   rS   rT   _build_transport   s&   z'KafkaConnectionManager._build_transportc              
      sx  d }zz6| j ||dI d H }|js| jr)td| W W |d ur'|  d S d S || d }|j|dI d H  W n> tyx } z2td| |	| | 
|j t|tjtjfrb|| j|j< W Y d }~W |d urr|  d S d S d }~ww W |d ur|  n
|d ur|  w w | jr| jj  |r| |j | j|jd  |jd ur| j|jr|j| _d S d S d S )N)r   z3%s: closed during connect; discarding new transportzConnection failed: %s)r   rN   r?   r   r{   connection_made
initializery   rx   connection_lostr|   rm   
isinstancerr   SaslAuthenticationFailedErrorAuthorizationErrorrG   rM   connection_createdrecordr   rz   rJ   rB   is_bootstrap)rO   r   r[   rf   r   r]   r   rS   rS   rT   _connect   sX   




zKafkaConnectionManager._connectc                    s2  |d u r	t d| |dkrt ||jv r$j| S j|d u r3t |t	d|j
jj tjf|jdj |rZ jfdd jrg jfdd |rs jfdd  j|< |d u r|}t |d	  j fd
d  S )NzNo node_id providedr   z9Initializing connection for node_id %s at %s:%s (rack=%s))rm   rJ   c                    s   j  jd S rV   )rD   rz   rm   _)r   rO   rS   rT   r\     s    z7KafkaConnectionManager.get_connection.<locals>.<lambda>c                    s    j j S rV   )rM   connection_closedr   r   rX   rS   rT   r\     r   c                    s
    j  S rV   )rB   request_updater   rX   rS   rT   r\   !  s   
 r8   c                      s   j  dS )N)rf   r   )r   rS   r[   r   rf   rO   r   rS   rT   r\   &      )rr   rs   maybe_raise_auth_failurert   rD   rB   broker_metadataUnknownBrokerIdErrorr?   rl   rn   ro   rp   r   rA   rJ   r;   close_futureadd_bothrM   r   r   rg   rh   r   )rO   rm   rc   rd   re   rf   rS   r   rT   rq     s0   







z%KafkaConnectionManager.get_connectionc              
   C   sd   |d ur|n|   }z| |}W n tjy* } zt |W  Y d }~S d }~ww |j||dS )N)r   )least_loaded_noderq   rr   rs   r   failuresend_request)rO   requestrm   r   r[   r   rS   rS   rT   send)  s   zKafkaConnectionManager.sendc           	      C   s   dd | j  D }t| td}d}|D ]>}| j|}|duo)|jo)|j }|r/|jp5| 	|dk}|dur?t
|jnd}|rK|dkrK|  S |sU||k rU|}|}q|S )a  Choose the node with fewest outstanding requests, with fallbacks.

        This method will prefer a node with an existing connection (not throttled)
        with no in-flight-requests. If no such node is found, a node will be chosen
        randomly from all nodes that are not throttled or "blacked out" (i.e.,
        are not subject to a reconnect backoff). If no node metadata has been
        obtained, will return a bootstrap node.

        Returns:
            node_id or None if no suitable node was found
        c                 S   s   g | ]}|j qS rS   )rm   )r   brokerrS   rS   rT   
<listcomp>>  r   z<KafkaConnectionManager.least_loaded_node.<locals>.<listcomp>infNr   )rB   r~   ri   shufflefloatrD   r>   rY   pausedrt   lenin_flight_requests)	rO   nodesinflightfoundrm   r[   rY   blacked_outcurr_inflightrS   rS   rT   r   2  s    
z(KafkaConnectionManager.least_loaded_nodec                 C   s$   z| j |= W d S  ty   Y d S w rV   )rF   KeyErrorrO   rm   rS   rS   rT   r   R  s
   z$KafkaConnectionManager.reset_backoffc                 C   s   t ddS )Ng?g333333?)ri   uniformrX   rS   rS   rT   
jitter_pctX  s   z!KafkaConnectionManager.jitter_pctc                 C   sL   ddd}| j | d|d   }||v r | j ||  }t||}||   S )Nr   r   )r   r      r   )r;   ru   r   )rO   rR   failuresmax_keysrc   max_msrS   rS   rT   _calculate_exp_timeout[  s   
z-KafkaConnectionManager._calculate_exp_timeoutc                 C   sj   | j |d\}}}|d7 }| d|}| d|}td|||| t |d  }|||f| j |< |S )N)r   r   r   r   r   r   zD%s reconnect backoff %d ms / connect timeout %d ms after %s failuresr8   )rF   r>   r   r?   r   rg   rh   )rO   rm   r   r   r   
connect_msbackoff_until_timerS   rS   rT   r|   f  s   z%KafkaConnectionManager.update_backoffc                 C   s*   || j vrdS td| j | d t  S )zgConnection delay in seconds.

        Uses exponential backoff/retry with jitter. See KIP-144.
        r   r   )rF   rv   rg   rh   r   rS   rS   rT   rt   q  s   
z'KafkaConnectionManager.connection_delayc                 C   s"   || j vr
| jd S | j | d S )Nr   r   )rF   r;   r   rS   rS   rT   r   z  s   

z9KafkaConnectionManager.socket_connection_setup_timeout_msc                 C   s   | j |S )zsReturn the most recent auth-class failure for ``node_id``,
        or None if there is no sticky failure on record.rG   r>   r   rS   rS   rT   auth_failure  s   z#KafkaConnectionManager.auth_failurec                 C   s   | j |}|dur|dS )z;Raise the cached auth-class failure for ``node_id`` if any.Nr   )rO   rm   r   rS   rS   rT   r     s   z/KafkaConnectionManager.maybe_raise_auth_failurec                 C   sp   |d ur| j |}|d ur|  d S d S | js6d| _| j  t| j  D ]}|  q(| j  d S d S NT)	rD   r>   r{   rN   rL   notifylistra   rB   )rO   rm   rc   r[   rS   rS   rT   r{     s   

zKafkaConnectionManager.closec              	      s    du r
|I dH S t  fdd}fdd}|| ||  fdd}| j d |}zI dH W | j| S | j| w )	a  Await `future` with a timeout in ms. Raises KafkaTimeoutError on timeout.

        Must be awaited from a coroutine running on this loop. The underlying
        future is not cancelled on timeout - it continues to run; the timeout
        only unblocks the awaiter.
        Nc                        j s
 |  d S d S rV   )r   success)r   wrapperrS   rT   _on_success     z4KafkaConnectionManager.wait_for.<locals>._on_successc                    r   rV   )r   r   r   r   rS   rT   _on_failure  r   z4KafkaConnectionManager.wait_for.<locals>._on_failurec                      s"   j std   d S d S )NzTimed out after %s ms)r   r   rr   r   rS   rc   r   rS   rT   _on_timeout  s
   
z4KafkaConnectionManager.wait_for.<locals>._on_timeoutr8   )r   add_callbackr   rA   
call_latercancel)rO   futurerc   r   r   r   timerrS   r   rT   wait_for  s   



zKafkaConnectionManager.wait_forc                 G      | j j|g|R  S )zAccepts a coroutine / awaitable / function and schedules it on the event loop.

        Thread-safe.

        Returns: Future
        )rA   call_soon_with_futurerO   coroargsrS   rS   rT   r     s   z KafkaConnectionManager.call_soonc                 G   r   )a  Schedules coro on the event loop, blocks until complete, returns value or raises.

        If an IO thread is running (via start()), the caller thread blocks on
        a cross-thread Event while the coroutine runs on the IO thread. Safe
        to call concurrently from multiple caller threads.

        If no IO thread is running, falls back to driving the loop on the
        caller thread (legacy behavior).
        )rA   r   r   rS   rS   rT   r     s   
zKafkaConnectionManager.runr   rV   )TN)NTTT)NN)'__name__
__module____qualname__r   socketIPPROTO_TCPTCP_NODELAY
SOL_SOCKETSO_KEEPALIVEr:   r<   rU   propertyrW   rb   r   r   r   r   r   rI   r   r   r   rq   r   r   r   r   r   r|   rt   r   r   r   r{   r   r   r   rS   rS   rS   rT   r      s    	
 !"#$%'+

7
	



)

	 	
	r   ) r9   logginginspectri   r   rg   inetr   
connectionr   r0   r   r]   r   r   kafka.clusterr   kafka.errorserrorsrr   kafka.net.wakeup_notifierr	   "kafka.protocol.broker_version_datar
   kafka.futurer   kafka.versionr   	getLoggerr   r?   r   rS   rS   rS   rT   <module>   s$    
