[Bug] The kafka consumer plugin has a bug that it may mix up some messages and make those messages to another trace

[darkness-2nd]

Search before asking

• I had searched in the issues and found no similar issues.

Apache SkyWalking Component

Java Agent (apache/skywalking-java)

What happened

I make a demo consumer use spring annotation KafkaListener, and make a REST API to do send kafka message by KafkaTemplate ,but when I make a breakpoint at org.apache.kafka.clients.consumer.KafkaConsumer#poll(long, boolean) before call method "pollAndFetchs", and then I make 5 http request in browser, after 1 miniutes, I resume the breakpoint, the kafka agent may mix up some messages and make those mixed message to another trace.
the test code is as follows
@GetMapping("/spring/kafka")
public String testSpringKafka() {
String traceId = TraceContext.traceId();
kafkaTemplate.send("TOPIC_DEDUCT_MONEY", "key", "msg");
return "OK";
}
@KafkaListener(
topics = {"TOPIC_DEDUCT_MONEY"},
groupId = "study-account"
)
public void deductStorage(ConsumerRecord<String, String> consumerRecord) {
String value = consumerRecord.value();
System.out.printf("deduct money，msg：%s%n", value);
}
I expect that the 5 messages can be the child span of its own http request, but there are 2 messages were belonged to a request that didn't send it.

The traces "3480b2ffc8594b1ab8749fc2af82ef74.82.17219730707690001" and "3480b2ffc8594b1ab8749fc2af82ef74.84.17219730707740001" don't have consume span, but another 3 http requests have its child consume span.


And look at the dashboard, there are only 3 consume segments, What's even more amazing is that the Consume span "3480b2ffc8594b1ab8749fc2af82ef74.83.17219730707690001" has 2 traceIds in its detail info, and those traces are belongs to ""3480b2ffc8594b1ab8749fc2af82ef74.82.17219730707690001" and "3480b2ffc8594b1ab8749fc2af82ef74.84.17219730707740001", which are missing in the dashboard.

What you expected to happen

I hope these two missing consume spans can be displayed on the dashboard and in the parent http request span to which they belong

How to reproduce

@GetMapping("/spring/kafka")
public String testSpringKafka() {
String traceId = TraceContext.traceId();
kafkaTemplate.send("TOPIC_DEDUCT_MONEY", "key", "msg");
return "OK";
}
@KafkaListener(
topics = {"TOPIC_DEDUCT_MONEY"},
groupId = "study-account"
)
public void deductStorage(ConsumerRecord<String, String> consumerRecord) {
String value = consumerRecord.value();
System.out.printf("deduct money，msg：%s%n", value);
}
In kafka 2.0.x ~ 3.6.x, make a breakpoint at org.apache.kafka.clients.consumer.KafkaConsumer#poll(long, boolean).

In kafka 3.7.x, make a breakpoint at org.apache.kafka.clients.consumer.KafkaConsumer#poll(long) and org.apache.kafka.clients.consumer.KafkaConsumer#poll(java.time.Duration).

And then call more than 5 http requests, then wait about 1 minutes, resume the breakpoint, then access the dashboard.

Anything else

No response

Are you willing to submit a pull request to fix on your own?

• Yes I am willing to submit a pull request on my own!

Code of Conduct

• I agree to follow this project's Code of Conduct

[wu-sheng]

I don't know which is wrong, but as same as many async scenarios, if the poll operation doesn't keep the thread to the process only, but switch to another poll, then the tracing context in the thread context could be mixed for sure.

You need to check carefully, because, if the consuming of message in the bulk mode, N messages per process, then I would say the agent and UI are correct. It should not be attached to match the request, it would be merged as the multiple parent of the consuming span.

[darkness-2nd]

And, your way of describing the scenarios is a little strange. We could discuss based on very certain scenarios, but not your local operation, like a breakpoint or something else.

We can't and should not follow your local setup. That is out of our scope.

OK, I tried it again, it is not need to make any breakpoints, if we send messages to broker before the consumer start, then the next time consumers start, and if it polls more than 1 message, those polled messages will mixed to the trace that belongs to the first polled message.

[wu-sheng]

Please read what I was saying. I said, if this is consuming in bulk, then yes. They should be merged. Because the consuming happens once. This is a very basic tracing theory. I think you should try to understand that first, rather than assuming something is a bug.

[darkness-2nd]

Please read what I was saying. I said, if this is consuming in bulk, then yes. They should be merged. Because the consuming happens once. This is a very basic tracing theory. I think you should try to understand that first, rather than assuming something is a bug.

OK, thank you, I understand, one consuming make one segment. But I'm so puzzled, in actual business operations, if there are many messages overstock, after many consumers start, how can we find the consumer instance quickly?

[wu-sheng]

There isn't an easy way to do this. I think. We don't support multiple trace ID list for a segment.

You could find the parent segments from the consuming span, but not reverted.

[darkness-2nd]

There isn't an easy way to do this. I think. We don't support multiple trace ID list for a segment.

You could find the parent segments from the consuming span, but not reverted.

OK. thank you, agent as a plugin, it dosen't need to care actual business, so it merges messages in one consuming is correct, and we can find another way to resolve the question I asked right now..