Bulk helper not giving indications on failure when using stream as datasource

🐛 Bug Report

When using the bulk helper function, the function doesn't throw, or doesn't notify on failures using any other mechanism i have found (onDrop \ result errors). I have set the node value for a URL that doesn't have an elastic cluster listening on. When you use stream datasource nothing happens, but when the datasource is array an exception is thrown.

To Reproduce

Run the following code and swap between the datasources to see the different behaviours. make sure the client is pointed to an address with no cluster running.

When its stream nothing gets printed.
When its array a connection refused error gets printed to the console.

Paste your code here:

import stream from "stream";
import { Client } from "@elastic/elasticsearch";

const client = new Client({ node: "http://localhost:9200" });

const readStream = new stream.Readable({
  objectMode: true,
  read() {},
});

readStream.push({ id: 1, title: "foo" });
readStream.push({ id: 2, title: "bar" });

try {
  const result = await client.helpers.bulk({
    // datasource: [{ id: 1, title: "foo" }, { id: 2, title: "bar" }],
    datasource: readStream,
    onDocument(doc) {
      return {
        index: { _index: "my-index3", _id: doc.id },
      };
    },
    timeout: "1s",
    retries: 1,
    onDrop(doc) {
      console.log(doc);
    },
  });

  console.log(result.errors)
} catch (error) {
  console.error(error);
}

Expected behavior

You have a way to detect such errors, either if they are thrown or by an event handler.

Your Environment

node version: v20.9.0 | v18.18.2
@elastic/elasticsearch version: =8.10.0
os: Ubuntu 20.04.6 LTS

elastic / elasticsearch-js