This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26128-flatpack-split-rows in repository https://gitbox.apache.org/repos/asf/camel.git
commit 6863bf350d95cda0693a80fc165ff89eecac98be Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 16:53:56 2026 +0100 CAMEL-26128: camel-flatpack - with splitRows each row message has its 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 1bc188419d61..6c0e9df0a638 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 @@ -861,6 +861,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
