[stankiewicz]

Follow-up to #39150 (PubsubIO) and #39151 (KafkaIO) to properly set OpenTelemetry messaging span kinds and create producer spans in the write-side header propagation DoFns:

1. Read-Side Span Kind (SpanKind.CONSUMER):
   • In PubsubIO.Read.OpenTelemetryHeaderConsumer, set .setSpanKind(SpanKind.CONSUMER) on the "PubSubIO.Read" span.
   • In KafkaIO.OpenTelemetryHeaderConsumer, set .setSpanKind(SpanKind.CONSUMER) on the "KafkaIO.Read" span.
2. Write-Side Producer Span Creation (SpanKind.PRODUCER):
   • Previously, OpenTelemetryHeaderPropagator in both PubsubIO and KafkaIO injected Context.current() directly without creating a producer span. If no upstream span was active on the worker thread, no traceparent was injected and no write span was recorded.
   • In PubsubIO.Write.OpenTelemetryHeaderPropagator, initialize the "PubSubIO" Tracer via PipelineOptions (SdkHarnessOptions) in @Setup, start a "PubSubIO.Write" span with SpanKind.PRODUCER and parent Context.current(), and make it current before injecting W3C trace context (googclient_traceparent / googclient_tracestate) and outputting the PubsubMessage.
   • In KafkaIO.OpenTelemetryHeaderPropagator, initialize the "KafkaIO" Tracer via PipelineOptions (SdkHarnessOptions) in @Setup, start a "KafkaIO.Write" span with SpanKind.PRODUCER and parent Context.current(), and make it current before injecting W3C trace context into the ProducerRecord headers and outputting the record.

[github-actions]

Assigning reviewers:

R: @Abacn for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[Abacn]

Newly added tracer are transient. Should we make it transient here as well?

https://github.com/apache/beam/pull/40483/changes#diff-89f3e8695375f86f8de56e6244d51187030f9e764072c6de4458744caea0961cR1583