def send_bulk(data, tag, chunk, bulk_message_count, extracted_values, info)
      begin

        log.on_trace { log.trace "bulk request: #{data}" }
        response = client(info.host).bulk body: data, index: info.index
        log.on_trace { log.trace "bulk response: #{response}" }

        if response['errors']
          error = Fluent::Plugin::ElasticsearchErrorHandler.new(self)
          error.handle_error(response, tag, chunk, bulk_message_count, extracted_values)
        end
      rescue RetryStreamError => e
        emit_tag = @retry_tag ? @retry_tag : tag
        router.emit_stream(emit_tag, e.retry_stream)
      rescue => e
        ignore = @ignore_exception_classes.any? { |clazz| e.class <= clazz }

        log.warn "Exception ignored in tag #{tag}: #{e.class.name} #{e.message}" if ignore

        @_es = nil if @reconnect_on_error
        @_es_info = nil if @reconnect_on_error

        raise UnrecoverableRequestFailure if ignore && @exception_backup

        # FIXME: identify unrecoverable errors and raise UnrecoverableRequestFailure instead
        raise RecoverableRequestFailure, "could not push logs to Elasticsearch cluster (#{connection_options_description(info.host)}): #{e.message}" unless ignore
      end
    end
def send_bulk(data, tag, chunk, bulk_message_count, extracted_values, info)
      begin

        log.on_trace { log.trace "bulk request: #{data}" }
        response = client(info.host).bulk body: data, index: info.index
        log.on_trace { log.trace "bulk response: #{response}" }

        if response['errors']
          error = Fluent::Plugin::ElasticsearchErrorHandler.new(self)
          error.handle_error(response, tag, chunk, bulk_message_count, extracted_values)
        end
      rescue RetryStreamError => e
        emit_tag = @retry_tag ? @retry_tag : tag
        router.emit_stream(emit_tag, e.retry_stream)
      rescue => e
        ignore = @ignore_exception_classes.any? { |clazz| e.class <= clazz }

        log.warn "Exception ignored in tag #{tag}: #{e.class.name} #{e.message}" if ignore

        @_es = nil if @reconnect_on_error
        @_es_info = nil if @reconnect_on_error

        raise UnrecoverableRequestFailure if ignore && @exception_backup

        # FIXME: identify unrecoverable errors and raise UnrecoverableRequestFailure instead
        raise RecoverableRequestFailure, "could not push logs to Elasticsearch cluster (#{connection_options_description(info.host)}): #{e.message}" unless ignore
      end
    end
send_bulk given a specific bulk request, the original tag,
 chunk, and bulk_message_count