From 8071ce24690c357a76f05f1e110a6140b019a011 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Fri, 9 Oct 2026 11:48:48 +0000 Subject: [PATCH 1/2] Fix OpenTelemetry span kinds and add write spans in PubsubIO and KafkaIO --- .../beam/sdk/io/gcp/pubsub/PubsubIO.java | 47 +++++-- .../beam/sdk/io/gcp/pubsub/PubsubIOTest.java | 121 +++++++++++++++++ .../org/apache/beam/sdk/io/kafka/KafkaIO.java | 26 +++- .../apache/beam/sdk/io/kafka/KafkaIOTest.java | 125 ++++++++++++++++++ 4 files changed, 304 insertions(+), 15 deletions(-) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java index 3c63b3386e9a..6ce8be5b7673 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java @@ -29,6 +29,7 @@ import com.google.protobuf.InvalidProtocolBufferException; import com.google.protobuf.Message; import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; import io.opentelemetry.context.Context; @@ -1170,6 +1171,7 @@ public void processElement( Span span = checkArgumentNotNull(tracer) .spanBuilder("PubSubIO.Read") + .setSpanKind(SpanKind.CONSUMER) .setParent(context) .startSpan(); try (Scope s = span.makeCurrent()) { @@ -1573,24 +1575,41 @@ public void set(@Nullable Map attr, String key, String value) { W3CTraceContextPropagator.getInstance().inject(Context.current(), attr, inject); } + @Setup + public void setup(PipelineOptions po) { + tracer = po.as(SdkHarnessOptions.class).getOpenTelemetry().getTracer("PubSubIO"); + } + + private transient @MonotonicNonNull Tracer tracer = null; + @ProcessElement public void processElement( @Element PubsubMessage message, OutputReceiver output) { - Map attributeMap = message.getAttributeMap(); - Map attr = - attributeMap == null ? new HashMap<>() : new HashMap<>(attributeMap); - injectSpanContext(attr); - - // copy the message, multiple fields - PubsubMessage ps = - new PubsubMessage( - message.getPayload(), attr, message.getMessageId(), message.getOrderingKey()); - - // topic is copied seperately, not via constructor - if (message.getTopic() != null) { - ps = ps.withTopic(message.getTopic()); + Span span = + checkArgumentNotNull(tracer) + .spanBuilder("PubSubIO.Write") + .setSpanKind(SpanKind.PRODUCER) + .setParent(Context.current()) + .startSpan(); + try (Scope s = span.makeCurrent()) { + Map attributeMap = message.getAttributeMap(); + Map attr = + attributeMap == null ? new HashMap<>() : new HashMap<>(attributeMap); + injectSpanContext(attr); + + // copy the message, multiple fields + PubsubMessage ps = + new PubsubMessage( + message.getPayload(), attr, message.getMessageId(), message.getOrderingKey()); + + // topic is copied seperately, not via constructor + if (message.getTopic() != null) { + ps = ps.withTopic(message.getTopic()); + } + output.output(ps); + } finally { + span.end(); } - output.output(ps); } } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java index 4aefac0379a8..af20679ea7a6 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java @@ -32,9 +32,19 @@ import com.google.protobuf.ByteString; import com.google.protobuf.DynamicMessage; import com.google.protobuf.InvalidProtocolBufferException; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Context; import java.io.IOException; import java.io.Serializable; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -62,11 +72,13 @@ import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO.Read; import org.apache.beam.sdk.io.gcp.pubsub.PubsubTestClient.PubsubTestClientFactory; import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.SdkHarnessOptions; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.MapElements; import org.apache.beam.sdk.transforms.SimpleFunction; import org.apache.beam.sdk.transforms.display.DisplayData; @@ -82,6 +94,7 @@ import org.apache.beam.sdk.values.ValueInSingleWindow; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.commons.lang3.RandomStringUtils; import org.checkerframework.checker.nullness.qual.Nullable; @@ -98,6 +111,7 @@ import org.junit.runner.RunWith; import org.junit.runners.JUnit4; import org.junit.runners.model.Statement; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; /** Tests for PubsubIO Read and Write transforms. */ @@ -1083,4 +1097,111 @@ public void testWithoutValidation() throws IOException { write.validate(options); } + + @Test + public void testOpenTelemetryHeaderConsumer() { + String traceId = "0af7651916cd43dd8448eb211c80319c"; + String parentSpanId = "b7ad6b7169203331"; + String childSpanId = "1111111111111111"; + String traceparent = "00-" + traceId + "-" + parentSpanId + "-01"; + + OpenTelemetry mockOpenTelemetry = Mockito.mock(OpenTelemetry.class); + Tracer mockTracer = Mockito.mock(Tracer.class); + SpanBuilder mockSpanBuilder = Mockito.mock(SpanBuilder.class, Mockito.RETURNS_SELF); + Span mockSpan = Mockito.mock(Span.class); + SpanContext childSpanContext = + SpanContext.create(traceId, childSpanId, TraceFlags.getSampled(), TraceState.getDefault()); + Mockito.when(mockSpan.getSpanContext()).thenReturn(childSpanContext); + Mockito.when(mockSpan.storeInContext(Mockito.any())).thenCallRealMethod(); + Mockito.when(mockSpan.makeCurrent()).thenCallRealMethod(); + Mockito.when(mockOpenTelemetry.getTracer("PubSubIO")).thenReturn(mockTracer); + Mockito.when(mockTracer.spanBuilder("PubSubIO.Read")).thenReturn(mockSpanBuilder); + Mockito.when(mockSpanBuilder.startSpan()).thenReturn(mockSpan); + + PipelineOptions options = TestPipeline.testingPipelineOptions(); + options.as(SdkHarnessOptions.class).setOpenTelemetry(mockOpenTelemetry); + + PubsubIO.Read.OpenTelemetryHeaderConsumer consumer = + new PubsubIO.Read.OpenTelemetryHeaderConsumer(); + consumer.setup(options); + + PubsubMessage message = + new PubsubMessage( + "payload".getBytes(StandardCharsets.UTF_8), + ImmutableMap.of("googclient_traceparent", traceparent)); + List outputs = new ArrayList<>(); + @SuppressWarnings("unchecked") + DoFn.OutputReceiver receiver = Mockito.mock(DoFn.OutputReceiver.class); + Mockito.doAnswer( + invocation -> { + assertEquals(childSpanId, Span.current().getSpanContext().getSpanId()); + outputs.add(invocation.getArgument(0)); + return null; + }) + .when(receiver) + .output(Mockito.any()); + consumer.processElement(message, receiver); + + assertEquals(1, outputs.size()); + Mockito.verify(mockTracer).spanBuilder("PubSubIO.Read"); + Mockito.verify(mockSpanBuilder).setSpanKind(SpanKind.CONSUMER); + ArgumentCaptor parentContextCaptor = ArgumentCaptor.forClass(Context.class); + Mockito.verify(mockSpanBuilder).setParent(parentContextCaptor.capture()); + assertEquals( + parentSpanId, + Span.fromContext(parentContextCaptor.getValue()).getSpanContext().getSpanId()); + Mockito.verify(mockSpan).end(); + } + + @Test + public void testOpenTelemetryHeaderPropagator() { + String traceId = "0af7651916cd43dd8448eb211c80319c"; + String producerSpanId = "2222222222222222"; + + OpenTelemetry mockOpenTelemetry = Mockito.mock(OpenTelemetry.class); + Tracer mockTracer = Mockito.mock(Tracer.class); + SpanBuilder mockSpanBuilder = Mockito.mock(SpanBuilder.class, Mockito.RETURNS_SELF); + Span mockSpan = Mockito.mock(Span.class); + SpanContext producerSpanContext = + SpanContext.create( + traceId, producerSpanId, TraceFlags.getSampled(), TraceState.getDefault()); + Mockito.when(mockSpan.getSpanContext()).thenReturn(producerSpanContext); + Mockito.when(mockSpan.storeInContext(Mockito.any())).thenCallRealMethod(); + Mockito.when(mockSpan.makeCurrent()).thenCallRealMethod(); + Mockito.when(mockOpenTelemetry.getTracer("PubSubIO")).thenReturn(mockTracer); + Mockito.when(mockTracer.spanBuilder("PubSubIO.Write")).thenReturn(mockSpanBuilder); + Mockito.when(mockSpanBuilder.startSpan()).thenReturn(mockSpan); + + PipelineOptions options = TestPipeline.testingPipelineOptions(); + options.as(SdkHarnessOptions.class).setOpenTelemetry(mockOpenTelemetry); + + PubsubIO.Write.OpenTelemetryHeaderPropagator propagator = + new PubsubIO.Write.OpenTelemetryHeaderPropagator(); + propagator.setup(options); + + PubsubMessage message = + new PubsubMessage("payload".getBytes(StandardCharsets.UTF_8), ImmutableMap.of("k", "v")); + List outputs = new ArrayList<>(); + @SuppressWarnings("unchecked") + DoFn.OutputReceiver receiver = Mockito.mock(DoFn.OutputReceiver.class); + Mockito.doAnswer( + invocation -> { + assertEquals(producerSpanId, Span.current().getSpanContext().getSpanId()); + outputs.add(invocation.getArgument(0)); + return null; + }) + .when(receiver) + .output(Mockito.any()); + propagator.processElement(message, receiver); + + assertEquals(1, outputs.size()); + PubsubMessage propagated = outputs.get(0); + assertEquals("v", propagated.getAttribute("k")); + assertEquals( + "00-" + traceId + "-" + producerSpanId + "-01", + propagated.getAttribute("googclient_traceparent")); + Mockito.verify(mockTracer).spanBuilder("PubSubIO.Write"); + Mockito.verify(mockSpanBuilder).setSpanKind(SpanKind.PRODUCER); + Mockito.verify(mockSpan).end(); + } } diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java index 99f8a72e8d17..af534f5697c3 100644 --- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java +++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java @@ -27,6 +27,7 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import io.confluent.kafka.serializers.KafkaAvroDeserializer; import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; import io.opentelemetry.context.Context; @@ -2304,6 +2305,7 @@ public void processElement( Span span = Preconditions.checkArgumentNotNull(tracer) .spanBuilder("KafkaIO.Read") + .setSpanKind(SpanKind.CONSUMER) .setParent(context) .startSpan(); try (Scope ignored = span.makeCurrent()) { @@ -2316,6 +2318,18 @@ public void processElement( static class OpenTelemetryHeaderPropagator extends DoFn, ProducerRecord> { + transient @Nullable Tracer tracer = null; + + @Setup + public void setup(PipelineOptions options) { + // inject tracer via options + io.opentelemetry.api.OpenTelemetry openTelemetry = + options.as(SdkHarnessOptions.class).getOpenTelemetry(); + if (openTelemetry != null) { + tracer = openTelemetry.getTracer("KafkaIO"); + } + } + ProducerRecord injectTraceContext(ProducerRecord message) { org.apache.kafka.common.header.internals.RecordHeaders headers = new org.apache.kafka.common.header.internals.RecordHeaders(message.headers()); @@ -2340,7 +2354,17 @@ ProducerRecord injectTraceContext(ProducerRecord message) { @ProcessElement public void processElement( @Element ProducerRecord element, OutputReceiver> receiver) { - receiver.output(injectTraceContext(element)); + Span span = + Preconditions.checkArgumentNotNull(tracer) + .spanBuilder("KafkaIO.Write") + .setSpanKind(SpanKind.PRODUCER) + .setParent(Context.current()) + .startSpan(); + try (Scope ignored = span.makeCurrent()) { + receiver.output(injectTraceContext(element)); + } finally { + span.end(); + } } } diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java index 1e3206814144..8e56f08f7a1b 100644 --- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java +++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java @@ -42,6 +42,15 @@ import io.confluent.kafka.schemaregistry.testutil.MockSchemaRegistry; import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig; import io.confluent.kafka.serializers.KafkaAvroSerializer; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Context; import java.io.IOException; import java.lang.reflect.Method; import java.nio.ByteBuffer; @@ -88,7 +97,9 @@ import org.apache.beam.sdk.metrics.MetricsFilter; import org.apache.beam.sdk.metrics.SinkMetrics; import org.apache.beam.sdk.metrics.SourceMetrics; +import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.apache.beam.sdk.options.SdkHarnessOptions; import org.apache.beam.sdk.options.StreamingOptions; import org.apache.beam.sdk.schemas.NoSuchSchemaException; import org.apache.beam.sdk.schemas.Schema; @@ -142,6 +153,7 @@ import org.apache.kafka.common.header.Header; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.common.serialization.ByteArrayDeserializer; import org.apache.kafka.common.serialization.Deserializer; @@ -164,6 +176,8 @@ import org.junit.rules.ExpectedException; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -2544,6 +2558,117 @@ public void testKafkaHeaderSchema() throws NoSuchSchemaException { assertEquals(Schema.Field.nullable("value", Schema.FieldType.BYTES), schema.getField(1)); } + @Test + public void testOpenTelemetryHeaderConsumer() { + String traceId = "0af7651916cd43dd8448eb211c80319c"; + String parentSpanId = "b7ad6b7169203331"; + String childSpanId = "1111111111111111"; + String traceparent = "00-" + traceId + "-" + parentSpanId + "-01"; + + OpenTelemetry mockOpenTelemetry = Mockito.mock(OpenTelemetry.class); + Tracer mockTracer = Mockito.mock(Tracer.class); + SpanBuilder mockSpanBuilder = Mockito.mock(SpanBuilder.class, Mockito.RETURNS_SELF); + Span mockSpan = Mockito.mock(Span.class); + SpanContext childSpanContext = + SpanContext.create(traceId, childSpanId, TraceFlags.getSampled(), TraceState.getDefault()); + Mockito.when(mockSpan.getSpanContext()).thenReturn(childSpanContext); + Mockito.when(mockSpan.storeInContext(Mockito.any())).thenCallRealMethod(); + Mockito.when(mockSpan.makeCurrent()).thenCallRealMethod(); + Mockito.when(mockOpenTelemetry.getTracer("KafkaIO")).thenReturn(mockTracer); + Mockito.when(mockTracer.spanBuilder("KafkaIO.Read")).thenReturn(mockSpanBuilder); + Mockito.when(mockSpanBuilder.startSpan()).thenReturn(mockSpan); + + PipelineOptions options = PipelineOptionsFactory.create(); + options.as(SdkHarnessOptions.class).setOpenTelemetry(mockOpenTelemetry); + + KafkaIO.OpenTelemetryHeaderConsumer consumer = + new KafkaIO.OpenTelemetryHeaderConsumer<>(); + consumer.setup(options); + + RecordHeaders headers = new RecordHeaders(); + headers.add("traceparent", traceparent.getBytes(StandardCharsets.UTF_8)); + KafkaRecord record = + new KafkaRecord<>( + "topic", 0, 0L, 0L, KafkaTimestampType.CREATE_TIME, headers, "key", "value"); + + List> outputs = new ArrayList<>(); + @SuppressWarnings("unchecked") + DoFn.OutputReceiver> receiver = + Mockito.mock(DoFn.OutputReceiver.class); + Mockito.doAnswer( + invocation -> { + assertEquals(childSpanId, Span.current().getSpanContext().getSpanId()); + outputs.add(invocation.getArgument(0)); + return null; + }) + .when(receiver) + .output(Mockito.any()); + consumer.processElement(record, receiver); + + assertEquals(1, outputs.size()); + Mockito.verify(mockTracer).spanBuilder("KafkaIO.Read"); + Mockito.verify(mockSpanBuilder).setSpanKind(SpanKind.CONSUMER); + ArgumentCaptor parentContextCaptor = ArgumentCaptor.forClass(Context.class); + Mockito.verify(mockSpanBuilder).setParent(parentContextCaptor.capture()); + assertEquals( + parentSpanId, + Span.fromContext(parentContextCaptor.getValue()).getSpanContext().getSpanId()); + Mockito.verify(mockSpan).end(); + } + + @Test + public void testOpenTelemetryHeaderPropagator() { + String traceId = "0af7651916cd43dd8448eb211c80319c"; + String producerSpanId = "2222222222222222"; + + OpenTelemetry mockOpenTelemetry = Mockito.mock(OpenTelemetry.class); + Tracer mockTracer = Mockito.mock(Tracer.class); + SpanBuilder mockSpanBuilder = Mockito.mock(SpanBuilder.class, Mockito.RETURNS_SELF); + Span mockSpan = Mockito.mock(Span.class); + SpanContext producerSpanContext = + SpanContext.create( + traceId, producerSpanId, TraceFlags.getSampled(), TraceState.getDefault()); + Mockito.when(mockSpan.getSpanContext()).thenReturn(producerSpanContext); + Mockito.when(mockSpan.storeInContext(Mockito.any())).thenCallRealMethod(); + Mockito.when(mockSpan.makeCurrent()).thenCallRealMethod(); + Mockito.when(mockOpenTelemetry.getTracer("KafkaIO")).thenReturn(mockTracer); + Mockito.when(mockTracer.spanBuilder("KafkaIO.Write")).thenReturn(mockSpanBuilder); + Mockito.when(mockSpanBuilder.startSpan()).thenReturn(mockSpan); + + PipelineOptions options = PipelineOptionsFactory.create(); + options.as(SdkHarnessOptions.class).setOpenTelemetry(mockOpenTelemetry); + + KafkaIO.OpenTelemetryHeaderPropagator propagator = + new KafkaIO.OpenTelemetryHeaderPropagator<>(); + propagator.setup(options); + + ProducerRecord record = new ProducerRecord<>("topic", "key", "value"); + List> outputs = new ArrayList<>(); + @SuppressWarnings("unchecked") + DoFn.OutputReceiver> receiver = + Mockito.mock(DoFn.OutputReceiver.class); + Mockito.doAnswer( + invocation -> { + assertEquals(producerSpanId, Span.current().getSpanContext().getSpanId()); + outputs.add(invocation.getArgument(0)); + return null; + }) + .when(receiver) + .output(Mockito.any()); + propagator.processElement(record, receiver); + + assertEquals(1, outputs.size()); + ProducerRecord propagated = outputs.get(0); + Header traceparentHeader = propagated.headers().lastHeader("traceparent"); + assertNotNull(traceparentHeader); + assertEquals( + "00-" + traceId + "-" + producerSpanId + "-01", + new String(traceparentHeader.value(), StandardCharsets.UTF_8)); + Mockito.verify(mockTracer).spanBuilder("KafkaIO.Write"); + Mockito.verify(mockSpanBuilder).setSpanKind(SpanKind.PRODUCER); + Mockito.verify(mockSpan).end(); + } + private static void verifyProducerRecords( MockProducer mockProducer, String topic, From a90d3ce0120baaf2dd4bb37eb17fd7e4a2d17697 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Tue, 13 Oct 2026 13:23:17 +0200 Subject: [PATCH 2/2] address comments --- .../java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java | 2 +- .../src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java index 6ce8be5b7673..0eb2b44bb734 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java @@ -1602,7 +1602,7 @@ public void processElement( new PubsubMessage( message.getPayload(), attr, message.getMessageId(), message.getOrderingKey()); - // topic is copied seperately, not via constructor + // topic is copied separately, not via constructor if (message.getTopic() != null) { ps = ps.withTopic(message.getTopic()); } diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java index af534f5697c3..a41db9fbeaf4 100644 --- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java +++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java @@ -143,6 +143,7 @@ import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.utils.AppInfoParser; +import org.checkerframework.checker.nullness.qual.MonotonicNonNull; import org.checkerframework.checker.nullness.qual.Nullable; import org.checkerframework.dataflow.qual.Pure; import org.joda.time.Duration; @@ -2252,7 +2253,7 @@ public void populateDisplayData(DisplayData.Builder builder) { static class OpenTelemetryHeaderConsumer extends DoFn, KafkaRecord> { - @Nullable Tracer tracer = null; + private transient @MonotonicNonNull Tracer tracer = null; @Setup public void setup(PipelineOptions options) { @@ -2318,7 +2319,7 @@ public void processElement( static class OpenTelemetryHeaderPropagator extends DoFn, ProducerRecord> { - transient @Nullable Tracer tracer = null; + transient @MonotonicNonNull Tracer tracer = null; @Setup public void setup(PipelineOptions options) {