Refactor KafkaInstrumenterFactory a bit
MODIFY
instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaSingletons.java
instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/KafkaSingletons.java
package io.opentelemetry.javaagent.instrumentation.kafkaclients;
+import io.opentelemetry.api.GlobalOpenTelemetry;
import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
import io.opentelemetry.instrumentation.kafka.internal.KafkaInstrumenterFactory;
import io.opentelemetry.instrumentation.kafka.internal.ReceivedRecords;
@@ -14,12 +15,17 @@ 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 =
- KafkaInstrumenterFactory.createProducerInstrumenter(INSTRUMENTATION_NAME);
- private static final Instrumenter<ReceivedRecords, Void> CONSUMER_RECEIVE_INSTRUMENTER =
- KafkaInstrumenterFactory.createConsumerReceiveInstrumenter(INSTRUMENTATION_NAME);
- private static final Instrumenter<ConsumerRecord<?, ?>, Void> CONSUMER_PROCESS_INSTRUMENTER =
- KafkaInstrumenterFactory.createConsumerProcessInstrumenter(INSTRUMENTATION_NAME);
+ private static final Instrumenter<ProducerRecord<?, ?>, Void> PRODUCER_INSTRUMENTER;
+ private static final Instrumenter<ReceivedRecords, Void> CONSUMER_RECEIVE_INSTRUMENTER;
+ private static final Instrumenter<ConsumerRecord<?, ?>, Void> CONSUMER_PROCESS_INSTRUMENTER;
+
+ static {
+ KafkaInstrumenterFactory instrumenterFactory =
+ new KafkaInstrumenterFactory(GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME);
+ PRODUCER_INSTRUMENTER = instrumenterFactory.createProducerInstrumenter();
+ CONSUMER_RECEIVE_INSTRUMENTER = instrumenterFactory.createConsumerReceiveInstrumenter();
+ CONSUMER_PROCESS_INSTRUMENTER = instrumenterFactory.createConsumerProcessInstrumenter();
+ }
public static Instrumenter<ProducerRecord<?, ?>, Void> producerInstrumenter() {
return PRODUCER_INSTRUMENTER;

MODIFY
instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/KafkaTelemetryBuilder.java
instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/KafkaTelemetryBuilder.java
@@ -41,14 +41,13 @@ public final class KafkaTelemetryBuilder {
}
public KafkaTelemetry build() {
+ KafkaInstrumenterFactory instrumenterFactory =
+ new KafkaInstrumenterFactory(openTelemetry, INSTRUMENTATION_NAME);
+
return new KafkaTelemetry(
openTelemetry,
- KafkaInstrumenterFactory.createProducerInstrumenter(
- INSTRUMENTATION_NAME, openTelemetry, producerAttributesExtractors),
- KafkaInstrumenterFactory.createConsumerOperationInstrumenter(
- INSTRUMENTATION_NAME,
- openTelemetry,
- MessageOperation.RECEIVE,
- consumerAttributesExtractors));
+ instrumenterFactory.createProducerInstrumenter(producerAttributesExtractors),
+ instrumenterFactory.createConsumerOperationInstrumenter(
+ MessageOperation.RECEIVE, consumerAttributesExtractors));
}
}

