Lightstep Tracing on Message Processing

pom.xml

@@ -111,7 +111,34 @@
             <version>1.0</version>
             <scope>compile</scope>
         </dependency>
-
+		<!-- Start - Dependencies for Opentelemetry lightstep -->
+		<!-- https://mvnrepository.com/artifact/com.google.protobuf/protobuf-java -->
+		<dependency>
+			<groupId>com.google.protobuf</groupId>
+			<artifactId>protobuf-java</artifactId>
+			<version>3.19.1</version>
+		</dependency>
+		<dependency>
+			<groupId>com.google.guava</groupId>
+			<artifactId>guava</artifactId>
+			<version>31.0.1-jre</version>
+		</dependency>
+		<dependency>
+			<groupId>com.lightstep.opentelemetry</groupId>
+			<artifactId>opentelemetry-launcher</artifactId>
+			<version>1.5.0</version>
+		</dependency>
+		<dependency>
+			<groupId>io.opentelemetry</groupId>
+			<artifactId>opentelemetry-api</artifactId>
+			<version>1.7.1</version>
+		</dependency>
+		<dependency>
+			<groupId>io.opentelemetry</groupId>
+			<artifactId>opentelemetry-extension-annotations</artifactId>
+			<version>1.7.1</version>
+		</dependency>
+		<!-- End - Dependencies for Opentelemetry lightstep -->
     </dependencies>
 
 	<build>

src/main/java/com/uci/outbound/Application/ReactiveKafkaConfiguration.java

@@ -1,6 +1,7 @@
 package com.uci.outbound.Application;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
@@ -38,12 +39,12 @@ Map<String, Object> kafkaConsumerConfiguration() {
     ReceiverOptions<String, String> kafkaReceiverOptions(@Value("${outbound}") String[] inTopicName) {
         ReceiverOptions<String, String> options = ReceiverOptions.create(kafkaConsumerConfiguration());
         return options.subscription(Arrays.asList(inTopicName))
-                .withKeyDeserializer(new JsonDeserializer<>())
+                .withKeyDeserializer(new StringDeserializer())
                 .withValueDeserializer(new JsonDeserializer());
     }
 
     @Bean
-    Flux<ReceiverRecord<String, String>> reactiveKafkaReceiver(ReceiverOptions<String, String> kafkaReceiverOptions) {
-        return KafkaReceiver.create(kafkaReceiverOptions).receive();
+    Flux<ConsumerRecord<String, String>> reactiveKafkaReceiver(ReceiverOptions<String, String> kafkaReceiverOptions) {
+        return KafkaReceiver.create(kafkaReceiverOptions).receiveAtmostOnce();
     }
 }

src/main/java/com/uci/outbound/consumers/OutboundKafkaController.java

@@ -8,7 +8,17 @@
 import lombok.extern.slf4j.Slf4j;
 import messagerosa.core.model.XMessage;
 import com.uci.dao.utils.XMessageDAOUtils;
+import com.uci.utils.kafka.adapter.TextMapGetterAdapter;
+import com.uci.utils.kafka.adapter.TextMapSetterAdapter;
+
+import io.grpc.Context;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
 import messagerosa.xml.XMessageParser;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.event.ApplicationStartedEvent;
 import org.springframework.context.event.EventListener;
@@ -24,23 +34,30 @@
 @Slf4j
 public class OutboundKafkaController {
 
-    private final Flux<ReceiverRecord<String, String>> reactiveKafkaReceiver;
+    private final Flux<ConsumerRecord<String, String>> reactiveKafkaReceiver;
 
     @Autowired
     private ProviderFactory factoryProvider;
 
     @Autowired
     private XMessageRepository xMessageRepo;
+
+    @Autowired
+	private Tracer tracer;
 
     @EventListener(ApplicationStartedEvent.class)
     public void onMessage() {
         reactiveKafkaReceiver
-                .doOnNext(new Consumer<ReceiverRecord<String, String>>() {
+                .doOnNext(new Consumer<ConsumerRecord<String, String>>() {
                     @Override
-                    public void accept(ReceiverRecord<String, String> msg) {
-                    	log.info("kafka message receieved");
-                        XMessage currentXmsg = null;
-                        try {
+                    public void accept(ConsumerRecord<String, String> msg) {
+                    	log.info("kafka message receieved");                    	
+                    	XMessage currentXmsg = null;
+//                    	Context extractedContext = GlobalOpenTelemetry.getPropagators().getTextMapPropagator().extract(Context.current(), msg.headers(), TextMapGetterAdapter.getter);
+//                        log.info("Opentelemetry extracted context : "+extractedContext);
+
+//                		try (Scope scope = extractedContext.makeCurrent()) {
+                    	try {
                             currentXmsg = XMessageParser.parse(new ByteArrayInputStream(msg.value().getBytes()));
                             String channel = currentXmsg.getChannelURI();
                             String provider = currentXmsg.getProviderURI();
@@ -74,4 +91,4 @@ public void accept(Throwable e) {
                 })
                 .subscribe();
     }
-}
+}

src/main/resources/application.properties

@@ -54,11 +54,11 @@ caffeine.cache.max.size=0
 caffeine.cache.exprie.duration.seconds=${CAFFEINE_CACHE_EXPIRE_DURATION:#{300}}
 
 #Opentelemetry Lighstep Config
-opentelemetry.lightstep.tracer=${OPENTELEMETERY_LIGHTSTEP_TRACER}
-opentelemetry.lightstep.tracer.version=${OPENTELEMETERY_LIGHTSTEP_TRACER_VERSION}
-opentelemetry.lightstep.service=${OPENTELEMETERY_LIGHTSTEP_SERVICE}
-opentelemetry.lightstep.access.token=${OPENTELEMETERY_LIGHTSTEP_ACCESS_TOKEN}
-opentelemetry.lightstep.end.point=${OPENTELEMETERY_LIGHTSTEP_END_POINT}
+opentelemetry.lightstep.tracer=${LS_TRACER_NAME}
+opentelemetry.lightstep.tracer.version=${LS_TRACER_VERSION}
+opentelemetry.lightstep.service=${LS_SERVICE_NAME}
+opentelemetry.lightstep.access.token=${LS_ACCESS_TOKEN}
+opentelemetry.lightstep.end.point=${LS_END_POINT}
 
 #Sunbird Adapater Outbound URL
 adapter.sunbird.transport.url=${TRANSPORT_SOCKET_BASE_URL}