Migrate Asn1DecodedDataRouter to use Spring Kafka #131

mcook42 · 2024-12-11T22:41:33Z

PR Details

Description

Sealed MessageSerializer and MessageDeserializer classes in preparation for marking them final after the full Spring Kafka migration is complete
Removed now-obsolete filters from KafkaConsumerConfig. All Asn1DecodedData is routed through one listener again, so the filter isn't needed anymore
Replaced the Asn1DecodedDataRouter with the Spring Kafka implementation (previously named Asn1DecodedDataListener)
Refactored all Kafka-related tests to use a more standard, less bug-prone approach to topic, consumer, and producer creation
(Optional) Deleted and deprecated unnecessary MessageSerializer and MessageDeserializer generic class extensions.

Related Issue

Motivation and Context

Implementing Spring Kafka gives us better lifecycle management of producers and consumers, more reusable producer/consumer code, easier testability, and a more robust production-ready Kafka library. This is part of a more significant effort to replace our hand-rolled Kafka implementation with Spring Kafka. The previous changesets related to this effort are #118, #116, #123, and #129

How Has This Been Tested?

Unit and Integration tests were added before making any functional changes. These continue to pass after the code changes are made. Some additional unit and integration tests were added after the functionality was changed to increase coverage. I have also run all data from the udpsender_[msgType].py scripts found under scripts/tests through a live local system started up with make rebuild. I confirmed there were no errors in the logs. I also confirmed that all expected messages were produced to and consumed from the correct queues by using the kafka-ui container available at localhost:8001 (on my local machine, of course).

Types of changes

Defect fix (non-breaking change that fixes an issue)
New feature (non-breaking change that adds functionality)
Breaking change (fix or feature that cause existing functionality to change)

Checklist:

I have added any new packages to the sonar-scanner.properties file
My change requires a change to the documentation.
I have updated the documentation accordingly.
I have read the CONTRIBUTING document.
ODE Contributing Guide
I have added tests to cover my changes.
All new and existing tests passed.

…merConfig

…cerConfig.java

… payload string in UdpHexDecoder.java

…r something to confirm testing

…removing from both, and comparing the expected and actual JSON

…/spring-kafka/udp-receivers

… cost of integration test

…NTest

…outerApprovalTest

…cted.json to use expected encoded value

…Test_ValidSSM.txt and the resulting SsmReceiverTest_ValidSSM_expected.json

…erTest_ValidSPAT.txt and the resulting SpatReceiverTest_ValidSPAT_expected.json

…Test_ValidPSM.txt and the resulting PsmReceiverTest_ValidPSM_expected.json

…Test_ValidSRM.txt and the resulting SrmReceiverTest_ValidSRM_expected.json

…t runs

…risk of race conditions

…etup

Since this test suite produces to and consumes from the same topics in multiple tests, we needed a better way than `getSingleRecord` to perform test assertions. By using uuids as keys, we are able to reuse the same topics for multiple tests and select the correct records for test assertions without risk of tests conflicting with one another.

This change helps simplify managing unique topic names for the disabledTopics set provided by OdeKafkaProperties. It is necessary to prevent data from other tests leaking onto the disabled topics and causing intermittent test failures in this suite

…NTest

…igTest

mcook42 · 2024-12-12T21:59:43Z

.github/workflows/ci.yml

@@ -27,4 +27,4 @@ jobs:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
        run: |
          ls -la && pwd
-          mvn -e -X clean org.jacoco:jacoco-maven-plugin:prepare-agent package sonar:sonar -Dsonar.projectKey=usdot.jpo.ode:jpo-ode -Dsonar.projectName=jpo-ode -Dsonar.organization=usdot-jpo-ode -Dsonar.host.url=https://sonarcloud.io -Dsonar.branch.name=$GITHUB_REF_NAME


Note: -X enables debug logging. It generated so much text that GH's runners were struggling to render test failures in the browser. Turning off debug logging doesn't mean we'll lose the information needed to determine the cause of test failures. It just makes finding failures in the logs easier

