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 41d9dd8ab401 CAMEL-26128: camel-flatpack - with splitRows each row
message has its own row, and a failing row fails the producer (#28906)
41d9dd8ab401 is described below
commit 41d9dd8ab4011426542a8914ab089e9ee1fd7e5d
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:40:34 2026 +0100
CAMEL-26128: camel-flatpack - with splitRows each row message has its own
row, and a failing row fails the producer (#28906)
… own row, and a failing row fails the producer
With splitRows every row message shared the one parsed DataSet, so
converting a
body to a List read all the rows and the producer stopped sending, and a
body
read after the next row (such as after seda) failed. Each row message now
gets a
read-only DataSetRow with only its row.
An exception in the route that consumes the rows was lost; it now fails the
producer, which stops at that row. Converting a DataSet that is not on a
row to
a Map fails with a clear IllegalStateException instead of a
NullPointerException,
and a java.io.File or java.nio.file.Path body is read through a
BufferedReader.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/components/flatpack.json | 4 +-
.../camel/catalog/docs/flatpack-component.adoc | 12 +-
.../apache/camel/component/flatpack/flatpack.json | 4 +-
.../src/main/docs/flatpack-component.adoc | 12 +-
.../camel/component/flatpack/DataSetRow.java | 343 +++++++++++++++++++++
.../component/flatpack/FlatpackConverter.java | 4 +
.../camel/component/flatpack/FlatpackEndpoint.java | 32 +-
.../camel/component/flatpack/FlatpackProducer.java | 6 +-
.../component/flatpack/FlatpackSplitRowsTest.java | 201 ++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 +
10 files changed, 612 insertions(+), 20 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/flatpack.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/flatpack.json
index 85ec8dfaf75c..83e0c7698252 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/flatpack.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/flatpack.json
@@ -32,8 +32,8 @@
"CamelFlatpackCounter": { "index": 0, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType": "int",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The current row index. For splitRows=false the counter
is the total number of rows.", "constantName":
"org.apache.camel.component.flatpack.FlatpackConstants#FLATPACK_COUNTER" }
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "pojo" ], "javaType":
"net.sf.flatpack.DataSet", "description": "The Flatpack DataSet the producer
parsed. With splitRows, one message per row, with the DataSet at that row:
convert it to a Map before any asynchronous step, as the next row moves it; a
List reads all the rows and ends the split. With splitRows=false, the whole
DataSet: convert it to a List of Maps" },
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"format": "text", "description": "The fixed-width or delimited data to parse,
such as a file from the file consumer; a java.io.File, a Map or a List fails.
The rows go to the consumer of the same endpoint" },
+ "consumerOut": { "carriers": [ "pojo" ], "javaType":
"net.sf.flatpack.DataSet", "description": "The Flatpack DataSet the producer
parsed. With splitRows, one message per row, with a DataSet of that one row
that converts to a Map. With splitRows=false, the whole DataSet: convert it to
a List of Maps" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"format": "text", "description": "The fixed-width or delimited data to parse,
such as a file from the file consumer, a java.io.File or a java.nio.file.Path;
a Map or a List fails. The rows go to the consumer of the same endpoint, and an
exception there fails the producer" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flatpack-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flatpack-component.adoc
index 24cf42f90647..be76a7646a8a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flatpack-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/flatpack-component.adoc
@@ -68,9 +68,15 @@ For example, to get the firstname from the sample below:
String firstName = row.get("FIRSTNAME");
----
-With `splitRows=true`, every message shares the same `DataSet`, positioned at
its row.
-Convert the body to a `Map` before any asynchronous step (such as `seda` or
`wireTap`), as the next row moves the `DataSet`.
-Do not convert it to a `List`: that reads all the rows and the remaining rows
are then not sent.
+With `splitRows=true`, every message has its own read-only `DataSet` with only
its row
+(`org.apache.camel.component.flatpack.DataSetRow`), so the body can go to an
asynchronous step (such as `seda` or `wireTap`)
+before it is converted. Converting it to a `List` gives a `List` with that one
row.
+
+With `splitRows=false`, the body is the whole `DataSet` and is not on a row:
convert it to a `List`, as converting it
+to a `Map` fails.
+
+The producer accepts the same bodies as the file consumer sends, and a
`java.io.File` or `java.nio.file.Path` too.
+When the route that consumes the rows fails on a row, the exception fails the
producer, and the next rows are not sent.
=== Header and Trailer records
diff --git
a/components/camel-flatpack/src/generated/resources/META-INF/org/apache/camel/component/flatpack/flatpack.json
b/components/camel-flatpack/src/generated/resources/META-INF/org/apache/camel/component/flatpack/flatpack.json
index 85ec8dfaf75c..83e0c7698252 100644
---
a/components/camel-flatpack/src/generated/resources/META-INF/org/apache/camel/component/flatpack/flatpack.json
+++
b/components/camel-flatpack/src/generated/resources/META-INF/org/apache/camel/component/flatpack/flatpack.json
@@ -32,8 +32,8 @@
"CamelFlatpackCounter": { "index": 0, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType": "int",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The current row index. For splitRows=false the counter
is the total number of rows.", "constantName":
"org.apache.camel.component.flatpack.FlatpackConstants#FLATPACK_COUNTER" }
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "pojo" ], "javaType":
"net.sf.flatpack.DataSet", "description": "The Flatpack DataSet the producer
parsed. With splitRows, one message per row, with the DataSet at that row:
convert it to a Map before any asynchronous step, as the next row moves it; a
List reads all the rows and ends the split. With splitRows=false, the whole
DataSet: convert it to a List of Maps" },
- "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"format": "text", "description": "The fixed-width or delimited data to parse,
such as a file from the file consumer; a java.io.File, a Map or a List fails.
The rows go to the consumer of the same endpoint" },
+ "consumerOut": { "carriers": [ "pojo" ], "javaType":
"net.sf.flatpack.DataSet", "description": "The Flatpack DataSet the producer
parsed. With splitRows, one message per row, with a DataSet of that one row
that converts to a Map. With splitRows=false, the whole DataSet: convert it to
a List of Maps" },
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file" ],
"format": "text", "description": "The fixed-width or delimited data to parse,
such as a file from the file consumer, a java.io.File or a java.nio.file.Path;
a Map or a List fails. The rows go to the consumer of the same endpoint, and an
exception there fails the producer" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
diff --git a/components/camel-flatpack/src/main/docs/flatpack-component.adoc
b/components/camel-flatpack/src/main/docs/flatpack-component.adoc
index 24cf42f90647..be76a7646a8a 100644
--- a/components/camel-flatpack/src/main/docs/flatpack-component.adoc
+++ b/components/camel-flatpack/src/main/docs/flatpack-component.adoc
@@ -68,9 +68,15 @@ For example, to get the firstname from the sample below:
String firstName = row.get("FIRSTNAME");
----
-With `splitRows=true`, every message shares the same `DataSet`, positioned at
its row.
-Convert the body to a `Map` before any asynchronous step (such as `seda` or
`wireTap`), as the next row moves the `DataSet`.
-Do not convert it to a `List`: that reads all the rows and the remaining rows
are then not sent.
+With `splitRows=true`, every message has its own read-only `DataSet` with only
its row
+(`org.apache.camel.component.flatpack.DataSetRow`), so the body can go to an
asynchronous step (such as `seda` or `wireTap`)
+before it is converted. Converting it to a `List` gives a `List` with that one
row.
+
+With `splitRows=false`, the body is the whole `DataSet` and is not on a row:
convert it to a `List`, as converting it
+to a `Map` fails.
+
+The producer accepts the same bodies as the file consumer sends, and a
`java.io.File` or `java.nio.file.Path` too.
+When the route that consumes the rows fails on a row, the exception fails the
producer, and the next rows are not sent.
=== Header and Trailer records
diff --git
a/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/DataSetRow.java
b/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/DataSetRow.java
new file mode 100644
index 000000000000..49bf69dedafa
--- /dev/null
+++
b/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/DataSetRow.java
@@ -0,0 +1,343 @@
+/*
+ * 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.flatpack;
+
+import java.math.BigDecimal;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.function.DoubleSupplier;
+import java.util.function.IntSupplier;
+import java.util.function.LongSupplier;
+import java.util.function.Supplier;
+
+import net.sf.flatpack.DataError;
+import net.sf.flatpack.DataSet;
+import net.sf.flatpack.Record;
+import net.sf.flatpack.ordering.OrderBy;
+
+/**
+ * A read-only {@link DataSet} with one row: the row the parsed {@link
DataSet} is on when this is created. With
+ * splitRows, every message gets its own {@link DataSetRow}, so a message
keeps its row while the next rows are sent. It
+ * starts on its row, like the parsed {@link DataSet} at that row.
+ */
+public class DataSetRow implements DataSet {
+ private final Record row;
+ private final List<DataError> errors;
+ private int pointer;
+ private Record current;
+
+ public DataSetRow(DataSet dataSet) {
+ this.row = dataSet.getRecord()
+ .orElseThrow(() -> new IllegalArgumentException("The DataSet
is not on a row"));
+ this.errors = new ArrayList<>(dataSet.getErrors());
+ this.current = row;
+ }
+
+ private Record current() {
+ if (current == null) {
+ throw new IllegalStateException("The DataSet is not on a row");
+ }
+ return current;
+ }
+
+ // cursor over the one row
+ // --------------------------------------------------------------
+
+ @Override
+ public boolean next() {
+ if (pointer < 0) {
+ pointer = 0;
+ current = row;
+ return true;
+ }
+ current = null;
+ return false;
+ }
+
+ @Override
+ public boolean previous() {
+ current = null;
+ return false;
+ }
+
+ @Override
+ public void goTop() {
+ pointer = -1;
+ current = null;
+ }
+
+ @Override
+ public void goBottom() {
+ pointer = 0;
+ current = row;
+ }
+
+ @Override
+ public void absolute(int localPointer) {
+ if (localPointer != 0) {
+ throw new IndexOutOfBoundsException("INVALID POINTER LOCATION: " +
localPointer);
+ }
+ pointer = 0;
+ current = row;
+ }
+
+ @Override
+ public int getIndex() {
+ return pointer;
+ }
+
+ @Override
+ public int getRowCount() {
+ return 1;
+ }
+
+ @Override
+ public Optional<Record> getRecord() {
+ return Optional.ofNullable(current);
+ }
+
+ @Override
+ public void orderRows(OrderBy ob) {
+ // one row is always in order
+ goTop();
+ }
+
+ @Override
+ public List<DataError> getErrors() {
+ return errors;
+ }
+
+ @Override
+ public int getErrorCount() {
+ return errors.size();
+ }
+
+ @Override
+ public boolean isAnError(int lineNo) {
+ for (DataError error : errors) {
+ if (error.getLineNo() == lineNo && error.getErrorLevel() > 1) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public void clearErrors() {
+ errors.clear();
+ }
+
+ // read-only
+ // --------------------------------------------------------------
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException("A row of a split DataSet is
read-only");
+ }
+
+ @Override
+ public void setLowerCase() {
+ throw new UnsupportedOperationException("A row of a split DataSet is
read-only");
+ }
+
+ @Override
+ public void setUpperCase() {
+ throw new UnsupportedOperationException("A row of a split DataSet is
read-only");
+ }
+
+ @Override
+ public void setStrictNumericParse(boolean strictNumericParse) {
+ throw new UnsupportedOperationException("A row of a split DataSet is
read-only");
+ }
+
+ @Override
+ public void setPZConvertProps(Properties props) {
+ throw new UnsupportedOperationException("A row of a split DataSet is
read-only");
+ }
+
+ @Override
+ public void setValue(String column, String value) {
+ throw new UnsupportedOperationException("A row of a split DataSet is
read-only");
+ }
+
+ @Override
+ public void clearRows() {
+ throw new UnsupportedOperationException("A row of a split DataSet is
read-only");
+ }
+
+ @Override
+ public void clearAll() {
+ throw new UnsupportedOperationException("A row of a split DataSet is
read-only");
+ }
+
+ // the values of the row
+ // --------------------------------------------------------------
+
+ @Override
+ public String[] getColumns() {
+ return row.getColumns();
+ }
+
+ @Override
+ public String[] getColumns(String recordID) {
+ return row.getColumns(recordID);
+ }
+
+ @Override
+ public int getRowNo() {
+ return current().getRowNo();
+ }
+
+ @Override
+ public boolean isRecordID(String recordID) {
+ return current().isRecordID(recordID);
+ }
+
+ @Override
+ public String getRecordID() {
+ return current().getRecordID();
+ }
+
+ @Override
+ public boolean contains(String column) {
+ return current().contains(column);
+ }
+
+ @Override
+ public boolean isRowEmpty() {
+ return current().isRowEmpty();
+ }
+
+ @Override
+ public String getRawData() {
+ return current().getRawData();
+ }
+
+ @Override
+ public String getString(String column) {
+ return current().getString(column);
+ }
+
+ @Override
+ public String getString(String column, Supplier<String> defaultSupplier) {
+ return current().getString(column, defaultSupplier);
+ }
+
+ @Override
+ public double getDouble(String column) {
+ return current().getDouble(column);
+ }
+
+ @Override
+ public double getDouble(String column, DoubleSupplier defaultSupplier) {
+ return current().getDouble(column, defaultSupplier);
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(String column) {
+ return current().getBigDecimal(column);
+ }
+
+ @Override
+ public BigDecimal getBigDecimal(String column, Supplier<BigDecimal>
defaultSupplier) {
+ return current().getBigDecimal(column, defaultSupplier);
+ }
+
+ @Override
+ public int getInt(String column) {
+ return current().getInt(column);
+ }
+
+ @Override
+ public int getInt(String column, IntSupplier defaultSupplier) {
+ return current().getInt(column, defaultSupplier);
+ }
+
+ @Override
+ public long getLong(String column) {
+ return current().getLong(column);
+ }
+
+ @Override
+ public long getLong(String column, LongSupplier defaultSupplier) {
+ return current().getLong(column, defaultSupplier);
+ }
+
+ @Override
+ public Date getDate(String column) throws ParseException {
+ return current().getDate(column);
+ }
+
+ @Override
+ public Date getDate(String column, SimpleDateFormat sdf) throws
ParseException {
+ return current().getDate(column, sdf);
+ }
+
+ @Override
+ public Date getDate(String column, Supplier<Date> defaultSupplier) throws
ParseException {
+ return current().getDate(column, defaultSupplier);
+ }
+
+ @Override
+ public Date getDate(String column, SimpleDateFormat sdf, Supplier<Date>
defaultSupplier)
+ throws ParseException {
+ return current().getDate(column, sdf, defaultSupplier);
+ }
+
+ @Override
+ public LocalDate getLocalDate(String column) throws ParseException {
+ return current().getLocalDate(column);
+ }
+
+ @Override
+ public LocalDate getLocalDate(String column, Supplier<LocalDate>
defaultSupplier) throws ParseException {
+ return current().getLocalDate(column, defaultSupplier);
+ }
+
+ @Override
+ public LocalDate getLocalDate(String column, DateTimeFormatter formatter)
throws ParseException {
+ return current().getLocalDate(column, formatter);
+ }
+
+ @Override
+ public LocalDate getLocalDate(String column, String dateFormat) throws
ParseException {
+ return current().getLocalDate(column, dateFormat);
+ }
+
+ @Override
+ public LocalDate getLocalDate(String column, String dateFormat,
Supplier<LocalDate> defaultSupplier)
+ throws ParseException {
+ return current().getLocalDate(column, dateFormat, defaultSupplier);
+ }
+
+ @Override
+ public Object getObject(String column, Class<?> classToConvertTo) {
+ return current().getObject(column, classToConvertTo);
+ }
+
+ @Override
+ public String toString() {
+ return "DataSetRow[rowNo=" + row.getRowNo() + ", errors=" +
errors.size() + "]";
+ }
+}
diff --git
a/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackConverter.java
b/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackConverter.java
index b234c3ef7799..a7a86b2045ed 100644
---
a/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackConverter.java
+++
b/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackConverter.java
@@ -43,6 +43,10 @@ public final class FlatpackConverter {
public static Map<String, Object> toMap(Record recordObj) {
Map<String, Object> map = new HashMap<>();
if (recordObj instanceof DataSet dataSet) {
+ if (dataSet.getRecord().isEmpty()) {
+ throw new IllegalStateException(
+ "The DataSet is not on a row, such as the whole
DataSet with splitRows=false: convert it to a List");
+ }
putValues(map, dataSet);
} else {
putValues(map, recordObj);
diff --git
a/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackEndpoint.java
b/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackEndpoint.java
index fa364470ad65..34466974f654 100644
---
a/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackEndpoint.java
+++
b/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackEndpoint.java
@@ -16,10 +16,13 @@
*/
package org.apache.camel.component.flatpack;
+import java.io.BufferedReader;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
+import java.nio.file.Path;
import net.sf.flatpack.DataSet;
import net.sf.flatpack.DefaultParserFactory;
@@ -55,14 +58,13 @@ import org.apache.camel.util.ObjectHelper;
remote = false, category = { Category.TRANSFORMATION },
headersClass = FlatpackConstants.class)
@BodyTypes(consumerOut = @BodyType(value = Carrier.POJO, javaType =
"net.sf.flatpack.DataSet",
description = "The Flatpack DataSet the
producer parsed. With splitRows, one message per row,"
- + " with the DataSet at that
row: convert it to a Map before any asynchronous"
- + " step, as the next row
moves it; a List reads all the rows and ends the"
- + " split. With
splitRows=false, the whole DataSet: convert it to a List of"
- + " Maps"),
+ + " with a DataSet of that
one row that converts to a Map. With"
+ + " splitRows=false, the
whole DataSet: convert it to a List of Maps"),
producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE }, format = "text",
description = "The fixed-width or delimited
data to parse, such as a file from the file"
- + " consumer; a java.io.File,
a Map or a List fails. The rows go to the"
- + " consumer of the same
endpoint"),
+ + " consumer, a java.io.File
or a java.nio.file.Path; a Map or a List fails."
+ + " The rows go to the
consumer of the same endpoint, and an exception there"
+ + " fails the producer"),
producerOut = @BodyType(value = Carrier.UNCHANGED))
public class FlatpackEndpoint extends DefaultPollingEndpoint {
@@ -122,13 +124,17 @@ public class FlatpackEndpoint extends
DefaultPollingEndpoint {
in.setBody(dataSet);
in.setHeader(FlatpackConstants.FLATPACK_COUNTER, counter);
loadBalancer.process(exchange);
+ if (exchange.getException() != null && originalExchange.getException()
== null) {
+ // a failure in the consumer route fails the original exchange
+ originalExchange.setException(exchange.getException());
+ }
}
public Parser createParser(Exchange exchange) throws Exception {
Reader bodyReader = null;
try {
if (FlatpackType.fixed == type) {
- bodyReader = exchange.getIn().getMandatoryBody(Reader.class);
+ bodyReader = getBodyReader(exchange);
return createFixedParser(resourceUri, bodyReader);
} else {
return createDelimitedParser(exchange);
@@ -156,7 +162,7 @@ public class FlatpackEndpoint extends
DefaultPollingEndpoint {
}
public Parser createDelimitedParser(Exchange exchange) throws
InvalidPayloadException, IOException {
- Reader bodyReader = exchange.getIn().getMandatoryBody(Reader.class);
+ Reader bodyReader = getBodyReader(exchange);
Parser parser;
if (ObjectHelper.isEmpty(getResourceUri())) {
@@ -179,6 +185,16 @@ public class FlatpackEndpoint extends
DefaultPollingEndpoint {
return parser;
}
+ private static Reader getBodyReader(Exchange exchange) throws
InvalidPayloadException {
+ Message in = exchange.getIn();
+ Object body = in.getBody();
+ if (body instanceof File || body instanceof Path) {
+ // there is no type converter from a File or Path to a Reader,
only to a BufferedReader
+ return in.getMandatoryBody(BufferedReader.class);
+ }
+ return in.getMandatoryBody(Reader.class);
+ }
+
// Properties
//-------------------------------------------------------------------------
diff --git
a/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackProducer.java
b/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackProducer.java
index dd0f427946c3..478ea015317e 100644
---
a/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackProducer.java
+++
b/components/camel-flatpack/src/main/java/org/apache/camel/component/flatpack/FlatpackProducer.java
@@ -43,8 +43,10 @@ class FlatpackProducer extends DefaultProducer {
if (endpoint.isSplitRows()) {
int counter = 0;
- while (dataSet.next()) {
- endpoint.processDataSet(exchange, dataSet, counter++);
+ // stop at the first row that fails in the consumer route
+ while (exchange.getException() == null && dataSet.next()) {
+ // each row gets its own DataSet, as the next row moves the
cursor of the parsed DataSet
+ endpoint.processDataSet(exchange, new DataSetRow(dataSet),
counter++);
}
} else {
endpoint.processDataSet(exchange, dataSet, dataSet.getRowCount());
diff --git
a/components/camel-flatpack/src/test/java/org/apache/camel/component/flatpack/FlatpackSplitRowsTest.java
b/components/camel-flatpack/src/test/java/org/apache/camel/component/flatpack/FlatpackSplitRowsTest.java
new file mode 100644
index 000000000000..5718fa6c9124
--- /dev/null
+++
b/components/camel-flatpack/src/test/java/org/apache/camel/component/flatpack/FlatpackSplitRowsTest.java
@@ -0,0 +1,201 @@
+/*
+ * 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.flatpack;
+
+import java.io.File;
+import java.io.StringReader;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+import net.sf.flatpack.DataSet;
+import net.sf.flatpack.DefaultParserFactory;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.TypeConversionException;
+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.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.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With splitRows every row message has its own row, an exception in the
consumer route fails the producer, a DataSet
+ * that is not on a row does not convert to a Map, and a File or Path body is
read.
+ */
+class FlatpackSplitRowsTest extends CamelTestSupport {
+
+ private static final String SPLIT =
"flatpack:delim:INVENTORY-Delimited.pzmap.xml";
+ private static final String NO_SPLIT =
"flatpack:delim:INVENTORY-Delimited.pzmap.xml?splitRows=false";
+ private static final String FAIL =
"flatpack:delim:INVENTORY-Delimited.pzmap.xml?allowShortLines=true";
+ private static final File DATA = new
File("src/test/data/delim/INVENTORY-CommaDelimitedWithQualifier.txt");
+ private static final List<String> ITEMS = List.of("SOME VALVE", "AN
ENGINE", "A BELT", "A BOLT");
+
+ @Test
+ void eachRowKeepsItsRowAfterTheNextRows() throws Exception {
+ MockEndpoint rows = getMockEndpoint("mock:rows");
+ rows.expectedMessageCount(4);
+
+ template.sendBody(SPLIT, DATA);
+
+ rows.assertIsSatisfied();
+ // all the rows were sent before the bodies are read, as with a seda
queue between
+ for (int i = 0; i < ITEMS.size(); i++) {
+ Message in = rows.getReceivedExchanges().get(i).getIn();
+ assertEquals(i, in.getHeader(FlatpackConstants.FLATPACK_COUNTER));
+ assertInstanceOf(DataSet.class, in.getBody());
+ assertEquals(ITEMS.get(i), in.getBody(Map.class).get("ITEM_DESC"));
+ assertEquals(ITEMS.get(i),
in.getBody(DataSet.class).getString("ITEM_DESC"));
+ }
+ }
+
+ @Test
+ void listOfOneRowDoesNotDropTheOtherRows() throws Exception {
+ MockEndpoint lists = getMockEndpoint("mock:lists");
+ lists.expectedMessageCount(4);
+
+ template.sendBody("direct:list", DATA);
+
+ lists.assertIsSatisfied();
+ for (int i = 0; i < ITEMS.size(); i++) {
+ List<?> list =
lists.getReceivedExchanges().get(i).getIn().getBody(List.class);
+ assertEquals(1, list.size());
+ assertEquals(ITEMS.get(i), ((Map<?, ?>)
list.get(0)).get("ITEM_DESC"));
+ }
+ }
+
+ @Test
+ void exceptionInConsumerRouteFailsProducer() throws Exception {
+ MockEndpoint failRows = getMockEndpoint("mock:fail-rows");
+ failRows.expectedMessageCount(2);
+
+ CamelExecutionException e = assertThrows(CamelExecutionException.class,
+ () -> template.sendBody(FAIL, DATA));
+
+ assertInstanceOf(IllegalArgumentException.class, e.getCause());
+ assertEquals("Cannot process AN ENGINE", e.getCause().getMessage());
+ // the rows after the failing row are not sent
+ failRows.assertIsSatisfied();
+ assertEquals(1,
failRows.getReceivedExchanges().get(1).getIn().getHeader(FlatpackConstants.FLATPACK_COUNTER));
+ }
+
+ @Test
+ void dataSetNotOnRowDoesNotConvertToMap() throws Exception {
+ MockEndpoint all = getMockEndpoint("mock:all");
+ all.expectedMessageCount(1);
+
+ template.sendBody(NO_SPLIT, DATA);
+
+ all.assertIsSatisfied();
+ Message in = all.getReceivedExchanges().get(0).getIn();
+ assertEquals(4, in.getBody(List.class).size());
+ TypeConversionException e =
assertThrows(TypeConversionException.class, () -> in.getBody(Map.class));
+ assertInstanceOf(IllegalStateException.class, e.getCause());
+ assertTrue(e.getCause().getMessage().contains("not on a row"),
e.getCause().getMessage());
+ }
+
+ @Test
+ void emptyInputWithoutSplitRows() throws Exception {
+ MockEndpoint all = getMockEndpoint("mock:all");
+ all.expectedMessageCount(1);
+
+ template.sendBody(NO_SPLIT,
"\"ITEM_DESC\",\"IN_STOCK\",\"PRICE\",\"LAST_RECV_DT\"\n");
+
+ all.assertIsSatisfied();
+ Message in = all.getReceivedExchanges().get(0).getIn();
+ assertEquals(0, in.getHeader(FlatpackConstants.FLATPACK_COUNTER));
+ assertTrue(in.getBody(List.class).isEmpty());
+ TypeConversionException e =
assertThrows(TypeConversionException.class, () -> in.getBody(Map.class));
+ assertInstanceOf(IllegalStateException.class, e.getCause());
+ }
+
+ @Test
+ void pathBody() throws Exception {
+ MockEndpoint rows = getMockEndpoint("mock:rows");
+ rows.expectedMessageCount(4);
+
+ template.sendBody(SPLIT, DATA.toPath());
+
+ rows.assertIsSatisfied();
+ assertEquals("A BOLT",
rows.getReceivedExchanges().get(3).getIn().getBody(Map.class).get("ITEM_DESC"));
+ }
+
+ @Test
+ void fileBodyWithFixedLength() throws Exception {
+ MockEndpoint fixed = getMockEndpoint("mock:fixed");
+ fixed.expectedMessageCount(4);
+
+ template.sendBody("flatpack:fixed:PEOPLE-FixedLength.pzmap.xml",
+
Path.of("src/test/data/fixed/PEOPLE-FixedLength.txt").toFile());
+
+ fixed.assertIsSatisfied();
+ Exchange last = fixed.getReceivedExchanges().get(3);
+ assertEquals("FRED", last.getIn().getBody(Map.class).get("FIRSTNAME"));
+ }
+
+ @Test
+ void dataSetRowIsOneReadOnlyRow() {
+ DataSet dataSet = DefaultParserFactory.getInstance()
+ .newDelimitedParser(new StringReader("A,B\n1,2\n3,4\n"), ',',
'"').parse();
+ assertTrue(dataSet.next());
+ DataSetRow row = new DataSetRow(dataSet);
+ assertTrue(dataSet.next());
+
+ assertEquals(0, row.getIndex());
+ assertEquals(1, row.getRowCount());
+ assertEquals("1", row.getString("A"));
+ assertFalse(row.next());
+ assertTrue(row.getRecord().isEmpty());
+ row.goTop();
+ assertEquals(-1, row.getIndex());
+ assertThrows(IllegalStateException.class, () -> row.getString("A"));
+ assertTrue(row.next());
+ assertEquals("2", row.getString("B"));
+ assertThrows(UnsupportedOperationException.class, () ->
row.setValue("A", "5"));
+ assertEquals("3", dataSet.getString("A"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(SPLIT).to("mock:rows");
+
+
from("direct:list").to("flatpack:delim:INVENTORY-Delimited.pzmap.xml?ignoreExtraColumns=true");
+
from("flatpack:delim:INVENTORY-Delimited.pzmap.xml?ignoreExtraColumns=true")
+ .convertBodyTo(List.class)
+ .to("mock:lists");
+
+ from(FAIL)
+ .to("mock:fail-rows")
+ .filter(exchange -> "AN
ENGINE".equals(exchange.getIn().getBody(DataSet.class).getString("ITEM_DESC")))
+ .throwException(new IllegalArgumentException("Cannot
process AN ENGINE"));
+
+ from(NO_SPLIT).to("mock:all");
+
+
from("flatpack:fixed:PEOPLE-FixedLength.pzmap.xml").to("mock:fixed");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 87cacf8cb5cc..f6c816a7f2f8 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -913,6 +913,20 @@ as soon as the encoding changed a byte, a body read with
another charset (such a
characters. The checksum is now calculated from the bytes written to the file,
and the body is decoded as it is
without `checksumFileAlgorithm`. Checksums of such files differ from the ones
written before.
+=== camel-flatpack - with splitRows each message has its own row, and a
failing row fails the producer
+
+With `splitRows=true` (the default), every row message shared the one parsed
`DataSet`, so the next row moved the row
+of a message that was not yet converted. The body of each row message is now
its own read-only `DataSet` with only that
+row (`org.apache.camel.component.flatpack.DataSetRow`): it converts to a `Map`
as before, converting it to a `List`
+gives that one row instead of all the rows, and the methods that change the
`DataSet` (such as `setValue` or
+`remove`) throw an `UnsupportedOperationException`.
+
+An exception in the route that consumes the rows was lost, and the producer
carried on with the next rows. The
+exception now fails the producer, and the rows after the failing row are not
sent.
+
+Converting a `DataSet` that is not on a row (the whole `DataSet` with
`splitRows=false`) to a `Map` now fails with an
+`IllegalStateException` instead of a `NullPointerException`; convert it to a
`List`.
+
=== camel-ftp, camel-mina-sftp - the SFTP consumers resolve listed symbolic
links for jailStartingDirectory
With `jailStartingDirectory` enabled (the default), the `sftp` and `mina-sftp`
consumers now resolve a listed file that
