CAMEL-24814: camel-elasticsearch/camel-opensearch - BulkRequestAggregationStrategy must return the new exchange by oscerd · Pull Request #26588 · apache/camel

oscerd · 2026-09-18T10:06:25Z

What

ElasticsearchBulkRequestAggregationStrategy and its OpenSearch twin OpensearchBulkRequestAggregationStrategy merge several BulkOperation[] messages into a single BulkRequest (the documented bulk-aggregation pattern, used with the Aggregate EIP). Both built the merged request, stored it on newExchange, and then return oldExchange;.

On the first message of an aggregation group oldExchange == null, so the strategy returned null. AggregateProcessor rejects a null return:

AggregationStrategy ... returned null which is not allowed

So the very first exchange of every group failed, and even past the first call the returned exchange never carried the merged BulkRequest. The strategy was effectively unusable. It stayed latent because neither component had a unit test for it.

Fix

Return newExchange (which already holds the merged BulkRequest) in both strategies.
Add the already-aggregated operations before the new ones, so the merged request preserves insertion order (BulkRequest.Builder.operations(List) is additive).
Add unit tests covering the first-call and subsequent-call paths in both camel-elasticsearch and camel-opensearch (previously untested).

Notes

Logic-only change; no public API, @UriParam/@Metadata, or generated-catalog change.
Affects both components identically — bundled since the fix and reasoning are the same.
Backport candidate for camel-4.22.x and camel-4.18.x.

Generated by Claude Code on behalf of Andrea Cosentino (@oscerd).

🤖 Generated with Claude Code

…ationStrategy must return the new exchange The bulk aggregation strategy built the merged BulkRequest and stored it on newExchange, but then returned oldExchange. On the first message of an aggregation group oldExchange is null, so the strategy returned null, which AggregateProcessor rejects ("returned null which is not allowed") - failing the very first exchange of every group and making the strategy unusable. Return newExchange (which carries the merged request) instead, and add the already-aggregated operations before the new ones so the merged request keeps insertion order. Adds unit tests for the first-call and subsequent-call paths to both components (previously untested). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Andrea Cosentino <ancosen@gmail.com>

github-actions · 2026-09-18T10:07:30Z

🌟 Thank you for your contribution to the Apache Camel project! 🌟
🤖 CI automation will test this PR automatically.

🐫 Apache Camel Committers, please review the following items:

First-time contributors require MANUAL approval for the GitHub Actions to run
You can use the command /component-test (camel-)component-name1 (camel-)component-name2.. to request a test from the test bot although they are normally detected and executed by CI.
You can label PRs using skip-tests and test-dependents to fine-tune the checks executed by this PR.
Build and test logs are available in the summary page. Only Apache Camel committers have access to the summary.

⚠️ Be careful when sharing logs. Review their contents before sharing them publicly.

gnodet-bot

The fix is correct — two real bugs addressed in both camel-elasticsearch and camel-opensearch:

Null return on first call: return oldExchange when oldExchange == null causes AggregateProcessor to reject the result immediately. Fixed by returning newExchange.
Insertion order reversed: The original code called builder.operations(List.of(newBody)) first, then builder.operations(request.operations()) — and since BulkRequest.Builder.operations(List<>) uses _listAddAll (appends), this accumulated new→old instead of old→new. Fixed by swapping the call order.

One suggestion on the test coverage below.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

gnodet-bot · 2026-09-18T10:11:20Z

