Send Elasticsearch bulk bodies without copying them and filter their responses - #5025

Fixes #5024

An Elasticsearch bulk request held its body in memory up to four times to send it, parsed every field of a response of which the client reads two for each item, and built a map for the action line of each document. Its action also depended on the JVM's default locale.

Changes

• BulkRequestEntity (new, in RestElasticSearchClient): the body of a bulk request, which writes the items' serialized bytes as they are (writeTo, which the gzip wrapper of index.[X].elasticsearch.compression uses) and reads them through a stream (getContent, which the HTTP client uses otherwise). It knows its length and is repeatable, so the client can send it again to another node; a reattempt builds a new one over the failed items. Its stream behaves as the ByteArrayInputStream of the entity it replaces did: a read fills the buffer across items, available() counts what is left, and it supports mark and reset, which request signers such as those of the AWS SDK use to hash the body before it is sent. It replaces buildBulkRequestInput, which copied the items into a ByteArrayOutputStream and from there into an array, on every attempt. With index.[X].elasticsearch.compression the client still compresses the body into a buffer of its own.
• Bulk responses: a bulk request asks for filter_path=errors,items.index.status,items.index.error,items.update.status,items.update.error,items.delete.status,items.delete.error: each item's status and error, which the client reads, under the actions JanusGraph sends, and errors, which keeps RestBulkResponse.isErrors() meaningful. Every item keeps its status, so the items line up with the requests they answer; filtering by the error alone leaves out the items which succeeded (checked against Elasticsearch 9.5.4). The actions are named from ElasticSearchMutation.RequestType rather than matched with *, so that the query string holds no asterisk, which a request signer would have to percent-encode the way the cluster does. The response to 1,000 index requests is 25,026 bytes instead of 172,842. RestBulkItemResponse.getResult(), which nothing read, is deprecated, as the response no longer carries it.
• Action lines: written field by field with a JsonGenerator, instead of as a HashMap in an ImmutableMap serialized by the ObjectMapper, and the action is lower-cased in the root locale. Under a Turkish default locale the action of an index request was ındex, with a dotless i, and Elasticsearch rejected the whole bulk request: Malformed action/metadata line [1], expected field [create], [delete], [index] or [update] but found [ındex].
• Changelog entry.

Measurements

A harness (not shipped) sends the same bulk of index requests through RestElasticSearchClient.bulkRequest to Elasticsearch 9.5.4 in Docker (one shard, nothing indexed but the source), with master's RestElasticSearchClient and RestBulkResponse against this branch's on the same classpath, in two rounds of alternating order. The time is the median per bulk; the allocation counts every thread, the HTTP client's I/O threads included.

The smallest heap in which the bulk of 50,000 documents of 1 KB completes is 320 MB on master in one run and 384 MB in the other, below which it runs out of memory in ByteArrayOutputStream. On this branch it is 144 MB, below which the serialization of the documents themselves runs out.

Measured alone, building the item of a 100-byte document takes 229 ns instead of about 400, and of a 1 KB document about 790 ns instead of 1,440. Parsing a response takes 52 ns per item instead of 200.

Tests

• RestClientBulkRequestsTest: the entity's body is the items' bytes in order, through writeTo and through its stream read twice, and its length is theirs; the stream gives the same bytes whatever the size of the buffer and the offset it reads into, filling each buffer across items, with available() counting down, and reads the rest again after a reset from a mark at every position, or from the start without one; a bulk request carries the filter and the entity; the action line of each operation, with characters which JSON escapes, the mapping type for Elasticsearch 6, and a Turkish default locale, which yields ındex when the action is lower-cased in the default locale.
• RestClientRetryTest: the reattempt of a bulk request sends the failed item alone.
• ElasticsearchIndexTest.testTheFailedItemsOfABulkResponseAreReportedForTheirOwnDocuments (new, Elasticsearch in a container): updates of missing documents between index requests which succeed fail with a 404, and the documents reported are exactly theirs. With a filter of the errors alone it fails, as the successful items are left out and the failures shift onto other documents.
• ElasticsearchIndexTest.testCompressedRequests sends bulk requests through the gzip wrapper, which writes the entity with writeTo.
• Locally, Java 11, against Elasticsearch 9.5.4 in a container: ElasticsearchIndexTest (288) and the RestClient*Test classes (77) pass on the final code. BerkeleyElasticsearchTest (92, 1 skipped), ElasticsearchConfigTest (12) and ElasticSearchIndexReattemptTest (7) passed before the last two changes, mark and reset, and the actions named in the filter.

🤖 Generated with Claude Code