[CAMEL-20614] KameletConsumerNotAvailableException is thrown when two or more threads call toD - ASF JIRA

XML

Word

Printable

JSON

Details

Type: Bug
Status: Resolved
Priority: Major
Resolution: Fixed
Affects Version/s: 3.22.1, 4.4.1
Fix Version/s: 4.0.5, 4.4.2, 4.6.0
Component/s: camel-core
Labels:
None

Patch Info:

Patch Available
Estimated Complexity:
Unknown

Description

When two or more threads try to instantiate a route template / kamelet parallelly with toD, the following exception is thrown:

Thread[#77,Camel (camel-1) thread #15 - Split,5,main]
org.apache.camel.component.kamelet.KameletConsumerNotAvailableException: No consumers available on endpoint: kamelet://my-kamelet?someParameter=someValue. Exchange[17F2F4997569083-000000000000000C]
	at org.apache.camel.component.kamelet.KameletProducer.process(KameletProducer.java:78)
	at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:172)
	at org.apache.camel.processor.errorhandler.NoErrorHandler.process(NoErrorHandler.java:46)
	at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:383)
	at org.apache.camel.processor.Pipeline$PipelineTask.run(Pipeline.java:102)
	at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.executeReactiveWork(DefaultReactiveExecutor.java:196)
	at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:164)
	at org.apache.camel.impl.engine.DefaultReactiveExecutor.schedule(DefaultReactiveExecutor.java:54)
	at org.apache.camel.processor.MulticastProcessor.lambda$schedule$1(MulticastProcessor.java:369)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:577)
	at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:317)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1589)

The sequence of events leading to this exception is as follows:

1. Thread A creates an endpoint with org.apache.camel.impl.engine.AbstractCamelContext#doGetEndpoint
2. Thread A initiates the endpoint with org.apache.camel.component.kamelet.KameletEndpoint#doInit and sets a key, e.g. my-kamelet-1
3. Thread A creates a new instance of a route template / kamelet with org.apache.camel.impl.DefaultModel#addRouteFromTemplate. This method delegates to org.apache.camel.model.RouteTemplateDefinition#asRouteDefinition which copies RouteDefinition from the template, but it is only a shallow copy (not a deep copy), so any referenced fields (e.g. RouteDefinition#input, RouteDefinition#outputs) are shared between threads.
4. Thread A updates the id of RouteDefinition (routeId=1) and URI-s of RouteDefinition#input (kamelet://source?routeId=1) and RouteDefinition#outputs (kamelet://sink?routeId=1) (see org.apache.camel.component.kamelet.Kamelet#templateToRoute) which is also reflected in the template's RouteDefinition.
5. Thread B executes steps 1-4 and overwrites the URI-s of in the template's RouteDefinition (my-kamelet-2), hence in the shallow-copied RouteDefinition referenced by Thread A.
6. Thread A instantiates a Consumer for the URI from RouteDefinition#input, but because it has been modified by Thread B, it will create a Consumer for my-kamelet-2 instead of my-kamelet-1.
7. Thread A tries to send an Exchange to the route my-kamelet-1 and creates a Producer with the same key (my-kamelet-1). This Producer searches for a Consumer with the same key (my-kamelet-1), but it can't find it because in the previous step a Consumer with a different key (my-kamelet-2) was created and it throws org.apache.camel.component.kamelet.KameletConsumerNotAvailableException.

As a solution org.apache.camel.component.kamelet.Kamelet#templateToRoute can be synchronised and referenced fields that are modified in this method should also be copied.