#NEWLINE# def _send_messages( self , topic , partition , * msg , **kwargs ) : #NEWLINE# #INDENT# key = kwargs . pop ( 'key' , None ) #NEWLINE# if ( not isinstance ( msg , ( list , tuple ) ) ) : #NEWLINE# #INDENT# raise TypeError ( 'msg is not a list or tuple!' ) #NEWLINE# #UNINDENT# for m in msg : #NEWLINE# #INDENT# if ( m is None ) : #NEWLINE# #INDENT# if ( key is None ) : #NEWLINE# #INDENT# raise TypeError ( "key and payload can't be null in one" ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if ( not isinstance ( m , six . binary_type ) ) : #NEWLINE# #INDENT# raise TypeError ( 'all produce message payloads must be null or type bytes' ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# if ( ( key is not None ) and ( not isinstance ( key , six . binary_type ) ) ) : #NEWLINE# #INDENT# raise TypeError ( 'the key must be type bytes' ) #NEWLINE# #UNINDENT# if self . async : #NEWLINE# #INDENT# for ( idx , m ) in enumerate ( msg ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# item = ( TopicPartition ( topic , partition ) , m , key ) #NEWLINE# if ( self . async_queue_put_timeout == 0 ) : #NEWLINE# #INDENT# self . queue . put_nowait ( item ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# self . queue . put ( item , True , self . async_queue_put_timeout ) #NEWLINE# #UNINDENT# #UNINDENT# except Full : #NEWLINE# #INDENT# raise AsyncProducerQueueFull ( msg [ idx : ] , ( 'Producer async queue overfilled. Current queue size %d.' % self . queue . qsize ( ) ) ) #NEWLINE# #UNINDENT# #UNINDENT# resp = [ ] #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# messages = create_message_set ( [ ( m , key ) for m in msg ] , self . codec , key , self . codec_compresslevel ) #NEWLINE# req = ProduceRequestPayload ( topic , partition , messages ) #NEWLINE# try : #NEWLINE# #INDENT# resp = self . client . send_produce_request ( [ req ] , acks = self . req_acks , timeout = self . ack_timeout , fail_on_error = self . sync_fail_on_error ) #NEWLINE# #UNINDENT# except Exception : #NEWLINE# #INDENT# log . exception ( 'Unable to send messages' ) #NEWLINE# raise #NEWLINE# #UNINDENT# #UNINDENT# return resp