Migrate Asn1EncodedDataRouter to use Spring Kafka

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/OdeTimJsonTopology.java

@@ -15,24 +15,23 @@
 
 
 /**
- * The OdeTimJsonTopology class sets up and manages a Kafka Streams topology
- * for processing TIM (Traveler Information Message) JSON data from the
- * OdeTimJson Kafka topic.
- * This class creates a K-Table that houses TMC-generated TIMs which can be
- * queried by UUID.
+ * The OdeTimJsonTopology class sets up and manages a Kafka Streams topology for processing TIM
+ * (Traveler Information Message) JSON data from the OdeTimJson Kafka topic. This class creates a
+ * K-Table that houses TMC-generated TIMs which can be queried by UUID.
  **/
 @Slf4j
 public class OdeTimJsonTopology {
 
   private final KafkaStreams streams;
 
   /**
-   * Constructs an instance of OdeTimJsonTopology to set up and manage a Kafka Streams
-   * topology for processing TIM JSON data.
+   * Constructs an instance of OdeTimJsonTopology to set up and manage a Kafka Streams topology for
+   * processing TIM JSON data.
    *
-   * @param odeKafkaProps the properties containing Kafka configuration, including brokers
-   *                      and optional Confluent-specific configuration for authentication.
-   * @param topic the Kafka topic from which TIM JSON data is consumed to build the topology.
+   * @param odeKafkaProps the properties containing Kafka configuration, including brokers and
+   *                      optional Confluent-specific configuration for authentication.
+   * @param topic         the Kafka topic from which TIM JSON data is consumed to build the
+   *                      topology.
    */
   public OdeTimJsonTopology(OdeKafkaProperties odeKafkaProps, String topic) {
 
@@ -52,20 +51,15 @@ public OdeTimJsonTopology(OdeKafkaProperties odeKafkaProps, String topic) {
     streams.start();
   }
 
-  public void stop() {
-    log.info("Stopping Ode Tim Json Topology");
-    streams.close();
-  }
-
   public boolean isRunning() {
-    return streams.state().isRunningOrRebalancing();
+    return streams.state().equals(KafkaStreams.State.RUNNING);
   }
 
   /**
    * Builds a Kafka Streams topology for processing TIM JSON data.
    *
-   * @param topic the Kafka topic from which TIM JSON data is consumed and used
-   *              to build the topology.
+   * @param topic the Kafka topic from which TIM JSON data is consumed and used to build the
+   *              topology.
    * @return the constructed Kafka Streams topology.
    */
   public Topology buildTopology(String topic) {

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/http/WebClientConfig.java

@@ -0,0 +1,42 @@
+package us.dot.its.jpo.ode.http;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * This class provides a configuration for creating and managing
+ * a {@link RestTemplate} bean, which is used for making HTTP requests
+ * to external services.
+ *
+ * <p><strong>NOTE:</strong> As of 5.0 the {@link RestTemplate} class is in maintenance mode, with
+ * only minor requests for changes and bugs to be accepted going forward. Please,
+ * consider using the {@code org.springframework.web.reactive.client.WebClient}
+ * which has a more modern API and supports sync, async, and streaming scenarios.
+ * Whenever we the time or resources to update our Spring version,
+ * we should replace usages of RestTemplate with WebClient.</p>
+ */
+@Configuration
+public class WebClientConfig {
+
+  /**
+   * Creates and configures a {@link RestTemplate} bean with a custom
+   * {@link MappingJackson2HttpMessageConverter} to use the provided
+   * {@link ObjectMapper} for JSON serialization and deserialization.
+   *
+   * @param mapper the {@link ObjectMapper} to be used for configuring
+   *               JSON message conversion.
+   * @return a configured {@link RestTemplate} instance that includes
+   *         the custom JSON message converter.
+   */
+  @Bean
+  public RestTemplate restTemplate(ObjectMapper mapper) {
+    var template = new RestTemplate();
+    MappingJackson2HttpMessageConverter customConverter = new MappingJackson2HttpMessageConverter();
+    customConverter.setObjectMapper(mapper);
+    template.getMessageConverters().add(customConverter);
+    return template;
+  }
+}

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/kafka/KafkaStreamsConfig.java

@@ -0,0 +1,21 @@
+package us.dot.its.jpo.ode.kafka;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import us.dot.its.jpo.ode.OdeTimJsonTopology;
+
+/**
+ * KafkaStreamsConfig is a Spring configuration class that provides
+ * beans related to Kafka Streams topology setup.
+ */
+@Configuration
+public class KafkaStreamsConfig {
+
+  @Bean
+  public OdeTimJsonTopology odeTimJsonTopology(
+      @Value("${ode.kafka.topics.json.tim}") String timTopic,
+      OdeKafkaProperties odeKafkaProperties) {
+    return new OdeTimJsonTopology(odeKafkaProperties, timTopic);
+  }
+}

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/kafka/listeners/asn1/Asn1DecodedDataRouter.java

@@ -1,5 +1,7 @@
 package us.dot.its.jpo.ode.kafka.listeners.asn1;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import joptsimple.internal.Strings;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.json.JSONObject;
@@ -30,8 +32,8 @@
  * processing and forwarding it to different topics based on specific criteria.
  *
  * <p>This listener is specifically designed to handle decoded data produced by the asn1_codec.
- * Upon receiving a payload, it transforms the payload and then determines the appropriate
- * Kafka topic to forward the processed data.</p>
+ * Upon receiving a payload, it transforms the payload and then determines the appropriate Kafka
+ * topic to forward the processed data.</p>
  *
  * <p>The class utilizes Spring Kafka's annotation-driven listener configuration,
  * allowing it to automatically consume messages from a configured Kafka topic.</p>
@@ -68,7 +70,8 @@ public Asn1DecodedDataRouter(KafkaTemplate<String, String> kafkaTemplate,
       id = "Asn1DecodedDataRouter",
       topics = "${ode.kafka.topics.asn1.decoder-output}"
   )
-  public void listen(ConsumerRecord<String, String> consumerRecord) throws XmlUtilsException {
+  public void listen(ConsumerRecord<String, String> consumerRecord)
+      throws XmlUtilsException, JsonProcessingException {
     log.debug("Key: {} payload: {}", consumerRecord.key(), consumerRecord.value());
 
     JSONObject consumed = XmlUtils.toJSONObject(consumerRecord.value())
@@ -80,16 +83,23 @@ public void listen(ConsumerRecord<String, String> consumerRecord) throws XmlUtil
             .getInt("messageId")
     );
 
+    var metadataJson = XmlUtils.toJSONObject(consumerRecord.value())
+        .getJSONObject(OdeAsn1Data.class.getSimpleName())
+        .getJSONObject(AppContext.METADATA_STRING);
     OdeLogMetadata.RecordType recordType = OdeLogMetadata.RecordType
-        .valueOf(XmlUtils.toJSONObject(consumerRecord.value())
-            .getJSONObject(OdeAsn1Data.class.getSimpleName())
-            .getJSONObject(AppContext.METADATA_STRING)
-            .getString("recordType")
-        );
+        .valueOf(metadataJson.getString("recordType"));
+
+    String streamId;
+    if (Strings.isNullOrEmpty(consumerRecord.key())
+        || "null".equalsIgnoreCase(consumerRecord.key())) {
+      streamId = metadataJson.getJSONObject("serialId").getString("streamId");
+    } else {
+      streamId = consumerRecord.key();
+    }
 
     switch (messageId) {
       case BasicSafetyMessage -> routeBSM(consumerRecord, recordType);
-      case TravelerInformation -> routeTIM(consumerRecord, recordType);
+      case TravelerInformation -> routeTIM(consumerRecord, streamId, recordType);
       case SPATMessage -> routeSPAT(consumerRecord, recordType);
       case MAPMessage -> routeMAP(consumerRecord, recordType);
       case SSMMessage -> routeSSM(consumerRecord, recordType);
@@ -156,17 +166,18 @@ private void routeMAP(ConsumerRecord<String, String> consumerRecord, RecordType
     kafkaTemplate.send(jsonTopics.getMap(), odeMapData);
   }
 
-  private void routeTIM(ConsumerRecord<String, String> consumerRecord, RecordType recordType)
-      throws XmlUtilsException {
+  private void routeTIM(ConsumerRecord<String, String> consumerRecord,
+      String streamId,
+      RecordType type) throws XmlUtilsException {
     String odeTimData =
         OdeTimDataCreatorHelper.createOdeTimDataFromDecoded(consumerRecord.value()).toString();
-    switch (recordType) {
+    switch (type) {
       case dnMsg -> kafkaTemplate.send(jsonTopics.getDnMessage(), consumerRecord.key(), odeTimData);
       case rxMsg -> kafkaTemplate.send(jsonTopics.getRxTim(), consumerRecord.key(), odeTimData);
-      default -> log.trace("Consumed TIM data with record type: {}", recordType);
+      default -> log.trace("Consumed TIM data with record type: {}", type);
     }
     // Send all TIMs also to OdeTimJson
-    kafkaTemplate.send(jsonTopics.getTim(), consumerRecord.key(), odeTimData);
+    kafkaTemplate.send(jsonTopics.getTim(), streamId, odeTimData);
   }
 
   private void routeBSM(ConsumerRecord<String, String> consumerRecord, RecordType recordType)