[rursprung]

Based on #3352 — please review only the commits after 31ce183f. I will rebase once #3352 is merged.

(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 #3345

What this does

Bulk operations can now be split into several bulk requests by number of operations and/or size in bytes:

BulkOptions.builder()
    .withMaxOperations(1000)
    .withMaxSize(5 * 1024 * 1024)
    .build();

Both limits are unset by default, so nothing changes unless they are used.

Design (as proposed in the issue)

• No BulkIngester. The splitting is done in client.elc (BulkRequestSplitter, RequestConverter#documentBulkRequests()). The ingester starts its own threads, can't stop at the first failed request, refreshes after every request, can go over maxSize by one operation and blocks, so it can't be used from the reactive template.
• Size measurement: each document (or update action) is serialized once with BinaryData.of(...) and put back into the operation, so it is not serialized again when the request is sent. Same idea as the ingester's IngesterOperation. The action/metadata line is measured exactly. This is only done when maxSize is set.
• Splitting: a new request is started before an operation would go over a limit. An operation that is larger than maxSize on its own is sent alone.
• Failures: requests are sent one after the other (reactive: concatMap). A request that fails as a whole stops the processing. Item failures of all requests are collected into one BulkFailureException at the end. Successful requests are not rolled back (documented).
• Result order: the returned IndexedObjectInformations / response items cover all queries in their original order, so ids, versions etc. are still written back.
• Refresh: the refresh policy is only applied to the last request; earlier requests use refresh=false.
• Template default: setBulkOptions()/getBulkOptions() on both abstract templates (kept in copies) and a bulkOptions() hook in ElasticsearchConfigurationSupport, used by all four ELC configurations. If the BulkOptions of a call don't set a limit, the template default limit is used. This makes the limits work for save(Iterable), saveAll(...) and reactive save(Flux, ...).

Open questions

1. Observations: there is still one BULK observation around the whole call. Should there also be one per request?
2. Refresh only on the last request: OK, or should every request get the refresh policy?
3. Template default: only maxOperations/maxSize are taken from the template default right now. Should the other fields (timeout, pipeline, …) also fall back to it?
4. Reuse by spring-data-opensearch: the splitting is package-private in client.elc. spring-data-opensearch keeps compiling and working unchanged, but it ignores the new limits until it implements them (opensearch-java has the same BinaryData/BulkOperation structure, so a port is straightforward). Should the client-independent batching logic be a public helper in core so it can be reused, or is a per-client copy preferred?

Compatibility

Only additions to the public API (BulkOptions builder methods and getters, template setter/getter, configuration hook), all with @since 6.2. BulkOptions stays client independent. spring-data-opensearch (which extends the abstract templates and ElasticsearchConfigurationSupport) compiles and behaves as before; the limits are simply not applied there yet. The new splitting code is package-private.

Tests

New tests: BulkOptionsUnitTests, BulkRequestSplitterTest (count/size limits, oversized operation, order, exact size calculation, properties kept when the document is replaced), RequestConverterTest (one request without limits, refresh only on the last request, template default fallback), and integration tests for split by count/size, ids written back, failures aggregated across requests with earlier writes kept, and the template default applied to save(Iterable)/saveAll (checked with a spied client: 3 requests, refresh false, false, true).

[rursprung]

CC @sothawo & @reta

[reta]

I am wondering if we keep checkForBulkOperationFailure signature unchanged and:

		return Flux.fromIterable(bulkRequests) //
				.concatMap(bulkRequest -> client.bulk(bulkRequest) //
						.onErrorMap(e -> new UncategorizedElasticsearchException("Error executing bulk request", e))) //
				.flatMap(this::checkForBulkOperationFailure) //
				.flatMap(response -> Flux.fromIterable(response.items()));

(so we don't need to collect to list before moving on)

[rursprung]

acting as a meat proxy:

Thanks! I thought about that, but it changes the behaviour on item failures:

• checkForBulkOperationFailure returns Mono.error for the first response with item failures. concatMap then cancels, so the remaining requests are not sent.
• The BulkFailureException would only contain the failures of that one request, not of all requests.

In the issue we agreed to send all requests and report all item failures in one exception at the end. The imperative template does the same, so both behave alike.

The collectList() only keeps the BulkResponses (ids, status, versions), not the documents, so the memory cost is small compared to the requests.

If you and @sothawo prefer fail-fast (stop at the first request with item failures), I'm happy to switch. But then I'd change the imperative template too, so both stay consistent. What do you think?

[reta]

Thanks @rursprung , this is a good point, I think the behavior that would be desired here (my opinion):

• do collect all individual failures for each bulk request item (so checkForBulkOperationFailure should collect but not fail with the exception)
• fail on first bulk response error (fe 429)

[reta]

Could we make those non-nullable:

private final int maxOperations = Integer.MAX_VALUE;
private final long maxSize = Long.MAX_VALUE;

I think it will simplify a lot the split part (no need to check nulls etc)

[rursprung]

AI response:

Good point about the null checks. I've applied it to the splitter: BulkRequestSplitter#split() now takes int/long with Integer.MAX_VALUE/Long.MAX_VALUE as "no limit", so the loop has no null checks anymore.

I kept BulkOptions nullable though. null means "not set", which is needed for the template default (setBulkOptions() / bulkOptions() in the configuration). If a call doesn't set a limit, the template default is used. With MAX_VALUE as the default we couldn't tell "not set" from "explicitly no limit", so a call couldn't switch off a limit configured on the template. RequestConverter now resolves this in one place: per-call value → template default → no limit.

[reta]

I am wondering why do we need to make a copy of operation to binaryOperation (vs just using operation)?

[rursprung]

AI response:

To measure the size, the document has to be serialized. If we kept operation, the client would serialize the document a second time when sending the request. That's double the work, and it can be a lot for large documents.

BulkOperation is immutable, so we build a copy that holds the already serialized bytes (BinaryData) instead of the document object. The client writes BinaryData as it is, so each document is serialized only once. The client's own BulkIngester does the same (IngesterOperation). We can't use rebuild() because in 9.5.4 it doesn't keep id/index/routing; there's a test for that.

I added a comment in the code to explain it (88e00039).

see elastic/elasticsearch-java#1353 for the rebuild bug in elasticsearch-java

[rursprung]

since #3352 is now merged this is ready for review

[sothawo]

Nice work, thanks a lot for that. I have just one minor comment/change