Refactor kafka-clients to use Instrumenter API
DELETE
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaClientsConfig.java
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package io.opentelemetry.javaagent.instrumentation.kafkaclients;
-
-import io.opentelemetry.instrumentation.api.config.Config;
-
-public final class KafkaClientsConfig {
-
- private static final boolean CLIENT_PROPAGATION_ENABLED =
- Config.get().getBoolean("otel.instrumentation.kafka.client-propagation.enabled", true);
-
- private static final boolean CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES =
- Config.get().getBoolean("otel.instrumentation.kafka.experimental-span-attributes", false);
-
- public static boolean isPropagationEnabled() {
- return CLIENT_PROPAGATION_ENABLED;
- }
-
- public static boolean captureExperimentalSpanAttributes() {
- return CAPTURE_EXPERIMENTAL_SPAN_ATTRIBUTES;
- }
-
- private KafkaClientsConfig() {}
-}

MODIFY
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaConsumerInstrumentation.java
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaConsumerInstrumentation.java
package io.opentelemetry.javaagent.instrumentation.kafkaclients;
-import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaConsumerTracer.tracer;
+import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaSingletons.consumerInstrumenter;
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
import static net.bytebuddy.matcher.ElementMatchers.isPublic;
import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -61,7 +61,7 @@ public class KafkaConsumerInstrumentation implements TypeInstrumentation {
public static void wrap(
@Advice.Return(readOnly = false) Iterable<ConsumerRecord<?, ?>> iterable) {
if (iterable != null) {
- iterable = new TracingIterable(iterable, tracer());
+ iterable = new TracingIterable(iterable, consumerInstrumenter());
}
}
}
@@ -72,7 +72,7 @@ public class KafkaConsumerInstrumentation implements TypeInstrumentation {
@Advice.OnMethodExit(suppress = Throwable.class)
public static void wrap(@Advice.Return(readOnly = false) List<ConsumerRecord<?, ?>> iterable) {
if (iterable != null) {
- iterable = new TracingList(iterable, tracer());
+ iterable = new TracingList(iterable, consumerInstrumenter());
}
}
}
@@ -84,7 +84,7 @@ public class KafkaConsumerInstrumentation implements TypeInstrumentation {
public static void wrap(
@Advice.Return(readOnly = false) Iterator<ConsumerRecord<?, ?>> iterator) {
if (iterator != null) {
- iterator = new TracingIterator(iterator, tracer());
+ iterator = new TracingIterator(iterator, consumerInstrumenter());
}
}
}

DELETE
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaConsumerTracer.java
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package io.opentelemetry.javaagent.instrumentation.kafkaclients;
-
-import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
-
-import io.opentelemetry.api.trace.Span;
-import io.opentelemetry.context.Context;
-import io.opentelemetry.instrumentation.api.tracer.BaseTracer;
-import io.opentelemetry.javaagent.instrumentation.kafka.KafkaHeadersGetter;
-import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
-import java.util.concurrent.TimeUnit;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.common.record.TimestampType;
-
-public class KafkaConsumerTracer extends BaseTracer {
- private static final KafkaConsumerTracer TRACER = new KafkaConsumerTracer();
- private static final KafkaHeadersGetter GETTER = new KafkaHeadersGetter();
-
- public static KafkaConsumerTracer tracer() {
- return TRACER;
- }
-
- public Context startSpan(ConsumerRecord<?, ?> record) {
- long now = System.currentTimeMillis();
-
- Context parentContext = extractParent(record);
- Span span =
- spanBuilder(parentContext, spanNameOnConsume(record), CONSUMER)
- .setStartTimestamp(now, TimeUnit.MILLISECONDS)
- .setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "kafka")
- .setAttribute(SemanticAttributes.MESSAGING_DESTINATION, record.topic())
- .setAttribute(SemanticAttributes.MESSAGING_DESTINATION_KIND, "topic")
- .setAttribute(SemanticAttributes.MESSAGING_OPERATION, "process")
- .setAttribute(
- SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_SIZE_BYTES,
- (long) record.serializedValueSize())
- .startSpan();
-
- onConsume(span, now, record);
- return withConsumerSpan(parentContext, span);
- }
-
- private Context extractParent(ConsumerRecord<?, ?> record) {
- if (KafkaClientsConfig.isPropagationEnabled()) {
- return extract(record, GETTER);
- } else {
- return Context.current();
- }
- }
-
- public String spanNameOnConsume(ConsumerRecord<?, ?> record) {
- return record.topic() + " process";
- }
-
- public void onConsume(Span span, long startTimeMillis, ConsumerRecord<?, ?> record) {
- // TODO should we set topic + offset as messaging.message_id?
- span.setAttribute(SemanticAttributes.MESSAGING_KAFKA_PARTITION, record.partition());
- if (record.value() == null) {
- span.setAttribute(SemanticAttributes.MESSAGING_KAFKA_TOMBSTONE, true);
- }
-
- if (KafkaClientsConfig.captureExperimentalSpanAttributes()) {
- span.setAttribute("kafka.offset", record.offset());
-
- // don't record a duration if the message was sent from an old Kafka client
- if (record.timestampType() != TimestampType.NO_TIMESTAMP_TYPE) {
- long produceTime = record.timestamp();
- // this attribute shows how much time elapsed between the producer and the consumer of this
- // message, which can be helpful for identifying queue bottlenecks
- span.setAttribute(
- "kafka.record.queue_time_ms", Math.max(0L, startTimeMillis - produceTime));
- }
- }
- }
-
- @Override
- protected String getInstrumentationName() {
- return "io.opentelemetry.kafka-clients-0.11";
- }
-}

