Consumer tracer to replace the use of Vert.x Kafka lib in base dispat…

…cher module. (#3269) * New Consumer Tracer Signed-off-by: debasishbsws <debasishbsws.abc@gmail.com> Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com> * Consumer tracer Impl Signed-off-by: debasishbsws <debasishbsws.abc@gmail.com> Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com> * Formating via Spotless. Signed-off-by: debasishbsws <debasishbsws.abc@gmail.com> Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com> * making the promises final in loom Producer Signed-off-by: debasishbsws <debasishbsws.abc@gmail.com> Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com> * replace with knative licence header Signed-off-by: debasishbsws <debasishbsws.abc@gmail.com> Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com> * Remove unwanted formatting Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com> * Added specific links Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com> * Update codegen Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com> --------- Signed-off-by: debasishbsws <debasishbsws.abc@gmail.com> Signed-off-by: debasishbsws <debasishbsws.dev@gmail.com>
knative-extensions · Aug 21, 2023 · 21c92e7 · 21c92e7
1 parent c402ae9
commit 21c92e7
Show file tree

Hide file tree

Showing 7 changed files with 128 additions and 14 deletions.
diff --git a/...re/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/kafka/ConsumerTracer.java b/...re/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/kafka/ConsumerTracer.java
@@ -0,0 +1,115 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * This piece of code is inspired from vert-x3/vertx-kafka-client project.
+ * The original source code can be found here:
+ * https://github.com/vert-x3/vertx-kafka-client/blob/a0e349fca33d3bb4f003ac53e6e0def42a76e8ab/src/main/java/io/vertx/kafka/client/common/tracing/ConsumerTracer.java
+ */
+package dev.knative.eventing.kafka.broker.core.tracing.kafka;
+
+import io.vertx.core.Context;
+import io.vertx.core.spi.tracing.SpanKind;
+import io.vertx.core.spi.tracing.TagExtractor;
+import io.vertx.core.spi.tracing.VertxTracer;
+import io.vertx.core.tracing.TracingPolicy;
+import java.util.AbstractMap;
+import java.util.Collections;
+import java.util.Map;
+import java.util.stream.StreamSupport;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.utils.Utils;
+
+/**
+ * Tracer for Kafka consumer, wrapping the generic tracer.
+ */
+public class ConsumerTracer<S> {
+    private final VertxTracer<S, Void> tracer;
+    private final String address;
+    private final String hostname;
+    private final String port;
+    private final TracingPolicy policy;
+
+    /**
+     * Creates a ConsumerTracer, which provides an opinionated facade for using {@link io.vertx.core.spi.tracing.VertxTracer}
+     * with a Kafka Consumer use case.
+     * The method will return {@code null} if Tracing is not setup in Vert.x.
+     * {@code TracingPolicy} is always set to {@code TracingPolicy.ALWAYS}.
+     * @param tracer the generic tracer object
+     * @param config Kafka client configuration
+     * @param <S> the type of spans that is going to be generated, depending on the tracing system (zipkin, opentracing ...)
+     * @return a new instance of {@code ConsumerTracer}, or {@code null}
+     */
+    public static <S> ConsumerTracer create(VertxTracer tracer, Map<String, Object> config, TracingPolicy policy) {
+        if (tracer == null) {
+            return null;
+        }
+        policy = policy == null ? TracingPolicy.ALWAYS : policy;
+        String address =
+                config.getOrDefault(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "").toString();
+        return new ConsumerTracer<S>(tracer, policy, address);
+    }
+
+    private ConsumerTracer(VertxTracer<S, Void> tracer, TracingPolicy policy, String bootstrapServer) {
+        this.tracer = tracer;
+        this.address = bootstrapServer;
+        this.hostname = Utils.getHost(bootstrapServer);
+        Integer port = Utils.getPort(bootstrapServer);
+        this.port = port == null ? null : port.toString();
+        this.policy = policy;
+    }
+
+    private static Iterable<Map.Entry<String, String>> convertHeaders(Headers headers) {
+        if (headers == null) {
+            return Collections.emptyList();
+        }
+        return () -> StreamSupport.stream(headers.spliterator(), false)
+                .map(h -> (Map.Entry<String, String>) new AbstractMap.SimpleEntry<>(h.key(), new String(h.value())))
+                .iterator();
+    }
+
+    public StartedSpan prepareMessageReceived(Context context, ConsumerRecord rec) {
+        TraceContext tc = new TraceContext("consumer", address, hostname, port, rec.topic());
+        S span = tracer.receiveRequest(
+                context,
+                SpanKind.MESSAGING,
+                policy,
+                tc,
+                "kafka_receive",
+                convertHeaders(rec.headers()),
+                TraceTags.TAG_EXTRACTOR);
+        return new StartedSpan(span);
+    }
+
+    public class StartedSpan {
+        private final S span;
+
+        private StartedSpan(S span) {
+            this.span = span;
+        }
+
+        public void finish(Context context) {
+            // We don't add any new tag to the span here, just stop span timer
+            tracer.sendResponse(context, null, span, null, TagExtractor.empty());
+        }
+
+        public void fail(Context context, Throwable failure) {
+            tracer.sendResponse(context, null, span, failure, TagExtractor.empty());
+        }
+    }
+}
diff --git a/...re/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/kafka/ProducerTracer.java b/...re/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/kafka/ProducerTracer.java
@@ -16,7 +16,8 @@
 
 /**
  * This piece of code is inspired from vert-x3/vertx-kafka-client project.
- * The original source code can be found here: https://github.com/vert-x3/vertx-kafka-client
+ * The original source code can be found here:
+ * https://github.com/vert-x3/vertx-kafka-client/blob/a0e349fca33d3bb4f003ac53e6e0def42a76e8ab/src/main/java/io/vertx/kafka/client/common/tracing/ProducerTracer.java
  */
 package dev.knative.eventing.kafka.broker.core.tracing.kafka;
 

diff --git a/...core/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/kafka/TraceContext.java b/...core/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/kafka/TraceContext.java
@@ -16,7 +16,8 @@
 
 /**
  * This piece of code is inspired from vert-x3/vertx-kafka-client project.
- * The original source code can be found here: https://github.com/vert-x3/vertx-kafka-client
+ * The original source code can be found here:
+ * https://github.com/vert-x3/vertx-kafka-client/blob/a0e349fca33d3bb4f003ac53e6e0def42a76e8ab/src/main/java/io/vertx/kafka/client/common/tracing/TraceContext.java
  */
 package dev.knative.eventing.kafka.broker.core.tracing.kafka;
 

diff --git a/...ne/core/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/kafka/TraceTags.java b/...ne/core/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/kafka/TraceTags.java
@@ -16,7 +16,8 @@
 
 /**
  * This piece of code is inspired from vert-x3/vertx-kafka-client project.
- * The original source code can be found here: https://github.com/vert-x3/vertx-kafka-client
+ * The original source code can be found here:
+ * https://github.com/vert-x3/vertx-kafka-client/blob/a0e349fca33d3bb4f003ac53e6e0def42a76e8ab/src/main/java/io/vertx/kafka/client/common/tracing/TraceTags.java
  */
 package dev.knative.eventing.kafka.broker.core.tracing.kafka;
 

diff --git a/...src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/...src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -19,6 +19,7 @@
 
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
+import dev.knative.eventing.kafka.broker.core.tracing.kafka.ConsumerTracer;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcher;
@@ -39,7 +40,6 @@
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.client.HttpResponse;
-import io.vertx.kafka.client.common.tracing.ConsumerTracer;
 import java.util.Base64;
 import java.util.HashMap;
 import java.util.Map;

diff --git a/.../main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleBuilder.java b/.../main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleBuilder.java
@@ -23,6 +23,7 @@
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.security.Credentials;
 import dev.knative.eventing.kafka.broker.core.security.KafkaClientsAuth;
+import dev.knative.eventing.kafka.broker.core.tracing.kafka.ConsumerTracer;
 import dev.knative.eventing.kafka.broker.dispatcher.CloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.DeliveryOrder;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
@@ -55,8 +56,6 @@
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.ext.web.client.WebClient;
 import io.vertx.ext.web.client.WebClientOptions;
-import io.vertx.kafka.client.common.KafkaClientOptions;
-import io.vertx.kafka.client.common.tracing.ConsumerTracer;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -117,11 +116,8 @@ private void build(final Vertx vertx, final ConsumerVerticle consumerVerticle, f
                         offsetManager,
                         ConsumerTracer.create(
                                 ((VertxInternal) vertx).tracer(),
-                                new KafkaClientOptions()
-                                        .setConfig(consumerVerticleContext.getConsumerConfigs())
-                                        // Make sure the policy is propagate for the manually instantiated consumer
-                                        // tracer
-                                        .setTracingPolicy(TracingPolicy.PROPAGATE)),
+                                consumerVerticleContext.getConsumerConfigs(),
+                                TracingPolicy.PROPAGATE),
                         Metrics.getRegistry()),
                 new CloudEventOverridesMutator(
                         consumerVerticleContext.getResource().getCloudEventOverrides()));

diff --git a/...-loom/src/main/java/dev/knative/eventing/kafka/broker/receiverloom/LoomKafkaProducer.java b/...-loom/src/main/java/dev/knative/eventing/kafka/broker/receiverloom/LoomKafkaProducer.java
@@ -61,7 +61,7 @@ public LoomKafkaProducer(Vertx v, Producer<K, V> producer) {
 
     @Override
     public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
-        Promise<RecordMetadata> promise = Promise.promise();
+        final Promise<RecordMetadata> promise = Promise.promise();
         if (isClosed.get()) {
             promise.fail("Producer is closed");
         } else {
@@ -110,7 +110,7 @@ private void sendFromQueue() {
 
     @Override
     public Future<Void> close() {
-        Promise<Void> promise = Promise.promise();
+        final Promise<Void> promise = Promise.promise();
         this.isClosed.set(true);
         Thread.ofVirtual().start(() -> {
             try {
@@ -130,7 +130,7 @@ public Future<Void> close() {
 
     @Override
     public Future<Void> flush() {
-        Promise<Void> promise = Promise.promise();
+        final Promise<Void> promise = Promise.promise();
         Thread.ofVirtual().start(() -> {
             try {
                 producer.flush();