Code after Splitter with aggregation strategy is not executed if exception in inner route were handled (Apache Camel)

user_5624378 0
Viewed 865

I've faced with behavior that I can't understand. This issue happens when Split with AggregationStrategy is executed and during one of the iterations, an exception occurs. An exception occurs inside of Splitter in another route (direct endpoint which is called for each iteration). Seems like route execution stops just after Splitter.

Here is sample code.

This is a route that builds one report per each client and collects names of files for internal statistics.


    @Component
    @RequiredArgsConstructor
    @FieldDefaults(level = PRIVATE, makeFinal = true)
    public class ReportRouteBuilder extends RouteBuilder {
    
      ClientRepository clientRepository;
    
      @Override
      public void configure() throws Exception {
        errorHandler(deadLetterChannel("direct:handleError")); //handles an error, adds error message to internal error collector for statistic and writes log
    
        from("direct:generateReports")
        
            .setProperty("reportTask", body()) //at this point there is in the body an object of type ReportTask, containig all data required for building report
    
            .bean(clientRepository, "getAllClients") // Body is a List<Client>
            
            .split(body())
              .aggregationStrategy(new FileNamesListAggregationStrategy())
              .to("direct:generateReportForClient") // creates report which is saved in the file system. uses the same error handler
            .end()
            
            //when an exception occurs during split then code after splitter is not executed
    
            .log("Finished generating reports. Files created ${body}"); // Body has to be List<String> with file names.
      }
    }

AggregationStrategy is pretty simple - it just extracts the name of the file. If the header is absent it returns NULL.


    public class FileNamesListAggregationStrategy extends AbstractListAggregationStrategy<String> {
    
      @Override
      public String getValue(Exchange exchange) {
        Message inMessage = exchange.getIn();
        return inMessage.getHeader(Exchange.FILE_NAME, String.class);
      }
    }

When everything goes smoothly after splitting there is in the Body List with all file names. But when in the route "direct:generateReportForClient" some exception occurred (I've added error simulation for one client) than aggregated body just contains one less file name -it's OK (everything was aggregated correctly).

BUT just after Split after route execution stops and result that is in the body at this point (List with file names) is returned to the client (FluentProducer) which expects ReportTask as a response body.

and it tries to convert value - List (aggregated result) to ReportTask and it causes org.apache.camel.NoTypeConversionAvailableException: No type converter available to convert from type

Why route breaks after split? All errors were handled and aggregation finished correctly.

PS I've read Camel In Action book and Documentation about Splitter but I haven't found the answer.

PPS project runs on Spring Boot 2.3.1 and Camel 3.3.0

UPDATE This route is started by FluentProducerTemplate

    ReportTask processedReportTask = producer.to("direct:generateReports")
            .withBody(reportTask)
            .request(ReportTask.class);
apache-camel
1 Answers

The problem is error handler + custom aggregation strategy in the split.

From Camel in Action book (5.3.5):

WARNING When using a custom AggregationStrategy with the Splitter, it’s important to know that you’re responsible for handling exceptions. If you don’t propagate the exception back, the Splitter will assume you’ve handled the exception and will ignore it.

In your code, you use the aggregation strategy extended from AbstractListAggregationStrategy. Let's look to aggregate method in AbstractListAggregationStrategy:

@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
    List<V> list;

    if (oldExchange == null) {
        list = getList(newExchange);
    } else {
        list = getList(oldExchange);
    }

    if (newExchange != null) {
        V value = getValue(newExchange);
        if (value != null) {
            list.add(value);
        }
    }

    return oldExchange != null ? oldExchange : newExchange;
}

If a first exchange is handled by error handler we will have in result exchange (newExchange) number of properties set by Error Handler (Exchange.EXCEPTION_CAUGHT, Exchange.FAILURE_ENDPOINT, Exchange.ERRORHANDLER_HANDLED and Exchange.FAILURE_HANDLED) and exchange.errorHandlerHandled=true. Methods getErrorHandlerHandled()/setErrorHandlerHandled(Boolean errorHandlerHandled) are available in ExtendedExchange interface.

In this case, your split finishes with an exchange with errorHandlerHandled=true and it breaks the route.

The reason is described in camel exception clause manual

If handled is true, then the thrown exception will be handled and Camel will not continue routing in the original route, but break out.

To prevent this behaviour you can cast your exchange to ExtendedExchange and set errorHandlerHandled=false in the aggregation strategy aggregate method. And your route won't be broken but will be continued.

  @Override
  public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
    Exchange aggregatedExchange = super.aggregate(oldExchange, newExchange);
    ((ExtendedExchange) aggregatedExchange).setErrorHandlerHandled(false);
    return aggregatedExchange;
  }

The tricky situation is that if you have exchange handled by Error Handler as not a first one in your aggregation strategy you won't face any issue. Because camel will use the first exchange(without errorHandlerHandled=true) as a base for aggregation.

user_32754450
Related