Merge c1b9cfb into 9835fc3

opentracing-contrib · Aug 30, 2018 · 3fc9b6b · 3fc9b6b
2 parents 9835fc3 + c1b9cfb
commit 3fc9b6b
Show file tree

Hide file tree

Showing 6 changed files with 112 additions and 29 deletions.
diff --git a/README.md b/README.md
@@ -54,6 +54,8 @@ pom.xml
 // Instantiate tracer
 Tracer tracer = ...
 
+// Optionally register tracer with GlobalTracer
+GlobalTracer.register(tracer);
 ```
 
 ### Kafka Client

diff --git a/...tracing-kafka-client/src/main/java/io/opentracing/contrib/kafka/TracingKafkaConsumer.java b/...tracing-kafka-client/src/main/java/io/opentracing/contrib/kafka/TracingKafkaConsumer.java
@@ -15,6 +15,14 @@
 
 
 import io.opentracing.Tracer;
+import io.opentracing.util.GlobalTracer;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiFunction;
+import java.util.regex.Pattern;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -27,14 +35,6 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.function.BiFunction;
-import java.util.regex.Pattern;
-
 public class TracingKafkaConsumer<K, V> implements Consumer<K, V> {
 
   private final Tracer tracer;
@@ -48,15 +48,30 @@ public TracingKafkaConsumer(Consumer<K, V> consumer, Tracer tracer) {
     this.consumerSpanNameProvider = ClientSpanNameProvider.CONSUMER_OPERATION_NAME;
   }
 
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingKafkaConsumer(Consumer<K, V> consumer) {
+    this(consumer, GlobalTracer.get());
+  }
+
   public TracingKafkaConsumer(Consumer<K, V> consumer, Tracer tracer,
-                BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider) {
+      BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider) {
     this.consumer = consumer;
     this.tracer = tracer;
     this.consumerSpanNameProvider = (consumerSpanNameProvider == null)
         ? ClientSpanNameProvider.CONSUMER_OPERATION_NAME
         : consumerSpanNameProvider;
   }
 
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingKafkaConsumer(Consumer<K, V> consumer,
+      BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider) {
+    this(consumer, GlobalTracer.get(), consumerSpanNameProvider);
+  }
+
   @Override
   public Set<TopicPartition> assignment() {
     return consumer.assignment();
@@ -130,7 +145,7 @@ public void commitAsync(OffsetCommitCallback callback) {
 
   @Override
   public void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets,
-              OffsetCommitCallback callback) {
+      OffsetCommitCallback callback) {
     consumer.commitAsync(offsets, callback);
   }
 

diff --git a/...tracing-kafka-client/src/main/java/io/opentracing/contrib/kafka/TracingKafkaProducer.java b/...tracing-kafka-client/src/main/java/io/opentracing/contrib/kafka/TracingKafkaProducer.java
@@ -16,12 +16,12 @@
 
 import io.opentracing.Scope;
 import io.opentracing.Tracer;
+import io.opentracing.util.GlobalTracer;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
-
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.Producer;
@@ -49,12 +49,28 @@ public TracingKafkaProducer(Producer<K, V> producer, Tracer tracer) {
     this.producerSpanNameProvider = ClientSpanNameProvider.PRODUCER_OPERATION_NAME;
   }
 
-  public TracingKafkaProducer(Producer<K, V> producer, Tracer tracer, BiFunction<String, ProducerRecord, String> producerSpanNameProvider) {
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingKafkaProducer(Producer<K, V> producer) {
+    this(producer, GlobalTracer.get());
+  }
+
+  public TracingKafkaProducer(Producer<K, V> producer, Tracer tracer,
+      BiFunction<String, ProducerRecord, String> producerSpanNameProvider) {
     this.producer = producer;
     this.tracer = tracer;
     this.producerSpanNameProvider = (producerSpanNameProvider == null)
-      ? ClientSpanNameProvider.PRODUCER_OPERATION_NAME
-      : producerSpanNameProvider;
+        ? ClientSpanNameProvider.PRODUCER_OPERATION_NAME
+        : producerSpanNameProvider;
+  }
+
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingKafkaProducer(Producer<K, V> producer,
+      BiFunction<String, ProducerRecord, String> producerSpanNameProvider) {
+    this(producer, GlobalTracer.get(), producerSpanNameProvider);
   }
 
   @Override
@@ -100,7 +116,8 @@ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callbac
         record.headers());
     */
 
-    try (Scope scope = TracingKafkaUtils.buildAndInjectSpan(record, tracer, producerSpanNameProvider)) {
+    try (Scope scope = TracingKafkaUtils
+        .buildAndInjectSpan(record, tracer, producerSpanNameProvider)) {
       Callback wrappedCallback = new TracingCallback(callback, scope);
       return producer.send(record, wrappedCallback);
     }

diff --git a/...afka-spring/src/main/java/io/opentracing/contrib/kafka/spring/TracingConsumerFactory.java b/...afka-spring/src/main/java/io/opentracing/contrib/kafka/spring/TracingConsumerFactory.java
@@ -16,9 +16,9 @@
 import io.opentracing.Tracer;
 import io.opentracing.contrib.kafka.ClientSpanNameProvider;
 import io.opentracing.contrib.kafka.TracingKafkaConsumer;
+import io.opentracing.util.GlobalTracer;
 import java.util.Map;
 import java.util.function.BiFunction;
-
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.serialization.Deserializer;
@@ -36,23 +36,40 @@ public TracingConsumerFactory(ConsumerFactory<K, V> consumerFactory, Tracer trac
     this.consumerSpanNameProvider = ClientSpanNameProvider.CONSUMER_OPERATION_NAME;
   }
 
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingConsumerFactory(ConsumerFactory<K, V> consumerFactory) {
+    this(consumerFactory, GlobalTracer.get());
+  }
+
   public TracingConsumerFactory(ConsumerFactory<K, V> consumerFactory, Tracer tracer,
-                                BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider) {
+      BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider) {
     this.tracer = tracer;
     this.consumerFactory = consumerFactory;
     this.consumerSpanNameProvider = (consumerSpanNameProvider == null)
-      ? ClientSpanNameProvider.CONSUMER_OPERATION_NAME
-      : consumerSpanNameProvider;
+        ? ClientSpanNameProvider.CONSUMER_OPERATION_NAME
+        : consumerSpanNameProvider;
+  }
+
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingConsumerFactory(ConsumerFactory<K, V> consumerFactory,
+      BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider) {
+    this(consumerFactory, GlobalTracer.get(), consumerSpanNameProvider);
   }
 
   @Override
   public Consumer<K, V> createConsumer() {
-    return new TracingKafkaConsumer<>(consumerFactory.createConsumer(), tracer, consumerSpanNameProvider);
+    return new TracingKafkaConsumer<>(consumerFactory.createConsumer(), tracer,
+        consumerSpanNameProvider);
   }
 
   @Override
   public Consumer<K, V> createConsumer(String clientIdSuffix) {
-    return new TracingKafkaConsumer<>(consumerFactory.createConsumer(clientIdSuffix), tracer, consumerSpanNameProvider);
+    return new TracingKafkaConsumer<>(consumerFactory.createConsumer(clientIdSuffix), tracer,
+        consumerSpanNameProvider);
   }
 
   @Override

diff --git a/...afka-spring/src/main/java/io/opentracing/contrib/kafka/spring/TracingProducerFactory.java b/...afka-spring/src/main/java/io/opentracing/contrib/kafka/spring/TracingProducerFactory.java
@@ -16,6 +16,7 @@
 import io.opentracing.Tracer;
 import io.opentracing.contrib.kafka.ClientSpanNameProvider;
 import io.opentracing.contrib.kafka.TracingKafkaProducer;
+import io.opentracing.util.GlobalTracer;
 import java.util.function.BiFunction;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -34,6 +35,13 @@ public TracingProducerFactory(ProducerFactory<K, V> producerFactory, Tracer trac
     this.producerSpanNameProvider = ClientSpanNameProvider.PRODUCER_OPERATION_NAME;
   }
 
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingProducerFactory(ProducerFactory<K, V> producerFactory) {
+    this(producerFactory, GlobalTracer.get());
+  }
+
   public TracingProducerFactory(ProducerFactory<K, V> producerFactory, Tracer tracer,
       BiFunction<String, ProducerRecord, String> producerSpanNameProvider) {
     this.producerFactory = producerFactory;
@@ -43,6 +51,14 @@ public TracingProducerFactory(ProducerFactory<K, V> producerFactory, Tracer trac
         : producerSpanNameProvider;
   }
 
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingProducerFactory(ProducerFactory<K, V> producerFactory,
+      BiFunction<String, ProducerRecord, String> producerSpanNameProvider) {
+    this(producerFactory, GlobalTracer.get(), producerSpanNameProvider);
+  }
+
   @Override
   public Producer<K, V> createProducer() {
     return new TracingKafkaProducer<>(producerFactory.createProducer(), tracer,

diff --git a/...treams/src/main/java/io/opentracing/contrib/kafka/streams/TracingKafkaClientSupplier.java b/...treams/src/main/java/io/opentracing/contrib/kafka/streams/TracingKafkaClientSupplier.java
@@ -17,9 +17,9 @@
 import io.opentracing.contrib.kafka.ClientSpanNameProvider;
 import io.opentracing.contrib.kafka.TracingKafkaConsumer;
 import io.opentracing.contrib.kafka.TracingKafkaProducer;
+import io.opentracing.util.GlobalTracer;
 import java.util.Map;
 import java.util.function.BiFunction;
-
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -43,16 +43,32 @@ public TracingKafkaClientSupplier(Tracer tracer) {
     this.producerSpanNameProvider = ClientSpanNameProvider.PRODUCER_OPERATION_NAME;
   }
 
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingKafkaClientSupplier() {
+    this(GlobalTracer.get());
+  }
+
   public TracingKafkaClientSupplier(Tracer tracer,
-                                    BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider,
-                                    BiFunction<String, ProducerRecord, String> producerSpanNameProvider) {
+      BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider,
+      BiFunction<String, ProducerRecord, String> producerSpanNameProvider) {
     this.tracer = tracer;
     this.consumerSpanNameProvider = (consumerSpanNameProvider == null)
-            ? ClientSpanNameProvider.CONSUMER_OPERATION_NAME
-            : consumerSpanNameProvider;
+        ? ClientSpanNameProvider.CONSUMER_OPERATION_NAME
+        : consumerSpanNameProvider;
     this.producerSpanNameProvider = (producerSpanNameProvider == null)
-            ? ClientSpanNameProvider.PRODUCER_OPERATION_NAME
-            : producerSpanNameProvider;
+        ? ClientSpanNameProvider.PRODUCER_OPERATION_NAME
+        : producerSpanNameProvider;
+  }
+
+  /**
+   * GlobalTracer is used to get tracer
+   */
+  public TracingKafkaClientSupplier(
+      BiFunction<String, ConsumerRecord, String> consumerSpanNameProvider,
+      BiFunction<String, ProducerRecord, String> producerSpanNameProvider) {
+    this(GlobalTracer.get(), consumerSpanNameProvider, producerSpanNameProvider);
   }
 
   // This method is required by Kafka Streams >=1.1, and optional for Kafka Streams <1.1
@@ -65,7 +81,7 @@ public AdminClient getAdminClient(final Map<String, Object> config) {
   public Producer<byte[], byte[]> getProducer(Map<String, Object> config) {
     return new TracingKafkaProducer<>(
         new KafkaProducer<>(config, new ByteArraySerializer(), new ByteArraySerializer()),
-            tracer, producerSpanNameProvider);
+        tracer, producerSpanNameProvider);
   }
 
   @Override