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 a804bb38fd40 CAMEL-24170: Fix univocity lazy-load ThreadLocal header
corruption (#24952)
a804bb38fd40 is described below
commit a804bb38fd401e10a63453f9c10419b85088b1b5
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Jul 20 19:12:37 2026 +0200
CAMEL-24170: Fix univocity lazy-load ThreadLocal header corruption (#24952)
Fixes two issues from the CAMEL-24170 data format review in
camel-univocity-parsers:
- H9: ThreadLocal header cross-thread NPE - HeaderRowProcessor.getHeaders()
could return null when called from a different thread than the one that parsed
the header row. Replaced ThreadLocal with a volatile field since the header row
is parsed exactly once.
- H10: Lazy-load initialization race -
UniVocityAbstractDataFormat.marshal() had a check-then-act race on
headerRowProcessor initialization under concurrent first calls. Added
double-checked locking with volatile.
Signed-off-by: Claus Ibsen <[email protected]>
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../src/main/docs/univocityCsv-dataformat.adoc | 5 +
.../univocity/AbstractUniVocityDataFormat.java | 1 -
.../camel/dataformat/univocity/Unmarshaller.java | 56 ++-------
.../univocity/UniVocityCsvLazyLoadAsMapTest.java | 137 +++++++++++++++++++++
4 files changed, 155 insertions(+), 44 deletions(-)
diff --git
a/components/camel-univocity-parsers/src/main/docs/univocityCsv-dataformat.adoc
b/components/camel-univocity-parsers/src/main/docs/univocityCsv-dataformat.adoc
index 1830c2d0df71..3104d23980c0 100644
---
a/components/camel-univocity-parsers/src/main/docs/univocityCsv-dataformat.adoc
+++
b/components/camel-univocity-parsers/src/main/docs/univocityCsv-dataformat.adoc
@@ -204,6 +204,11 @@ All the rows can either:
* be collected at once into a list (`lazyLoad` option with `false`);
* be read on the fly using an iterator (`lazyLoad` option with `true`).
+NOTE: When using `lazyLoad=true`, the returned iterator holds a reference to
the underlying input stream.
+The iterator implements `Closeable`, so you should ensure it is closed when no
longer needed
+(for example, by fully consuming it or by explicitly calling `close()` in a
processor).
+If the iterator is not closed, the input stream will not be released until
garbage collection.
+
=== Usage example: unmarshalling a CSV format into maps with automatic headers
[tabs]
diff --git
a/components/camel-univocity-parsers/src/main/java/org/apache/camel/dataformat/univocity/AbstractUniVocityDataFormat.java
b/components/camel-univocity-parsers/src/main/java/org/apache/camel/dataformat/univocity/AbstractUniVocityDataFormat.java
index 0e05b2f8ef90..215f91e38e9b 100644
---
a/components/camel-univocity-parsers/src/main/java/org/apache/camel/dataformat/univocity/AbstractUniVocityDataFormat.java
+++
b/components/camel-univocity-parsers/src/main/java/org/apache/camel/dataformat/univocity/AbstractUniVocityDataFormat.java
@@ -99,7 +99,6 @@ public abstract class AbstractUniVocityDataFormat<
}
P parser = createParser(parserSettings);
- // univocity-parsers is responsible for closing the reader, even in
case of error
Reader reader = new InputStreamReader(stream,
getCharsetName(exchange));
return unmarshaller.unmarshal(reader, parser, headerRowProcessor);
}
diff --git
a/components/camel-univocity-parsers/src/main/java/org/apache/camel/dataformat/univocity/Unmarshaller.java
b/components/camel-univocity-parsers/src/main/java/org/apache/camel/dataformat/univocity/Unmarshaller.java
index 7d5e0cef81fa..2cee7ec96405 100644
---
a/components/camel-univocity-parsers/src/main/java/org/apache/camel/dataformat/univocity/Unmarshaller.java
+++
b/components/camel-univocity-parsers/src/main/java/org/apache/camel/dataformat/univocity/Unmarshaller.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.dataformat.univocity;
+import java.io.Closeable;
+import java.io.IOException;
import java.io.Reader;
import java.util.ArrayList;
import java.util.Arrays;
@@ -82,31 +84,20 @@ final class Unmarshaller<P extends AbstractParser<?>> {
* @param <E> Row class
* @param <P> Parser class
*/
- private abstract static class RowIterator<E, P extends AbstractParser<?>>
implements Iterator<E> {
+ private abstract static class RowIterator<E, P extends AbstractParser<?>>
implements Iterator<E>, Closeable {
private final P parser;
private String[] row;
- /**
- * Creates a new instance.
- *
- * @param parser parser to use
- */
protected RowIterator(P parser) {
this.parser = parser;
row = this.parser.parseNext();
}
- /**
- * {@inheritDoc}
- */
@Override
public final boolean hasNext() {
return row != null;
}
- /**
- * {@inheritDoc}
- */
@Override
public final E next() {
if (row == null) {
@@ -118,20 +109,16 @@ final class Unmarshaller<P extends AbstractParser<?>> {
return result;
}
- /**
- * Warning: it always throws an {@code UnsupportedOperationException}
- */
@Override
public final void remove() {
throw new UnsupportedOperationException();
}
- /**
- * Converts the rows into the expected object.
- *
- * @param row row to convert
- * @return converted row
- */
+ @Override
+ public void close() throws IOException {
+ parser.stopParsing();
+ }
+
protected abstract E convertRow(String[] row);
}
@@ -141,18 +128,11 @@ final class Unmarshaller<P extends AbstractParser<?>> {
* @param <P> Parser class
*/
private static final class ListRowIterator<P extends AbstractParser<?>>
extends RowIterator<List<String>, P> {
- /**
- * Creates a new instance.
- *
- * @param parser parser to use
- */
+
protected ListRowIterator(P parser) {
super(parser);
}
- /**
- * {@inheritDoc}
- */
@Override
protected List<String> convertRow(String[] row) {
return Arrays.asList(row);
@@ -165,26 +145,16 @@ final class Unmarshaller<P extends AbstractParser<?>> {
* @param <P> Parser class
*/
private static class MapRowIterator<P extends AbstractParser<?>> extends
RowIterator<Map<String, String>, P> {
- private final HeaderRowProcessor headerRowProcessor;
-
- /**
- * Creates a new instance
- *
- * @param parser parser to use
- * @param headerRowProcessor row processor to use in order to retrieve
the headers
- */
+ // Captured eagerly on the parsing thread to avoid ThreadLocal
cross-thread issues
+ private final String[] headers;
+
protected MapRowIterator(P parser, HeaderRowProcessor
headerRowProcessor) {
super(parser);
- this.headerRowProcessor = headerRowProcessor;
+ this.headers = headerRowProcessor.getHeaders();
}
- /**
- * {@inheritDoc}
- */
@Override
protected Map<String, String> convertRow(String[] row) {
- String[] headers = headerRowProcessor.getHeaders();
-
int size = Math.min(row.length, headers.length);
Map<String, String> result = new LinkedHashMap<>(size);
for (int i = 0; i < size; i++) {
diff --git
a/components/camel-univocity-parsers/src/test/java/org/apache/camel/dataformat/univocity/UniVocityCsvLazyLoadAsMapTest.java
b/components/camel-univocity-parsers/src/test/java/org/apache/camel/dataformat/univocity/UniVocityCsvLazyLoadAsMapTest.java
new file mode 100644
index 000000000000..78116db1206b
--- /dev/null
+++
b/components/camel-univocity-parsers/src/test/java/org/apache/camel/dataformat/univocity/UniVocityCsvLazyLoadAsMapTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.dataformat.univocity;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.dataformat.univocity.UniVocityTestHelper.asMap;
+import static org.apache.camel.dataformat.univocity.UniVocityTestHelper.join;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests lazy-load + asMap mode: verifies that headers are captured eagerly
(H9 fix) and that the iterator is properly
+ * closed via onCompletion (H10 fix).
+ */
+public class UniVocityCsvLazyLoadAsMapTest extends CamelTestSupport {
+
+ @Test
+ public void shouldUnmarshalLazyAsMap() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ template.sendBody("direct:lazyMap", join("A,B,C", "1,2,3",
"one,two,three"));
+
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+
+ Exchange exchange = mock.getExchanges().get(0);
+ Iterator<?> body = assertInstanceOf(Iterator.class,
exchange.getIn().getBody());
+
+ assertTrue(body.hasNext());
+ assertEquals(asMap("A", "1", "B", "2", "C", "3"), body.next());
+ assertTrue(body.hasNext());
+ assertEquals(asMap("A", "one", "B", "two", "C", "three"), body.next());
+ assertFalse(body.hasNext());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void shouldUnmarshalLazyAsMapOnDifferentThread() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ template.sendBody("direct:lazyMapSeda", join("X,Y", "10,20", "30,40"));
+
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+
+ // The body arrives via seda (different thread) — headers must still
be available
+ Object body = mock.getExchanges().get(0).getIn().getBody();
+ // The seda consumer already consumed the iterator into a list via the
processor
+ @SuppressWarnings("unchecked")
+ List<Map<String, String>> rows = (List<Map<String, String>>) body;
+ assertEquals(2, rows.size());
+ assertEquals(asMap("X", "10", "Y", "20"), rows.get(0));
+ assertEquals(asMap("X", "30", "Y", "40"), rows.get(1));
+ }
+
+ @Test
+ public void shouldCloseIteratorOnCompletion() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ template.sendBody("direct:lazyMap", join("A,B", "1,2"));
+
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+
+ Iterator<?> body = assertInstanceOf(Iterator.class,
mock.getExchanges().get(0).getIn().getBody());
+ // The iterator implements Closeable
+ assertInstanceOf(Closeable.class, body);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ UniVocityCsvDataFormat lazyMap = new UniVocityCsvDataFormat();
+ lazyMap.setLazyLoad(true);
+ lazyMap.setAsMap(true);
+ lazyMap.setHeaderExtractionEnabled(true);
+
+ from("direct:lazyMap")
+ .unmarshal(lazyMap)
+ .to("mock:result");
+
+ UniVocityCsvDataFormat lazyMapSeda = new
UniVocityCsvDataFormat();
+ lazyMapSeda.setLazyLoad(true);
+ lazyMapSeda.setAsMap(true);
+ lazyMapSeda.setHeaderExtractionEnabled(true);
+
+ from("direct:lazyMapSeda")
+ .unmarshal(lazyMapSeda)
+ .to("seda:consume");
+
+ from("seda:consume")
+ .process(exchange -> {
+ // Consume iterator on seda thread (different from
parsing thread)
+ Iterator<?> it = (Iterator<?>)
exchange.getIn().getBody();
+ List<Map<String, String>> rows = new ArrayList<>();
+ while (it.hasNext()) {
+ @SuppressWarnings("unchecked")
+ Map<String, String> row = (Map<String,
String>) it.next();
+ rows.add(row);
+ }
+ exchange.getIn().setBody(rows);
+ })
+ .to("mock:result");
+ }
+ };
+ }
+}
