You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@camel.apache.org by "Andre Weickel (Jira)" <ji...@apache.org> on 2024/08/23 12:39:00 UTC
[jira] [Updated] (CAMEL-21114) ZipSplitter with AggregationStrategy does not aggregate all splits
[ https://issues.apache.org/jira/browse/CAMEL-21114?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Andre Weickel updated CAMEL-21114:
----------------------------------
Description:
A transacted route with ZipSplitter and Aggregation Strategy does not aggregate the last zip file entry. The issue only occurs for transacted routes.
Example:
_Zip Archive_
* _A.xml_
* _B.xml_
Both splits are processed but only for the first exchange (A.xml) the aggregate method is called.
For a zip archive with two entries the doRun() method of MulticastTransactedTask is called three times. The third time iterator.next() returns null although hasNext() was true. As a result the doDone() method is called but there is still a task in the queue (with the second exchange). This task is processed after doDone() was executed but it’s not aggregated because of a done check in aggregate() of MulticastTransactedTask.
We found the problem in Camel 3.14, but it is still present in Camel 3.22.
It can be reproduced with the following test (it works if you remove the transacted tag from the route)
{code:java}
import org.apache.camel.AggregationStrategy;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.dataformat.zipfile.ZipSplitter;
import org.apache.camel.spring.spi.SpringTransactionPolicy;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.Test;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;
public class ZipSplitterTest extends CamelTestSupport {
String zipArchiveWithTwoFiles ="UEsDBBQAAAAIAFlrtFDFAfecUAAAAB4BAAALAAAAT3JkZXJzMS54bWyzyS9KSS0qtuPl4oQwQSxOm8wUOxMb/cwUCK+gKD+lNLkEzOG0yUvMTbWDCik42uiD+WB1+kgKbfThxqEZbEqUwU6kG2xGlMHOhA2GsortAFBLAwQUAAAACABBW9hQgBf0tVgAAAAqAQAACwAAAE9yZGVyczIueG1ss8kvSkktKrbj5eKEMEEsTpvMFDtDQ0Mb/cwUCL+gKD+lNLkEzOG0yUvMTbWDCimA1YFFwCr1kZTa6MONRDPcyMiIKMPB6kg13NjYmCjDweoIGQ5lFdsBAFBLAQIfABQAAAAIAFlrtFDFAfecUAAAAB4BAAALACQAAAAAAAAAIAAAAAAAAABPcmRlcnMxLnhtbAoAIAAAAAAAAQAYAAD57I2ZLtYBg97kuHn02gEA+eyNmS7WAVBLAQIfABQAAAAIAEFb2FCAF/S1WAAAACoBAAALACQAAAAAAAAAIAAAAHkAAABPcmRlcnMyLnhtbAoAIAAAAAAAAQAYAAAxPXoJStYBjn3iuHn02gEAMT16CUrWAVBLBQYAAAAAAgACALoAAAD6AAAAAAA=";
@Test
public void testIfAllSplitsAggregated() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
template.sendBody("direct:start", "");
// Check if second file was processed in aggregate() method of AggregationStrategy
assertEquals("Orders2.xml", mock.getExchanges().get(0).getMessage().getHeader("CamelFileName", String.class));
}
@Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL("jdbc:h2:mem:testdb;DB_CLOSE_DELAY=-1");
dataSource.setUser("sa");
dataSource.setPassword("");
DataSourceTransactionManager txManager = new
DataSourceTransactionManager(dataSource);
TransactionTemplate transactionTemplate = new TransactionTemplate(txManager);
transactionTemplate.setPropagationBehaviorName("PROPAGATION_REQUIRED");
transactionTemplate.setIsolationLevelName("ISOLATION_READ_COMMITTED");
transactionTemplate.setTimeout(1800); SpringTransactionPolicy springTransactionPolicy = new SpringTransactionPolicy();
springTransactionPolicy.setTransactionManager(txManager);
springTransactionPolicy.setTransactionTemplate(transactionTemplate);
getContext().getRegistry().bind("transacted", springTransactionPolicy);
getContext().getRegistry().bind("zipSplitter", new ZipSplitter()); from("direct:start")
.transacted("transacted")
.setBody().simple(zipArchiveWithTwoFiles)
.unmarshal().base64()
.split().ref("zipSplitter").streaming().aggregationStrategy(new StringAggregationStrategy())
.log("Splitted")
.end()
.to("mock:result");
}
};
}
private static class StringAggregationStrategy implements AggregationStrategy {
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
System.out.println(newExchange.getMessage().getHeader("CamelFileName"));
return newExchange;
}
}
}{code}
was:
A transacted route with ZipSplitter and Aggregation Strategy does not aggregate the last zip file entry. The issue only occurs for transacted routes.
Example:
_Zip Archive_
* _A.xml_
* _B.xml_
Both splits are processed but only for the first exchange (A.xml) the aggregate method is called.
For a zip archive with two entries the doRun() method of MulticastTransactedTask is called three times. The third time iterator.next() returns null although hasNext() was true. As a result the doDone() method is called but there is still a task in the queue (with the second exchange). This task is processed after doDone() was executed but it’s not aggregated because of a done check in aggregate() of MulticastTransactedTask.
We found the problem in Camel 3.14, but it is still present in Camel 3.22.
It can be reproduced with the following test (it works if you remove the transacted tag from the route)
{code:java}
import org.apache.camel.AggregationStrategy;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.dataformat.zipfile.ZipSplitter;
import org.apache.camel.spring.spi.SpringTransactionPolicy;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.h2.jdbcx.JdbcDataSource;
import org.junit.Test;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;
public class ZipSplitterTest extends CamelTestSupport {
String zipArchiveWithTwoFiles ="UEsDBBQAAAAIAFlrtFDFAfecUAAAAB4BAAALAAAAT3JkZXJzMS54bWyzyS9KSS0qtuPl4oQwQSxOm8wUOxMb/cwUCK+gKD+lNLkEzOG0yUvMTbWDCik42uiD+WB1+kgKbfThxqEZbEqUwU6kG2xGlMHOhA2GsortAFBLAwQUAAAACABBW9hQgBf0tVgAAAAqAQAACwAAAE9yZGVyczIueG1ss8kvSkktKrbj5eKEMEEsTpvMFDtDQ0Mb/cwUCL+gKD+lNLkEzOG0yUvMTbWDCimA1YFFwCr1kZTa6MONRDPcyMiIKMPB6kg13NjYmCjDweoIGQ5lFdsBAFBLAQIfABQAAAAIAFlrtFDFAfecUAAAAB4BAAALACQAAAAAAAAAIAAAAAAAAABPcmRlcnMxLnhtbAoAIAAAAAAAAQAYAAD57I2ZLtYBg97kuHn02gEA+eyNmS7WAVBLAQIfABQAAAAIAEFb2FCAF/S1WAAAACoBAAALACQAAAAAAAAAIAAAAHkAAABPcmRlcnMyLnhtbAoAIAAAAAAAAQAYAAAxPXoJStYBjn3iuHn02gEAMT16CUrWAVBLBQYAAAAAAgACALoAAAD6AAAAAAA=";
@Test
public void testIfAllSplitsAggregated() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:result");
template.sendBody("direct:start", ""); // Check if second file was processed in aggregate() method of AggregationStrategy
assertEquals("Orders2.xml", mock.getExchanges().get(0).getMessage().getHeader("CamelFileName", String.class));
} @Override
protected RouteBuilder createRouteBuilder() throws Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
JdbcDataSource dataSource = new JdbcDataSource();
dataSource.setURL("jdbc:h2:mem:testdb;DB_CLOSE_DELAY=-1");
dataSource.setUser("sa");
dataSource.setPassword(""); DataSourceTransactionManager txManager = new DataSourceTransactionManager(dataSource); TransactionTemplate transactionTemplate = new TransactionTemplate(txManager);
transactionTemplate.setPropagationBehaviorName("PROPAGATION_REQUIRED");
transactionTemplate.setIsolationLevelName("ISOLATION_READ_COMMITTED");
transactionTemplate.setTimeout(1800); SpringTransactionPolicy springTransactionPolicy = new SpringTransactionPolicy();
springTransactionPolicy.setTransactionManager(txManager);
springTransactionPolicy.setTransactionTemplate(transactionTemplate);
getContext().getRegistry().bind("transacted", springTransactionPolicy);
getContext().getRegistry().bind("zipSplitter", new ZipSplitter()); from("direct:start")
.transacted("transacted")
.setBody().simple(zipArchiveWithTwoFiles)
.unmarshal().base64()
.split().ref("zipSplitter").streaming().aggregationStrategy(new StringAggregationStrategy())
.log("Splitted")
.end()
.to("mock:result");
}
};
}
private static class StringAggregationStrategy implements AggregationStrategy {
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
System.out.println(newExchange.getMessage().getHeader("CamelFileName"));
return newExchange;
}
}
}{code}
> ZipSplitter with AggregationStrategy does not aggregate all splits
> ------------------------------------------------------------------
>
> Key: CAMEL-21114
> URL: https://issues.apache.org/jira/browse/CAMEL-21114
> Project: Camel
> Issue Type: Bug
> Affects Versions: 3.14.10
> Reporter: Andre Weickel
> Priority: Major
>
> A transacted route with ZipSplitter and Aggregation Strategy does not aggregate the last zip file entry. The issue only occurs for transacted routes.
>
> Example:
>
> _Zip Archive_
> * _A.xml_
> * _B.xml_
>
> Both splits are processed but only for the first exchange (A.xml) the aggregate method is called.
> For a zip archive with two entries the doRun() method of MulticastTransactedTask is called three times. The third time iterator.next() returns null although hasNext() was true. As a result the doDone() method is called but there is still a task in the queue (with the second exchange). This task is processed after doDone() was executed but it’s not aggregated because of a done check in aggregate() of MulticastTransactedTask.
>
> We found the problem in Camel 3.14, but it is still present in Camel 3.22.
>
> It can be reproduced with the following test (it works if you remove the transacted tag from the route)
> {code:java}
> import org.apache.camel.AggregationStrategy;
> import org.apache.camel.Exchange;
> import org.apache.camel.builder.RouteBuilder;
> import org.apache.camel.component.mock.MockEndpoint;
> import org.apache.camel.dataformat.zipfile.ZipSplitter;
> import org.apache.camel.spring.spi.SpringTransactionPolicy;
> import org.apache.camel.test.junit4.CamelTestSupport;
> import org.h2.jdbcx.JdbcDataSource;
> import org.junit.Test;
> import org.springframework.jdbc.datasource.DataSourceTransactionManager;
> import org.springframework.transaction.support.TransactionTemplate;
> public class ZipSplitterTest extends CamelTestSupport {
> String zipArchiveWithTwoFiles ="UEsDBBQAAAAIAFlrtFDFAfecUAAAAB4BAAALAAAAT3JkZXJzMS54bWyzyS9KSS0qtuPl4oQwQSxOm8wUOxMb/cwUCK+gKD+lNLkEzOG0yUvMTbWDCik42uiD+WB1+kgKbfThxqEZbEqUwU6kG2xGlMHOhA2GsortAFBLAwQUAAAACABBW9hQgBf0tVgAAAAqAQAACwAAAE9yZGVyczIueG1ss8kvSkktKrbj5eKEMEEsTpvMFDtDQ0Mb/cwUCL+gKD+lNLkEzOG0yUvMTbWDCimA1YFFwCr1kZTa6MONRDPcyMiIKMPB6kg13NjYmCjDweoIGQ5lFdsBAFBLAQIfABQAAAAIAFlrtFDFAfecUAAAAB4BAAALACQAAAAAAAAAIAAAAAAAAABPcmRlcnMxLnhtbAoAIAAAAAAAAQAYAAD57I2ZLtYBg97kuHn02gEA+eyNmS7WAVBLAQIfABQAAAAIAEFb2FCAF/S1WAAAACoBAAALACQAAAAAAAAAIAAAAHkAAABPcmRlcnMyLnhtbAoAIAAAAAAAAQAYAAAxPXoJStYBjn3iuHn02gEAMT16CUrWAVBLBQYAAAAAAgACALoAAAD6AAAAAAA=";
>
> @Test
> public void testIfAllSplitsAggregated() throws Exception {
> MockEndpoint mock = getMockEndpoint("mock:result");
> template.sendBody("direct:start", "");
> // Check if second file was processed in aggregate() method of AggregationStrategy
> assertEquals("Orders2.xml", mock.getExchanges().get(0).getMessage().getHeader("CamelFileName", String.class));
> }
> @Override
> protected RouteBuilder createRouteBuilder() throws Exception {
> return new RouteBuilder() {
> @Override
> public void configure() throws Exception {
>
> JdbcDataSource dataSource = new JdbcDataSource();
> dataSource.setURL("jdbc:h2:mem:testdb;DB_CLOSE_DELAY=-1");
> dataSource.setUser("sa");
> dataSource.setPassword("");
> DataSourceTransactionManager txManager = new
> DataSourceTransactionManager(dataSource);
> TransactionTemplate transactionTemplate = new TransactionTemplate(txManager);
> transactionTemplate.setPropagationBehaviorName("PROPAGATION_REQUIRED");
> transactionTemplate.setIsolationLevelName("ISOLATION_READ_COMMITTED");
> transactionTemplate.setTimeout(1800); SpringTransactionPolicy springTransactionPolicy = new SpringTransactionPolicy();
> springTransactionPolicy.setTransactionManager(txManager);
> springTransactionPolicy.setTransactionTemplate(transactionTemplate);
>
> getContext().getRegistry().bind("transacted", springTransactionPolicy);
> getContext().getRegistry().bind("zipSplitter", new ZipSplitter()); from("direct:start")
> .transacted("transacted")
> .setBody().simple(zipArchiveWithTwoFiles)
> .unmarshal().base64()
> .split().ref("zipSplitter").streaming().aggregationStrategy(new StringAggregationStrategy())
> .log("Splitted")
> .end()
> .to("mock:result");
> }
> };
> }
>
> private static class StringAggregationStrategy implements AggregationStrategy {
> @Override
> public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
> System.out.println(newExchange.getMessage().getHeader("CamelFileName"));
> return newExchange;
> }
> }
> }{code}
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)