public < T extends Serializable > int enqueue ( final String topic , List < T > objects ) { int messagesBuffered = 0 ; for ( T object : objects ) { final String value ; if ( String . class . isAssignableFrom ( object . getClass ( ) ) ) { value = String . class . cast ( object ) ; } else { try { value = _mapper . writeValueAsString ( object ) ; } catch ( JsonProcessingException e ) { _logger . warn ( "Exception while serializing the object to a string. Skipping this object." , e ) ; continue ; } } try { boolean addedToBuffer = _executorService . submit ( new ProducerWorker ( topic , value ) ) . get ( ) ; if ( addedToBuffer ) { messagesBuffered ++ ; } } catch ( InterruptedException e ) { _logger . warn ( "Enqueue operation was interrupted by calling code." ) ; Thread . currentThread ( ) . interrupt ( ) ; } catch ( ExecutionException e ) { throw new SystemException ( e ) ; } } return messagesBuffered ; }
Adds the messages to the Producer Buffer which will later be batched by Kafka and sent to the brokers .