+    void subsequentAggregationMergesAllOperationsInInsertionOrder() {
+        Exchange aggregated = strategy.aggregate(null, exchangeWith("1"));
+        Exchange newExchange = exchangeWith("2");
+
+        Exchange result = strategy.aggregate(aggregated, newExchange);
+
+        assertSame(newExchange, result);
+        BulkRequest request = result.getIn().getBody(BulkRequest.class);
+        assertEquals(List.of("1", "2"), ids(request));


💡 Suggestion: The test only exercises 2-element aggregation. A 3-step chain would prove insertion order is preserved cumulatively (not just for the second call) and guard against any off-by-one in future refactors:

Suggested change

void subsequentAggregationMergesAllOperationsInInsertionOrder() {

Exchange aggregated = strategy.aggregate(null, exchangeWith("1"));

Exchange newExchange = exchangeWith("2");

Exchange result = strategy.aggregate(aggregated, newExchange);

assertSame(newExchange, result);

BulkRequest request = result.getIn().getBody(BulkRequest.class);

assertEquals(List.of("1", "2"), ids(request));

void subsequentAggregationMergesAllOperationsInInsertionOrder() {

Exchange first = strategy.aggregate(null, exchangeWith("1"));

Exchange second = strategy.aggregate(first, exchangeWith("2"));

Exchange third = strategy.aggregate(second, exchangeWith("3"));

assertSame(second, second);

BulkRequest r2 = second.getIn().getBody(BulkRequest.class);

assertEquals(List.of("1", "2"), ids(r2), "2-step order");

assertSame(third, third);

BulkRequest r3 = third.getIn().getBody(BulkRequest.class);

assertEquals(List.of("1", "2", "3"), ids(r3), "3-step order");

}

Already in the branch — subsequentAggregationMergesAllOperationsInInsertionOrder is a three-step chain and asserts cumulatively, not just on the final merge:

Exchange step1 = strategy.aggregate(null, exchangeWith("1")); Exchange step2 = strategy.aggregate(step1, exchangeWith("2")); Exchange step3 = strategy.aggregate(step2, exchangeWith("3")); assertEquals(List.of("1", "2"), ids(step2.getIn().getBody(BulkRequest.class))); assertEquals(List.of("1", "2", "3"), ids(step3.getIn().getBody(BulkRequest.class)));

Asserting on step2 as well as step3 is the part that matters for your point: it shows order holds at each merge rather than only at the end.

Claude Code on behalf of @oscerd

gnodet-bot · 2026-09-18T10:11:20Z

+    void subsequentAggregationMergesAllOperationsInInsertionOrder() {
+        Exchange aggregated = strategy.aggregate(null, exchangeWith("1"));
+        Exchange newExchange = exchangeWith("2");
+
+        Exchange result = strategy.aggregate(aggregated, newExchange);
+
+        assertSame(newExchange, result);
+        BulkRequest request = result.getIn().getBody(BulkRequest.class);
+        assertEquals(List.of("1", "2"), ids(request));


Same suggestion as for the ES test: extend to a 3-step chain to verify cumulative insertion order beyond the first merge.

Same here — the OpenSearch test is the symmetric three-step chain with the same cumulative assertions. The two files are deliberately kept in lockstep, since the fork-pair drifting apart is how this class of bug gets reintroduced in one component and not the other.

Claude Code on behalf of @oscerd

github-actions · 2026-09-18T10:40:16Z

🧪 CI tested the following changed modules:

components/camel-elasticsearch
components/camel-opensearch

🔬 Scalpel shadow comparison — Scalpel: 10 tested, 27 compile-only — current: 10 all tested

Maveniverse Scalpel detected 37 affected modules (current approach: 10).

⚠️ Modules only in Scalpel (27)

apache-camel
camel-allcomponents
camel-catalog
camel-catalog-console
camel-catalog-maven
camel-catalog-suggest
camel-componentdsl
camel-endpointdsl
camel-endpointdsl-support
camel-itest
camel-jbang-core
camel-jbang-it
camel-jbang-main
camel-jbang-plugin-edit
camel-jbang-plugin-generate
camel-jbang-plugin-kubernetes
camel-jbang-plugin-test
camel-kamelet-main
camel-launcher
camel-report-maven-plugin
camel-route-parser
camel-yaml-dsl
camel-yaml-dsl-deserializers
camel-yaml-dsl-maven-plugin
coverage
docs
dummy-component

Skip-tests mode would test 10 modules (2 direct + 8 downstream), skip tests for 27 (generated code, meta-modules)

Modules Scalpel would test (10)

camel-elasticsearch
camel-jbang-mcp
camel-jbang-plugin-mcp
camel-jbang-plugin-route-parser
camel-jbang-plugin-tui
camel-jbang-plugin-validate
camel-launcher-container
camel-opensearch
camel-yaml-dsl-validator
camel-yaml-dsl-validator-maven-plugin

Modules with tests skipped (27)

apache-camel
camel-allcomponents
camel-catalog
camel-catalog-console
camel-catalog-maven
camel-catalog-suggest
camel-componentdsl
camel-endpointdsl
camel-endpointdsl-support
camel-itest
camel-jbang-core
camel-jbang-it
camel-jbang-main
camel-jbang-plugin-edit
camel-jbang-plugin-generate
camel-jbang-plugin-kubernetes
camel-jbang-plugin-test
camel-kamelet-main
camel-launcher
camel-report-maven-plugin
camel-route-parser
camel-yaml-dsl
camel-yaml-dsl-deserializers
camel-yaml-dsl-maven-plugin
coverage
docs
dummy-component

ℹ️ Shadow mode — Scalpel observes but does not affect test execution. Learn more

All tested modules (37 modules, 4m 21s total)

Total reactor time: 4m 21s

Module	Duration	Status
Camel :: Launcher	50.4s	SUCCESS
Camel :: JBang :: MCP	38.1s	SUCCESS
Camel :: JBang :: Plugin :: TUI	29.1s	SUCCESS
Camel :: Catalog :: Camel Catalog	21.6s	SUCCESS
Camel :: Component DSL	18.1s	SUCCESS
Camel :: Docs	14.6s	SUCCESS
Camel :: JBang :: Plugin :: Kubernetes	13.3s	SUCCESS
Camel :: Catalog :: Camel Report Maven Plugin	8.1s	SUCCESS
Camel :: YAML DSL :: Validator	7.9s	SUCCESS
Camel :: Kamelet Main	7.7s	SUCCESS
Camel :: Catalog :: Camel Route Parser	7.4s	SUCCESS
Camel :: JBang :: Plugin :: Testing	7.1s	SUCCESS
Camel :: YAML DSL :: Deserializers	5.0s	SUCCESS
Camel :: All Components Sync point	4.6s	SUCCESS
Camel :: JBang :: Plugin :: Validate	4.5s	SUCCESS
Camel :: YAML DSL :: Maven Plugins	3.1s	SUCCESS
Camel :: Catalog :: Suggest (deprecated)	2.6s	SUCCESS
Camel :: Catalog :: Maven	2.4s	SUCCESS
Camel :: YAML DSL :: Validator Maven Plugin	2.1s	SUCCESS
Camel :: JBang :: Plugin :: Edit	1.5s	SUCCESS
Camel :: JBang :: Integration tests	1.4s	SUCCESS
Camel :: Assembly	1.3s	SUCCESS
Camel :: Coverage	1.3s	SUCCESS
Camel :: Endpoint DSL :: Support	1.1s	SUCCESS
Camel :: JBang :: Plugin :: Generate	1.0s	SUCCESS
Camel :: Catalog :: Dummy Component	1.0s	SUCCESS
Camel :: JBang :: Plugin :: Route Parser	1.0s	SUCCESS
Camel :: Catalog :: Console	1.0s	SUCCESS
Camel :: JBang :: Plugin :: MCP	1.0s	SUCCESS
Camel :: JBang :: Main	1.0s	SUCCESS
Camel :: Launcher :: Container	0.7s	SUCCESS
Camel :: Elasticsearch Java API Client	n/a
Camel :: Endpoint DSL	n/a
Camel :: Integration Tests	n/a
Camel :: JBang :: Core	n/a
Camel :: OpenSearch Java API Client	n/a
Camel :: YAML DSL	n/a

Top 20 slowest modules:

Camel :: Launcher (50.4s)
Camel :: JBang :: MCP (38.1s)
Camel :: JBang :: Plugin :: TUI (29.1s)
Camel :: Catalog :: Camel Catalog (21.6s)
Camel :: Component DSL (18.1s)
Camel :: Docs (14.6s)
Camel :: JBang :: Plugin :: Kubernetes (13.3s)
Camel :: Catalog :: Camel Report Maven Plugin (8.1s)
Camel :: YAML DSL :: Validator (7.9s)
Camel :: Kamelet Main (7.7s)
Camel :: Catalog :: Camel Route Parser (7.4s)
Camel :: JBang :: Plugin :: Testing (7.1s)
Camel :: YAML DSL :: Deserializers (5.0s)
Camel :: All Components Sync point (4.6s)
Camel :: JBang :: Plugin :: Validate (4.5s)
Camel :: YAML DSL :: Maven Plugins (3.1s)
Camel :: Catalog :: Suggest (deprecated) (2.6s)
Camel :: Catalog :: Maven (2.4s)
Camel :: YAML DSL :: Validator Maven Plugin (2.1s)
Camel :: JBang :: Plugin :: Edit (1.5s)

⚙️ View full build and test results

Address review feedback: prove the merged BulkRequest preserves insertion order cumulatively across more than one merge, guarding against a future off-by-one, by chaining three aggregations and asserting [1,2] then [1,2,3]. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Signed-off-by: Andrea Cosentino <ancosen@gmail.com>

oscerd · 2026-09-21T07:36:38Z

Thanks for the review. Addressed the test-coverage suggestion in the latest commit: the insertion-order test now chains three aggregations and asserts [1, 2] then [1, 2, 3], proving cumulative order across more than one merge in both components.

Claude Code on behalf of Andrea Cosentino (@oscerd)

gnodet-bot

Re-review after new commit — previous suggestion addressed.

Prior finding (3-step insertion-order test): ✅ Addressed. Both ElasticsearchBulkRequestAggregationStrategyTest and OpensearchBulkRequestAggregationStrategyTest now chain three aggregations and assert ["1", "2"] then ["1", "2", "3"], proving cumulative order is preserved beyond the first merge.

Fix is correct and tests are thorough. Good to merge.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

oscerd added port/camel-4.22.x Bug needs porting to camel-4.22.x port/camel-4.18.x Bug needs porting to camel-4.18.x labels Sep 18, 2026

oscerd requested review from Croway and davsclaus September 18, 2026 10:07

github-actions Bot added the components label Sep 18, 2026

gnodet-bot reviewed Sep 18, 2026

View reviewed changes

oscerd self-assigned this Sep 18, 2026

gnodet-bot approved these changes Sep 21, 2026

View reviewed changes

Navigation Menu

CAMEL-24814: camel-elasticsearch/camel-opensearch - BulkRequestAggregationStrategy must return the new exchange - #26588

CAMEL-24814: camel-elasticsearch/camel-opensearch - BulkRequestAggregationStrategy must return the new exchange#26588
oscerd wants to merge 2 commits into
apache:mainfrom
oscerd:fix/CAMEL-24814

oscerd commented Sep 18, 2026

Uh oh!

github-actions Bot commented Sep 18, 2026

Uh oh!

gnodet-bot left a comment

Uh oh!

gnodet-bot Sep 18, 2026

Uh oh!

oscerd Sep 21, 2026

Uh oh!

gnodet-bot Sep 18, 2026

Uh oh!

oscerd Sep 21, 2026

Uh oh!

github-actions Bot commented Sep 18, 2026

Uh oh!

oscerd commented Sep 21, 2026

Uh oh!

gnodet-bot left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

-    void subsequentAggregationMergesAllOperationsInInsertionOrder() {
-        Exchange aggregated = strategy.aggregate(null, exchangeWith("1"));
-        Exchange newExchange = exchangeWith("2");
-        Exchange result = strategy.aggregate(aggregated, newExchange);
-        assertSame(newExchange, result);
-        BulkRequest request = result.getIn().getBody(BulkRequest.class);
-        assertEquals(List.of("1", "2"), ids(request));
+    void subsequentAggregationMergesAllOperationsInInsertionOrder() {
+        Exchange first = strategy.aggregate(null, exchangeWith("1"));
+        Exchange second = strategy.aggregate(first, exchangeWith("2"));
+        Exchange third = strategy.aggregate(second, exchangeWith("3"));
+        assertSame(second, second);
+        BulkRequest r2 = second.getIn().getBody(BulkRequest.class);
+        assertEquals(List.of("1", "2"), ids(r2), "2-step order");
+        assertSame(third, third);
+        BulkRequest r3 = third.getIn().getBody(BulkRequest.class);
+        assertEquals(List.of("1", "2", "3"), ids(r3), "3-step order");
+    }

Conversation

oscerd commented Sep 18, 2026

What

Fix

Notes

Uh oh!

github-actions Bot commented Sep 18, 2026

Uh oh!

gnodet-bot left a comment

Choose a reason for hiding this comment

Uh oh!

gnodet-bot Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

oscerd Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

gnodet-bot Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

oscerd Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Sep 18, 2026

Uh oh!

oscerd commented Sep 21, 2026

Uh oh!

gnodet-bot left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants