Fix hang in bulk helper semaphore when server responses are slower than flushInterval (#2027)

* Set version to 8.10.1

* Add tests for bulk helper with various flush and server timeouts

* Copy and empty bulkBody when flushBytes is reached

Before it was waiting until after semaphore resolved, then sending with
a reference to bulkBody. If flushInterval is reached after `await
semaphore()` but before `send(bulkBody)`, onFlushTimeout is "stealing"
bulkBody so that there is nothing left in bulkBody for the flushBytes
block to send, causing an indefinite hang for a promise that does not
resolve.

* comment typo fixes

---------

Co-authored-by: Quentin Pradet <quentin.pradet@elastic.co>

test/unit/helpers/bulk.test.ts

@@ -23,9 +23,11 @@ import { createReadStream } from 'fs'
 import * as http from 'http'
 import { join } from 'path'
 import split from 'split2'
+import { Readable } from 'stream'
 import { test } from 'tap'
 import { Client, errors } from '../../../'
 import { buildServer, connection } from '../../utils'
+const { sleep } = require('../../integration/helper')
 
 let clientVersion: string = require('../../../package.json').version // eslint-disable-line
 if (clientVersion.includes('-')) {
@@ -1594,3 +1596,150 @@ test('Flush interval', t => {
 
   t.end()
 })
+
+test(`flush timeout does not lock process when flushInterval is less than server timeout`, async t => {
+  const flushInterval = 500
+
+  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+    setTimeout(() => {
+      res.writeHead(200, { 'content-type': 'application/json' })
+      res.end(JSON.stringify({ errors: false, items: [{}] }))
+    }, 1000)
+  }
+
+  const [{ port }, server] = await buildServer(handler)
+  const client = new Client({ node: `http://localhost:${port}` })
+
+  async function * generator () {
+    const data = dataset.slice()
+    for (const doc of data) {
+      await sleep(flushInterval)
+      yield doc
+    }
+  }
+
+  const result = await client.helpers.bulk({
+    datasource: Readable.from(generator()),
+    flushBytes: 1,
+    flushInterval: flushInterval,
+    concurrency: 1,
+    onDocument (_) {
+      return {
+        index: { _index: 'test' }
+      }
+    },
+    onDrop (_) {
+      t.fail('This should never be called')
+    }
+  })
+
+  t.type(result.time, 'number')
+  t.type(result.bytes, 'number')
+  t.match(result, {
+    total: 3,
+    successful: 3,
+    retry: 0,
+    failed: 0,
+    aborted: false
+  })
+
+  server.stop()
+})
+
+test(`flush timeout does not lock process when flushInterval is greater than server timeout`, async t => {
+  const flushInterval = 500
+
+  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+    setTimeout(() => {
+      res.writeHead(200, { 'content-type': 'application/json' })
+      res.end(JSON.stringify({ errors: false, items: [{}] }))
+    }, 250)
+  }
+
+  const [{ port }, server] = await buildServer(handler)
+  const client = new Client({ node: `http://localhost:${port}` })
+
+  async function * generator () {
+    const data = dataset.slice()
+    for (const doc of data) {
+      await sleep(flushInterval)
+      yield doc
+    }
+  }
+
+  const result = await client.helpers.bulk({
+    datasource: Readable.from(generator()),
+    flushBytes: 1,
+    flushInterval: flushInterval,
+    concurrency: 1,
+    onDocument (_) {
+      return {
+        index: { _index: 'test' }
+      }
+    },
+    onDrop (_) {
+      t.fail('This should never be called')
+    }
+  })
+
+  t.type(result.time, 'number')
+  t.type(result.bytes, 'number')
+  t.match(result, {
+    total: 3,
+    successful: 3,
+    retry: 0,
+    failed: 0,
+    aborted: false
+  })
+
+  server.stop()
+})
+
+test(`flush timeout does not lock process when flushInterval is equal to server timeout`, async t => {
+  const flushInterval = 500
+
+  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+    setTimeout(() => {
+      res.writeHead(200, { 'content-type': 'application/json' })
+      res.end(JSON.stringify({ errors: false, items: [{}] }))
+    }, flushInterval)
+  }
+
+  const [{ port }, server] = await buildServer(handler)
+  const client = new Client({ node: `http://localhost:${port}` })
+
+  async function * generator () {
+    const data = dataset.slice()
+    for (const doc of data) {
+      await sleep(flushInterval)
+      yield doc
+    }
+  }
+
+  const result = await client.helpers.bulk({
+    datasource: Readable.from(generator()),
+    flushBytes: 1,
+    flushInterval: flushInterval,
+    concurrency: 1,
+    onDocument (_) {
+      return {
+        index: { _index: 'test' }
+      }
+    },
+    onDrop (_) {
+      t.fail('This should never be called')
+    }
+  })
+
+  t.type(result.time, 'number')
+  t.type(result.bytes, 'number')
+  t.match(result, {
+    total: 3,
+    successful: 3,
+    retry: 0,
+    failed: 0,
+    aborted: false
+  })
+
+  server.stop()
+})