Lightstep Tracing on Message Processing

pom.xml

@@ -129,6 +129,34 @@
 			<groupId>org.springframework.data</groupId>
 			<artifactId>spring-data-cassandra</artifactId>
 		</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/orchestrator/Application/AppConfigOrchestrator.java

@@ -6,7 +6,9 @@
 import com.uci.utils.kafka.ReactiveProducer;
 import io.fusionauth.client.FusionAuthClient;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.kie.api.io.Resource;
 import org.kie.api.runtime.KieSession;
 import org.kie.internal.io.ResourceFactory;
@@ -100,7 +102,7 @@ Map<String, Object> kafkaProducerConfiguration() {
     ReceiverOptions<String, String> kafkaReceiverOptions(@Value("${inboundProcessed}") String[] inTopicName) {
         ReceiverOptions<String, String> options = ReceiverOptions.create(kafkaConsumerConfiguration());
         return options.subscription(Arrays.asList(inTopicName))
-                .withKeyDeserializer(new JsonDeserializer<>())
+                .withKeyDeserializer(new StringDeserializer())
                 .withValueDeserializer(new JsonDeserializer());
     }
 
@@ -110,8 +112,8 @@ SenderOptions<Integer, String> kafkaSenderOptions() {
     }
 
     @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();
     }
 
     @Bean

src/main/resources/application.properties

@@ -52,3 +52,10 @@ spring.r2dbc.password=${FORMS_DB_PASSWORD}
 caffeine.cache.max.size=0
 caffeine.cache.exprie.duration.seconds=${CAFFEINE_CACHE_EXPIRE_DURATION:#{300}}
 
+#Opentelemetry Lighstep Config
+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}
+

src/main/resources/log4j2.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Configuration packages="com.uci.orchestrator" status="info" name="orchestrator-log">
+	<Appenders>
+		<Console name="info" target="SYSTEM_OUT">
+			<PatternLayout
+				pattern="%d{HH:mm:ss.SSS} info %highlight(%-5p) [%-7t] %F:%L - %m%n" />
+		</Console>
+	</Appenders>
+	<Loggers>
+		<Root level="INFO">
+			<AppenderRef ref="info" />
+		</Root>
+	</Loggers>
+</Configuration>