chore: add map/sink tracing examples

.github/workflows/build-push.yaml

@@ -19,7 +19,8 @@ jobs:
           "batch-map-flatmap", "mapt-event-time-filter-function", "flat-map-stream", "map-flatmap",
           "even-odd", "simple-sink", "reduce-sum", "reduce-stream-sum",
           "map-forward-message", "reduce-counter", "sideinput-example",
-          "udf-sideinput-example", "source-simple-source", "session-reduce-count", "stream-sorter"
+          "udf-sideinput-example", "source-simple-source", "session-reduce-count", "stream-sorter",
+          "map-tracing", "sink-tracing"
         ]
 
     steps:

examples/pom.xml

@@ -11,8 +11,37 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <docker.tag>stable</docker.tag>
+        <opentelemetry.version>1.62.0</opentelemetry.version>
+        <kotlin.version>2.1.0</kotlin.version>
     </properties>
 
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>io.opentelemetry</groupId>
+                <artifactId>opentelemetry-bom</artifactId>
+                <version>${opentelemetry.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.jetbrains.kotlin</groupId>
+                <artifactId>kotlin-stdlib</artifactId>
+                <version>${kotlin.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.jetbrains.kotlin</groupId>
+                <artifactId>kotlin-stdlib-jdk7</artifactId>
+                <version>${kotlin.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.jetbrains.kotlin</groupId>
+                <artifactId>kotlin-stdlib-jdk8</artifactId>
+                <version>${kotlin.version}</version>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
     <dependencies>
         <dependency>
             <groupId>io.numaproj.numaflow</groupId>
@@ -39,6 +68,31 @@
             <version>5.10.2</version>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.36</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-api</artifactId>
+            <version>${opentelemetry.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk</artifactId>
+            <version>${opentelemetry.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-otlp</artifactId>
+            <version>${opentelemetry.version}</version>
+        </dependency>
     </dependencies>
 
     <build>
@@ -403,6 +457,46 @@
                             </to>
                         </configuration>
                     </execution>
+                    <execution>
+                        <id>map-tracing</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>dockerBuild</goal>
+                        </goals>
+                        <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
+                            <container>
+                                <mainClass>
+                                    io.numaproj.numaflow.examples.map.tracing.TracingMapFunction
+                                </mainClass>
+                            </container>
+                            <to>
+                                <image>numaflow-java-examples/map-tracing:${docker.tag}</image>
+                            </to>
+                        </configuration>
+                    </execution>
+                    <execution>
+                        <id>sink-tracing</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>dockerBuild</goal>
+                        </goals>
+                        <configuration>
+                            <from>
+                                <image>amazoncorretto:11</image>
+                            </from>
+                            <container>
+                                <mainClass>
+                                    io.numaproj.numaflow.examples.sink.tracing.TracingSink
+                                </mainClass>
+                            </container>
+                            <to>
+                                <image>numaflow-java-examples/sink-tracing:${docker.tag}</image>
+                            </to>
+                        </configuration>
+                    </execution>
                 </executions>
             </plugin>
         </plugins>
@@ -417,6 +511,13 @@
                     <version>3.10.1</version>
                     <configuration>
                         <release>11</release>
+                        <annotationProcessorPaths>
+                            <path>
+                                <groupId>org.projectlombok</groupId>
+                                <artifactId>lombok</artifactId>
+                                <version>1.18.36</version>
+                            </path>
+                        </annotationProcessorPaths>
                     </configuration>
                 </plugin>
             </plugins>

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.
+ *
+ * <p>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}.
+ *
+ * <p>Expected trace tree for a MonoVertex {@code source -> map (this UDF) -> sink}:
+ *
+ * <pre>
+ * numaflow.vertex.process
+ * ├── numaflow.monovertex.source.dispatch
+ * ├── numaflow.monovertex.map
+ * │   └── user.work                          ← emitted by this example
+ * └── numaflow.monovertex.sink.write
+ * </pre>
+ *
+ * <p>Required environment variables (set via Pipeline/MonoVertex {@code containerTemplate.env}):
+ *
+ * <ul>
+ *   <li>{@code OTEL_EXPORTER_OTLP_TRACES_ENDPOINT} or {@code OTEL_EXPORTER_OTLP_ENDPOINT}</li>
+ *   <li>{@code OTEL_SERVICE_NAME} (optional; defaults to {@code numaflow-udf})</li>
+ * </ul>
+ */
+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();
+        }
+    }
+}

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.
+ *
+ * <p>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.
+ *
+ * <p>Required environment variables (set via Pipeline/MonoVertex {@code containerTemplate.env}):
+ *
+ * <ul>
+ *   <li>{@code OTEL_EXPORTER_OTLP_TRACES_ENDPOINT} or {@code OTEL_EXPORTER_OTLP_ENDPOINT}</li>
+ *   <li>{@code OTEL_SERVICE_NAME} (optional; defaults to {@code numaflow-udf})</li>
+ * </ul>
+ */
+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();
+    }
+}