Yeah I like this

mcook42 · 2024-12-12T22:05:04Z

.github/workflows/ci.yml

+on:
+  pull_request:
+  push:
+    branches:
+      - "develop"
+      - "master"
+      - "release/*"


Note: this change will stop us from running the ci workflow every time a push to any branch happens. This will restrict the ci workflow to only the push to our main branches and any time a PR is opened or updated. It should reduce unnecessary costs associated with running the workflow

mcook42 · 2024-12-12T22:08:21Z

jpo-ode-core/src/main/java/us/dot/its/jpo/ode/wrapper/serdes/MessagingDeserializer.java

 import us.dot.its.jpo.ode.util.SerializationUtils;

-public class MessagingDeserializer<T> implements Deserializer<T> {
-
-    SerializationUtils<T> deserializer = new SerializationUtils<T>();


note: default methods are provided for both configure and close, so overriding them is unnecessary

mcook42 · 2024-12-12T22:08:44Z

jpo-ode-core/src/main/java/us/dot/its/jpo/ode/wrapper/serdes/MessagingSerializer.java

-
-    SerializationUtils<T> serializer = new SerializationUtils<T>();
-
-    @Override


note: default methods are provided for both configure and close, so overriding them is unnecessary

mcook42 · 2024-12-12T22:13:08Z

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

note: much of this code comes from the original implementations at Asn1DecodedDataListener and services/Asn1DecodedDataRouter

drewjj

Looks great! All the tests look good and pass. Each message type's decoded sample data is accurate for the input/output. I just have one minor comment about a debug log.

However.. decoded BSM messages don't seem to show up in the topic.OdeBsmJson or topic.OdeBsmPojo topics. The tests evidently pass for the BSM decode routing but I wonder what the difference is between the test and a deployed environment.

drewjj · 2024-12-12T22:59:11Z

.github/workflows/ci.yml

@@ -27,4 +27,4 @@ jobs:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
        run: |
          ls -la && pwd
-          mvn -e -X clean org.jacoco:jacoco-maven-plugin:prepare-agent package sonar:sonar -Dsonar.projectKey=usdot.jpo.ode:jpo-ode -Dsonar.projectName=jpo-ode -Dsonar.organization=usdot-jpo-ode -Dsonar.host.url=https://sonarcloud.io -Dsonar.branch.name=$GITHUB_REF_NAME


Yeah I like this

drewjj · 2024-12-12T23:12:08Z

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

+      id = "Asn1DecodedDataRouter",
+      topics = "${ode.kafka.topics.asn1.decoder-output}"
+  )
+  public void listen(ConsumerRecord<String, String> consumerRecord) throws XmlUtilsException {


I really like this now. Very clean!

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

…oggingProducerListener.java Co-authored-by: Drew Johnston <[email protected]>

dmccoystephenson · 2024-12-12T23:42:36Z

jpo-ode-core/src/main/java/us/dot/its/jpo/ode/wrapper/serdes/MessagingDeserializer.java

+/**
+ * MessagingDeserializer is a generic base class implementing the Kafka Deserializer interface to
+ * provide serialization of objects for use in Kafka messages.
+ *
+ * <p>This class uses a generic type parameter, allowing it to handle serialization
+ * of various types. Internal serialization is performed using an instance of the SerializationUtils
+ * class, which leverages Kryo for efficient object serialization.</p>
+ *
+ * <p>The class is declared as sealed, restricting which other classes can directly extend it. It
+ * will
+ * soon be marked as final to prevent incorrect usage through unnecessary subtyping</p>
+ *
+ * @param <T> the type of data to be serialized
+ */


(blocking) These comments indicate that the class is for serialization, but the class is for deserialization.

corrected in d7f2bbe

dmccoystephenson · 2024-12-12T23:47:04Z

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

+ * data by 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 uses transforms the payload and then determines the appropriate


(blocking) Looks like a typo here: 'uses transforms' should probably just be 'transforms'

corrected in d7f2bbe

dmccoystephenson · 2024-12-12T23:51:36Z

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

+   *
+   * @param kafkaTemplate the KafkaTemplate used for sending messages to Kafka topics.
+   */
+  public Asn1DecodedDataRouter(KafkaTemplate<String, String> kafkaTemplate,


(non-blocking) Is the @Autowired annotation necessary here, or does the class @Component annotation allow Spring to wire everything up without it?

mcook42 · 2024-12-13T00:13:05Z

marking as draft until I can figure out a nice way to handle the disabled topic exceptions blocking the rest of the code execution

…t program flow

…eptingKafkaTemplate This allows us to prevent sending to disabled topics without throwing exceptions. It allows graceful blocking of sends to disabled topics where the previous exception-driven approach was interrupting normal code paths.

…thub.com/CDOT-CV/jpo-ode into mcook42/spring-kafka/asn1-decoded-router

mcook42 · 2024-12-13T01:07:00Z

@drewjj I'm able to confirm that the updates in da34d26 allow us to publish to the topic.OdeBsmJson after blocking the attempt to send to topicOdeBsmTxPojo when running locally.

mcook42 added 30 commits November 26, 2024 12:40

extract method for generating kafka consumer properties in KafkaConsu…

d0a40eb

…merConfig

extract method for generating kafka producer properties in KafkaProdu…

ac70ab4

…cerConfig.java

hex encode payload start flag to correctly find start flag in encoded…

83d58e5

… payload string in UdpHexDecoder.java

hex encode payload start flag to correctly find start flag in encoded…

9d3b8e9

… payload string in UdpHexDecoder.java

get a working bsm through the Bsm reciever. needs stream-id removal o…

b151bdf

…r something to confirm testing

complete test for BsmReceiverTest by testing serial-id is not equal, …

d3fb75b

…removing from both, and comparing the expected and actual JSON

swap Spring Kafka in for BsmReceiver and BsmReceiverTest

29dcbdd

remove unused OdeKafkaProperties bean from BsmReceiverTest

30d50c9

Merge branch 'dev' of https://github.com/CDOT-CV/jpo-ode into mcook42…

874bac7

…/spring-kafka/udp-receivers

reformat UperUtil with checkstyle.xml

b315b13

use hex encoded versions of the start flag in UdpHexDecoder.java

3fb27c4

protect against missing start flag in stripDot3Header

20544ca

use SpringBootTest slices in MapReceiverTest to reduce setup/teardown…

1cfb5fd

… cost of integration test

reformat UdpHexDecoder.java

7519ada

remove dead comment from MapReceiverTest after introduction of slices

841f386

use SpringBootTest slices for smaller footprint with Asn1DecodeMAPJSO…

c754a3b

…NTest

use SpringBootTest slices for smaller footprint with Asn1DecodedDataR…

bffd8ea

…outerApprovalTest

update BsmReceiverTest_ValidBSM.txt and BsmReceiverTest_ValidBSM_expe…

37457d1

…cted.json to use expected encoded value

add TimReceiverTest with inputs and expected value files

6eaad52

swap KafkaTemplate<String, String> in for TimReceiver

c4d291e

reformat TimReceiver

038e24e

reformat UdpHexDecoder.java

4a6132d

use value from scripts/tests/udpsender_ssm.py to populate SsmReceiver…

643019c

…Test_ValidSSM.txt and the resulting SsmReceiverTest_ValidSSM_expected.json

swap KafkaTemplate<String, String> in for SsmReceiver

04e4de3

use value from scripts/tests/udpsender_spat.py to populate SpatReceiv…

6ab6fa0

…erTest_ValidSPAT.txt and the resulting SpatReceiverTest_ValidSPAT_expected.json

swap KafkaTemplate<String, String> in for SpatReceiver

179f569

use value from scripts/tests/udpsender_psm.py to populate PsmReceiver…

69d4a4b

…Test_ValidPSM.txt and the resulting PsmReceiverTest_ValidPSM_expected.json

swap KafkaTemplate<String, String> in for PsmReceiver

7fc441c

use value from scripts/tests/udpsender_srm.py to populate SrmReceiver…

db3eb20

…Test_ValidSRM.txt and the resulting SrmReceiverTest_ValidSRM_expected.json

swap KafkaTemplate<String, String> in for SrmReceiver

f7f1e46

mcook42 added 11 commits December 12, 2024 09:41

test: standardized Asn1DecodeMAPJSONTest to reduce variability in tes…

565fd12

…t runs

test: annotate KafkaProducerConfigTest with DirtiesContext to reduce …

e4244ea

…risk of race conditions

test: standardize ReceiverTest annotations and setup

edda6a5

test: standardize Asn1DecodedDataRouterApprovalTest annotations and s…

bb5e256

…etup

ci: remove -X flag from mvn command to reduce noise in output

d3807e3

test: use correct deserializers for test consumer in Asn1DecodeMAPJSO…

4887d33

…NTest

chore: add debug logging to addTopics in EmbeddedKafkaHolder

4ae479b

test: enable simple config of unique topic names in KafkaProducerConf…

885e477

…igTest

ci: isolate ci.yml runs to pull_request and specific push events

66c71c3

mcook42 commented Dec 12, 2024

View reviewed changes

mcook42 marked this pull request as ready for review December 12, 2024 22:24

mcook42 requested review from payneBrandon, dmccoystephenson, drewjj, Michael7371 and mwodahl December 12, 2024 22:25

drewjj requested changes Dec 12, 2024

View reviewed changes

Update jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/kafka/producer/L…

71b6b12

…oggingProducerListener.java Co-authored-by: Drew Johnston <[email protected]>

dmccoystephenson reviewed Dec 12, 2024

View reviewed changes

mcook42 marked this pull request as draft December 13, 2024 00:12

mcook42 added 3 commits December 12, 2024 17:18

test: add failing test to ensure disabledTopic send does not interrup…

c32ac70

…t program flow

Merge branch 'mcook42/spring-kafka/asn1-decoded-router' of https://gi…

855d628

…thub.com/CDOT-CV/jpo-ode into mcook42/spring-kafka/asn1-decoded-router

docs: correct wording in Javadocs

d7f2bbe

mcook42 requested review from dmccoystephenson and drewjj December 13, 2024 01:11

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Migrate Asn1DecodedDataRouter to use Spring Kafka #131

Migrate Asn1DecodedDataRouter to use Spring Kafka #131

mcook42 commented Dec 11, 2024 •

edited

Loading

mcook42 Dec 12, 2024

drewjj Dec 12, 2024

mcook42 Dec 12, 2024

mcook42 Dec 12, 2024

mcook42 Dec 12, 2024

mcook42 Dec 12, 2024

drewjj left a comment

drewjj Dec 12, 2024

drewjj Dec 12, 2024

dmccoystephenson Dec 12, 2024

mcook42 Dec 13, 2024

dmccoystephenson Dec 12, 2024

mcook42 Dec 13, 2024

dmccoystephenson Dec 12, 2024

mcook42 commented Dec 13, 2024

mcook42 commented Dec 13, 2024


		SerializationUtils<T> serializer = new SerializationUtils<T>();

		@Override

Migrate Asn1DecodedDataRouter to use Spring Kafka #131

Are you sure you want to change the base?

Migrate Asn1DecodedDataRouter to use Spring Kafka #131

Conversation

mcook42 commented Dec 11, 2024 • edited Loading

PR Details

Description

Related Issue

Motivation and Context

How Has This Been Tested?

Types of changes

Checklist:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

drewjj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mcook42 commented Dec 13, 2024

mcook42 commented Dec 13, 2024

mcook42 commented Dec 11, 2024 •

edited

Loading