MODIFY
instrumentation/kafka/kafka-clients/kafka-clients-common/library/src/main/java/io/opentelemetry/instrumentation/kafka/internal/KafkaInstrumenterFactory.java
instrumentation/kafka/kafka-clients/kafka-clients-common/library/src/main/java/io/opentelemetry/instrumentation/kafka/internal/KafkaInstrumenterFactory.java
@@ -27,15 +27,25 @@ import org.apache.kafka.clients.producer.ProducerRecord;
*/
public final class KafkaInstrumenterFactory {
- public static Instrumenter<ProducerRecord<?, ?>, Void> createProducerInstrumenter(
- String instrumentationName) {
- return createProducerInstrumenter(
- instrumentationName, GlobalOpenTelemetry.get(), Collections.emptyList());
+ private final OpenTelemetry openTelemetry;
+ private final String instrumentationName;
+ private ErrorCauseExtractor errorCauseExtractor = ErrorCauseExtractor.jdk();
+
+ public KafkaInstrumenterFactory(OpenTelemetry openTelemetry, String instrumentationName) {
+ this.openTelemetry = openTelemetry;
+ this.instrumentationName = instrumentationName;
+ }
+
+ public KafkaInstrumenterFactory setErrorCauseExtractor(ErrorCauseExtractor errorCauseExtractor) {
+ this.errorCauseExtractor = errorCauseExtractor;
+ return this;
}
- public static Instrumenter<ProducerRecord<?, ?>, Void> createProducerInstrumenter(
- String instrumentationName,
- OpenTelemetry openTelemetry,
+ public Instrumenter<ProducerRecord<?, ?>, Void> createProducerInstrumenter() {
+ return createProducerInstrumenter(Collections.emptyList());
+ }
+
+ public Instrumenter<ProducerRecord<?, ?>, Void> createProducerInstrumenter(
Iterable<AttributesExtractor<ProducerRecord<?, ?>, Void>> extractors) {
KafkaProducerAttributesGetter getter = KafkaProducerAttributesGetter.INSTANCE;
@@ -48,18 +58,15 @@ public final class KafkaInstrumenterFactory {
.addAttributesExtractor(MessagingAttributesExtractor.create(getter, operation))
.addAttributesExtractors(extractors)
.addAttributesExtractor(new KafkaProducerAdditionalAttributesExtractor())
+ .setErrorCauseExtractor(errorCauseExtractor)
.newInstrumenter(SpanKindExtractor.alwaysProducer());
}
- public static Instrumenter<ReceivedRecords, Void> createConsumerReceiveInstrumenter(
- String instrumentationName) {
- return createConsumerReceiveInstrumenter(
- instrumentationName, GlobalOpenTelemetry.get(), Collections.emptyList());
+ public Instrumenter<ReceivedRecords, Void> createConsumerReceiveInstrumenter() {
+ return createConsumerReceiveInstrumenter(Collections.emptyList());
}
- public static Instrumenter<ReceivedRecords, Void> createConsumerReceiveInstrumenter(
- String instrumentationName,
- OpenTelemetry openTelemetry,
+ public Instrumenter<ReceivedRecords, Void> createConsumerReceiveInstrumenter(
Iterable<AttributesExtractor<ReceivedRecords, Void>> extractors) {
KafkaReceiveAttributesGetter getter = KafkaReceiveAttributesGetter.INSTANCE;
@@ -71,46 +78,19 @@ public final class KafkaInstrumenterFactory {
MessagingSpanNameExtractor.create(getter, operation))
.addAttributesExtractor(MessagingAttributesExtractor.create(getter, operation))
.addAttributesExtractors(extractors)
+ .setErrorCauseExtractor(errorCauseExtractor)
.setTimeExtractor(new KafkaConsumerTimeExtractor())
.setEnabled(ExperimentalConfig.get().messagingReceiveInstrumentationEnabled())
.newInstrumenter(SpanKindExtractor.alwaysConsumer());
}
- public static Instrumenter<ConsumerRecord<?, ?>, Void> createConsumerProcessInstrumenter(
- String instrumentationName) {
- return createConsumerOperationInstrumenter(
- instrumentationName,
- GlobalOpenTelemetry.get(),
- MessageOperation.PROCESS,
- Collections.emptyList(),
- ErrorCauseExtractor.jdk());
+ public Instrumenter<ConsumerRecord<?, ?>, Void> createConsumerProcessInstrumenter() {
+ return createConsumerOperationInstrumenter(MessageOperation.PROCESS, Collections.emptyList());
}
- public static Instrumenter<ConsumerRecord<?, ?>, Void> createConsumerProcessInstrumenter(
- String instrumentationName, ErrorCauseExtractor errorCauseExtractor) {
- return createConsumerOperationInstrumenter(
- instrumentationName,
- GlobalOpenTelemetry.get(),
- MessageOperation.PROCESS,
- Collections.emptyList(),
- errorCauseExtractor);
- }
-
- public static Instrumenter<ConsumerRecord<?, ?>, Void> createConsumerOperationInstrumenter(
- String instrumentationName,
- OpenTelemetry openTelemetry,
+ public Instrumenter<ConsumerRecord<?, ?>, Void> createConsumerOperationInstrumenter(
MessageOperation operation,
Iterable<AttributesExtractor<ConsumerRecord<?, ?>, Void>> extractors) {
- return createConsumerOperationInstrumenter(
- instrumentationName, openTelemetry, operation, extractors, ErrorCauseExtractor.jdk());
- }
-
- private static Instrumenter<ConsumerRecord<?, ?>, Void> createConsumerOperationInstrumenter(
- String instrumentationName,
- OpenTelemetry openTelemetry,
- MessageOperation operation,
- Iterable<AttributesExtractor<ConsumerRecord<?, ?>, Void>> extractors,
- ErrorCauseExtractor errorCauseExtractor) {
KafkaConsumerAttributesGetter getter = KafkaConsumerAttributesGetter.INSTANCE;
@@ -138,6 +118,4 @@ public final class KafkaInstrumenterFactory {
return builder.newConsumerInstrumenter(KafkaConsumerRecordGetter.INSTANCE);
}
}
-
- private KafkaInstrumenterFactory() {}
}

MODIFY
instrumentation/kafka/kafka-streams-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkastreams/KafkaStreamsSingletons.java
instrumentation/kafka/kafka-streams-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkastreams/KafkaStreamsSingletons.java
package io.opentelemetry.javaagent.instrumentation.kafkastreams;
+import io.opentelemetry.api.GlobalOpenTelemetry;
import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
import io.opentelemetry.instrumentation.kafka.internal.KafkaInstrumenterFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -13,11 +14,9 @@ public final class KafkaStreamsSingletons {
private static final String INSTRUMENTATION_NAME = "io.opentelemetry.kafka-streams-0.11";
- private static final Instrumenter<ConsumerRecord<?, ?>, Void> INSTRUMENTER = buildInstrumenter();
-
- private static Instrumenter<ConsumerRecord<?, ?>, Void> buildInstrumenter() {
- return KafkaInstrumenterFactory.createConsumerProcessInstrumenter(INSTRUMENTATION_NAME);
- }
+ private static final Instrumenter<ConsumerRecord<?, ?>, Void> INSTRUMENTER =
+ new KafkaInstrumenterFactory(GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME)
+ .createConsumerProcessInstrumenter();
public static Instrumenter<ConsumerRecord<?, ?>, Void> instrumenter() {
return INSTRUMENTER;

MODIFY
instrumentation/spring/spring-kafka-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/kafka/SpringKafkaSingletons.java
instrumentation/spring/spring-kafka-2.7/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/kafka/SpringKafkaSingletons.java
@@ -22,8 +22,9 @@ public final class SpringKafkaSingletons {
private static final Instrumenter<ConsumerRecords<?, ?>, Void> BATCH_PROCESS_INSTRUMENTER =
buildBatchProcessInstrumenter();
private static final Instrumenter<ConsumerRecord<?, ?>, Void> PROCESS_INSTRUMENTER =
- KafkaInstrumenterFactory.createConsumerProcessInstrumenter(
- INSTRUMENTATION_NAME, SpringKafkaErrorCauseExtractor.INSTANCE);
+ new KafkaInstrumenterFactory(GlobalOpenTelemetry.get(), INSTRUMENTATION_NAME)
+ .setErrorCauseExtractor(SpringKafkaErrorCauseExtractor.INSTANCE)
+ .createConsumerProcessInstrumenter();
private static Instrumenter<ConsumerRecords<?, ?>, Void> buildBatchProcessInstrumenter() {
KafkaBatchProcessAttributesGetter getter = KafkaBatchProcessAttributesGetter.INSTANCE;

Refactor KafkaInstrumenterFactory a bit (#5911)