KafkaServer and KafkaClient parsers java processes as consumer or producer spans with the resource is the kafka broker or the client is the kafka broker

[esara]

With both consumer and product java application and the kafka broker running as a containerized service in k8s, beyla generates the following traces

2024-11-30 01:58:09.11301589 (500.3197ms[500.3197ms]) KafkaServer 0 process * [10.233.102.129 as kafka-client.namespace:52388]->[10.233.102.161 as kafka.namespace:9092] size:0B svc=[namespace/kafka java] traceparent=[]
2024-11-30 02:41:18.11302411 (500.5914ms[500.5914ms]) KafkaServer 0 process namespace.suspend-engine-processing [10.233.102.148 as suspend.namespace:57146]->[10.233.102.161 as kafka.namespace:9092] size:0B svc=[namespace/kafka java] traceparent=[]

and

2024-11-30 01:59:29.113015929 (500.275ms[500.275ms]) KafkaClient 0 process namespace.topology-on-demand-cost [10.233.102.148 as suspend.namespace:57146]->[10.233.57.197 as kafka.namespace:9092] size:0B svc=[namespace/suspend go] traceparent=[]
2024-11-30 01:59:30.113015930 (169.528ms[169.528µs]) KafkaClient 0 process * [10.233.102.136 as metrics-processor.namespace:35984]->[10.233.57.197 as kafka.namespace:9092] size:0B svc=[namespace/metrics-processor java] traceparent=[]

or queried using a tempo traceql

{resource.telemetry.sdk.name="beyla" && resource.service.name="kafka" && span.messaging.system="kafka"}

This is not the correct, because the consumer and the producer is supposed to have the application as the resource, not the kafka broker.
because we are parsing the tcp connection for the peer address for kafka messages
https://github.com/grafana/beyla/blob/main/pkg/internal/ebpf/common/kafka_detect_transform.go#L351