CAMEL-23338: added OpenSearchClient as parameter to pass custom instance#24320
CAMEL-23338: added OpenSearchClient as parameter to pass custom instance#24320renjth-81 wants to merge 6 commits into
Conversation
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
|
you need to also rebuild camel-catalog, and some code in component and endpoint dsl |
|
There are uncommitted changes |
oscerd
left a comment
oscerd
left a comment
There was a problem hiding this comment.
Thanks for the contribution, @renjth-81 — being able to pass a custom OpenSearchClient is a useful addition, and the generated catalog/DSL files are all correctly regenerated. However, there's a blocking correctness problem in the current implementation that CI is also catching, so I'm requesting changes.
Blocker: the cached transport captures a null RestClient → NPE on every exchange
OpensearchProducer now builds and caches openSearchClient in its constructor:
- the constructor sets
this.client = endpoint.getClient();thenthis.openSearchClient = createDefaultOpenSearchClient(); createDefaultOpenSearchClient()buildsnew OpenSearchClient(new RestClientTransport(client, ...)), but only callsstartClient()whenconfiguration.isDisconnect() && client == null.
In the standard configuration (a RestClient built from hostAddresses, no autowired client), endpoint.getClient() is null at construction time — the component creates the endpoint with a null client by default (OpensearchComponent.java:91), and the real RestClient is created lazily later in doStart() → startClient() (OpensearchProducer.java, the double-checked-locking block around lines 452–467). disconnect defaults to false (OpensearchConfiguration.java:60), so createDefaultOpenSearchClient() does not start the client and ends up wrapping a RestClientTransport around a null RestClient.
process() then does OpenSearchTransport transport = openSearchClient._transport(); and reuses that cached transport, so every exchange dereferences the null RestClient:
java.lang.NullPointerException: Cannot invoke
"org.opensearch.client.RestClient.performRequestAsync(...)" because "this.restClient" is null
This is exactly what CI is failing on — build (17, false) and build (25, false) are both red, with OpensearchPingIT.testPing and all OpensearchBulkIT.* failing with this NPE. The user-supplied-client path (endpoint.getOpenSearchClient() != null) works, but the default path — which all of the component's own integration tests exercise — is broken.
Suggested direction
Resolve the transport lazily rather than in the constructor: keep using the user-supplied OpenSearchClient when one is provided, but for the default case build the transport in process() / after doStart() once client is non-null (as the original code did). Please also keep the disconnect=true semantics in mind — the old process() rebuilt the transport per call and cleanup() closes and nulls the client after each exchange (OpensearchProducer.java, around lines 437–445), so a single cached transport would also break disconnect mode after the first exchange.
Other items
- Tests: please add a test proving a custom
OpenSearchClientinstance is actually honored at runtime — the new option currently has no coverage, and the existing integration tests regressed. - Jira: apache/camel changes should be tracked by a
CAMEL-XXXXXissue with the commit/PR title prefixed accordingly. Could you create one (componentcamel-opensearch) and update the title? Theadded missing filescommit subject could also be made more descriptive.
For context: this PR currently shows as approved, but that review was on the very first commit (62b5af1), before the generated-files commit and before CI ran — so it predates the failures above.
Thanks again for working on this — happy to re-review once the transport init is made lazy and CI is green.
Reviewed with Claude Code on behalf of Andrea Cosentino (@oscerd). This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
|
creating a new client should ideally not be done in constructors but in doInit or doStart |
renjth-81
changed the title
renjth-81
changed the title
…nt' into opensearch-custom-openSearchClient
|
@davsclaus @oscerd I have pushed the fixes and integration test. Please could you review |
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 12 tested, 26 compile-only — current: 38 all testedMaveniverse Scalpel detected 38 affected modules (current approach: 38). Skip-tests mode would test 12 modules (4 direct + 8 downstream), skip tests for 26 (generated code, meta-modules) Modules Scalpel would test (12)
Modules with tests skipped (26)
All tested modules (38 modules)
|
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks for addressing the NPE blocker from the previous review, @renjth-81 — the lazy transport resolution in process() is the right approach. However, there are a couple of remaining issues that need fixing before this can be merged.
Blocker: Missing Mockito dependency
The new test CustomOpensearchClientPingIT.java uses Mockito.mock(), Mockito.when(), and RETURNS_SMART_NULLS, but components/camel-opensearch/pom.xml does not declare mockito-core as a test dependency. No transitive path brings it in either (camel-test-junit6, camel-test-infra-core, and camel-test-infra-opensearch do not depend on Mockito). The module will fail to compile.
Please add to pom.xml:
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>${mockito-version}</version>
<scope>test</scope>
</dependency>Bug: Transport closed on user-provided client when disconnect=true
In OpensearchProducer.cleanup(), IOHelper.close(ctx.transport()) is called when disconnect=true. For the default path this is safe — a fresh RestClientTransport is created per process() call. But when a custom OpenSearchClient is provided, ctx.transport() is openSearchClient._transport() — the user's own transport. Closing it will corrupt the user's client for future use.
The test exercises this path (disconnect=true + custom client), but since the mock's close() is a no-op, the bug is masked.
Suggested fix: skip closing the transport and client in cleanup() when a custom OpenSearchClient was provided.
Minor: doStart() logs spurious warning with custom client
doStart() unconditionally calls startClient() when disconnect=false. When a user provides only an OpenSearchClient (no hostAddresses), startClient() logs: "Incorrect ip address and port parameters settings for OpenSearch cluster". Consider skipping startClient() when a custom OpenSearchClient is set.
Minor: Commit hygiene
- Per project conventions, commits should be prefixed with
CAMEL-23338:. - There's a merge commit and duplicate commit messages — please squash/clean up before merge.
- A spurious blank line was added in
OpensearchProducer.javabetweenInteger from = message.getHeader(...)andif (from == null)— please remove.
Note: this review does not replace specialized AI review tools (CodeRabbit, Sourcery) or static analysis (SonarCloud).
This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
|
||
| import org.apache.camel.builder.RouteBuilder; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.mockito.Mockito; |
There was a problem hiding this comment.
Missing dependency: mockito-core is not declared as a test dependency in components/camel-opensearch/pom.xml and is not transitively available. This import (and all Mockito usage in this file) will cause a compilation failure.
Please add to pom.xml:
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>${mockito-version}</version>
<scope>test</scope>
</dependency>| final ObjectMapper mapper = new ObjectMapper(); | ||
| mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false); | ||
| OpenSearchTransport transport = new RestClientTransport(client, new JacksonJsonpMapper(mapper)); | ||
| // 2. Index and type will be set by: |
There was a problem hiding this comment.
Transport lifecycle bug with disconnect=true: When a custom OpenSearchClient is provided and disconnect=true, cleanup() (line ~438) calls IOHelper.close(ctx.transport()). In this branch, transport is openSearchClient._transport() — the user's own transport. Closing it will corrupt the user's client.
The default path creates a fresh RestClientTransport per call, so closing it is safe. But the custom client's transport should not be closed by the component.
Suggested fix: pass a flag into ActionContext indicating whether the transport is user-owned, and skip IOHelper.close(ctx.transport()) in cleanup() when it is.
| @@ -188,6 +195,7 @@ public boolean process(Exchange exchange, AsyncCallback callback) { | |||
| } | |||
There was a problem hiding this comment.
Nit: this blank line was not present before and is not needed — please remove to keep the diff clean.
3 participants
Description
Added OpenSearchClient as a parameter to allow custom OpenSearchClient instance to be passed.
Target
mainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.