[FEATURE] Adaptive maxOperations for BulkInjester

[gaobinlong]

Is your feature request related to a problem?

BulkIngester requires users to manually tune maxOperations or maxByteSize to achieve optimal indexing throughput. The optimal value depends on cluster capacity, document size, shard count, and current load — all of which vary at runtime and differ across deployments. Setting it too low leaves throughput on the table; setting it too high can cause memory pressure or trigger 429s from an overloaded cluster. Users currently have to run offline benchmarks, pick a static number, and hope it holds under production conditions.

What solution would you like?

An optional AdaptiveOperationsController that automatically tunes maxOperations at runtime by observing BulkResponse.took() after every bulk response. Users opt in with one line:

BulkIngester ingester = BulkIngester.of(b -> b
.client(client)
.adaptiveOperations(new AdaptiveOperationsController(200)) // initialOps: 200, optional
);

The controller finds the throughput-optimal maxOperations without any manual tuning. initialOps can be optional set as an initial value for maxOperations; the controller takes over from the first response onward.

Algorithm: Hill Climbing + AIMD Decrease

Two mechanisms handle two separate signals:

• Hill climbing (increase path) — after each bulk response, computes throughput = batchSize / took * 1000 (ops/s), smoothed with EWMA. Increases maxOperations by a step while throughput keeps improving. Stops at the plateau where further increases yield less than a configurable minimum gain (default 0.5%). Step size halves on each regression, enabling coarse-grained ramp-up and fine-grained search near the optimum.

• AIMD decrease (overload path) — when 429s or transport failures are detected, immediately multiplies maxOperations by a decrease factor (default 0.75) and resets the hill climb from the new lower baseline.

Controller trace:
ops=200 → improving → ops=400
ops=400 → improving → ops=600
... (fast ramp, step=initialOps)
ops=5000 → improving → ops=5200
ops=5200 → gain=0.04% < 0.5% → PLATEAU → hold at ops=5000

What alternatives have you considered?

Users do multiple rounds tests to find the optimal value for the maxOperations parameter or maxSize parameter against a cluster having same configuration with the production cluster, and then change the client code to apply that value, this should be done again when switching to another OpenSearch cluster.

[gaobinlong]

cc @reta

[reta]

@gaobinlong two things here.

First I think the adaptive bulk policy would be really useful (I would suggest to designed it as a policy, similarly to backoff fe but that is minor).

Secondly, the bulk levers include: maxConcurrentRequests, maxSize and maxOperations. All three do impact the bulk ingestion and I believe all three have to be taken into account for adaptive policy to be useful, not only operations.

Hope it makes sense.

[gaobinlong]

Thanks @reta , I meant to implement adaptive controllers for the 3 parameters one by one since they have separate signals:

• maxOperations / maxSize → driven by resp.took() (server latency)
• maxConcurrentRequests → driven by requestContentionsCount() (client-side contention)

You're right that all three levers interact and an adaptive policy should account for all of them, but we can start with maxOperations at the first step?

[reta]

You're right that all three levers interact and an adaptive policy should account for all of them, but we can start with maxOperations at the first step?

Thanks @gaobinlong , this is the gist of the problem I think, we cannot just pick one. Here are a few examples:

• assume no concurrent bulk request, maxOperations and maxSize are set (which ever triggers first): reducing maxOperations should also reduce maxSize appropriately, right?
• assume concurrent bulk request, should we reduce maxOperations or/and maxSize (from above), or reduce the concurrency first?

I think about all three as a coherent set, not separate parameters, so starting with just one does not play well here, does it make sense or I am missing something?

Thank you

[gaobinlong]

Hi @reta , I meant to do is disable either maxSize or maxOperations and tune another, like this:

BulkIngester.of(b -> b
       .client(client)
       .maxSize(-1)            // disabled — never triggers
       .adaptiveOperations(new AdaptiveController(1000)) // initial value set to 1000, and have upper bound like 100_000
   );

, when one parameter is disabled and the other is active, there is exactly one binding constraint and one knob to tune. The adaptive controller always knows what it is controlling, the signal (resp.took()) is unambiguous, and the hill climbing + AIMD logic applies directly based on a stable metric(docs/s or MB/s), and it can converge fast with an optimal value found.

When both paras are active, there are two competing thresholds — whichever fires first determines the actual batch. The adaptive controller cannot know which one triggered, so it has one signal (took) but two knobs. This is under-determined: Scaling both proportionally is one workaround, but it assumes the initial ratio between the two was already optimal, which may not be true.

For maxConcurrentRequests: this is a separate dimension with its own independent signal (client-side contention via requestContentionsCount()), so it can be adapted independently in a follow-up without interfering with the adaptive logic of maxSize/maxOperations.

Overall, I think the three parameters should be tuned by separate signals: docs/s, MB/s or contentionsCount/s, I don't know which signals can be used to tune all three parameters simultaneously, and seems it's a hard problem.

Thanks!

[reta]

Hi @reta , I meant to do is disable either maxSize or maxOperations and tune another, like this:

Ah I see, thanks @gaobinlong , let's take some time to design an API so it won't be clunky and stay intuitive / easy to use (and also open for customizations), some early ideas:

• static sizing (current approach)

BulkIngester.of(b -> b
       .client(client)
       .maxSize(1000)
   );

• adaptive sizing (new)

BulkIngester.of(b -> b
       .client(client)
       .maxSize(new AdaptiveSizing(1000, 10_000))
   );

The gist of it - having two different knobs to the control the behaviour of the same property is counter-intuive (and error prone), hope it makes sense.