Kafka producer span ends before send completion, causing incorrect duration and Kafka metadata

[spike83]

Kafka producer spans created by the Kafka connector are completed before the asynchronous Kafka send operation completes.

In KafkaSink, outgoing tracing is performed before client.send(...) is invoked:

if (isTracingEnabled) {
    KafkaTrace kafkaTrace = new KafkaTrace.Builder()
            .withPartition(record.partition() != null ? record.partition() : -1)
            .withTopic(record.topic())
            .withHeaders(record.headers())
            .withGroupId(...)
            .withClientId(...)
            .build();

    kafkaInstrumenter.traceOutgoing(message, kafkaTrace);
}

Uni<RecordMetadata> sendUni = client.send((ProducerRecord) record);

traceOutgoing(...) starts and ends the producer span synchronously. Consequently, the span is already finished before the actual Kafka send operation starts.

This has several observable consequences:

1. The producer span duration does not represent the Kafka send operation.
2. Broker-assigned metadata is unavailable when the span ends, in particular the actual partition and offset.
3. messaging.kafka.offset is reported as 0, even though the actual offset is unknown at that point.
4. Asynchronous send failures cannot be reflected by the already completed producer span.

Environment

Observed with:

Quarkus: 3.36.2
OpenTelemetry SDK: 1.60.1
SmallRye Reactive Messaging version: managed by Quarkus 3.36.2

Observed span

For example:

{
  "name": "synthetic-features publish",
  "kind": "SPAN_KIND_PRODUCER",
  "startTimeUnixNano": 1786102981960590800,
  "endTimeUnixNano": 1786102981960602600,
  "attributes": [
    {
      "key": "messaging.system",
      "value": {
        "stringValue": "kafka"
      }
    },
    {
      "key": "messaging.kafka.offset",
      "value": {
        "intValue": 0
      }
    },
    {
      "key": "messaging.client_id",
      "value": {
        "stringValue": "kafka-producer-generated"
      }
    },
    {
      "key": "messaging.destination.name",
      "value": {
        "stringValue": "synthetic-features"
      }
    },
    {
      "key": "messaging.operation",
      "value": {
        "stringValue": "publish"
      }
    }
  ]
}

The span duration is approximately 11.8 µs.

This duration cannot represent the Kafka publish operation because the span has already ended before client.send(...) is invoked.

The span also reports:

messaging.kafka.offset = 0

while no partition attribute is present.

Kafka send completion already provides the missing information

KafkaSink already receives the result of the asynchronous send as RecordMetadata:

Uni<RecordMetadata> sendUni = client.send((ProducerRecord) record);

Uni<Void> uni = sendUni.onItem().transformToUni(recordMetadata -> {
    OutgoingMessageMetadata.setResultOnMessage(message, recordMetadata);

    log.successfullyToTopic(
            message,
            channel,
            recordMetadata.topic(),
            recordMetadata.partition(),
            recordMetadata.offset());

    return Uni.createFrom().completionStage(message.ack());
});

At this point the actual partition and offset are available.

This completion also represents a much more useful endpoint for the producer span: the span duration would represent the client-observed Kafka publish latency rather than only the time required to create and close the tracing span.

Incorrect offset

There is an additional issue caused by creating KafkaTrace before RecordMetadata is available.

The outgoing trace does not call:

.withOffset(...)

The offset in KafkaTrace is backed by a primitive long, so an unset value defaults to 0.

As 0 is a valid Kafka offset, the resulting telemetry is misleading:

messaging.kafka.offset = 0

cannot be distinguished from a record that was actually written at offset 0.

Expected behaviour

The producer span should cover the asynchronous Kafka send operation.

Conceptually:

start producer span
        |
        v
client.send(record)
        |
        | asynchronous Kafka send
        | partitioning / batching / network / acknowledgement
        |
        v
RecordMetadata / failure
        |
        v
add resulting Kafka attributes
end producer span

On successful completion, information available from RecordMetadata could be added to the span, including:

partition
offset

The resulting span duration would then provide useful information about Kafka producer send/publish latency.

On failure, the asynchronous producer failure could also be recorded on the span.

Expected metadata behaviour

If the actual offset is available from RecordMetadata, it should be reported.

If it is not available, for example because of the producer acknowledgement configuration, the offset attribute should be omitted rather than reported as 0.

The same applies to a broker/producer-selected partition: the partition from RecordMetadata should be used once the send completes.

Suggested direction

Instead of traceOutgoing(...) starting and immediately ending the span, producer tracing could expose separate start/end operations so that KafkaSink can keep the tracing context until sendUni terminates.

For example, conceptually:

TracingContext context = kafkaInstrumenter.startOutgoing(message, kafkaTrace);

Uni<RecordMetadata> sendUni = client.send(record);

return sendUni
        .invoke(metadata ->
                kafkaInstrumenter.endOutgoing(context, metadata, null))
        .onFailure().invoke(error ->
                kafkaInstrumenter.endOutgoing(context, null, error));

The exact implementation may of course differ depending on the tracing abstraction.

A minimal change that only changes the default offset from 0 to -1 would avoid emitting incorrect offset data, but it would not address the underlying producer span lifecycle issue:

• producer duration would still not represent the send operation
• actual partition would still be unavailable when Kafka chooses it
• actual offset would still be unavailable
• asynchronous producer failures could not be represented

For that reason, I think handling the asynchronous producer span lifecycle would address the underlying issue more completely.

Reproducer

A minimal reproducer can be provided if required.

Producer tracing also uses consumer-specific configuration

While looking at the producer tracing code, I also noticed that the outgoing KafkaTrace is populated using ConsumerConfig constants:

.withGroupId((String) client.configuration().get(ConsumerConfig.GROUP_ID_CONFIG))
.withClientId((String) client.configuration().get(ConsumerConfig.CLIENT_ID_CONFIG))

This seems unusual for a producer path.

For client.id, this is mostly a semantic/code clarity issue because both ConsumerConfig.CLIENT_ID_CONFIG and ProducerConfig.CLIENT_ID_CONFIG refer to the same Kafka configuration key, client.id. On the producer path, however, using ProducerConfig.CLIENT_ID_CONFIG would better reflect which client is being instrumented.

group.id is more questionable. A Kafka producer does not have a standard group.id configuration; consumer groups are a consumer concept. Likewise, the OpenTelemetry Kafka semantic conventions describe the consumer group attribute as the Kafka consumer group id.

It therefore seems that producer tracing should probably not populate a group id at all, while the client id should come from the producer configuration:

.withClientId((String) client.configuration().get(ProducerConfig.CLIENT_ID_CONFIG))

This may be another indication that the producer tracing path currently reuses tracing structures designed around the consumer path rather than modelling the Kafka producer operation independently.

I do not think this is the main issue by itself, but it seems worth reviewing together with the producer span lifecycle, since correcting the producer instrumentation would be a good opportunity to ensure that only producer-relevant attributes are attached.