RENAME
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/TextMapInjectAdapter.java
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaHeadersSetter.java
@@ -9,9 +9,7 @@ import io.opentelemetry.context.propagation.TextMapSetter;
import java.nio.charset.StandardCharsets;
import org.apache.kafka.common.header.Headers;
-public class TextMapInjectAdapter implements TextMapSetter<Headers> {
-
- public static final TextMapInjectAdapter SETTER = new TextMapInjectAdapter();
+public final class KafkaHeadersSetter implements TextMapSetter<Headers> {
@Override
public void set(Headers headers, String key, String value) {

ADD
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaProducerAdditionalAttributesExtractor.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaclients;
+
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+public final class KafkaProducerAdditionalAttributesExtractor
+ extends AttributesExtractor<ProducerRecord<?, ?>, Void> {
+ @Override
+ protected void onStart(AttributesBuilder attributes, ProducerRecord<?, ?> producerRecord) {
+ Integer partition = producerRecord.partition();
+ if (partition != null) {
+ set(attributes, SemanticAttributes.MESSAGING_KAFKA_PARTITION, partition.longValue());
+ }
+ if (producerRecord.value() == null) {
+ set(attributes, SemanticAttributes.MESSAGING_KAFKA_TOMBSTONE, true);
+ }
+ }
+
+ @Override
+ protected void onEnd(
+ AttributesBuilder attributes, ProducerRecord<?, ?> producerRecord, @Nullable Void unused) {}
+}

ADD
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaProducerAttributesExtractor.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaclients;
+
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessageOperation;
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingAttributesExtractor;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+public final class KafkaProducerAttributesExtractor
+ extends MessagingAttributesExtractor<ProducerRecord<?, ?>, Void> {
+ @Override
+ protected String system(ProducerRecord<?, ?> producerRecord) {
+ return "kafka";
+ }
+
+ @Override
+ protected String destinationKind(ProducerRecord<?, ?> producerRecord) {
+ return SemanticAttributes.MessagingDestinationKindValues.TOPIC;
+ }
+
+ @Override
+ protected String destination(ProducerRecord<?, ?> producerRecord) {
+ return producerRecord.topic();
+ }
+
+ @Override
+ protected boolean temporaryDestination(ProducerRecord<?, ?> producerRecord) {
+ return false;
+ }
+
+ @Override
+ protected @Nullable String protocol(ProducerRecord<?, ?> producerRecord) {
+ return null;
+ }
+
+ @Override
+ protected @Nullable String protocolVersion(ProducerRecord<?, ?> producerRecord) {
+ return null;
+ }
+
+ @Override
+ protected @Nullable String url(ProducerRecord<?, ?> producerRecord) {
+ return null;
+ }
+
+ @Override
+ protected @Nullable String conversationId(ProducerRecord<?, ?> producerRecord) {
+ return null;
+ }
+
+ @Override
+ protected @Nullable Long messagePayloadSize(ProducerRecord<?, ?> producerRecord) {
+ return null;
+ }
+
+ @Override
+ protected @Nullable Long messagePayloadCompressedSize(ProducerRecord<?, ?> producerRecord) {
+ return null;
+ }
+
+ @Override
+ protected MessageOperation operation(ProducerRecord<?, ?> producerRecord) {
+ return MessageOperation.SEND;
+ }
+
+ @Override
+ protected @Nullable String messageId(ProducerRecord<?, ?> producerRecord, @Nullable Void unused) {
+ return null;
+ }
+}

MODIFY
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaProducerInstrumentation.java
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaProducerInstrumentation.java
package io.opentelemetry.javaagent.instrumentation.kafkaclients;
-import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaProducerTracer.tracer;
-import static io.opentelemetry.javaagent.instrumentation.kafkaclients.TextMapInjectAdapter.SETTER;
+import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaSingletons.producerInstrumenter;
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
import static net.bytebuddy.matcher.ElementMatchers.isPublic;
import static net.bytebuddy.matcher.ElementMatchers.named;
@@ -54,42 +53,33 @@ public class KafkaProducerInstrumentation implements TypeInstrumentation {
@Advice.Local("otelScope") Scope scope) {
Context parentContext = Java8BytecodeBridge.currentContext();
+ if (!producerInstrumenter().shouldStart(parentContext, record)) {
+ return;
+ }
- context = tracer().startProducerSpan(parentContext, record);
-
- callback = new ProducerCallback(callback, parentContext, context);
-
- if (tracer().shouldPropagate(apiVersions)) {
- try {
- tracer().inject(context, record.headers(), SETTER);
- } catch (IllegalStateException e) {
- // headers must be read-only from reused record. try again with new one.
- record =
- new ProducerRecord<>(
- record.topic(),
- record.partition(),
- record.timestamp(),
- record.key(),
- record.value(),
- record.headers());
+ context = producerInstrumenter().start(parentContext, record);
+ scope = context.makeCurrent();
- tracer().inject(context, record.headers(), SETTER);
- }
+ if (KafkaPropagation.shouldPropagate(apiVersions)) {
+ record = KafkaPropagation.propagateContext(context, record);
}
- scope = context.makeCurrent();
+ callback = new ProducerCallback(callback, parentContext, context, record);
}
@Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
public static void stopSpan(
+ @Advice.Argument(value = 0, readOnly = false) ProducerRecord<?, ?> record,
@Advice.Thrown Throwable throwable,
@Advice.Local("otelContext") Context context,
@Advice.Local("otelScope") Scope scope) {
-
+ if (scope == null) {
+ return;
+ }
scope.close();
if (throwable != null) {
- tracer().endExceptionally(context, throwable);
+ producerInstrumenter().end(context, record, null, throwable);
}
// span finished by ProducerCallback
}

DELETE
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaProducerTracer.java
-/*
- * Copyright The OpenTelemetry Authors
- * SPDX-License-Identifier: Apache-2.0
- */
-
-package io.opentelemetry.javaagent.instrumentation.kafkaclients;
-
-import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
-
-import io.opentelemetry.api.trace.SpanBuilder;
-import io.opentelemetry.context.Context;
-import io.opentelemetry.instrumentation.api.tracer.BaseTracer;
-import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
-import org.apache.kafka.clients.ApiVersions;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.record.RecordBatch;
-
-public class KafkaProducerTracer extends BaseTracer {
- private static final KafkaProducerTracer TRACER = new KafkaProducerTracer();
-
- public static KafkaProducerTracer tracer() {
- return TRACER;
- }
-
- public Context startProducerSpan(Context parentContext, ProducerRecord<?, ?> record) {
- SpanBuilder span = spanBuilder(parentContext, spanNameOnProduce(record), PRODUCER);
- onProduce(span, record);
- return parentContext.with(span.startSpan());
- }
-
- // Do not inject headers for batch versions below 2
- // This is how similar check is being done in Kafka client itself:
- // https://github.com/apache/kafka/blob/05fcfde8f69b0349216553f711fdfc3f0259c601/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java#L411-L412
- // Also, do not inject headers if specified by JVM option or environment variable
- // This can help in mixed client environments where clients < 0.11 that do not support
- // headers attempt to read messages that were produced by clients > 0.11 and the magic
- // value of the broker(s) is >= 2
- public boolean shouldPropagate(ApiVersions apiVersions) {
- return apiVersions.maxUsableProduceMagic() >= RecordBatch.MAGIC_VALUE_V2
- && KafkaClientsConfig.isPropagationEnabled();
- }
-
- public String spanNameOnProduce(ProducerRecord<?, ?> record) {
- return record.topic() + " send";
- }
-
- public void onProduce(SpanBuilder span, ProducerRecord<?, ?> record) {
- span.setAttribute(SemanticAttributes.MESSAGING_SYSTEM, "kafka");
- span.setAttribute(SemanticAttributes.MESSAGING_DESTINATION_KIND, "topic");
- span.setAttribute(SemanticAttributes.MESSAGING_DESTINATION, record.topic());
-
- Integer partition = record.partition();
- if (partition != null) {
- span.setAttribute(SemanticAttributes.MESSAGING_KAFKA_PARTITION, partition.longValue());
- }
- if (record.value() == null) {
- span.setAttribute(SemanticAttributes.MESSAGING_KAFKA_TOMBSTONE, true);
- }
- }
-
- @Override
- protected String getInstrumentationName() {
- return "io.opentelemetry.kafka-clients-0.11";
- }
-}

ADD
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaPropagation.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaclients;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.api.config.Config;
+import org.apache.kafka.clients.ApiVersions;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.record.RecordBatch;
+
+public final class KafkaPropagation {
+
+ private static final KafkaHeadersSetter SETTER = new KafkaHeadersSetter();
+
+ private static final boolean CLIENT_PROPAGATION_ENABLED =
+ Config.get().getBoolean("otel.instrumentation.kafka.client-propagation.enabled", true);
+
+ public static boolean isPropagationEnabled() {
+ return CLIENT_PROPAGATION_ENABLED;
+ }
+
+ // Do not inject headers for batch versions below 2
+ // This is how similar check is being done in Kafka client itself:
+ // https://github.com/apache/kafka/blob/05fcfde8f69b0349216553f711fdfc3f0259c601/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java#L411-L412
+ // Also, do not inject headers if specified by JVM option or environment variable
+ // This can help in mixed client environments where clients < 0.11 that do not support
+ // headers attempt to read messages that were produced by clients > 0.11 and the magic
+ // value of the broker(s) is >= 2
+ public static boolean shouldPropagate(ApiVersions apiVersions) {
+ return apiVersions.maxUsableProduceMagic() >= RecordBatch.MAGIC_VALUE_V2
+ && isPropagationEnabled();
+ }
+
+ public static <K, V> ProducerRecord<K, V> propagateContext(
+ Context context, ProducerRecord<K, V> record) {
+ try {
+ inject(context, record);
+ } catch (IllegalStateException e) {
+ // headers must be read-only from reused record. try again with new one.
+ record =
+ new ProducerRecord<>(
+ record.topic(),
+ record.partition(),
+ record.timestamp(),
+ record.key(),
+ record.value(),
+ record.headers());
+
+ inject(context, record);
+ }
+ return record;
+ }
+
+ private static <K, V> void inject(Context context, ProducerRecord<K, V> record) {
+ GlobalOpenTelemetry.getPropagators()
+ .getTextMapPropagator()
+ .inject(context, record.headers(), SETTER);
+ }
+
+ private KafkaPropagation() {}
+}

ADD
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaSingletons.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.kafkaclients;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
+import io.opentelemetry.instrumentation.api.instrumenter.InstrumenterBuilder;
+import io.opentelemetry.instrumentation.api.instrumenter.SpanKindExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.SpanNameExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.messaging.MessagingSpanNameExtractor;
+import io.opentelemetry.javaagent.instrumentation.kafka.KafkaConsumerAdditionalAttributesExtractor;
+import io.opentelemetry.javaagent.instrumentation.kafka.KafkaConsumerAttributesExtractor;
+import io.opentelemetry.javaagent.instrumentation.kafka.KafkaConsumerExperimentalAttributesExtractor;
+import io.opentelemetry.javaagent.instrumentation.kafka.KafkaHeadersGetter;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+public final class KafkaSingletons {
+ private static final String INSTRUMENTATION_NAME = "io.opentelemetry.kafka-clients-0.11";
+
+ private static final Instrumenter<ProducerRecord<?, ?>, Void> PRODUCER_INSTRUMENTER =
+ buildProducerInstrumenter();
+ private static final Instrumenter<ConsumerRecord<?, ?>, Void> CONSUMER_INSTRUMENTER =
+ buildConsumerInstrumenter();
+
+ private static Instrumenter<ProducerRecord<?, ?>, Void> buildProducerInstrumenter() {
+ KafkaProducerAttributesExtractor attributesExtractor = new KafkaProducerAttributesExtractor();
+ SpanNameExtractor<ProducerRecord<?, ?>> spanNameExtractor =
+ MessagingSpanNameExtractor.create(attributesExtractor);
+
+ return Instrumenter.<ProducerRecord<?, ?>, Void>newBuilder(
+ GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME, spanNameExtractor)
+ .addAttributesExtractor(attributesExtractor)
+ .addAttributesExtractor(new KafkaProducerAdditionalAttributesExtractor())
+ .newInstrumenter(SpanKindExtractor.alwaysProducer());
+ }
+
+ private static Instrumenter<ConsumerRecord<?, ?>, Void> buildConsumerInstrumenter() {
+ KafkaConsumerAttributesExtractor attributesExtractor = new KafkaConsumerAttributesExtractor();
+ SpanNameExtractor<ConsumerRecord<?, ?>> spanNameExtractor =
+ MessagingSpanNameExtractor.create(attributesExtractor);
+
+ InstrumenterBuilder<ConsumerRecord<?, ?>, Void> builder =
+ Instrumenter.<ConsumerRecord<?, ?>, Void>newBuilder(
+ GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME, spanNameExtractor)
+ .addAttributesExtractor(attributesExtractor)
+ .addAttributesExtractor(new KafkaConsumerAdditionalAttributesExtractor());
+ if (KafkaConsumerExperimentalAttributesExtractor.isEnabled()) {
+ builder.addAttributesExtractor(new KafkaConsumerExperimentalAttributesExtractor());
+ }
+ return KafkaPropagation.isPropagationEnabled()
+ ? builder.newConsumerInstrumenter(new KafkaHeadersGetter())
+ : builder.newInstrumenter(SpanKindExtractor.alwaysConsumer());
+ }
+
+ public static Instrumenter<ProducerRecord<?, ?>, Void> producerInstrumenter() {
+ return PRODUCER_INSTRUMENTER;
+ }
+
+ public static Instrumenter<ConsumerRecord<?, ?>, Void> consumerInstrumenter() {
+ return CONSUMER_INSTRUMENTER;
+ }
+
+ private KafkaSingletons() {}
+}

MODIFY
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/ProducerCallback.java
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/ProducerCallback.java
package io.opentelemetry.javaagent.instrumentation.kafkaclients;
-import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaProducerTracer.tracer;
+import static io.opentelemetry.javaagent.instrumentation.kafkaclients.KafkaSingletons.producerInstrumenter;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
public class ProducerCallback implements Callback {
private final Callback callback;
private final Context parentContext;
private final Context context;
+ private final ProducerRecord<?, ?> request;
- public ProducerCallback(Callback callback, Context parentContext, Context context) {
+ public ProducerCallback(
+ Callback callback, Context parentContext, Context context, ProducerRecord<?, ?> request) {
this.callback = callback;
this.parentContext = parentContext;
this.context = context;
+ this.request = request;
}
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
- if (exception != null) {
- tracer().endExceptionally(context, exception);
- } else {
- tracer().end(context);
- }
+ producerInstrumenter().end(context, request, null, exception);
if (callback != null) {
try (Scope ignored = parentContext.makeCurrent()) {

MODIFY
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/TracingIterable.java
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/TracingIterable.java
package io.opentelemetry.javaagent.instrumentation.kafkaclients;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
import java.util.Iterator;
import org.apache.kafka.clients.consumer.ConsumerRecord;
public class TracingIterable implements Iterable<ConsumerRecord<?, ?>> {
private final Iterable<ConsumerRecord<?, ?>> delegate;
- private final KafkaConsumerTracer tracer;
+ private final Instrumenter<ConsumerRecord<?, ?>, Void> instrumenter;
private boolean firstIterator = true;
- public TracingIterable(Iterable<ConsumerRecord<?, ?>> delegate, KafkaConsumerTracer tracer) {
+ public TracingIterable(
+ Iterable<ConsumerRecord<?, ?>> delegate,
+ Instrumenter<ConsumerRecord<?, ?>, Void> instrumenter) {
this.delegate = delegate;
- this.tracer = tracer;
+ this.instrumenter = instrumenter;
}
@Override
@@ -25,7 +28,7 @@ public class TracingIterable implements Iterable<ConsumerRecord<?, ?>> {
// However, this is not thread-safe, but usually the first (hopefully only) traversal of
// ConsumerRecords is performed in the same thread that called poll()
if (firstIterator) {
- it = new TracingIterator(delegate.iterator(), tracer);
+ it = new TracingIterator(delegate.iterator(), instrumenter);
firstIterator = false;
} else {
it = delegate.iterator();

MODIFY
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/TracingIterator.java
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/TracingIterator.java
@@ -7,6 +7,7 @@ package io.opentelemetry.javaagent.instrumentation.kafkaclients;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
import java.util.Iterator;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -14,20 +15,24 @@ import org.checkerframework.checker.nullness.qual.Nullable;
public class TracingIterator implements Iterator<ConsumerRecord<?, ?>> {
private final Iterator<ConsumerRecord<?, ?>> delegateIterator;
- private final KafkaConsumerTracer tracer;
+ private final Instrumenter<ConsumerRecord<?, ?>, Void> instrumenter;
+ private final Context parentContext;
/**
* Note: this may potentially create problems if this iterator is used from different threads. But
* at the moment we cannot do much about this.
*/
- @Nullable private Context currentContext;
+ @Nullable private ConsumerRecord<?, ?> currentRequest;
+ @Nullable private Context currentContext;
@Nullable private Scope currentScope;
public TracingIterator(
- Iterator<ConsumerRecord<?, ?>> delegateIterator, KafkaConsumerTracer tracer) {
+ Iterator<ConsumerRecord<?, ?>> delegateIterator,
+ Instrumenter<ConsumerRecord<?, ?>, Void> instrumenter) {
this.delegateIterator = delegateIterator;
- this.tracer = tracer;
+ this.instrumenter = instrumenter;
+ parentContext = Context.current();
}
@Override
@@ -42,9 +47,9 @@ public class TracingIterator implements Iterator<ConsumerRecord<?, ?>> {
closeScopeAndEndSpan();
ConsumerRecord<?, ?> next = delegateIterator.next();
-
- if (next != null) {
- currentContext = tracer.startSpan(next);
+ if (next != null && instrumenter.shouldStart(parentContext, next)) {
+ currentRequest = next;
+ currentContext = instrumenter.start(parentContext, currentRequest);
currentScope = currentContext.makeCurrent();
}
return next;
@@ -53,8 +58,9 @@ public class TracingIterator implements Iterator<ConsumerRecord<?, ?>> {
private void closeScopeAndEndSpan() {
if (currentScope != null) {
currentScope.close();
+ instrumenter.end(currentContext, currentRequest, null, null);
currentScope = null;
- tracer.end(currentContext);
+ currentRequest = null;
currentContext = null;
}
}

MODIFY
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/TracingList.java
instrumentation/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/TracingList.java
package io.opentelemetry.javaagent.instrumentation.kafkaclients;
+import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
import java.util.Collection;
import java.util.List;
import java.util.ListIterator;
@@ -13,8 +14,9 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
public class TracingList extends TracingIterable implements List<ConsumerRecord<?, ?>> {
private final List<ConsumerRecord<?, ?>> delegate;
- public TracingList(List<ConsumerRecord<?, ?>> delegate, KafkaConsumerTracer tracer) {
- super(delegate, tracer);
+ public TracingList(
+ List<ConsumerRecord<?, ?>> delegate, Instrumenter<ConsumerRecord<?, ?>, Void> instrumenter) {
+ super(delegate, instrumenter);
this.delegate = delegate;
}

Refactor kafka-clients to use Instrumenter API (#3883)