kafka-python instrumentation: span lifetime does not fully encompass the instrumented operation, making synchronous failures invisible

Summary

While investigating the kafka-python instrumentation, I found two cases where
the span lifetime does not fully encompass the operation being instrumented.
Although they affect different code paths (producer and consumer), both stem
from the same structural pattern: the instrumented operation executes outside
the lifetime of the span intended to represent it. This causes synchronous
failures to be missing from tracing.

I verified this against current main using the real OpenTelemetry SDK,
wrote regression tests reproducing the behavior, compared the implementation
with sibling Kafka instrumentations, and reviewed git history to determine
whether this was intentional or introduced later.

---

Producer

Current behavior

In utils.py (_wrap_send), the producer span is created, propagation
headers are injected, and the span exits before the wrapped send()
operation is executed.

with tracer.start_as_current_span(...):
    propagate.inject(...)
    ...

return func(*args, **kwargs)

If the underlying send() raises synchronously (for example,
ConnectionError or a serialization error), the exception occurs after the
producer span has already ended.

Reproducer

Using the real OpenTelemetry SDK:

ConnectionError propagated.
Exported spans: 1
StatusCode.UNSET
events=[]

The operation failed, but the exported producer span completed successfully
(StatusCode.UNSET) and contains no recorded exception.

Proposed behavior

The producer span should remain active while the synchronous send() call
executes so that failures are recorded on the span.

As a local experiment, I temporarily moved:

return func(*args, **kwargs)

inside the span's with block.

Without any other changes, the regression test changed from failing to
passing, suggesting that the current span lifetime is responsible for the
observed behavior.

---

Consumer

Current behavior

In utils.py (_wrap_next), the wrapped receive operation executes before
any consumer span is created.

record = func(*args, **kwargs)

if record:
    _create_consumer_span(record)

return record

If the wrapped receive call raises, execution exits immediately and no
consumer span is created.

Reproducer

Using the real OpenTelemetry SDK:

ValueError propagated.
Exported spans: 0

No receive span is created for the failed operation.

Question for maintainers

Is this the intended behavior for exceptions raised during receive?

Receive/poll spans represent the receive operation itself, and OpenTelemetry's
general guidance recommends recording exceptions from instrumented
operations. Based on that, I expected synchronous receive failures to be
represented by an errored receive span, but I'd appreciate confirmation that
this matches the intended behavior.

This report concerns only exceptions raised by the receive call—not
idle polling that returns no message, which is a separate, already-settled
case in the sibling confluent-kafka instrumentation.

---

Relation to prior Kafka discussions

I reviewed #1674 and #1678.

Those discussions focused on span links, context propagation, and
batch-receiver semantics for idle/empty polls.

This issue intentionally focuses only on synchronous exception visibility and
does not propose changes to the context propagation or span-linking behavior
discussed there.

---

Investigation performed

• Reproduced on current main using the real OpenTelemetry SDK (not mocks)
• Confirmed the runtime behavior by stepping through _wrap_send and
  _wrap_next with the VS Code debugger
• Wrote regression tests (test_exception_visibility_gap.py)
  • Both regression tests fail on current main (2 failed)
  • The package's 7 existing tests continue to pass unchanged (7 passed)
• Performed a local experiment moving the producer func() call inside the
  span's with block; the producer regression test changed from failing to
  passing with no other code changes
• Compared kafka-python, aiokafka, and confluent-kafka
  • confluent-kafka keeps the synchronous producer call inside the span
  • aiokafka independently exhibits the same producer and consumer pattern
    (I'll file a separate issue there)
• Reviewed git blame; both patterns originate from the initial
  kafka-python instrumentation PR (#814, Jan 2022), rather than a later
  regression
• Searched existing issues and PRs; I couldn't find one covering this
  specific behavior

---

Proposed direction

For the producer, keeping send() inside the span's with block appears to
be the minimal change that addresses the reproduced behavior in my local
testing.

For the consumer, I'd appreciate guidance on whether receive-operation
exceptions should produce an errored receive span.

If this aligns with the intended behavior, I'd be happy to submit a PR
including the regression tests.

Start in utils.py at _wrap_send and _wrap_next, then run test_exception_visibility_gap.py with the real OpenTelemetry SDK and the existing package tests. Confirm the producer send and consumer receive exceptions are represented by spans without changing the separate idle-poll or propagation behavior; the consumer exception semantics still need maintainer confirmation.