code: def callback(self, filename, lines, **kwargs):
          
          timestamp = self.get_timestamp(**kwargs)
          if kwargs.get('timestamp', False):
              del kwargs['timestamp']
  
          for line in lines:
              try:
                  import warnings
                  with warnings.catch_warnings():
                      warnings.simplefilter('error')
                      #produce message
                      if self._key is None:
                          response = self._prod.send_messages(self._kafka_config['topic'], self.format(filename, line, timestamp, **kwargs))
                      else:
                          response = self._prod.send_messages(self._kafka_config['topic'], self._key, self.format(filename, line, timestamp, **kwargs))
  
                      if response:
                          if response[0].error:
                              self._logger.info('message error: {0}'.format(response[0].error))
                              self._logger.info('message offset: {0}'.format(response[0].offset))
  
              except Exception as e:
                  try:
                      self._logger.error('Exception caught sending message/s : ' + str(e))
                      raise TransportException(e.strerror)
                  except AttributeError:
                      raise TransportException('Unspecified exception encountered')
docs: publishes lines one by one to the given topic