This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch feature/CAMEL-23338-opensearch-custom-client in repository https://gitbox.apache.org/repos/asf/camel.git
commit 1b2c6286610089efd4a5fa7454ffa4e1c672e925 Author: Claus Ibsen <[email protected]> AuthorDate: Tue Jul 21 14:39:09 2026 +0200 CAMEL-23338: camel-opensearch - Add OpenSearchClient as component option Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/catalog/components/opensearch.json | 15 +++++++------- .../opensearch/OpensearchComponentConfigurer.java | 8 +++++++- .../camel/component/opensearch/opensearch.json | 15 +++++++------- .../src/main/docs/opensearch-component.adoc | 19 ++++++++++++++++++ .../component/opensearch/OpensearchComponent.java | 17 +++++++++++++++- .../component/opensearch/OpensearchEndpoint.java | 13 +++++++++++- .../component/opensearch/OpensearchProducer.java | 23 ++++++++++++++-------- .../dsl/OpensearchComponentBuilderFactory.java | 18 +++++++++++++++++ 8 files changed, 103 insertions(+), 25 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/opensearch.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/opensearch.json index 0cfb8f1dc670..f9ebde3ecace 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/opensearch.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/opensearch.json @@ -32,13 +32,14 @@ "autowiredEnabled": { "index": 5, "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching t [...] "client": { "index": 6, "kind": "property", "displayName": "Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.opensearch.client.RestClient", "deprecated": false, "autowired": true, "secret": false, "description": "To use an existing configured OpenSearch client, instead of creating a client per endpoint. This allows customizing the client with specific settings." }, "enableSniffer": { "index": 7, "kind": "property", "displayName": "Enable Sniffer", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable automatically discover nodes from a running OpenSearch cluster. If this option is used in conjunction with Spring Boot, then it's managed by the Spring Boot configuration (see: Disable Sniffer in S [...] - "sniffAfterFailureDelay": { "index": 8, "kind": "property", "displayName": "Sniff After Failure Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 60000, "description": "The delay of a sniff execution scheduled after a failure (in milliseconds)" }, - "snifferInterval": { "index": 9, "kind": "property", "displayName": "Sniffer Interval", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 300000, "description": "The interval between consecutive ordinary sniff executions in milliseconds. Will be honoured when sniffOnFailure is disabled or when there are no failures between consecutive sniff executions" }, - "enableSSL": { "index": 10, "kind": "property", "displayName": "Enable SSL", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable SSL" }, - "password": { "index": 11, "kind": "property", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Password for authenticating" }, - "sslContextParameters": { "index": 12, "kind": "property", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "autowired": false, "secret": false, "description": "To configure security using SSLContextParameters. When configured, this takes precedence over the certificatePath option." }, - "useGlobalSslContextParameters": { "index": 13, "kind": "property", "displayName": "Use Global Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable usage of global SSL context parameters." }, - "user": { "index": 14, "kind": "property", "displayName": "User", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Basic authenticate user" } + "openSearchClient": { "index": 8, "kind": "property", "displayName": "Open Search Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.opensearch.client.opensearch.OpenSearchClient", "deprecated": false, "autowired": true, "secret": false, "description": "To use a custom configured OpenSearchClient instance. When set, this takes precedence over the RestClient-based client option." }, + "sniffAfterFailureDelay": { "index": 9, "kind": "property", "displayName": "Sniff After Failure Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 60000, "description": "The delay of a sniff execution scheduled after a failure (in milliseconds)" }, + "snifferInterval": { "index": 10, "kind": "property", "displayName": "Sniffer Interval", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 300000, "description": "The interval between consecutive ordinary sniff executions in milliseconds. Will be honoured when sniffOnFailure is disabled or when there are no failures between consecutive sniff executions" }, + "enableSSL": { "index": 11, "kind": "property", "displayName": "Enable SSL", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable SSL" }, + "password": { "index": 12, "kind": "property", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Password for authenticating" }, + "sslContextParameters": { "index": 13, "kind": "property", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "autowired": false, "secret": false, "description": "To configure security using SSLContextParameters. When configured, this takes precedence over the certificatePath option." }, + "useGlobalSslContextParameters": { "index": 14, "kind": "property", "displayName": "Use Global Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable usage of global SSL context parameters." }, + "user": { "index": 15, "kind": "property", "displayName": "User", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Basic authenticate user" } }, "headers": { "CamelOpensearchOperation": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "org.apache.camel.component.opensearch.OpensearchOperation", "enum": [ "Index", "Update", "Bulk", "GetById", "MultiGet", "MultiSearch", "Delete", "DeleteIndex", "Search", "Exists", "Ping" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The operation to perform", "constantName": "org.apache.c [...] diff --git a/components/camel-opensearch/src/generated/java/org/apache/camel/component/opensearch/OpensearchComponentConfigurer.java b/components/camel-opensearch/src/generated/java/org/apache/camel/component/opensearch/OpensearchComponentConfigurer.java index 17188936c36b..a586671bd3d6 100644 --- a/components/camel-opensearch/src/generated/java/org/apache/camel/component/opensearch/OpensearchComponentConfigurer.java +++ b/components/camel-opensearch/src/generated/java/org/apache/camel/component/opensearch/OpensearchComponentConfigurer.java @@ -38,6 +38,8 @@ public class OpensearchComponentConfigurer extends PropertyConfigurerSupport imp case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true; case "maxretrytimeout": case "maxRetryTimeout": target.setMaxRetryTimeout(property(camelContext, int.class, value)); return true; + case "opensearchclient": + case "openSearchClient": target.setOpenSearchClient(property(camelContext, org.opensearch.client.opensearch.OpenSearchClient.class, value)); return true; case "password": target.setPassword(property(camelContext, java.lang.String.class, value)); return true; case "sniffafterfailuredelay": case "sniffAfterFailureDelay": target.setSniffAfterFailureDelay(property(camelContext, int.class, value)); return true; @@ -56,7 +58,7 @@ public class OpensearchComponentConfigurer extends PropertyConfigurerSupport imp @Override public String[] getAutowiredNames() { - return new String[]{"client"}; + return new String[]{"client", "openSearchClient"}; } @Override @@ -77,6 +79,8 @@ public class OpensearchComponentConfigurer extends PropertyConfigurerSupport imp case "lazyStartProducer": return boolean.class; case "maxretrytimeout": case "maxRetryTimeout": return int.class; + case "opensearchclient": + case "openSearchClient": return org.opensearch.client.opensearch.OpenSearchClient.class; case "password": return java.lang.String.class; case "sniffafterfailuredelay": case "sniffAfterFailureDelay": return int.class; @@ -112,6 +116,8 @@ public class OpensearchComponentConfigurer extends PropertyConfigurerSupport imp case "lazyStartProducer": return target.isLazyStartProducer(); case "maxretrytimeout": case "maxRetryTimeout": return target.getMaxRetryTimeout(); + case "opensearchclient": + case "openSearchClient": return target.getOpenSearchClient(); case "password": return target.getPassword(); case "sniffafterfailuredelay": case "sniffAfterFailureDelay": return target.getSniffAfterFailureDelay(); diff --git a/components/camel-opensearch/src/generated/resources/META-INF/org/apache/camel/component/opensearch/opensearch.json b/components/camel-opensearch/src/generated/resources/META-INF/org/apache/camel/component/opensearch/opensearch.json index 0cfb8f1dc670..f9ebde3ecace 100644 --- a/components/camel-opensearch/src/generated/resources/META-INF/org/apache/camel/component/opensearch/opensearch.json +++ b/components/camel-opensearch/src/generated/resources/META-INF/org/apache/camel/component/opensearch/opensearch.json @@ -32,13 +32,14 @@ "autowiredEnabled": { "index": 5, "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching t [...] "client": { "index": 6, "kind": "property", "displayName": "Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.opensearch.client.RestClient", "deprecated": false, "autowired": true, "secret": false, "description": "To use an existing configured OpenSearch client, instead of creating a client per endpoint. This allows customizing the client with specific settings." }, "enableSniffer": { "index": 7, "kind": "property", "displayName": "Enable Sniffer", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable automatically discover nodes from a running OpenSearch cluster. If this option is used in conjunction with Spring Boot, then it's managed by the Spring Boot configuration (see: Disable Sniffer in S [...] - "sniffAfterFailureDelay": { "index": 8, "kind": "property", "displayName": "Sniff After Failure Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 60000, "description": "The delay of a sniff execution scheduled after a failure (in milliseconds)" }, - "snifferInterval": { "index": 9, "kind": "property", "displayName": "Sniffer Interval", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 300000, "description": "The interval between consecutive ordinary sniff executions in milliseconds. Will be honoured when sniffOnFailure is disabled or when there are no failures between consecutive sniff executions" }, - "enableSSL": { "index": 10, "kind": "property", "displayName": "Enable SSL", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable SSL" }, - "password": { "index": 11, "kind": "property", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Password for authenticating" }, - "sslContextParameters": { "index": 12, "kind": "property", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "autowired": false, "secret": false, "description": "To configure security using SSLContextParameters. When configured, this takes precedence over the certificatePath option." }, - "useGlobalSslContextParameters": { "index": 13, "kind": "property", "displayName": "Use Global Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable usage of global SSL context parameters." }, - "user": { "index": 14, "kind": "property", "displayName": "User", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Basic authenticate user" } + "openSearchClient": { "index": 8, "kind": "property", "displayName": "Open Search Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.opensearch.client.opensearch.OpenSearchClient", "deprecated": false, "autowired": true, "secret": false, "description": "To use a custom configured OpenSearchClient instance. When set, this takes precedence over the RestClient-based client option." }, + "sniffAfterFailureDelay": { "index": 9, "kind": "property", "displayName": "Sniff After Failure Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 60000, "description": "The delay of a sniff execution scheduled after a failure (in milliseconds)" }, + "snifferInterval": { "index": 10, "kind": "property", "displayName": "Sniffer Interval", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 300000, "description": "The interval between consecutive ordinary sniff executions in milliseconds. Will be honoured when sniffOnFailure is disabled or when there are no failures between consecutive sniff executions" }, + "enableSSL": { "index": 11, "kind": "property", "displayName": "Enable SSL", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable SSL" }, + "password": { "index": 12, "kind": "property", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Password for authenticating" }, + "sslContextParameters": { "index": 13, "kind": "property", "displayName": "Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "object", "javaType": "org.apache.camel.support.jsse.SSLContextParameters", "deprecated": false, "autowired": false, "secret": false, "description": "To configure security using SSLContextParameters. When configured, this takes precedence over the certificatePath option." }, + "useGlobalSslContextParameters": { "index": 14, "kind": "property", "displayName": "Use Global Ssl Context Parameters", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable usage of global SSL context parameters." }, + "user": { "index": 15, "kind": "property", "displayName": "User", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "security": "secret", "description": "Basic authenticate user" } }, "headers": { "CamelOpensearchOperation": { "index": 0, "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "org.apache.camel.component.opensearch.OpensearchOperation", "enum": [ "Index", "Update", "Bulk", "GetById", "MultiGet", "MultiSearch", "Delete", "DeleteIndex", "Search", "Exists", "Ping" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The operation to perform", "constantName": "org.apache.c [...] diff --git a/components/camel-opensearch/src/main/docs/opensearch-component.adoc b/components/camel-opensearch/src/main/docs/opensearch-component.adoc index cacdb0cefdb3..9ca368bb4eff 100644 --- a/components/camel-opensearch/src/main/docs/opensearch-component.adoc +++ b/components/camel-opensearch/src/main/docs/opensearch-component.adoc @@ -406,3 +406,22 @@ MsearchRequest.Builder builder = new MsearchRequest.Builder().index("twitter").s .body(new MultisearchBody.Builder().query(b -> b.matchAll(x -> x)).build()).build()); List<MultiSearchResponseItem<?>> response = template.requestBody("direct:multiSearch", builder, List.class); ---- + +=== Custom OpenSearchClient Example + +You can provide a pre-configured `OpenSearchClient` instance instead of letting the component +build one from host addresses. This is useful when you need a custom transport, for example +when connecting to AWS OpenSearch Service with IAM-based authentication. + +._Java-only: programmatic CamelContext configuration_ +[source,java] +---- +// Build a custom OpenSearchClient (e.g., with AWS SDK transport) +OpenSearchClient customClient = new OpenSearchClient(myCustomTransport); + +// Bind it to the registry so it can be autowired +context.getRegistry().bind("myOpenSearchClient", customClient); +---- + +The component will automatically pick up the `OpenSearchClient` from the registry via autowiring. +When both a `RestClient` and an `OpenSearchClient` are available, the `OpenSearchClient` takes precedence. diff --git a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchComponent.java b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchComponent.java index e4e69240c5a5..4fdcff614a65 100644 --- a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchComponent.java +++ b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchComponent.java @@ -30,6 +30,7 @@ import org.apache.camel.support.DefaultComponent; import org.apache.camel.support.jsse.SSLContextParameters; import org.apache.hc.core5.http.HttpHost; import org.opensearch.client.RestClient; +import org.opensearch.client.opensearch.OpenSearchClient; /** * Represents the component that manages {@link OpensearchEndpoint}. @@ -39,6 +40,8 @@ public class OpensearchComponent extends DefaultComponent implements SSLContextP @Metadata(label = "advanced", autowired = true) private RestClient client; + @Metadata(label = "advanced", autowired = true) + private OpenSearchClient openSearchClient; @Metadata private String hostAddresses; @Metadata(defaultValue = "" + OpensearchConstants.DEFAULT_SOCKET_TIMEOUT) @@ -88,7 +91,7 @@ public class OpensearchComponent extends DefaultComponent implements SSLContextP config.setClusterName(remaining); config.setSslContextParameters(this.getSslContextParameters()); - Endpoint endpoint = new OpensearchEndpoint(uri, this, config, client); + Endpoint endpoint = new OpensearchEndpoint(uri, this, config, client, openSearchClient); setProperties(endpoint, parameters); if (config.getSslContextParameters() == null) { config.setSslContextParameters(retrieveGlobalSslContextParameters()); @@ -130,6 +133,18 @@ public class OpensearchComponent extends DefaultComponent implements SSLContextP this.client = client; } + public OpenSearchClient getOpenSearchClient() { + return openSearchClient; + } + + /** + * To use a custom configured OpenSearchClient instance. When set, this takes precedence over the RestClient-based + * client option. + */ + public void setOpenSearchClient(OpenSearchClient openSearchClient) { + this.openSearchClient = openSearchClient; + } + /** * Comma separated list with ip:port formatted remote transport addresses to use. The ip and port options must be * left blank for hostAddresses to be considered instead. diff --git a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchEndpoint.java b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchEndpoint.java index f1fbc1d20125..ae0a484843f3 100644 --- a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchEndpoint.java +++ b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchEndpoint.java @@ -25,6 +25,7 @@ import org.apache.camel.spi.UriEndpoint; import org.apache.camel.spi.UriParam; import org.apache.camel.support.DefaultEndpoint; import org.opensearch.client.RestClient; +import org.opensearch.client.opensearch.OpenSearchClient; /** * Send requests to OpenSearch via Java Client API. @@ -38,12 +39,14 @@ public class OpensearchEndpoint extends DefaultEndpoint implements EndpointServi private final OpensearchConfiguration configuration; private final RestClient client; + private OpenSearchClient openSearchClient; public OpensearchEndpoint(String uri, OpensearchComponent component, OpensearchConfiguration config, - RestClient client) { + RestClient client, OpenSearchClient openSearchClient) { super(uri, component); this.configuration = config; this.client = client; + this.openSearchClient = openSearchClient; } public OpensearchConfiguration getConfiguration() { @@ -64,6 +67,14 @@ public class OpensearchEndpoint extends DefaultEndpoint implements EndpointServi return client; } + public OpenSearchClient getOpenSearchClient() { + return openSearchClient; + } + + public void setOpenSearchClient(OpenSearchClient openSearchClient) { + this.openSearchClient = openSearchClient; + } + @Override public String getServiceUrl() { return configuration.getHostAddresses(); diff --git a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java index f3570d6cb9fb..c25efe389fa2 100644 --- a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java +++ b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchProducer.java @@ -89,6 +89,7 @@ class OpensearchProducer extends DefaultAsyncProducer { private static final Logger LOG = LoggerFactory.getLogger(OpensearchProducer.class); protected final OpensearchConfiguration configuration; + private final OpenSearchClient openSearchClient; private volatile RestClient client; private Sniffer sniffer; @@ -96,6 +97,7 @@ class OpensearchProducer extends DefaultAsyncProducer { super(endpoint); this.configuration = configuration; this.client = endpoint.getClient(); + this.openSearchClient = endpoint.getOpenSearchClient(); } private OpensearchOperation resolveOperation(Exchange exchange) { @@ -154,12 +156,17 @@ class OpensearchProducer extends DefaultAsyncProducer { @Override public boolean process(Exchange exchange, AsyncCallback callback) { try { - if (configuration.isDisconnect() && client == null) { - startClient(); + OpenSearchTransport transport; + if (openSearchClient != null) { + transport = openSearchClient._transport(); + } else { + if (configuration.isDisconnect() && client == null) { + startClient(); + } + final ObjectMapper mapper = new ObjectMapper(); + mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false); + transport = new RestClientTransport(client, new JacksonJsonpMapper(mapper)); } - 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: // a. If the incoming body is already an action request // b. If the body is not an action request we will use headers if they @@ -434,7 +441,7 @@ class OpensearchProducer extends DefaultAsyncProducer { if (ctx.configWaitForActiveShards()) { message.removeHeader(OpensearchConstants.PARAM_WAIT_FOR_ACTIVE_SHARDS); } - if (configuration.isDisconnect()) { + if (configuration.isDisconnect() && openSearchClient == null) { IOHelper.close(ctx.transport()); if (configuration.isEnableSniffer()) { IOHelper.close(sniffer); @@ -451,7 +458,7 @@ class OpensearchProducer extends DefaultAsyncProducer { @Override protected void doStart() throws Exception { super.doStart(); - if (!configuration.isDisconnect()) { + if (!configuration.isDisconnect() && openSearchClient == null) { startClient(); } } @@ -548,7 +555,7 @@ class OpensearchProducer extends DefaultAsyncProducer { @Override protected void doStop() throws Exception { - if (client != null) { + if (client != null && openSearchClient == null) { LOG.info("Disconnecting from OpenSearch cluster: {}", configuration.getClusterName()); client.close(); if (sniffer != null) { diff --git a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/OpensearchComponentBuilderFactory.java b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/OpensearchComponentBuilderFactory.java index 081561ea90ac..88ac7e6ed78b 100644 --- a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/OpensearchComponentBuilderFactory.java +++ b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/OpensearchComponentBuilderFactory.java @@ -203,6 +203,23 @@ public interface OpensearchComponentBuilderFactory { return this; } + /** + * To use a custom configured OpenSearchClient instance. When set, this + * takes precedence over the RestClient-based client option. + * + * The option is a: + * <code>org.opensearch.client.opensearch.OpenSearchClient</code> type. + * + * Group: advanced + * + * @param openSearchClient the value to set + * @return the dsl builder + */ + default OpensearchComponentBuilder openSearchClient(org.opensearch.client.opensearch.OpenSearchClient openSearchClient) { + doSetProperty("openSearchClient", openSearchClient); + return this; + } + /** * The delay of a sniff execution scheduled after a failure (in @@ -343,6 +360,7 @@ public interface OpensearchComponentBuilderFactory { case "autowiredEnabled": ((OpensearchComponent) component).setAutowiredEnabled((boolean) value); return true; case "client": ((OpensearchComponent) component).setClient((org.opensearch.client.RestClient) value); return true; case "enableSniffer": ((OpensearchComponent) component).setEnableSniffer((boolean) value); return true; + case "openSearchClient": ((OpensearchComponent) component).setOpenSearchClient((org.opensearch.client.opensearch.OpenSearchClient) value); return true; case "sniffAfterFailureDelay": ((OpensearchComponent) component).setSniffAfterFailureDelay((int) value); return true; case "snifferInterval": ((OpensearchComponent) component).setSnifferInterval((int) value); return true; case "enableSSL": ((OpensearchComponent) component).setEnableSSL((boolean) value); return true;
