<s> from __future__ import absolute_import <EOL> import collections <EOL> import copy <EOL> import logging <EOL> import threading <EOL> import six <EOL> from . . import errors as Errors <EOL> from . . structs import TopicPartition <EOL> from . . version import __version__ <EOL> from . . protocol . produce import ProduceRequest <EOL> log = logging . getLogger ( __name__ ) <EOL> class Sender ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> _DEFAULT_CONFIG = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' + __version__ , <EOL> '<STR_LIT>' : ( <NUM_LIT:0> , <NUM_LIT:8> , <NUM_LIT:0> ) , <EOL> } <EOL> def __init__ ( self , client , metadata , accumulator , ** configs ) : <EOL> super ( Sender , self ) . __init__ ( ) <EOL> self . config = copy . copy ( self . _DEFAULT_CONFIG ) <EOL> for key in self . config : <EOL> if key in configs : <EOL> self . config [ key ] = configs . pop ( key ) <EOL> self . name = self . config [ '<STR_LIT>' ] + '<STR_LIT>' <EOL> self . _client = client <EOL> self . _accumulator = accumulator <EOL> self . _metadata = client . cluster <EOL> self . _running = True <EOL> self . _force_close = False <EOL> self . _topics_to_add = set ( ) <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> while self . _running : <EOL> try : <EOL> self . run_once ( ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" ) <EOL> log . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> while ( not self . _force_close <EOL> and ( self . _accumulator . has_unsent ( ) <EOL> or self . _client . in_flight_request_count ( ) > <NUM_LIT:0> ) ) : <EOL> try : <EOL> self . run_once ( ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" ) <EOL> if self . _force_close : <EOL> self . _accumulator . abort_incomplete_batches ( ) <EOL> try : <EOL> self . _client . close ( ) <EOL> except Exception : <EOL> log . exception ( "<STR_LIT>" ) <EOL> log . debug ( "<STR_LIT>" ) <EOL> def run_once ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . _topics_to_add : <EOL> self . _client . add_topic ( self . _topics_to_add . pop ( ) ) <EOL> result = self . _accumulator . ready ( self . _metadata ) <EOL> ready_nodes , next_ready_check_delay , unknown_leaders_exist = result <EOL> if unknown_leaders_exist : <EOL> log . debug ( '<STR_LIT>' ) <EOL> self . _metadata . request_update ( ) <EOL> not_ready_timeout = <NUM_LIT> <EOL> for node in list ( ready_nodes ) : <EOL> if not self . _client . ready ( node ) : <EOL> log . debug ( '<STR_LIT>' , node ) <EOL> ready_nodes . remove ( node ) <EOL> not_ready_timeout = min ( not_ready_timeout , <EOL> self . _client . connection_delay ( node ) ) <EOL> batches_by_node = self . _accumulator . drain ( <EOL> self . _metadata , ready_nodes , self . config [ '<STR_LIT>' ] ) <EOL> expired_batches = self . _accumulator . abort_expired_batches ( <EOL> self . config [ '<STR_LIT>' ] , self . _metadata ) <EOL> requests = self . _create_produce_requests ( batches_by_node ) <EOL> poll_timeout_ms = min ( next_ready_check_delay * <NUM_LIT:1000> , not_ready_timeout ) <EOL> if ready_nodes : <EOL> log . debug ( "<STR_LIT>" , ready_nodes ) <EOL> log . debug ( "<STR_LIT>" , len ( requests ) , requests ) <EOL> poll_timeout_ms = <NUM_LIT:0> <EOL> for node_id , request in six . iteritems ( requests ) : <EOL> batches = batches_by_node [ node_id ] <EOL> log . debug ( '<STR_LIT>' , request ) <EOL> ( self . _client . send ( node_id , request ) <EOL> . add_callback ( <EOL> self . _handle_produce_response , batches ) <EOL> . add_errback ( <EOL> self . _failed_produce , batches , node_id ) ) <EOL> self . _client . poll ( poll_timeout_ms , sleep = True ) <EOL> def initiate_close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _running = False <EOL> self . _accumulator . close ( ) <EOL> self . wakeup ( ) <EOL> def force_close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _force_close = True <EOL> self . initiate_close ( ) <EOL> def add_topic ( self , topic ) : <EOL> if topic not in self . _topics_to_add : <EOL> self . _topics_to_add . add ( topic ) <EOL> self . wakeup ( ) <EOL> def _failed_produce ( self , batches , node_id , error ) : <EOL> log . debug ( "<STR_LIT>" , node_id , error ) <EOL> for batch in batches : <EOL> self . _complete_batch ( batch , error , - <NUM_LIT:1> ) <EOL> def _handle_produce_response ( self , batches , response ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( '<STR_LIT>' , response ) <EOL> if response : <EOL> batches_by_partition = dict ( [ ( batch . topic_partition , batch ) <EOL> for batch in batches ] ) <EOL> for topic , partitions in response . topics : <EOL> for partition , error_code , offset in partitions : <EOL> tp = TopicPartition ( topic , partition ) <EOL> error = Errors . for_code ( error_code ) <EOL> batch = batches_by_partition [ tp ] <EOL> self . _complete_batch ( batch , error , offset ) <EOL> else : <EOL> for batch in batches : <EOL> self . _complete_batch ( batch , None , - <NUM_LIT:1> ) <EOL> def _complete_batch ( self , batch , error , base_offset ) : <EOL> """<STR_LIT>""" <EOL> if error is Errors . NoError : <EOL> error = None <EOL> if error is not None and self . _can_retry ( batch , error ) : <EOL> log . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> batch . topic_partition , <EOL> self . config [ '<STR_LIT>' ] - batch . attempts - <NUM_LIT:1> , <EOL> error ) <EOL> self . _accumulator . reenqueue ( batch ) <EOL> else : <EOL> if error is Errors . TopicAuthorizationFailedError : <EOL> error = error ( batch . topic_partition . topic ) <EOL> batch . done ( base_offset , error ) <EOL> self . _accumulator . deallocate ( batch ) <EOL> if getattr ( error , '<STR_LIT>' , False ) : <EOL> self . _metadata . request_update ( ) <EOL> def _can_retry ( self , batch , error ) : <EOL> """<STR_LIT>""" <EOL> return ( batch . attempts < self . config [ '<STR_LIT>' ] <EOL> and getattr ( error , '<STR_LIT>' , False ) ) <EOL> def _create_produce_requests ( self , collated ) : <EOL> """<STR_LIT>""" <EOL> requests = { } <EOL> for node_id , batches in six . iteritems ( collated ) : <EOL> requests [ node_id ] = self . _produce_request ( <EOL> node_id , self . config [ '<STR_LIT>' ] , <EOL> self . config [ '<STR_LIT>' ] , batches ) <EOL> return requests <EOL> def _produce_request ( self , node_id , acks , timeout , batches ) : <EOL> """<STR_LIT>""" <EOL> produce_records_by_partition = collections . defaultdict ( dict ) <EOL> for batch in batches : <EOL> topic = batch . topic_partition . topic <EOL> partition = batch . topic_partition . partition <EOL> buf = batch . records . buffer ( ) <EOL> produce_records_by_partition [ topic ] [ partition ] = buf <EOL> version = <NUM_LIT:1> if self . config [ '<STR_LIT>' ] >= ( <NUM_LIT:0> , <NUM_LIT:9> ) else <NUM_LIT:0> <EOL> return ProduceRequest [ version ] ( <EOL> required_acks = acks , <EOL> timeout = timeout , <EOL> topics = [ ( topic , list ( partition_info . items ( ) ) ) <EOL> for topic , partition_info <EOL> in six . iteritems ( produce_records_by_partition ) ] <EOL> ) <EOL> def wakeup ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _client . wakeup ( ) </s>
