def callback(self, filename, lines, **kwargs):
        """publishes lines one by one to the given topic"""
        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')
publishes lines one by one to the given topic