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 fae6c3323cdf CAMEL-23883: Kafka producer no longer silently drops
scalar Jackson ValueNode
fae6c3323cdf is described below
commit fae6c3323cdf97c794224a979fe482ac62c4984a
Author: matteominin <[email protected]>
AuthorDate: Mon Jul 6 18:46:41 2026 +0200
CAMEL-23883: Kafka producer no longer silently drops scalar Jackson
ValueNode
With useIterator=true (the default), the Kafka producer splits Iterable
bodies
into individual records. Since Jackson JsonNode implements Iterable, scalar
value
nodes (e.g. IntNode from a jq transform) were treated as empty iterables —
send()
was never called and the message was silently dropped.
Narrow isIterable() so that a JsonNode is only treated as iterable when it
is a
container node (ArrayNode/ObjectNode). Scalar nodes now take the
single-message
path. Includes unit tests for both scalar and container cases, an
integration
test, and an upgrade guide entry.
Closes #24467
---
.../camel/component/kafka/KafkaProducer.java | 5 ++
.../camel/component/kafka/KafkaProducerTest.java | 31 ++++++++++++
.../integration/KafkaProducerValueNodeIT.java | 59 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 11 ++++
4 files changed, 106 insertions(+)
diff --git
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index bfa56a42b898..92e1ab5d8e74 100755
---
a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++
b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -27,6 +27,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
+import com.fasterxml.jackson.databind.JsonNode;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
@@ -332,6 +333,10 @@ public class KafkaProducer extends DefaultAsyncProducer
implements RouteIdAware
}
private boolean isIterable(Object body) {
+ if (body instanceof JsonNode node) {
+ return node.isContainerNode();
+ }
+
if (body instanceof Iterable || body instanceof Iterator) {
return true;
}
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index 3a455bfa81a8..67db580eb5f4 100755
---
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -28,6 +28,9 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.IntNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import org.apache.camel.AggregationStrategy;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
@@ -480,6 +483,34 @@ public class KafkaProducerTest {
assertRecordMetadataExistsForEachAggregatedMessage();
}
+ @Test
+ public void processSendsMessageWithScalarJsonNode() throws Exception {
+ endpoint.getConfiguration().setTopic("sometopic");
+ Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getMessage()).thenReturn(in);
+
+ in.setBody(IntNode.valueOf(42));
+ producer.process(exchange);
+
+ verifySendMessage("sometopic");
+ }
+
+ @Test
+ public void processSendsMessagesWithJsonArrayNode() throws Exception {
+ endpoint.getConfiguration().setTopic("sometopic");
+ Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getMessage()).thenReturn(in);
+
+ ArrayNode node = JsonNodeFactory.instance.arrayNode();
+ node.add(1);
+ node.add(2);
+
+ in.setBody(node);
+ producer.process(exchange);
+
+ verifySendMessages(Arrays.asList("sometopic", "sometopic"), null);
+ }
+
@SuppressWarnings({ "unchecked", "rawtypes" })
protected void verifySendMessage(Integer partitionKey, String topic,
String messageKey) {
ArgumentCaptor<ProducerRecord> captor =
ArgumentCaptor.forClass(ProducerRecord.class);
diff --git
a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaProducerValueNodeIT.java
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaProducerValueNodeIT.java
new file mode 100644
index 000000000000..8457f5cee511
--- /dev/null
+++
b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/integration/KafkaProducerValueNodeIT.java
@@ -0,0 +1,59 @@
+/*
+ * 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.kafka.integration;
+
+import java.util.Collections;
+
+import com.fasterxml.jackson.databind.node.IntNode;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+public class KafkaProducerValueNodeIT extends BaseKafkaTestSupport {
+
+ private static final String TOPIC = "value-node";
+
+ private static final String FROM_URI = "kafka:" + TOPIC
+ +
"?groupId=KafkaProducerValueNodeIT&autoOffsetReset=earliest&keyDeserializer=org.apache.kafka.common.serialization.StringDeserializer&"
+ +
"valueDeserializer=org.apache.kafka.common.serialization.StringDeserializer"
+ +
"&autoCommitIntervalMs=1000&pollTimeoutMs=1000&autoCommitEnable=true";
+
+ @AfterEach
+ public void after() {
+ kafkaAdminClient.deleteTopics(Collections.singletonList(TOPIC));
+ }
+
+ @Test
+ public void scalarJsonNodeIsDelivered() throws Exception {
+ MockEndpoint mock = contextExtension.getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("5");
+ contextExtension.getProducerTemplate().sendBody("direct:start",
IntNode.valueOf(5));
+ mock.assertIsSatisfied(5000);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").to("kafka:" + TOPIC +
"?groupId=KafkaProducerValueNodeIT");
+ from(FROM_URI).to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 19879bae8b3a..b8ae87f73de7 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -109,3 +109,14 @@ the message, consistent with the inbound header filtering
already performed by t
Ordinary application headers are unaffected. If a route relied on `Camel*`
headers being propagated from the MIME
content, set them explicitly after unmarshalling.
+
+=== camel-kafka - Producer no longer silently drops scalar Jackson nodes
+
+With `useIterator=true`, the Kafka producer splits an `Iterable` body into one
record per element.
+A Jackson `JsonNode` implements `Iterable`, so a scalar value node (e.g. an
`IntNode` produced by
+`transform().jq(".my-value")`) was seen as an empty iterable and produced no
record, silently
+discarding the message with no exception or log.
+
+Scalar value nodes are now sent as a single record. Only container nodes
(`ArrayNode`, `ObjectNode`)
+are still split, which is unchanged. Any `convertBodyTo(...)` previously used
as a workaround is no
+longer required.
