Camel Kafka OpenTelemetry Duplicate traceId
- Dominant language
- Java
- Stars
- 302
- Forks
- 232
- Avg merge
- 1d 20h
- Merged PRs (30d)
- 114
Description
### Bug description
I have many routes like the one below. When it comes under heavy load, that is, when there are too many messages from Kafka, the route seems to get stuck. Each event starts appearing with the same traceId. He only does this sometimes. Sometimes It happens after 3-4 days. I expect to see 1 received log with 1 traceId.
@Inject
CamelContext camelContext;
route:
```
camelContext.setUseMDCLogging(true);
from("kafka:{{kafka.test.in.topic.name}}").routeId(getRouteId("test", RouteType.KAFKA.getValue()))
.log("Received : ${body}")
.to("bean:EventUtil?method=createEvent(*,test,in)") -> mongo save async
.to("bean:testServiceImp?method=process(*)") ->process and postgre save if same data exist return null
.choice().when(body().isNull())
.log("The route was terminated because body was null")
.endChoice().otherwise()
.to("bean:EventUtil?method=setHeader(*,grn-pnf, com.test.event.test)") -> kafka message header set
.process(exchange -> {
CloudEventDTO message = mapper.readValue(exchange.getIn().getBody(String.class), new TypeReference>() {
});
message.setId(exchange.getMessage().getHeader(CloudEvent.CE_ID.getValue(), String.class));
message.setSource(exchange.getMessage().getHeader(CloudEvent.CE_SOURCE.getValue(), String.class));
message.setType(exchange.getMessage().getHeader(CloudEvent.CE_TYPE.getValue(), String.class));
message.setSubject(exchange.getMessage().getHeader(CloudEvent.CE_SUBJECT.getValue(), String.class));
message.setSpecVersion(exchange.getMessage().getHeader(CloudEvent.CE_SPECVERSION.getValue(), String.class));
message.setDataContentType(exchange.getMessage().getHeader(CloudEvent.CONTENT_TYPE.getValue(), String.class));
message.setTime(LocalDateTime.now(Clock.systemUTC()));
exchange.getMessage().setBody(mapper.writeValueAsString(message));
})
.to("bean:EventUtil?method=createEvent(*,test,out)")
.to("kafka:{{kafka.test.topic.name}}")
.log("Sent: ${body} To: " + getTopicName("test", RouteDirection.OUT.getValue()))
.end();
```
pom:
```
3.11.0
false
0.2.0
17
1.5.5.Final
1.18.28
UTF-8
UTF-8
quarkus-bom
io.quarkus.platform
3.4.1
2.0.3
2.0.2
true
true
3.1.2
${quarkus.platform.group-id}
${quarkus.platform.artifact-id}
${quarkus.platform.version}
pom
import
${quarkus.platform.group-id}
quarkus-camel-bom
${quarkus.platform.version}
pom
import
org.apache.camel.quarkus
camel-quarkus-opentelemetry
io.opentelemetry.instrumentation
opentelemetry-jdbc
org.apache.camel.quarkus
camel-quarkus-core
org.apache.camel.quarkus
camel-quarkus-kafka
org.apache.camel.quarkus
camel-quarkus-direct
org.apache.camel.quarkus
camel-quarkus-http
org.apache.camel.quarkus
camel-quarkus-jackson
org.apache.camel.quarkus
camel-quarkus-bean
```
log example:
2023-09-29 01:14:04.376
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=09e679cba592a965, spanId=2e202a2bbe6d5360, sampled=true [testInfoKafkaRoute] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) Received : {"testJson}
2023-09-29 01:14:04.377
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=2e202a2bbe6d5360, spanId=d8fb606803bb3571, sampled=true [ri.gr.wo.se.pa.im.TestInfoServiceImp] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) Processing... Incoming Test Info
2023-09-29 01:14:04.378
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=2e202a2bbe6d5360, spanId=d8fb606803bb3571, sampled=true [ri.gr.wo.se.pa.im.TestInfoSsrServiceImp] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) Processing... Incoming Test Ssr Info
2023-09-29 01:14:04.378
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=2e202a2bbe6d5360, spanId=d8fb606803bb3571, sampled=true [ri.gr.wo.se.AbstractService] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) The same data exists. Entity id: 11,015,523
2023-09-29 01:14:04.379
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=09e679cba592a965, spanId=2e202a2bbe6d5360, sampled=true [testInfoKafkaRoute] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) The route was terminated because body was null
2023-09-29 01:14:04.379
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=09e679cba592a965, spanId=1bc2b6898c093d0f, sampled=true [testInfoKafkaRoute] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) Received : {"testJson"}
2023-09-29 01:14:04.379
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=1bc2b6898c093d0f, spanId=b34387e031536af4, sampled=true [ri.gr.wo.se.pa.im.TestInfoServiceImp] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) Processing... Incoming Test Info
2023-09-29 01:14:04.380
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=1bc2b6898c093d0f, spanId=b34387e031536af4, sampled=true [ri.gr.wo.se.pa.im.TestInfoSsrServiceImp] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) Processing... Incoming Test Ssr Info
2023-09-29 01:14:04.380
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=1bc2b6898c093d0f, spanId=b34387e031536af4, sampled=true [ri.gr.wo.se.AbstractService] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) The same data exists. Entity id: 11,015,766
2023-09-29 01:14:04.381
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=09e679cba592a965, spanId=1bc2b6898c093d0f, sampled=true [testInfoKafkaRoute] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) The route was terminated because body was null
2023-09-29 01:14:04.381
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=09e679cba592a965, spanId=5b9b07ed0b44243b, sampled=true [testInfoKafkaRoute] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) Received : {"testJson"}
2023-09-29 01:14:04.381
22:14:04 INFO traceId=6012f1fb50dadbf148ff61c7a764d7f7, parentId=5b9b07ed0b44243b, spanId=82c81dd9f9f2a097, sampled=true [ri.gr.wo.se.pa.im.TestInfoServiceImp] (Camel (camel-1) thread #6 - KafkaConsumer[com.test.topic.in.test.test]) Processing... Incoming Test Info
I don't know if it will solve it or have any effect, but I recently added the following properties. I'm monitoring it, if it increases again I'll write again, I'm not sure this is the problem.
quarkus.otel.traces.sampler=traceidratio
quarkus.otel.traces.sampler.arg=1.0
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with the Camel Kafka route shown in the report and the camel-quarkus-opentelemetry dependency and Quarkus tracing properties in the provided pom. Reproduce the behavior under heavy Kafka load and inspect how trace context is created and propagated between events. Done means consecutive received events no longer reuse one traceId and the result is covered by a regression test.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- backend, distributed-systems, observability
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100