Working wiht Apache Camel I had a problem of stucked thread in parallel processing
My Problem
Sometimes, with the parallel processing, my application ends to work and result stucked in the split without doing any operation
My Configuration
I have a springboot application with Springboot 3.2.0 and Apache Camel 4.2.0.
I found in the Apache Camel documentation that the best approach is to create a profile for each split or multicast with the parallel processing, so I created them in my
application.yml
camel:
threadpool:
config[first-split-profile]:
id: "first-split-profile"
pool-size: 5
max-pool-size: 5
max-queue-size: -1
config[second-split-profile]:
id: "second-split-profile"
pool-size: 20
max-pool-size: 20
max-queue-size: -1
config[third-split-profile]:
id: "third-split-profile"
pool-size: 50
max-pool-size: 50
max-queue-size: -1
config[multicast-profile]:
id: "multicast-profile"
pool-size: 50
max-pool-size: 50
max-queue-size: -1
Most of the times the application works without problem, but sometimes, suddenly stop to work.
It is not a stop with an exception or because is executing a long operation, it seems like the threads are all waiting and no one complete the operation.
I tried to replicate the structure of my project with simple operation and I can still face the problem.
Only when I remove the parallelProcessing() and the thread configuration everything work in the correct way.
The actual structure is a tree:
- I receive a List in the first route
- Split the body
- From each of the splitted element create a new List
- Send the new List to the second route
- split the body
- From each of the splitted element create a new List
- Send the new List to the third route
- split the body
- For each of the splitted element perfom e multicast parallel log
So with the actual values in the test class I have a List of 5 elements as the body of the first route (5 total elements).
Each of them send a List of 100 elements to the second route (5*1000 total elemens).
Each of them send a List of 10 elements to the third route (5*100*100 total elements).
I replaced the operation of my application (simple SELECT or INSERT operation of one record) with only some logs.
How to replicate the issue
Routes:
@Component
@Slf4j
@RequiredArgsConstructor
public class ParallelThreadsRoute extends RouteBuilder {
private final CallSecondRouteProcessor callSecondRouteProcessor;
private final CallThirdRouteProcessor callThirdRouteProcessor;
@Override
public void configure() throws Exception {
from("direct:route1")
.routeId("route1")
.log(LoggingLevel.INFO, "Start route 1 with n elements: [${body.size}]")
.split(body()).parallelProcessing().executorService("first-split-profile")
.process(callSecondRouteProcessor)
.end()
from("direct:route2")
.routeId("route2")
.log(LoggingLevel.INFO, "Start route 2 with n elements: [${body.size}]")
.split(body()).parallelProcessing().executorService("second-split-profile")
.process(callThirdRouteProcessor)
.end();
from("direct:route3")
.routeId("route3")
.log(LoggingLevel.INFO, "Start route 3 with n elements: [${body.size}]")
.split(body()).parallelProcessing().executorService("third-split-profile")
.process(new Processor(){
@Override
public void process(Exchange exchange){
log.info("Log in processor with body: [{}]", exchange.getIn().getBody(String.class));
}
})
.multicast().parallelProcessing().executorService("multicast-profile")
.process(new Processor(){
@Override
public void process(Exchange exchange){
log.info("Log in first multicast with body: [{}]", exchange.getIn().getBody(String.class));
}
})
.process(new Processor(){
@Override
public void process(Exchange exchange){
log.info("Log in second multicast with body: [{}]", exchange.getIn().getBody(String.class));
}
})
.end()
.end();
}
}
Service that start the operation:
@Service
@Slf4j
@RequiredArgsConstructor
public class CallFirstRouteService {
private final CamelContext context;
@Produce("direct:route1")
private ProducerTemplate route1Producer;
@Handler
public void callFirstRoute(List<String> firstRouteElements, int elementsSecondRoute,
int elementsThirdRoute) throws Exception {
Exchange newExchange = ExchangeBuilder.anExchange(context)
.withProperty("SECOND_ROUTE_ELEMENTS", elementsSecondRoute)
.withProperty("THIRD_ROUTE_ELEMENTS", elementsThirdRoute)
.withBody(firstRouteElements)
.build();
Exchange outputExchange = route1Producer.send(newExchange);
//ISOLATE LEVEL TO PERFORM SOME OPERATION WITH THE OUTPUT
//AND KEEP ORIGINAL HEADERS AND PROPERTIES
if (outputExchange.isFailed()) {
throw outputExchange.getException();
}
log.info("End of routes with [{}] elements", firstRouteElements);
}
}
Processors:
@Component
@Slf4j
@RequiredArgsConstructor
public class CallSecondRouteProcessor implements Processor{
private final CamelContext context;
@Produce("direct:route2")
private ProducerTemplate route2Producer;
@Override
public void process(Exchange exchange) throws Exception {
int elementsSecondRoute = exchange.getProperty("SECOND_ROUTE_ELEMENTS", Integer.class);
List<String> secondRouteElements = new ArrayList<>();
for(int i = 1; i <= elementsSecondRoute; i++){
secondRouteElements.add(exchange.getIn().getBody(String.class) + "_" + i);
}
Exchange newExchange = ExchangeBuilder.anExchange(context)
.withProperty("THIRD_ROUTE_ELEMENTS", elementsThirdRoute)
.withBody(secondRouteElements)
.build();
Exchange outputExchange = route2Producer.send(newExchange);
//ISOLATE LEVEL TO PERFORM SOME OPERATION WITH THE OUTPUT
//AND KEEP ORIGINAL HEADERS AND PROPERTIES
if (outputExchange.isFailed()) {
throw outputExchange.getException();
}
log.info("End of routes with [{}] elements", secondRouteElements);
}
}
@Component
@Slf4j
@RequiredArgsConstructor
public class CallThirdRouteProcessor implements Processor{
private final CamelContext context;
@Produce("direct:route3")
private ProducerTemplate route3Producer;
@Override
public void process(Exchange exchange) throws Exception {
int elementsThirdRoute = exchange.getProperty("THIRD_ROUTE_ELEMENTS", Integer.class);
List<String> thirdRouteElements = new ArrayList<>();
for(int i = 1; i <= elementsThirdRoute; i++){
thirdRouteElements.add(exchange.getIn().getBody(String.class) + "_" + i);
}
Exchange newExchange = ExchangeBuilder.anExchange(context)
.withBody(thirdRouteElements)
.build();
Exchange outputExchange = route3Producer.send(newExchange);
//ISOLATE LEVEL TO PERFORM SOME OPERATION WITH THE OUTPUT
//AND KEEP ORIGINAL HEADERS AND PROPERTIES
if (outputExchange.isFailed()) {
throw outputExchange.getException();
}
log.info("End of routes with [{}] elements", elementsThirdRoute);
}
}
Test class to reproduce the case:
@SpringBootTest
@EnableScheduling
@Slf4j
class TestParallelThreadsTest {
@Autowired
private CallFirstRouteService callFirstRouteService;
@Autowired
private ThreadService threadService;
@Test
void testCall(){
int elementsFirstRoute = 5;
int elementsSecondRoute = 1000;
int elementsThirdRoute = 100;
List<String> firstRouteElements = new ArrayList<>();
for(int i = 1; i <= elementsFirstRoute; i++){
firstRouteElements.add("Element_" + i));
}
callFirstRouteService.callFirstRoute(firstRouteElements, elementsSecondRoute, elementsThirdRoute);
}
@Scheduled(cron = "0 */2 * * * *")
void callThread() {
log.info("Active thread: [{}]", threadService.getThreadsInformations());
}
}
How to control what happens
For me the problem is something like:
“In the second-route, thread X from the element 1 of first-route and thread Y from the element 2 of first-route ended their operation in the same moment. Now only one thread is free. Which of them should use it?”
In this way without an orchestrator they remain stucked.
To be sure there are no operation to complete I created a service that log all the current threads of the application. It seems that when the application stop to work there aren’t other operation in running.
This is the Service class already injected in the test:
@Service
@Slf4j
@RequiredArgsConstructor
public class ThreadService {
public List<String> getThreadsInformations() {
Map<Thread, StackTraceElement[]> allThreads = Thread.getAllStackTraces();
List<String> threadsDetails = new ArrayList<>();
for (Map.Entry<Thread, StackTraceElement[]> entry : allThreads.entrySet()) {
Thread thread = entry.getKey();
threadsDetails.add(String.format("Thread: [%s] in state: [%s]", thread.getName(), thread.getName()));
}
return threadsDetails;
}
}
What I need
Does somebody knows why the application is stuck (only sometimes) and how can I avoid it in my implementation?
I still need to do operation in parallel
Thank you in advance for your help