OpenTelemetry Instrumentation Not Working for Apache Spark Structured Streaming Consumers

[patpatpat123]

Is your feature request related to a problem? Please describe.

📋 Issue Summary

Instrumentation is currently not working for Apache Spark Structured Streaming consumers, particularly when processing Kafka messages with trace context in headers. This creates a significant observability gap for distributed tracing in Spark-based streaming applications.

https://spark.apache.org/docs/latest/streaming/index.html#:~:text=Structured%20Streaming%20is%20a%20scalable%20and%20fault-tolerant%20stream,would%20express%20a%20batch%20computation%20on%20static%20data.

🔍 Problem Description

Current Behavior

• Spark Structured Streaming consumers cannot properly propagate OpenTelemetry trace context
• Trace context from Kafka message headers is not automatically extracted and propagated
• No built-in OpenTelemetry instrumentation exists for Apache Spark in streaming contexts
• Manual trace context extraction and span creation is required but not straightforward

Impact

• Observability Gap: Loss of distributed tracing across Spark streaming pipelines
• Debugging Difficulty: Inability to trace message flow through Spark transformations
• Performance Monitoring: Missing insights into processing latency and bottlenecks
• Compliance Issues: Difficulty meeting observability requirements in production environments

🛠️ Root Cause Analysis

Technical Challenges

1. Header Access Complexity: Kafka headers containing trace context require manual parsing and extraction
2. Span Lifecycle Management: Manual span creation, linking, and cleanup in streaming contexts
3. Resource Management: Proper OpenTelemetry SDK initialization and cleanup in Spark executor environments

Current Limitations

• No native OpenTelemetry support in Apache Spark Structured Streaming
• Manual trace context extraction from Kafka headers required
• Complex span linking and parent-child relationship management
• No automatic instrumentation for Spark transformations

Describe the solution you'd like

Hoping for a solution without manual propagation.

Hoping for a solution where users can just use ./spark-submit [...] --javaagent and propagation will work. (And this is achievable!)

Describe alternatives you've considered

No response

Additional context

No response

Tip

_{React with 👍 to help prioritize this issue. Please use comments to provide useful context, avoiding +1 or me too, to help us triage it. Learn more here.}

[patpatpat123]

Currently, in order to make it work, we need to use manual propagation, manual continuation.

Suppose the kafka producer has valid W3C open telemetry traces, the "secret" on the consumer side to make it work are:

• Step 1, allow Spark to get the Kafka message header: https://spark.apache.org/docs/latest/streaming/structured-streaming-kafka-integration.html#creating-a-kafka-source-for-streaming-queries

// Subscribe to 1 topic, with headers
Dataset<Row> df = spark
  .readStream()
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .option("includeHeaders", "true") //IMPORTANT HERE
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "headers");

• Step 2: Once the header with a valid traceparent is there, the manual code looks something like this:

Implementation Approach

The issue can be addressed through a custom UDF-based solution that manually handles OpenTelemetry instrumentation. Here's the recommended approach:

1. Trace Context Extraction

// Extract trace parent from Kafka headers
String importantTraceParent = new String(byteArray);
String[] traceSplit = importantTraceParent.split("-");
SpanContext remoteContext = SpanContext.createFromRemoteParent(
    traceSplit[1], 
    traceSplit[2], 
    TraceFlags.getSampled(), 
    TraceState.getDefault()
);

2. OpenTelemetry SDK Setup

// Configure OTLP exporter
OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder()
    .setEndpoint("https://trace-backend:443")
    .build();

// Create resource with service identification
Resource resource = Resource.create(Attributes.of(
    AttributeKey.stringKey("service.name"), 
    "spark-structured-streaming-service"
));

// Configure tracer provider
SdkTracerProvider tracerProvider = SdkTracerProvider.builder()
    .setResource(resource)
    .addSpanProcessor(SimpleSpanProcessor.create(spanExporter))
    .build();

3. Span Creation and Linking

// Create consumer span with proper linking
SpanBuilder spanBuilder = tracer.spanBuilder("kafka.consumer.process")
    .setSpanKind(SpanKind.CONSUMER)
    .addLink(remoteContext)
    .setParent(Context.current().with(Span.wrap(remoteContext)));

