This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 2d640a1b9226 CAMEL-24816: camel-elasticsearch/camel-opensearch - minor
robustness fixes
2d640a1b9226 is described below
commit 2d640a1b92265b62feca4e31e827c51e6730f81a
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Sep 22 09:14:32 2026 +0200
CAMEL-24816: camel-elasticsearch/camel-opensearch - minor robustness fixes
Three small, shared robustness issues in camel-elasticsearch and
camel-opensearch:
- ScrollRequestIterator.close() built ClearScrollRequest with
List.of(scrollId),
which throws NullPointerException when the initial search returned no
scroll id.
The clear-scroll is now issued only when a scroll id is present.
- ActionRequestConverter set the id from the CamelIndexId header
unconditionally on
a pre-built IndexRequest.Builder/UpdateRequest.Builder, overwriting a
caller-supplied
id with null when the header was absent. The id is now set only when the
header is
present. The elasticsearch document-only-mode check also uses
Boolean.TRUE.equals()
instead of reference equality.
- The producer defaults the size/from headers from the endpoint
configuration, but
cleanup() removed only the index-name and wait-for-active-shards headers,
so a
defaulted size/from leaked into subsequent endpoints. They are now
removed in
cleanup() too, tracked with the same configXxx pattern.
Adds converter unit tests in both components covering id preservation and
header
override for both index and update builders.
Closes #26590
Co-authored-by: Claude <[email protected]>
---
.../camel/component/es/ElasticsearchProducer.java | 27 ++++++-
.../es/ElasticsearchScrollRequestIterator.java | 13 ++--
.../ElasticsearchActionRequestConverter.java | 16 +++-
.../ElasticsearchActionRequestConverterTest.java | 86 ++++++++++++++++++++++
.../component/opensearch/OpensearchProducer.java | 15 +++-
.../OpensearchScrollRequestIterator.java | 13 ++--
.../OpensearchActionRequestConverter.java | 14 +++-
.../OpensearchActionRequestConverterTest.java | 86 ++++++++++++++++++++++
8 files changed, 251 insertions(+), 19 deletions(-)
diff --git
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
index 5b60d9a543e8..20027487a64a 100644
---
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
+++
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
@@ -166,14 +166,18 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
configIndexName = true;
}
+ boolean configSize = false;
Integer size =
message.getHeader(ElasticsearchConstants.PARAM_SIZE, Integer.class);
if (size == null) {
message.setHeader(ElasticsearchConstants.PARAM_SIZE,
configuration.getSize());
+ configSize = true;
}
+ boolean configFrom = false;
Integer from =
message.getHeader(ElasticsearchConstants.PARAM_FROM, Integer.class);
if (from == null) {
message.setHeader(ElasticsearchConstants.PARAM_FROM,
configuration.getFrom());
+ configFrom = true;
}
Boolean enableDocumentOnlyMode =
message.getHeader(ElasticsearchConstants.PARAM_DOCUMENT_MODE, Boolean.class);
@@ -193,7 +197,8 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
documentClass = configuration.getDocumentClass();
}
- ActionContext ctx = new ActionContext(exchange, callback,
transport, configIndexName, configWaitForActiveShards);
+ ActionContext ctx = new ActionContext(
+ exchange, callback, transport, configIndexName,
configWaitForActiveShards, configSize, configFrom);
switch (operation) {
case Index: {
@@ -455,6 +460,12 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
if (ctx.isConfigWaitForActiveShards()) {
message.removeHeader(ElasticsearchConstants.PARAM_WAIT_FOR_ACTIVE_SHARDS);
}
+ if (ctx.isConfigSize()) {
+ message.removeHeader(ElasticsearchConstants.PARAM_SIZE);
+ }
+ if (ctx.isConfigFrom()) {
+ message.removeHeader(ElasticsearchConstants.PARAM_FROM);
+ }
if (configuration.isDisconnect()) {
IOHelper.close(ctx.getTransport());
if (configuration.isEnableSniffer()) {
@@ -586,14 +597,18 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
private final ElasticsearchTransport transport;
private final boolean configIndexName;
private final boolean configWaitForActiveShards;
+ private final boolean configSize;
+ private final boolean configFrom;
ActionContext(Exchange exchange, AsyncCallback callback,
ElasticsearchTransport transport, boolean configIndexName,
- boolean configWaitForActiveShards) {
+ boolean configWaitForActiveShards, boolean configSize,
boolean configFrom) {
this.exchange = exchange;
this.callback = callback;
this.transport = transport;
this.configIndexName = configIndexName;
this.configWaitForActiveShards = configWaitForActiveShards;
+ this.configSize = configSize;
+ this.configFrom = configFrom;
}
ElasticsearchTransport getTransport() {
@@ -612,6 +627,14 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
return configWaitForActiveShards;
}
+ boolean isConfigSize() {
+ return configSize;
+ }
+
+ boolean isConfigFrom() {
+ return configFrom;
+ }
+
Exchange getExchange() {
return exchange;
}
diff --git
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchScrollRequestIterator.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchScrollRequestIterator.java
index b4ebf0d51910..b336998609e8 100644
---
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchScrollRequestIterator.java
+++
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchScrollRequestIterator.java
@@ -125,11 +125,14 @@ public class
ElasticsearchScrollRequestIterator<TDocument> implements Iterator<H
public void close() {
if (!closed) {
try {
- ClearScrollRequest clearScrollRequest = new
ClearScrollRequest.Builder()
- .scrollId(List.of(scrollId))
- .build();
-
- esClient.clearScroll(clearScrollRequest);
+ // scrollId can be null if the initial search returned no
scroll id; List.of(null) would NPE
+ if (scrollId != null) {
+ ClearScrollRequest clearScrollRequest = new
ClearScrollRequest.Builder()
+ .scrollId(List.of(scrollId))
+ .build();
+
+ esClient.clearScroll(clearScrollRequest);
+ }
closed = true;
exchange.setProperty(ElasticsearchConstants.PROPERTY_SCROLL_ES_QUERY_COUNT,
requestCount);
} catch (IOException e) {
diff --git
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverter.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverter.java
index 90b2fc8ff76e..7e18c719dd8e 100644
---
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverter.java
+++
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverter.java
@@ -85,7 +85,12 @@ public final class ElasticsearchActionRequestConverter {
@Converter
public static IndexRequest.Builder<?> toIndexRequestBuilder(Object
document, Exchange exchange) throws IOException {
if (document instanceof IndexRequest.Builder<?> indexReqBuilder) {
- return
indexReqBuilder.id(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_ID,
String.class));
+ // only override the id when the header is present, otherwise a
caller-supplied id would be cleared
+ String id =
exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_ID, String.class);
+ if (id != null) {
+ indexReqBuilder.id(id);
+ }
+ return indexReqBuilder;
}
IndexRequest.Builder<Object> builder = new IndexRequest.Builder<>();
if (document instanceof byte[] byteArray) {
@@ -115,12 +120,17 @@ public final class ElasticsearchActionRequestConverter {
@Converter
public static UpdateRequest.Builder<?, ?> toUpdateRequestBuilder(Object
document, Exchange exchange) throws IOException {
if (document instanceof UpdateRequest.Builder<?, ?> updateReqBuilder) {
- return
updateReqBuilder.id(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_ID,
String.class));
+ // only override the id when the header is present, otherwise a
caller-supplied id would be cleared
+ String id =
exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_ID, String.class);
+ if (id != null) {
+ updateReqBuilder.id(id);
+ }
+ return updateReqBuilder;
}
UpdateRequest.Builder<?, Object> builder = new
UpdateRequest.Builder<>();
Boolean enableDocumentOnlyMode
=
exchange.getIn().getHeader(ElasticsearchConstants.PARAM_DOCUMENT_MODE,
Boolean.FALSE, Boolean.class);
- Mode mode = enableDocumentOnlyMode == Boolean.TRUE ?
Mode.DOCUMENT_ONLY : Mode.DEFAULT;
+ Mode mode = Boolean.TRUE.equals(enableDocumentOnlyMode) ?
Mode.DOCUMENT_ONLY : Mode.DEFAULT;
if (document instanceof byte[] byteArray) {
mode.addDocToUpdateRequestBuilder(builder, new
ByteArrayInputStream(byteArray));
} else if (document instanceof InputStream inputStream) {
diff --git
a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterTest.java
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterTest.java
new file mode 100644
index 000000000000..796e859980b3
--- /dev/null
+++
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterTest.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.es.converter;
+
+import java.util.Map;
+
+import co.elastic.clients.elasticsearch.core.IndexRequest;
+import co.elastic.clients.elasticsearch.core.UpdateRequest;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.es.ElasticsearchConstants;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class ElasticsearchActionRequestConverterTest {
+
+ private CamelContext context;
+
+ @BeforeEach
+ void setUp() {
+ context = new DefaultCamelContext();
+ }
+
+ @AfterEach
+ void tearDown() {
+ context.stop();
+ }
+
+ private IndexRequest.Builder<Object> preBuiltIndexBuilder() {
+ return new
IndexRequest.Builder<>().index("idx").id("original").document(Map.of("k", "v"));
+ }
+
+ @Test
+ void preBuiltIndexBuilderKeepsItsIdWhenHeaderAbsent() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+
+ IndexRequest.Builder<?> result
+ =
ElasticsearchActionRequestConverter.toIndexRequestBuilder(preBuiltIndexBuilder(),
exchange);
+
+ // no CamelIndexId header -> the caller's id must be preserved, not
overwritten with null
+ assertEquals("original", result.build().id());
+ }
+
+ @Test
+ void preBuiltIndexBuilderHeaderOverridesId() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(ElasticsearchConstants.PARAM_INDEX_ID,
"fromHeader");
+
+ IndexRequest.Builder<?> result
+ =
ElasticsearchActionRequestConverter.toIndexRequestBuilder(preBuiltIndexBuilder(),
exchange);
+
+ assertEquals("fromHeader", result.build().id());
+ }
+
+ @Test
+ void preBuiltUpdateBuilderKeepsItsIdWhenHeaderAbsent() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ UpdateRequest.Builder<Object, Object> preBuilt
+ = new
UpdateRequest.Builder<>().index("idx").id("original").doc(Map.of("k", "v"));
+
+ UpdateRequest.Builder<?, ?> result
+ =
ElasticsearchActionRequestConverter.toUpdateRequestBuilder(preBuilt, exchange);
+
+ // no CamelIndexId header -> the caller's id must be preserved, not
overwritten with null
+ assertEquals("original", result.build().id());
+ }
+}
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 f90f6314a1ca..2d96335235bf 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
@@ -189,14 +189,18 @@ class OpensearchProducer extends DefaultAsyncProducer {
configIndexName = true;
}
+ boolean configSize = false;
Integer size = message.getHeader(OpensearchConstants.PARAM_SIZE,
Integer.class);
if (size == null) {
message.setHeader(OpensearchConstants.PARAM_SIZE,
configuration.getSize());
+ configSize = true;
}
+ boolean configFrom = false;
Integer from = message.getHeader(OpensearchConstants.PARAM_FROM,
Integer.class);
if (from == null) {
message.setHeader(OpensearchConstants.PARAM_FROM,
configuration.getFrom());
+ configFrom = true;
}
boolean configWaitForActiveShards = false;
@@ -211,7 +215,8 @@ class OpensearchProducer extends DefaultAsyncProducer {
documentClass = configuration.getDocumentClass();
}
- ActionContext ctx = new ActionContext(exchange, callback,
transport, configIndexName, configWaitForActiveShards);
+ ActionContext ctx = new ActionContext(
+ exchange, callback, transport, configIndexName,
configWaitForActiveShards, configSize, configFrom);
switch (operation) {
case Index -> processIndexAsync(ctx);
@@ -441,6 +446,12 @@ class OpensearchProducer extends DefaultAsyncProducer {
if (ctx.configWaitForActiveShards()) {
message.removeHeader(OpensearchConstants.PARAM_WAIT_FOR_ACTIVE_SHARDS);
}
+ if (ctx.configSize()) {
+ message.removeHeader(OpensearchConstants.PARAM_SIZE);
+ }
+ if (ctx.configFrom()) {
+ message.removeHeader(OpensearchConstants.PARAM_FROM);
+ }
if (configuration.isDisconnect() && openSearchClient == null) {
IOHelper.close(ctx.transport());
if (configuration.isEnableSniffer()) {
@@ -604,7 +615,7 @@ class OpensearchProducer extends DefaultAsyncProducer {
* An inner class providing all the information that an asynchronous
action could need.
*/
private record ActionContext(Exchange exchange, AsyncCallback callback,
OpenSearchTransport transport,
- boolean configIndexName, boolean configWaitForActiveShards) {
+ boolean configIndexName, boolean configWaitForActiveShards,
boolean configSize, boolean configFrom) {
OpenSearchAsyncClient getClient() {
return new OpenSearchAsyncClient(transport);
diff --git
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchScrollRequestIterator.java
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchScrollRequestIterator.java
index 70e19c9c0880..5c70ac93c61c 100644
---
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchScrollRequestIterator.java
+++
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/OpensearchScrollRequestIterator.java
@@ -125,11 +125,14 @@ public class OpensearchScrollRequestIterator<TDocument>
implements Iterator<Hit<
public void close() {
if (!closed) {
try {
- ClearScrollRequest clearScrollRequest = new
ClearScrollRequest.Builder()
- .scrollId(List.of(scrollId))
- .build();
-
- esClient.clearScroll(clearScrollRequest);
+ // scrollId can be null if the initial search returned no
scroll id; List.of(null) would NPE
+ if (scrollId != null) {
+ ClearScrollRequest clearScrollRequest = new
ClearScrollRequest.Builder()
+ .scrollId(List.of(scrollId))
+ .build();
+
+ esClient.clearScroll(clearScrollRequest);
+ }
closed = true;
exchange.setProperty(OpensearchConstants.PROPERTY_SCROLL_OPENSEARCH_QUERY_COUNT,
requestCount);
} catch (IOException e) {
diff --git
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverter.java
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverter.java
index 512c2aa118e8..bdb6c711ec9c 100644
---
a/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverter.java
+++
b/components/camel-opensearch/src/main/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverter.java
@@ -88,7 +88,12 @@ public final class OpensearchActionRequestConverter {
@Converter
public static IndexRequest.Builder<?> toIndexRequestBuilder(Object
document, Exchange exchange) throws IOException {
if (document instanceof IndexRequest.Builder<?> builder) {
- return
builder.id(exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_ID,
String.class));
+ // only override the id when the header is present, otherwise a
caller-supplied id would be cleared
+ String id =
exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_ID, String.class);
+ if (id != null) {
+ builder.id(id);
+ }
+ return builder;
}
JacksonJsonpMapper mapper = createMapper();
IndexRequest.Builder<Object> builder = new IndexRequest.Builder<>();
@@ -116,7 +121,12 @@ public final class OpensearchActionRequestConverter {
@Converter
public static UpdateRequest.Builder<?, ?> toUpdateRequestBuilder(Object
document, Exchange exchange) throws IOException {
if (document instanceof UpdateRequest.Builder<?, ?> builder) {
- return
builder.id(exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_ID,
String.class));
+ // only override the id when the header is present, otherwise a
caller-supplied id would be cleared
+ String id =
exchange.getIn().getHeader(OpensearchConstants.PARAM_INDEX_ID, String.class);
+ if (id != null) {
+ builder.id(id);
+ }
+ return builder;
}
JacksonJsonpMapper mapper = createMapper();
UpdateRequest.Builder<?, Object> builder = new
UpdateRequest.Builder<>();
diff --git
a/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterTest.java
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterTest.java
new file mode 100644
index 000000000000..87dede478c7d
--- /dev/null
+++
b/components/camel-opensearch/src/test/java/org/apache/camel/component/opensearch/converter/OpensearchActionRequestConverterTest.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.opensearch.converter;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.opensearch.OpensearchConstants;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.opensearch.client.opensearch.core.IndexRequest;
+import org.opensearch.client.opensearch.core.UpdateRequest;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class OpensearchActionRequestConverterTest {
+
+ private CamelContext context;
+
+ @BeforeEach
+ void setUp() {
+ context = new DefaultCamelContext();
+ }
+
+ @AfterEach
+ void tearDown() {
+ context.stop();
+ }
+
+ private IndexRequest.Builder<Object> preBuiltIndexBuilder() {
+ return new
IndexRequest.Builder<>().index("idx").id("original").document(Map.of("k", "v"));
+ }
+
+ @Test
+ void preBuiltIndexBuilderKeepsItsIdWhenHeaderAbsent() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+
+ IndexRequest.Builder<?> result
+ =
OpensearchActionRequestConverter.toIndexRequestBuilder(preBuiltIndexBuilder(),
exchange);
+
+ // no CamelOpensearchIndexId header -> the caller's id must be
preserved, not overwritten with null
+ assertEquals("original", result.build().id());
+ }
+
+ @Test
+ void preBuiltIndexBuilderHeaderOverridesId() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(OpensearchConstants.PARAM_INDEX_ID,
"fromHeader");
+
+ IndexRequest.Builder<?> result
+ =
OpensearchActionRequestConverter.toIndexRequestBuilder(preBuiltIndexBuilder(),
exchange);
+
+ assertEquals("fromHeader", result.build().id());
+ }
+
+ @Test
+ void preBuiltUpdateBuilderKeepsItsIdWhenHeaderAbsent() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ UpdateRequest.Builder<Object, Object> preBuilt
+ = new
UpdateRequest.Builder<>().index("idx").id("original").doc(Map.of("k", "v"));
+
+ UpdateRequest.Builder<?, ?> result
+ =
OpensearchActionRequestConverter.toUpdateRequestBuilder(preBuilt, exchange);
+
+ // no CamelOpensearchIndexId header -> the caller's id must be
preserved, not overwritten with null
+ assertEquals("original", result.build().id());
+ }
+}
