[rursprung]

(generated by copilot/claude, reviewed by me)

• You have read the Spring Data contribution guidelines.
• There is a ticket in the bug tracker for the project in our issue tracker. Add the issue number to the Closes #issue-number line below
• You use the code formatters provided here and have them applied to your changes. Don’t submit any formatting related changes.
• You submit test cases (unit or integration tests) that back your changes.
• You added yourself as author in the headers of the classes you touched. Amend the date range in the Apache license header if needed. For new types, add the license header (copy from another file and set the current year only).

Closes #2071

What and why

This adds a BulkIngester to ElasticsearchOperations. It collects index, update and delete operations one by one and sends them in bulk requests. It follows the direction agreed in #2071: a client-independent API, an implementation in elc, a default that throws UnsupportedClientOperationException for other clients, a mandatory listener, imperative only, no retries.

bulkIndex / bulkUpdate (also with the size limits from #3357) work on a list that is known in advance and block until it is done. The ingester covers the cases where that does not fit:

• Streaming: operations come one by one (files, messages, change feeds) and the total is unknown.
• Time-based flush: send pending operations after an interval, even if no limit is reached.
• Concurrent requests: more than one request in flight.
• Backpressure: add() blocks when too many requests are in flight, so memory use stays bounded.
• Mixed operations, including delete, in the same request.
• Per-item results with a context: each operation can carry an object that is passed back with its result, so failures can be traced to the source record.

Today users can use the native BulkIngester directly, but then they lose entity mapping, auditing, entity callbacks, write-back of id/version/seq_no/primary_term, exception translation and observations.

API overview

New package org.springframework.data.elasticsearch.core.bulk (no client types):

• BulkIngester<C>: add(IndexQuery[, C]), add(UpdateQuery[, C]), addDelete(id[, routing, C]), flush(), close(), statistics getters.
• BulkIngesterListener<C>: afterBulk(executionId, List<BulkOperationResult<C>>) and afterBulk(executionId, List<C> contexts, RuntimeException failure).
• BulkIngesterOptions: maxOperations, maxSize, maxConcurrentRequests, flushInterval and BulkOptions for the request-wide settings. Defaults are those of the native helper (1000 operations, 5 MiB, 1 concurrent request, no flush interval).
• BulkOperationResult<C> (record: type, context, id, index, status, IndexedObjectInformation, FailureDetails) and BulkOperationType.

Entry point: ElasticsearchOperations#bulkIngester(BulkIngesterOptions, IndexCoordinates, BulkIngesterListener) plus overloads with an entity class and/or default options.

var options = BulkIngesterOptions.builder()
		.withMaxOperations(500)
		.withFlushInterval(Duration.ofSeconds(5))
		.build();

try (BulkIngester<String> ingester = operations.bulkIngester(options, Person.class, listener)) {
	for (Person person : persons) {
		ingester.add(new IndexQueryBuilder().withObject(person).build(), person.getName());
	}
	ingester.addDelete("43", null, "delete 43");
}

Behaviour

• Callbacks: auditing and BeforeConvertCallback run on the calling thread in add(). For successful index/create operations the id, version, seq_no and primary term are written back to the entity and the AfterSaveCallback runs. Failed operations get neither.
• Threading: the listener and AfterSaveCallback run on threads of the ingester. With more than one concurrent request they can run concurrently. The listener must not call add/flush/close (that would deadlock in the native helper); the ingester detects this and throws IllegalStateException. Exceptions thrown by the listener are logged, because the native helper would lose them silently.
• Blocking: add() and flush() block when maxConcurrentRequests requests are in flight. Not for reactive threads.
• Limits: the native helper sends a request after the operation that reaches maxSize, so a request can be larger by one operation.
• Request settings: the BulkOptions and the refresh policy of the template are applied to every request.
• Failures: item failures are part of the results. A failed request is unwrapped from CompletionException/ExecutionException and translated with the client's exception translator before it is passed to the listener. Nothing is retried.
• Close: close() sends the rest, waits for all requests and listener calls, can be called more than once, and does not close the client. add after close throws IllegalStateException.
• Observations: one bulk observation per request with the real batch size. A request failure is recorded as error. Item failures are not (unlike bulkIndex, where they become a BulkFailureException).

Impact on spring-data-opensearch

The change is additive. OpenSearchTemplate gets the default bulkIngester that throws UnsupportedClientOperationException, so nothing breaks and nothing has to be done right away.

To keep the effort for spring-data-opensearch small, everything that does not depend on the client is in core.AbstractBulkIngester: argument checks, closed state, before-convert/auditing, id conversion and default routing for deletes, context bookkeeping, mapping of a client-neutral ItemResponse to BulkOperationResult, write-back and after-save callback, unwrapping and translation of request failures, and listener isolation. It lives in core next to AbstractElasticsearchTemplate so it can use the existing protected callback helpers without new API on the template. OpenSearchTemplate extends AbstractElasticsearchTemplate, so an adapter can pass it to the base class.

opensearch-java 3.x has the same helper (org.opensearch.client.opensearch._helpers.bulk.BulkIngester, same builder and listener methods). An OpenSearch adapter only has to:

• implement doAdd(IndexQuery, …), doAdd(UpdateQuery, …), doAddDelete(id, routing, …) by converting with its RequestConverter and calling the native add;
• implement doFlush(), doClose() and the seven statistics getters by delegating to the native helper;
• implement translateException(Throwable) with OpenSearchExceptionTranslator;
• in its native BulkListener, map each BulkResponseItem to AbstractBulkIngester.ItemResponse and call onResponse(...) / onFailure(...);
• override bulkIngester(...) in OpenSearchTemplate.

Its osc.RequestConverter would need the equivalents of bulkRequestBuilder, bulkOperation (#3352) and bulkDeleteOperation. The ELC adapter here is about 200 lines including the observation handling, which OpenSearch does not need.

spring-data-opensearch keeps its own copy of CallbackIntegrationTests, so the new test there does not affect it until it is synced.

Compatibility

Additive only: new types, new default methods on ElasticsearchOperations, a new method in ElasticsearchTemplate, a new package-private method in RequestConverter. ElasticsearchTemplate#createObservation changed from private to package-private. No behaviour change for existing operations.

Out of scope / follow-ups

• Reactive API.
• Retry/backoff (the native backoffPolicy).
• Configuring an external scheduler / thread pool.
• Repository-level API.
• Delete by entity (addDelete(Object entity)).

Open questions

1. Is core.AbstractBulkIngester (public but not official API, like AbstractElasticsearchTemplate) the right place for the shared logic, or would you prefer it in core.bulk with hooks passed in by the template?
2. Should an item failure mark the bulk observation as error, like bulkIndex does?
3. Since Size limited bulk requests #3357, BulkOptions also has maxOperations/maxSize, and the template can have default BulkOptions. The ingester uses only the limits from BulkIngesterOptions; it ignores the limits in the nested BulkOptions and the template default (documented). Is that acceptable, or should it fall back to them?
4. Exceptions from the AfterSaveCallback on ingester threads are logged and the item is still reported as successful (the document was written). Is that the expected behaviour?

Tests

• Unit: BulkIngesterOptionsUnitTests, ElasticsearchOperationsBulkIngesterUnitTests, AbstractBulkIngesterUnitTests (fake adapter: result mapping, callbacks only on success, listener exception isolation, re-entrance guard, closed state, failure translation), ElasticsearchBulkIngesterUnitTests (mocked transport: request failure translation, request settings, item mapping, listener required), RequestConverterTest (delete conversion).
• Integration (// #2071): ElasticsearchELCIntegrationTests (mixed index/create/update/delete with contexts, generated id and version write-back, partial 409 failure, request splitting, stats, idempotent close, flush interval), CallbackIntegrationTests (BeforeConvert, AfterSave, seq_no/primary_term write-back), ObservabilityIntegrationTests (one bulk observation per request with batch size).

Related

This is based on #3352 and related to #3357

[rursprung]

CC @sothawo, @reta, @siddeshbb & @patpatpat123

[rursprung]

since #3352 is now merged this is ready for review

[rursprung]

rebased to resolve merge conflicts now that #3357 has landed.