@@ -403,6 +457,46 @@
+
+ map-tracing
+ package
+
+ dockerBuild
+
+
+
+ amazoncorretto:11
+
+
+
+ io.numaproj.numaflow.examples.map.tracing.TracingMapFunction
+
+
+
+ numaflow-java-examples/map-tracing:${docker.tag}
+
+
+
+
+ sink-tracing
+ package
+
+ dockerBuild
+
+
+
+ amazoncorretto:11
+
+
+
+ io.numaproj.numaflow.examples.sink.tracing.TracingSink
+
+
+
+ numaflow-java-examples/sink-tracing:${docker.tag}
+
+
+
@@ -417,6 +511,13 @@
3.10.1
11
+
+
+ org.projectlombok
+ lombok
+ 1.18.36
+
+
diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/map/tracing/TracingMapFunction.java b/examples/src/main/java/io/numaproj/numaflow/examples/map/tracing/TracingMapFunction.java
new file mode 100644
index 00000000..fc2a1a65
--- /dev/null
+++ b/examples/src/main/java/io/numaproj/numaflow/examples/map/tracing/TracingMapFunction.java
@@ -0,0 +1,64 @@
+package io.numaproj.numaflow.examples.map.tracing;
+
+import io.numaproj.numaflow.examples.tracing.OtelTracing;
+import io.numaproj.numaflow.mapper.Datum;
+import io.numaproj.numaflow.mapper.Mapper;
+import io.numaproj.numaflow.mapper.Message;
+import io.numaproj.numaflow.mapper.MessageList;
+import io.numaproj.numaflow.mapper.Server;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+
+/**
+ * Tracing-aware pass-through map UDF example.
+ *
+ * Emits a {@code user.work} span nested under the Numaflow platform's per-message
+ * {@code numaflow.{topology}.map} span. Replace the body of {@link #processMessage} with
+ * your real work; any further spans started in that scope hang off {@code user.work}.
+ *
+ *
Expected trace tree for a MonoVertex {@code source -> map (this UDF) -> sink}:
+ *
+ *
+ * numaflow.vertex.process
+ * ├── numaflow.monovertex.source.dispatch
+ * ├── numaflow.monovertex.map
+ * │ └── user.work ← emitted by this example
+ * └── numaflow.monovertex.sink.write
+ *
+ *
+ * Required environment variables (set via Pipeline/MonoVertex {@code containerTemplate.env}):
+ *
+ *
+ * - {@code OTEL_EXPORTER_OTLP_TRACES_ENDPOINT} or {@code OTEL_EXPORTER_OTLP_ENDPOINT}
+ * - {@code OTEL_SERVICE_NAME} (optional; defaults to {@code numaflow-udf})
+ *
+ */
+public class TracingMapFunction extends Mapper {
+
+ private static final String TRACER_NAME = "numaflow-java-example/mapper-tracing";
+ private static final String USER_WORK_SPAN = "user.work";
+
+ public static void main(String[] args) throws Exception {
+ OtelTracing.initTracer();
+ Server server = new Server(new TracingMapFunction());
+ server.start();
+ server.awaitTermination();
+ }
+
+ @Override
+ public MessageList processMessage(String[] keys, Datum data) {
+ Context ctx = OtelTracing.extractContext(data.getSystemMetadata());
+ Span span = OtelTracing.getTracer(TRACER_NAME)
+ .spanBuilder(USER_WORK_SPAN)
+ .setParent(ctx)
+ .startSpan();
+ try (Scope scope = span.makeCurrent()) {
+ return MessageList.newBuilder()
+ .addMessage(new Message(data.getValue(), keys))
+ .build();
+ } finally {
+ span.end();
+ }
+ }
+}
diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/sink/tracing/TracingSink.java b/examples/src/main/java/io/numaproj/numaflow/examples/sink/tracing/TracingSink.java
new file mode 100644
index 00000000..79b2f511
--- /dev/null
+++ b/examples/src/main/java/io/numaproj/numaflow/examples/sink/tracing/TracingSink.java
@@ -0,0 +1,77 @@
+package io.numaproj.numaflow.examples.sink.tracing;
+
+import io.numaproj.numaflow.examples.tracing.OtelTracing;
+import io.numaproj.numaflow.sinker.Datum;
+import io.numaproj.numaflow.sinker.DatumIterator;
+import io.numaproj.numaflow.sinker.Response;
+import io.numaproj.numaflow.sinker.ResponseList;
+import io.numaproj.numaflow.sinker.Server;
+import io.numaproj.numaflow.sinker.Sinker;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tracing-aware log sink UDF example.
+ *
+ * Emits a {@code user.persist} span per message nested under the Numaflow platform's
+ * per-message {@code numaflow.{topology}.sink.write} span — typical place to span an
+ * external DB write, HTTP POST, or other persistence call.
+ *
+ *
Required environment variables (set via Pipeline/MonoVertex {@code containerTemplate.env}):
+ *
+ *
+ * - {@code OTEL_EXPORTER_OTLP_TRACES_ENDPOINT} or {@code OTEL_EXPORTER_OTLP_ENDPOINT}
+ * - {@code OTEL_SERVICE_NAME} (optional; defaults to {@code numaflow-udf})
+ *
+ */
+public class TracingSink extends Sinker {
+
+ private static final Logger log = LoggerFactory.getLogger(TracingSink.class);
+ private static final String TRACER_NAME = "numaflow-java-example/sinker-tracing";
+ private static final String USER_PERSIST_SPAN = "user.persist";
+
+ public static void main(String[] args) throws Exception {
+ OtelTracing.initTracer();
+ Server server = new Server(new TracingSink());
+ server.start();
+ server.awaitTermination();
+ }
+
+ @Override
+ public ResponseList processMessages(DatumIterator datumIterator) {
+ ResponseList.ResponseListBuilder responseListBuilder = ResponseList.newBuilder();
+ while (true) {
+ Datum datum;
+ try {
+ datum = datumIterator.next();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ continue;
+ }
+ if (datum == null) {
+ break;
+ }
+
+ Context ctx = OtelTracing.extractContext(datum.getSystemMetadata());
+ Span span = OtelTracing.getTracer(TRACER_NAME)
+ .spanBuilder(USER_PERSIST_SPAN)
+ .setParent(ctx)
+ .startSpan();
+ try (Scope scope = span.makeCurrent()) {
+ String msg = new String(datum.getValue());
+ log.info("Traced sink: {}, id: {}", msg, datum.getId());
+ responseListBuilder.addResponse(Response.responseOK(datum.getId()));
+ } catch (Exception e) {
+ responseListBuilder.addResponse(Response.responseFailure(
+ datum.getId(),
+ e.getMessage()));
+ } finally {
+ span.end();
+ }
+ }
+ return responseListBuilder.build();
+ }
+}
diff --git a/examples/src/main/java/io/numaproj/numaflow/examples/tracing/OtelTracing.java b/examples/src/main/java/io/numaproj/numaflow/examples/tracing/OtelTracing.java
new file mode 100644
index 00000000..a2578ebb
--- /dev/null
+++ b/examples/src/main/java/io/numaproj/numaflow/examples/tracing/OtelTracing.java
@@ -0,0 +1,170 @@
+package io.numaproj.numaflow.examples.tracing;
+
+import io.numaproj.numaflow.shared.SystemMetadata;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.propagation.TextMapGetter;
+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.BatchSpanProcessor;
+import io.opentelemetry.sdk.trace.samplers.Sampler;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Shared OpenTelemetry wiring for Numaflow UDF tracing examples.
+ *
+ * The Numaflow data plane injects the platform's per-stage trace context into
+ * {@code sys_metadata["tracing_udf"]} (W3C traceparent + optional tracestate) before
+ * calling the UDF. These helpers:
+ *
+ *
+ * - Initialise an OTLP gRPC tracer in the UDF process ({@link #initTracer()}).
+ * - Extract the platform parent context from each message ({@link #extractContext(SystemMetadata)}).
+ * - Allow user-defined child spans to nest under the platform stage span.
+ *
+ *
+ * Required environment variables (set on the Pipeline/MonoVertex containerTemplate):
+ *
+ *
+ * - {@code OTEL_EXPORTER_OTLP_TRACES_ENDPOINT} or {@code OTEL_EXPORTER_OTLP_ENDPOINT}
+ * - {@code OTEL_SERVICE_NAME} (optional; defaults to {@code numaflow-udf})
+ *
+ *
+ * When neither endpoint variable is set, {@link #initTracer()} is a no-op and span
+ * creation in the UDF body is essentially free.
+ */
+public final class OtelTracing {
+
+ private static final String TRACING_UDF_GROUP = "tracing_udf";
+
+ private static SdkTracerProvider tracerProvider;
+ private static OpenTelemetry openTelemetry;
+
+ private OtelTracing() {
+ }
+
+ /**
+ * Wires an OTLP gRPC tracer provider and the W3C propagator.
+ *
+ *
Registers a JVM shutdown hook so spans are flushed before exit.
+ */
+ public static void initTracer() {
+ String endpoint = System.getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT");
+ if (endpoint == null || endpoint.isEmpty()) {
+ endpoint = System.getenv("OTEL_EXPORTER_OTLP_ENDPOINT");
+ }
+ if (endpoint == null || endpoint.isEmpty()) {
+ System.out.println("[tracing] OTLP endpoint not set; UDF spans will be no-ops");
+ return;
+ }
+
+ String serviceName = System.getenv("OTEL_SERVICE_NAME");
+ if (serviceName == null || serviceName.isEmpty()) {
+ serviceName = "numaflow-udf";
+ }
+
+ OtlpGrpcSpanExporter exporter = OtlpGrpcSpanExporter.builder()
+ .setEndpoint(endpoint)
+ .build();
+
+ Resource resource = Resource.getDefault()
+ .merge(Resource.create(Attributes.of(
+ AttributeKey.stringKey("service.name"),
+ serviceName
+ )));
+
+ tracerProvider = SdkTracerProvider.builder()
+ .setSampler(Sampler.parentBased(Sampler.alwaysOn()))
+ .addSpanProcessor(BatchSpanProcessor.builder(exporter).build())
+ .setResource(resource)
+ .build();
+
+ openTelemetry = OpenTelemetrySdk.builder()
+ .setTracerProvider(tracerProvider)
+ .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
+ .buildAndRegisterGlobal();
+
+ System.out.printf(
+ "[tracing] OTLP exporter configured: endpoint=%s service=%s%n",
+ endpoint,
+ serviceName
+ );
+
+ Runtime.getRuntime().addShutdownHook(new Thread(OtelTracing::shutdown));
+ }
+
+ /**
+ * Shuts down the tracer provider so batched spans are flushed.
+ */
+ public static void shutdown() {
+ if (tracerProvider != null) {
+ tracerProvider.shutdown().join(5, TimeUnit.SECONDS);
+ tracerProvider = null;
+ openTelemetry = null;
+ }
+ }
+
+ /**
+ * Returns a tracer for the given instrumentation scope.
+ */
+ public static Tracer getTracer(String instrumentationName) {
+ if (openTelemetry != null) {
+ return openTelemetry.getTracer(instrumentationName);
+ }
+ return OpenTelemetry.noop().getTracer(instrumentationName);
+ }
+
+ /**
+ * Reads the W3C traceparent/tracestate the platform wrote into
+ * {@code sys_metadata["tracing_udf"]} and returns a context whose current span is
+ * the platform-side stage span.
+ *
+ *
Safe to call when tracing is disabled or when no parent context is present.
+ */
+ public static Context extractContext(SystemMetadata systemMetadata) {
+ if (systemMetadata == null || openTelemetry == null) {
+ return Context.current();
+ }
+
+ byte[] traceparentBytes = systemMetadata.getValue(TRACING_UDF_GROUP, "traceparent");
+ if (traceparentBytes == null || traceparentBytes.length == 0) {
+ return Context.current();
+ }
+
+ Map carrier = new HashMap<>();
+ carrier.put("traceparent", new String(traceparentBytes, StandardCharsets.UTF_8));
+
+ byte[] tracestateBytes = systemMetadata.getValue(TRACING_UDF_GROUP, "tracestate");
+ if (tracestateBytes != null && tracestateBytes.length > 0) {
+ carrier.put("tracestate", new String(tracestateBytes, StandardCharsets.UTF_8));
+ }
+
+ return openTelemetry.getPropagators().getTextMapPropagator()
+ .extract(Context.current(), carrier, MapGetter.INSTANCE);
+ }
+
+ private enum MapGetter implements TextMapGetter> {
+ INSTANCE;
+
+ @Override
+ public Iterable keys(Map carrier) {
+ return carrier.keySet();
+ }
+
+ @Override
+ public String get(Map carrier, String key) {
+ return carrier.get(key);
+ }
+ }
+}
diff --git a/examples/src/test/java/io/numaproj/numaflow/examples/map/tracing/TracingMapFunctionTest.java b/examples/src/test/java/io/numaproj/numaflow/examples/map/tracing/TracingMapFunctionTest.java
new file mode 100644
index 00000000..c5021873
--- /dev/null
+++ b/examples/src/test/java/io/numaproj/numaflow/examples/map/tracing/TracingMapFunctionTest.java
@@ -0,0 +1,64 @@
+package io.numaproj.numaflow.examples.map.tracing;
+
+import com.google.protobuf.ByteString;
+import common.MetadataOuterClass;
+import io.numaproj.numaflow.mapper.MapperTestKit;
+import io.numaproj.numaflow.mapper.Message;
+import io.numaproj.numaflow.mapper.MessageList;
+import io.numaproj.numaflow.shared.SystemMetadata;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+public class TracingMapFunctionTest {
+
+ private static final String TRACEPARENT =
+ "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01";
+
+ @Test
+ public void testPassThroughWithoutTracingMetadata() {
+ MapperTestKit.TestDatum datum = MapperTestKit.TestDatum.builder()
+ .value("hello".getBytes())
+ .build();
+
+ TracingMapFunction function = new TracingMapFunction();
+ MessageList result = function.processMessage(new String[]{}, datum);
+
+ List messages = result.getMessages();
+ Assertions.assertEquals(1, messages.size());
+ Assertions.assertEquals("hello", new String(messages.get(0).getValue()));
+ }
+
+ @Test
+ public void testPassThroughWithTracingMetadata() {
+ SystemMetadata systemMetadata = tracingSystemMetadata(TRACEPARENT, "vendor=value");
+ MapperTestKit.TestDatum datum = MapperTestKit.TestDatum.builder()
+ .value("hello".getBytes())
+ .systemMetadata(systemMetadata)
+ .build();
+
+ TracingMapFunction function = new TracingMapFunction();
+ MessageList result = function.processMessage(new String[]{"key-1"}, datum);
+
+ List messages = result.getMessages();
+ Assertions.assertEquals(1, messages.size());
+ Assertions.assertEquals("hello", new String(messages.get(0).getValue()));
+ Assertions.assertEquals("key-1", messages.get(0).getKeys()[0]);
+ }
+
+ private static SystemMetadata tracingSystemMetadata(String traceparent, String tracestate) {
+ MetadataOuterClass.KeyValueGroup.Builder groupBuilder =
+ MetadataOuterClass.KeyValueGroup.newBuilder()
+ .putKeyValue("traceparent", ByteString.copyFromUtf8(traceparent));
+ if (tracestate != null) {
+ groupBuilder.putKeyValue("tracestate", ByteString.copyFromUtf8(tracestate));
+ }
+
+ MetadataOuterClass.Metadata protoMetadata = MetadataOuterClass.Metadata.newBuilder()
+ .putSysMetadata("tracing_udf", groupBuilder.build())
+ .build();
+
+ return new SystemMetadata(protoMetadata);
+ }
+}
diff --git a/examples/src/test/java/io/numaproj/numaflow/examples/sink/tracing/TracingSinkTest.java b/examples/src/test/java/io/numaproj/numaflow/examples/sink/tracing/TracingSinkTest.java
new file mode 100644
index 00000000..f44be912
--- /dev/null
+++ b/examples/src/test/java/io/numaproj/numaflow/examples/sink/tracing/TracingSinkTest.java
@@ -0,0 +1,67 @@
+package io.numaproj.numaflow.examples.sink.tracing;
+
+import com.google.protobuf.ByteString;
+import common.MetadataOuterClass;
+import io.numaproj.numaflow.shared.SystemMetadata;
+import io.numaproj.numaflow.sinker.Response;
+import io.numaproj.numaflow.sinker.ResponseList;
+import io.numaproj.numaflow.sinker.SinkerTestKit;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TracingSinkTest {
+
+ private static final String TRACEPARENT =
+ "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01";
+
+ @Test
+ public void testTracingSink() {
+ int datumCount = 10;
+ TracingSink tracingSink = new TracingSink();
+ SinkerTestKit.TestListIterator testListIterator = new SinkerTestKit.TestListIterator();
+ for (int i = 0; i < datumCount; i++) {
+ testListIterator.addDatum(
+ SinkerTestKit.TestDatum.builder()
+ .id("id-" + i)
+ .value(("value-" + i).getBytes())
+ .build());
+ }
+
+ ResponseList responseList = tracingSink.processMessages(testListIterator);
+ Assertions.assertEquals(datumCount, responseList.getResponses().size());
+ for (Response response : responseList.getResponses()) {
+ Assertions.assertTrue(response.getSuccess());
+ }
+ }
+
+ @Test
+ public void testTracingSinkWithTracingMetadata() {
+ TracingSink tracingSink = new TracingSink();
+ SinkerTestKit.TestListIterator testListIterator = new SinkerTestKit.TestListIterator();
+ testListIterator.addDatum(
+ SinkerTestKit.TestDatum.builder()
+ .id("traced-id")
+ .value("traced-value".getBytes())
+ .systemMetadata(tracingSystemMetadata(TRACEPARENT, "vendor=value"))
+ .build());
+
+ ResponseList responseList = tracingSink.processMessages(testListIterator);
+ Assertions.assertEquals(1, responseList.getResponses().size());
+ Assertions.assertTrue(responseList.getResponses().get(0).getSuccess());
+ }
+
+ private static SystemMetadata tracingSystemMetadata(String traceparent, String tracestate) {
+ MetadataOuterClass.KeyValueGroup.Builder groupBuilder =
+ MetadataOuterClass.KeyValueGroup.newBuilder()
+ .putKeyValue("traceparent", ByteString.copyFromUtf8(traceparent));
+ if (tracestate != null) {
+ groupBuilder.putKeyValue("tracestate", ByteString.copyFromUtf8(tracestate));
+ }
+
+ MetadataOuterClass.Metadata protoMetadata = MetadataOuterClass.Metadata.newBuilder()
+ .putSysMetadata("tracing_udf", groupBuilder.build())
+ .build();
+
+ return new SystemMetadata(protoMetadata);
+ }
+}