Span span = spanBuilder.startSpan();
try {
    // Business logic processing
    processMessage(message);
} catch (Exception e) {
    span.setStatus(StatusCode.ERROR, e.getMessage());
    span.recordException(e);
} finally {
    span.end();
}

🔧 Implementation Steps

1. Add OpenTelemetry Dependencies to pom.xml:

   <dependency>
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-exporter-otlp</artifactId>
   </dependency>
   <dependency>
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-sdk</artifactId>
   </dependency>

2. Create Custom UDF for trace handling (see reference implementation)

3. Configure Kafka Consumer with header inclusion:

   .option("includeHeaders", "true")

4. Register UDF and apply to streaming data

(Full code)

package org.example;

import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.trace.*;
import io.opentelemetry.context.Context;
import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.resources.Resource;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF2;
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.types.DataTypes;
import scala.collection.Seq;
import java.time.Instant;

public final class Main {

    public static void main(String[] args) throws Exception {
        SparkSession session = SparkSession.builder().appName("question").getOrCreate();

        //Create UDF to handle trace in Headers
        session.udf().register("handleTraceInHeader", new UDF1<Seq<GenericRowWithSchema>, String>() {
            @Override
            public String call(Seq<GenericRowWithSchema> messageHeaders) throws Exception {
                GenericRowWithSchema genericRowWithSchema = messageHeaders.apply(0);
                byte[] byteArray = (byte[]) genericRowWithSchema.get(1);
                String importantTraceParent = new String(byteArray);
                String[] traceSplit = importantTraceParent.split("-");
                SpanContext remoteContext = SpanContext.createFromRemoteParent(traceSplit[1], traceSplit[2], TraceFlags.getSampled(), TraceState.getDefault());

                OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder()
                        .setEndpoint("https://trace-backendt:443")
                        .build();


                Resource myResourceBusinessService = Resource.create(Attributes.of(AttributeKey.stringKey("service.name"), "sparkstructuredstreamingservice"));

                SdkTracerProvider tracerProviderBusinessService = SdkTracerProvider.builder()
                        .setResource(myResourceBusinessService)
                        .addSpanProcessor(SimpleSpanProcessor.create(spanExporter))
                        .build();

                OpenTelemetry openTelemetryBusinessService = OpenTelemetrySdk.builder().setTracerProvider(tracerProviderBusinessService).build();

                Tracer tracerBusinessService = openTelemetryBusinessService.getTracer("businessService");

                SpanBuilder sbBusinessService = tracerBusinessService.spanBuilder("consumer").setSpanKind(SpanKind.CONSUMER);

                sbBusinessService.setParent(Context.current().with(Span.wrap(remoteContext)));

                Span serverSpanBusinessService = sbBusinessService.startSpan();
                try {
                    System.out.println("Do some business logic");
                } catch (Exception e) {
                    serverSpanBusinessService.setStatus(StatusCode.ERROR, e.getMessage());
                    serverSpanBusinessService.recordException(e);
                } finally {
                    serverSpanBusinessService.end();
                }
//                Thread.sleep(200);
                return importantTraceParent.toUpperCase();
            }
        }, DataTypes.StringType);

        Dataset<Row> df = session
                .readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "localhost:9093")
                .option("subscribe", "kafkatopic")
                .option("includeHeaders", "true") // VERY IMPORTANT
                .load();

        df.selectExpr("CAST(headers AS STRING)", "headers");
        df.createOrReplaceTempView("mytable");
        Dataset<Row> result = session.sql("select handleTraceInHeader(headers) as session_duration from mytable");

//...
        query.awaitTermination();
    }

}

[patpatpat123]

Past issues:

#9638

#8019

While it is possible to have distributed tracing directly baked into Apache Spark repo (with its very complex code base), a solution with java agent can be a viable answer.

Goal, visual of what to achieve:

The left hand side is the producer

The "gap" in the middle is the time spent inside kafka (there are couple of other possibilites for this, to discuss with the team)

The right hand side is what is missing. Currently, there are no traces from consumer

[patpatpat123]

@trask @breedx-splk For visibility.

Thank you for considering this!