Migrate Asn1EncodedDataRouter to use Spring Kafka #138

mcook42 · 2024-12-24T21:13:35Z

PR Details

Description

Migrated Asn1EncodedDataRouter to use Spring Kafka instead of homegrown Kafka implementation
Changed from swallowing all exceptions to explicitly handling the non-critical exceptions (where we can reasonably continue to process the data) and throwing specific exceptions where we cannot continue processing (where missing data would cause unexpected downstream issues)
Removed duplicated logging wherever EventLogger was used
Moved OdeTimJsonTopology to a Spring Configuration class to enable dependency injection for testing and Asn1DecodedDataRouter listener Bean initialization
Extracted interactions with the security services module into SecurityServicesClient and defined the request and response models based on the raw JSON manipulation done prior
Deleted Asn1CommandManager and moved logic to the Asn1EncodedDataRouter for clearer code and fewer classes to maintain
Moved class creation out of Asn1EncodedDataRouter into the relevant Beans to enable easier testing with Spring Dependency Injection
Replaced outdated (and buggy) usage of the Date class with usages of LocalDateTime (only in Asn1EncodedDataRouter) to allow for explicit usage of UTC timezone instead of relying on the system settings (they aren't consistent between local and CI, so it's reasonable to suspect there is inconsistency between hosting environments)

Related Issue

No related USDOT 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:

How Has This Been Tested?

Before making any code changes, I refactored the Asn1EncodedDataRouter class to check the output to the different topics is as expected.
Refactored Asn1EncodedDataRouter class to use Spring Kafka and confirmed the tests continued to behave as expected
I sent the 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).

For the live testing, I set the following environment variables on my system to enable signing and confirm all pieces played well together

DATA_SIGNING_ENABLED_RSU=true
DATA_SIGNING_ENABLED_SDW=true

Snippet from the logs during local testing:

2024-12-27 14:14:48 2024-12-27 21:14:48 [UdpReceiverManager] DEBUG UdpHexDecoder - Packet received from 172.18.0.1:56852 
2024-12-27 14:14:48 2024-12-27 21:14:48 [UdpReceiverManager] DEBUG UdpHexDecoder - Full TIM packet: 005f498718cca69ec1a04600000100105d9b46ec5be401003a0103810040038081d4001f80d07016da410000000000000bbc2b0f775d9b0309c271431fa166ee0a27fff93f136b8205a0a107fb2ef979f4c5bfaeec97e4ad70c2fb36cd9730becdb355cc2fd2a7556b160b98b46ab98ae62c185fa55efb468d5b4000000004e2863f42cddc144ff7980040401262cdd7b809c509f5c62cdd35519c507b9062cdcee129c505cf262cdca5ff9c50432c62cdc5d3d9c502e3e62cdc13e79c501e9262cdbca2d9c5013ee62cdb80359c500e6a62cdb36299c500bc862cdaec1d9c50093c62cdaa2109c5006ea1080203091a859eeebb36006001830001aad27f4ff7580001aad355e39b5880a30029d6585009ef808332d8d9f80c3855151b38c772f765007967ec1170bcb7937f5cb880a25a52863493bcb87570dbcb5abc6bfb2faec606cfa34eb95a24790b2017366d3aabe7729e00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 
2024-12-27 14:14:48 2024-12-27 21:14:48 [UdpReceiverManager] DEBUG UperUtil - Base payload: 001f80d07016da410000000000000bbc2b0f775d9b0309c271431fa166ee0a27fff93f136b8205a0a107fb2ef979f4c5bfaeec97e4ad70c2fb36cd9730becdb355cc2fd2a7556b160b98b46ab98ae62c185fa55efb468d5b4000000004e2863f42cddc144ff7980040401262cdd7b809c509f5c62cdd35519c507b9062cdcee129c505cf262cdca5ff9c50432c62cdc5d3d9c502e3e62cdc13e79c501e9262cdbca2d9c5013ee62cdb80359c500e6a62cdb36299c500bc862cdaec1d9c50093c62cdaa2109c5006ea1080203091a859eeebb36006001830001aad27f4ff7580001aad355e39b5880a30029d6585009ef808332d8d9f80c3855151b38c772f765007967ec1170bcb7937f5cb880a25a52863493bcb87570dbcb5abc6bfb2faec606cfa34eb95a24790b2017366d3aabe7729e00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 
2024-12-27 14:14:48 2024-12-27 21:14:48 [UdpReceiverManager] DEBUG UperUtil - Stripped payload: 001f80d07016da410000000000000bbc2b0f775d9b0309c271431fa166ee0a27fff93f136b8205a0a107fb2ef979f4c5bfaeec97e4ad70c2fb36cd9730becdb355cc2fd2a7556b160b98b46ab98ae62c185fa55efb468d5b4000000004e2863f42cddc144ff7980040401262cdd7b809c509f5c62cdd35519c507b9062cdcee129c505cf262cdca5ff9c50432c62cdc5d3d9c502e3e62cdc13e79c501e9262cdbca2d9c5013ee62cdb80359c500e6a62cdb36299c500bc862cdaec1d9c50093c62cdaa2109c5006ea1080203091a859eeebb36006001830001aad27f4ff7580001aad355e39b5880a30029d6585009ef808332d8d9f80c3855151b38c772f765007967ec1170bcb7937f5cb880a25a52863493bcb87570dbcb5abc6bfb2faec606cfa34eb95a24790b2017366d3aabe7729e00 
2024-12-27 14:14:48 2024-12-27 21:14:48 [UdpReceiverManager] DEBUG UdpHexDecoder - Stripped TIM packet: 03810040038081d4001f80d07016da410000000000000bbc2b0f775d9b0309c271431fa166ee0a27fff93f136b8205a0a107fb2ef979f4c5bfaeec97e4ad70c2fb36cd9730becdb355cc2fd2a7556b160b98b46ab98ae62c185fa55efb468d5b4000000004e2863f42cddc144ff7980040401262cdd7b809c509f5c62cdd35519c507b9062cdcee129c505cf262cdca5ff9c50432c62cdc5d3d9c502e3e62cdc13e79c501e9262cdbca2d9c5013ee62cdb80359c500e6a62cdb36299c500bc862cdaec1d9c50093c62cdaa2109c5006ea1080203091a859eeebb36006001830001aad27f4ff7580001aad355e39b5880a30029d6585009ef808332d8d9f80c3855151b38c772f765007967ec1170bcb7937f5cb880a25a52863493bcb87570dbcb5abc6bfb2faec606cfa34eb95a24790b2017366d3aabe7729e00 
2024-12-27 14:14:48 2024-12-27 21:14:48 [UdpReceiverManager] DEBUG TimReceiver - Waiting for UDP TIM packets... 
2024-12-27 14:14:48 2024-12-27 21:14:48 [kafka-producer-network-thread | producer-5] DEBUG LoggingProducerListener - Successfully produced to topic topic.OdeRawEncodedTIMJson with key 81787258-c75b-4cb5-9d8f-1cc6d998573c_1.0.0#0 and value {"metadata":{"recordType":"timMsg","securityResultCode":"success","payloadType":"us.dot.its.jpo.ode.model.OdeAsn1Payload","serialId":{"streamId":"81787258-c75b-4cb5-9d8f-1cc6d998573c","bundleSize":1,"bundleId":0,"recordId":0,"serialNumber":0},"odeReceivedAt":"2024-12-27T21:14:48.007Z","schemaVersion":7,"maxDurationTime":0,"recordGeneratedBy":"RSU","sanitized":false,"asn1":"03810040038081D4001F80D07016DA410000000000000BBC2B0F775D9B0309C271431FA166EE0A27FFF93F136B8205A0A107FB2EF979F4C5BFAEEC97E4AD70C2FB36CD9730BECDB355CC2FD2A7556B160B98B46AB98AE62C185FA55EFB468D5B4000000004E2863F42CDDC144FF7980040401262CDD7B809C509F5C62CDD35519C507B9062CDCEE129C505CF262CDCA5FF9C50432C62CDC5D3D9C502E3E62CDC13E79C501E9262CDBCA2D9C5013EE62CDB80359C500E6A62CDB36299C500BC862CDAEC1D9C50093C62CDAA2109C5006EA1080203091A859EEEBB36006001830001AAD27F4FF7580001AAD355E39B5880A30029D6585009EF808332D8D9F80C3855151B38C772F765007967EC1170BCB7937F5CB880A25A52863493BCB87570DBCB5ABC6BFB2FAEC606CFA34EB95A24790B2017366D3AABE7729E00","originIp":"172.18.0.1"},"payload":{"dataType":"us.dot.its.jpo.ode.model.OdeHexByteArray","data":{"bytes":"03810040038081D4001F80D07016DA410000000000000BBC2B0F775D9B0309C271431FA166EE0A27FFF93F136B8205A0A107FB2EF979F4C5BFAEEC97E4AD70C2FB36CD9730BECDB355CC2FD2A7556B160B98B46AB98AE62C185FA55EFB468D5B4000000004E2863F42CDDC144FF7980040401262CDD7B809C509F5C62CDD35519C507B9062CDCEE129C505CF262CDCA5FF9C50432C62CDC5D3D9C502E3E62CDC13E79C501E9262CDBCA2D9C5013EE62CDB80359C500E6A62CDB36299C500BC862CDAEC1D9C50093C62CDAA2109C5006EA1080203091A859EEEBB36006001830001AAD27F4FF7580001AAD355E39B5880A30029D6585009EF808332D8D9F80C3855151B38C772F765007967EC1170BCB7937F5CB880A25A52863493BCB87570DBCB5ABC6BFB2FAEC606CFA34EB95A24790B2017366D3AABE7729E00"}}}  
2024-12-27 14:14:48 2024-12-27 21:14:48 [kafka-producer-network-thread | producer-6] DEBUG LoggingProducerListener - Successfully produced to topic topic.Asn1DecoderInput with key 81787258-c75b-4cb5-9d8f-1cc6d998573c_1.0.0#0 and value {"metadata":{"recordType":"timMsg","securityResultCode":"success","encodings":[{"elementName":"unsecuredData","elementType":"MessageFrame","encodingRule":"UPER"}],"payloadType":"us.dot.its.jpo.ode.model.OdeAsn1Payload","serialId":{"streamId":"81787258-c75b-4cb5-9d8f-1cc6d998573c","bundleSize":1,"bundleId":0,"recordId":0,"serialNumber":0},"odeReceivedAt":"2024-12-27T21:14:48.007Z","schemaVersion":7,"maxDurationTime":0,"recordGeneratedBy":"RSU","sanitized":false,"asn1":"03810040038081D4001F80D07016DA410000000000000BBC2B0F775D9B0309C271431FA166EE0A27FFF93F136B8205A0A107FB2EF979F4C5BFAEEC97E4AD70C2FB36CD9730BECDB355CC2FD2A7556B160B98B46AB98AE62C185FA55EFB468D5B4000000004E2863F42CDDC144FF7980040401262CDD7B809C509F5C62CDD35519C507B9062CDCEE129C505CF262CDCA5FF9C50432C62CDC5D3D9C502E3E62CDC13E79C501E9262CDBCA2D9C5013EE62CDB80359C500E6A62CDB36299C500BC862CDAEC1D9C50093C62CDAA2109C5006EA1080203091A859EEEBB36006001830001AAD27F4FF7580001AAD355E39B5880A30029D6585009EF808332D8D9F80C3855151B38C772F765007967EC1170BCB7937F5CB880A25A52863493BCB87570DBCB5ABC6BFB2FAEC606CFA34EB95A24790B2017366D3AABE7729E00","originIp":"172.18.0.1"},"payload":{"dataType":"us.dot.its.jpo.ode.model.OdeHexByteArray","data":{"bytes":"001F80D07016DA410000000000000BBC2B0F775D9B0309C271431FA166EE0A27FFF93F136B8205A0A107FB2EF979F4C5BFAEEC97E4AD70C2FB36CD9730BECDB355CC2FD2A7556B160B98B46AB98AE62C185FA55EFB468D5B4000000004E2863F42CDDC144FF7980040401262CDD7B809C509F5C62CDD35519C507B9062CDCEE129C505CF262CDCA5FF9C50432C62CDC5D3D9C502E3E62CDC13E79C501E9262CDBCA2D9C5013EE62CDB80359C500E6A62CDB36299C500BC862CDAEC1D9C50093C62CDAA2109C5006EA1080203091A859EEEBB36006001830001AAD27F4FF7580001AAD355E39B5880A30029D6585009EF808332D8D9F80C3855151B38C772F765007967EC1170BCB7937F5CB880A25A52863493BCB87570DBCB5ABC6BFB2FAEC606CFA34EB95A24790B2017366D3AABE7729E00"}}}  
2024-12-27 14:14:48 2024-12-27 21:14:48 [Asn1DecodedDataRouter-0-C-1] DEBUG Asn1DecodedDataRouter - Key: null payload: <?xml version="1.0"?><OdeAsn1Data><metadata><logFileName/><recordType>timMsg</recordType><securityResultCode>success</securityResultCode><receivedMessageDetails/><encodings><encodings><elementName>unsecuredData</elementName><elementType>MessageFrame</elementType><encodingRule>UPER</encodingRule></encodings></encodings><payloadType>us.dot.its.jpo.ode.model.OdeAsn1Payload</payloadType><serialId><streamId>81787258-c75b-4cb5-9d8f-1cc6d998573c</streamId><bundleSize>1</bundleSize><bundleId>0</bundleId><recordId>0</recordId><serialNumber>0</serialNumber></serialId><odeReceivedAt>2024-12-27T21:14:48.007Z</odeReceivedAt><schemaVersion>7</schemaVersion><maxDurationTime>0</maxDurationTime><recordGeneratedAt/><recordGeneratedBy>RSU</recordGeneratedBy><sanitized>false</sanitized><odePacketID/><odeTimStartDateTime/><asn1>03810040038081D4001F80D07016DA410000000000000BBC2B0F775D9B0309C271431FA166EE0A27FFF93F136B8205A0A107FB2EF979F4C5BFAEEC97E4AD70C2FB36CD9730BECDB355CC2FD2A7556B160B98B46AB98AE62C185FA55EFB468D5B4000000004E2863F42CDDC144FF7980040401262CDD7B809C509F5C62CDD35519C507B9062CDCEE129C505CF262CDCA5FF9C50432C62CDC5D3D9C502E3E62CDC13E79C501E9262CDBCA2D9C5013EE62CDB80359C500E6A62CDB36299C500BC862CDAEC1D9C50093C62CDAA2109C5006EA1080203091A859EEEBB36006001830001AAD27F4FF7580001AAD355E39B5880A30029D6585009EF808332D8D9F80C3855151B38C772F765007967EC1170BCB7937F5CB880A25A52863493BCB87570DBCB5ABC6BFB2FAEC606CFA34EB95A24790B2017366D3AABE7729E00</asn1><originIp>172.18.0.1</originIp></metadata><payload><dataType>MessageFrame</dataType><data><MessageFrame><messageId>31</messageId><value><TravelerInformation><msgCnt>1</msgCnt><timeStamp>449089</timeStamp><packetID>0000000000000BBC2B</packetID><urlB>null</urlB><dataFrames><TravelerDataFrame><notUsed>1</notUsed><frameType><advisory/></frameType><msgId><roadSignID><position><lat>411269876</lat><long>-1047269563</long></position><viewAngle>1111111111111111</viewAngle><mutcdCode><warning/></mutcdCode></roadSignID></msgId><startYear>2018</startYear><startTime>448260</startTime><durationTime>1440</durationTime><priority>5</priority><notUsed1>1</notUsed1><regions><GeographicalPath><name>westbound_I-80_366.0_365.0_RSU-10.145.1.100_RW_4456</name><id><region>0</region><id>0</id></id><anchor><lat>411269876</lat><long>-1047269563</long></anchor><laneWidth>32700</laneWidth><directionality><both/></directionality><closedPath><false/></closedPath><direction>0000000000010000</direction><description><path><scale>0</scale><offset><xy><nodes><NodeXY><delta><node-LatLon><lon>-1047287423</lon><lat>411264686</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047305390</lon><lat>411260104</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047323629</lon><lat>411256185</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047342080</lon><lat>411252886</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047360706</lon><lat>411250207</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047379480</lon><lat>411248201</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047398354</lon><lat>411246839</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047417290</lon><lat>411246133</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047436246</lon><lat>411245796</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047455202</lon><lat>411245470</lat></node-LatLon></delta></NodeXY><NodeXY><delta><node-LatLon><lon>-1047474159</lon><lat>411245173</lat></node-LatLon></delta></NodeXY></nodes></xy></offset></path></description></GeographicalPath></regions><notUsed2>1</notUsed2><notUsed3>1</notUsed3><content><advisory><SEQUENCE><item><itis>777</itis></item></SEQUENCE><SEQUENCE><item><itis>13579</itis></item></SEQUENCE></advisory></content><url>null</url></TravelerDataFrame></dataFrames></TravelerInformation></value></MessageFrame></data></payload></OdeAsn1Data> 
2024-12-27 14:14:48 2024-12-27 21:14:48 [kafka-producer-network-thread | producer-5] DEBUG LoggingProducerListener - Successfully produced to topic topic.OdeTimJson with key 81787258-c75b-4cb5-9d8f-1cc6d998573c and value {"metadata":{"logFileName":"","recordType":"timMsg","securityResultCode":"success","receivedMessageDetails":{"rxSource":"NA"},"payloadType":"us.dot.its.jpo.ode.model.OdeTimPayload","serialId":{"streamId":"81787258-c75b-4cb5-9d8f-1cc6d998573c","bundleSize":1,"bundleId":0,"recordId":0,"serialNumber":0},"odeReceivedAt":"2024-12-27T21:14:48.007Z","schemaVersion":7,"maxDurationTime":0,"recordGeneratedAt":"","recordGeneratedBy":"RSU","sanitized":false,"odePacketID":"","odeTimStartDateTime":"","asn1":"03810040038081D4001F80D07016DA410000000000000BBC2B0F775D9B0309C271431FA166EE0A27FFF93F136B8205A0A107FB2EF979F4C5BFAEEC97E4AD70C2FB36CD9730BECDB355CC2FD2A7556B160B98B46AB98AE62C185FA55EFB468D5B4000000004E2863F42CDDC144FF7980040401262CDD7B809C509F5C62CDD35519C507B9062CDCEE129C505CF262CDCA5FF9C50432C62CDC5D3D9C502E3E62CDC13E79C501E9262CDBCA2D9C5013EE62CDB80359C500E6A62CDB36299C500BC862CDAEC1D9C50093C62CDAA2109C5006EA1080203091A859EEEBB36006001830001AAD27F4FF7580001AAD355E39B5880A30029D6585009EF808332D8D9F80C3855151B38C772F765007967EC1170BCB7937F5CB880A25A52863493BCB87570DBCB5ABC6BFB2FAEC606CFA34EB95A24790B2017366D3AABE7729E00","originIp":"172.18.0.1"},"payload":{"data":{"msgCnt":1,"timeStamp":449089,"packetID":"0000000000000BBC2B","urlB":"null","dataFrames":[{"doNotUse1":1,"frameType":"advisory","msgId":{"roadSignID":{"position":{"lat":411269876,"long":-1047269563},"viewAngle":{"from000-0to022-5degrees":true,"from022-5to045-0degrees":true,"from045-0to067-5degrees":true,"from067-5to090-0degrees":true,"from090-0to112-5degrees":true,"from112-5to135-0degrees":true,"from135-0to157-5degrees":true,"from157-5to180-0degrees":true,"from180-0to202-5degrees":true,"from202-5to225-0degrees":true,"from225-0to247-5degrees":true,"from247-5to270-0degrees":true,"from270-0to292-5degrees":true,"from292-5to315-0degrees":true,"from315-0to337-5degrees":true,"from337-5to360-0degrees":true},"mutcdCode":"warning"}},"startYear":2018,"startTime":448260,"durationTime":1440,"priority":5,"doNotUse2":1,"regions":[{"name":"westbound_I-80_366.0_365.0_RSU-10.145.1.100_RW_4456","id":{"region":0,"id":0},"anchor":{"lat":411269876,"long":-1047269563},"laneWidth":32700,"directionality":"both","closedPath":false,"direction":{"from000-0to022-5degrees":false,"from022-5to045-0degrees":false,"from045-0to067-5degrees":false,"from067-5to090-0degrees":false,"from090-0to112-5degrees":false,"from112-5to135-0degrees":false,"from135-0to157-5degrees":false,"from157-5to180-0degrees":false,"from180-0to202-5degrees":false,"from202-5to225-0degrees":false,"from225-0to247-5degrees":false,"from247-5to270-0degrees":true,"from270-0to292-5degrees":false,"from292-5to315-0degrees":false,"from315-0to337-5degrees":false,"from337-5to360-0degrees":false},"description":{"path":{"scale":0,"offset":{"xy":{"nodes":[{"delta":{"node-LatLon":{"lon":-1047287423,"lat":411264686}}},{"delta":{"node-LatLon":{"lon":-1047305390,"lat":411260104}}},{"delta":{"node-LatLon":{"lon":-1047323629,"lat":411256185}}},{"delta":{"node-LatLon":{"lon":-1047342080,"lat":411252886}}},{"delta":{"node-LatLon":{"lon":-1047360706,"lat":411250207}}},{"delta":{"node-LatLon":{"lon":-1047379480,"lat":411248201}}},{"delta":{"node-LatLon":{"lon":-1047398354,"lat":411246839}}},{"delta":{"node-LatLon":{"lon":-1047417290,"lat":411246133}}},{"delta":{"node-LatLon":{"lon":-1047436246,"lat":411245796}}},{"delta":{"node-LatLon":{"lon":-1047455202,"lat":411245470}}},{"delta":{"node-LatLon":{"lon":-1047474159,"lat":411245173}}}]}}}}}],"doNotUse3":1,"doNotUse4":1,"content":{"advisory":[{"item":{"itis":777}},{"item":{"itis":13579}}]},"url":"null"}]},"dataType":"us.dot.its.jpo.ode.plugin.j2735.travelerinformation.TravelerInformation"}}

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.

Helpful Documentation

…from clean slate

Generated by sending tim to deposit tim endpoint and grabbing value produced to the `topic.Asn1EncoderOutput` topic via kafka-ui

…gned-tim.xml

…TimUnsecured_depositsToSdxTopicAndTimTmcFiltered

…ve DI and testability

…tImpl These can be used to wrap the calls to security services and allow easier mocking of responses. It also allows us to better shield ourselves from API contract changes. users of the client don't need to be aware of changes to the external API as all interactions can be encapsulated within the client

This moves the calls to security services from Asn1CommandManager to SecurityServicesClientImpl which allows for more testability, flexibility (mocking client calls), and modularity

…ether (make data dynamic)

…essSignedMessage

…h refactor nor do they confirm behavior

The responsibility to submit data to topics is already contained within Asn1EncodedDataRouter, and Asn1CommandManager is not responsible for any kafka interactions otherwise, so the responsibility was moved to Asn1EncodedDataRouter

…ommandManager

Updated method signatures and instance usage to specify ResponseEvent with Address as the generic type. This change improves type safety and aligns with best practices for clearer code readability and maintainability.

…outer

…nce for topic

…ests - not working well with other tests

…lient

Refactored the SecurityServicesClient to use dependency-injected RestTemplate and improved the request payload structure with a dedicated SignatureRequestModel class. Added unit tests for signMessage functionality and introduced configuration for RestTemplate with WebClientConfig.

Added `RestClientException` to `signMessage` for better error handling. Implemented a new test case to verify behavior when the signing service returns a null response.

Simplified `getServiceRequest` method by directly passing the metadata JSON object instead of the full consumed object. Updated relevant calls to align with this change, improving clarity and reducing redundant operations.

Removed verbose and redundant JavaDoc comments, replacing them with concise inline comments to improve code readability. This change ensures that key processing logic remains documented without unnecessary detail duplication.

Removed redundant null checks before calling `sendToRsus` in the main logic. The null checks are properly handled within the `sendToRsus` method, ensuring clearer and more maintainable code.

Standardizing on the ObjectMapper provided in the app context will reduce variability and decrease risk of unintended differences in serdes behavior

This name better represents the behavior of the method. The Javadocs were updated to be accurate as well

mcook42

I would recommend reviewing the Asn1EncodedDataRouter code from within your IDE. There are a lot of changes in there that don't necessarily work well with GitHub's viewer. The processing flow should not have changed. The changes in that file should be in one of four categories:

Deduplication of JSON processing
Deduplication of message signing or encoding
Using the SerializationConfig-provided Object Mapper instead of JsonUtils or TimTransmorgrifier or other JSON serdes (there were a few weird ones).
Extracting logic into named methods to reduce cognitive complexity

mcook42 · 2024-12-26T23:17:08Z

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

-    return streams.state().isRunningOrRebalancing();
+    return streams.state().equals(KafkaStreams.State.RUNNING);


Note: this method is only used in tests for now, and in the tests, a state of REBALANCING would cause an error. Addressing this possible bug may be worth the effort in a follow-up PR or separate work item. Still, I don't believe the changes belong here since I didn't change how the OdeTimJsonTopology works in this PR.

mcook42 · 2024-12-26T23:22:29Z

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

-    kafkaTemplate.send(jsonTopics.getTim(), consumerRecord.key(), odeTimData);
+    kafkaTemplate.send(jsonTopics.getTim(), streamId, odeTimData);


Note: the key is empty when consuming from the asn1 decoder output topic. We need to include the streamId as the key when publishing TIM data so that the downstream consumers can use the streamId to lookup the TIM JSON from the K-Table in OdeTimJsonTopology. If we don't produce with a streamId (which we conditionally set a few lines above this method)) then we will never produce to the TMCFiltered topic in the Asn1EncodedDataRouter

mcook42 · 2024-12-26T23:36:19Z

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/services/asn1/Asn1EncodedDataRouter.java

-      if (metadata.has(TimTransmogrifier.REQUEST_STRING)) {
-        JSONObject request = metadata.getJSONObject(TimTransmogrifier.REQUEST_STRING);
-        if (request.has(TimTransmogrifier.RSUS_STRING)) {
-          Object rsus = request.get(TimTransmogrifier.RSUS_STRING);
-          if (rsus instanceof JSONObject) {
-            JSONObject rsusIn = (JSONObject) request.get(TimTransmogrifier.RSUS_STRING);
-            if (rsusIn.has(TimTransmogrifier.RSUS_STRING)) {
-              Object rsu = rsusIn.get(TimTransmogrifier.RSUS_STRING);
-              JSONArray rsusOut = new JSONArray();
-              if (rsu instanceof JSONArray) {
-                log.debug("Multiple RSUs exist in the request: {}", request);
-                JSONArray rsusInArray = (JSONArray) rsu;
-                for (int i = 0; i < rsusInArray.length(); i++) {
-                  rsusOut.put(rsusInArray.get(i));
-                }
-                request.put(TimTransmogrifier.RSUS_STRING, rsusOut);
-              } else if (rsu instanceof JSONObject) {
-                log.debug("Single RSU exists in the request: {}", request);
-                rsusOut.put(rsu);
-                request.put(TimTransmogrifier.RSUS_STRING, rsusOut);
-              } else {
-                log.debug("No RSUs exist in the request: {}", request);
-                request.remove(TimTransmogrifier.RSUS_STRING);
-              }
-            }
-          }


Note: as far as I can tell by reading the code and stepping through it with a debugger, this code block did nothing. The request variable is manipulated but never used after line 153. Please take a second look at this change. I'm fairly certain it can be safely deleted, but I may be missing something

mcook42 · 2024-12-26T23:44:53Z

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/services/asn1/Asn1EncodedDataRouter.java

-          + consumedData;
-      if (log.isDebugEnabled()) {
-        // print error message and stack trace
-        EventLogger.logger.error(msg, e);


Question: Does anyone know if there was a purpose to the EventLogger? Do we consume data from this logger differently?

mcook42 · 2024-12-27T00:13:08Z

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/services/asn1/Asn1EncodedDataRouter.java

+      Date requiredExpirationDate = new Date();
+      requiredExpirationDate.setTime(timTimestamp.getTime() + maxDurationTime);
+      timWithExpiration.put("requiredExpirationDate", dateFormat.format(requiredExpirationDate));
+    } catch (Exception e) {


Note: this is one of the rare times in this code where we can safely continue (as far as I can tell from the code) processing a message should an Exception be thrown. I intentionally handled the expectation here and allowed the processing to continue. I couldn't find any downstream issues in the code. Also, please note that we already handled the exception and continued processing in the previous implementation. This does not change the code flow.

mcook42 · 2024-12-27T00:13:19Z

jpo-ode-svcs/src/main/java/us/dot/its/jpo/ode/services/asn1/Asn1EncodedDataRouter.java

+    } catch (Exception e) {
+      log.error("Unable to get expiration date from signed messages response. Setting expirationData to 'null'", e);
+      timWithExpiration.put("expirationDate", "null");
+    }


Note: this is one of the rare times in this code where we can safely continue (as far as I can tell from the code) processing a message should an Exception be thrown. I intentionally handled the expectation here and allowed the processing to continue. I couldn't find any downstream issues in the code. Also, please note that we already handled the exception and continued processing in the previous implementation. This does not change the code flow.

mcook42 · 2024-12-27T00:31:54Z

jpo-ode-svcs/src/test/java/us/dot/its/jpo/ode/OdeTimJsonTopologyTest.java

Note: This code is heavily tested indirectly by the Asn1EncodedDataRouter tests. The existing tests were not adding much value (I wrote most of them, so my bad), and meaningful tests were difficult to write without duplicating much of the testing already performed in the Asn1EncodedDataRouter tests.

…for Kafka container setup Replaced ConcurrentMessageListenerContainer with KafkaMessageListenerContainer and removed KafkaConsumerConfig dependency. This fixes an issue where we would intermittently encounter the following error when running tests: java.lang.IllegalStateException: Expected 1 but got 0 partitions at org.springframework.kafka.test.utils.ContainerTestUtils.waitForAssignment(ContainerTestUtils.java:85)

Updated the expiration date calculation to use `Instant` and ensure accuracy by handling milliseconds properly. Adjusted test cases and mock classes to align with the updated logic. Fixed incorrect test data to reflect the corrected expiration date format.

Replaced the outdated `SimpleDateFormat` with the modern `DateTimeFormatter` to handle date formatting and parsing. This improves thread safety and code readability while ensuring alignment with Java's modern date/time APIs. Adjusted test resources to reflect updated timestamp formatting logic.

Add exception handling to catch RestClientException when signing TIM messages. Log detailed error messages, including specific handling for HttpClientErrorException.NotFound, to provide better debugging information and highlight potential misconfiguration of jpo-security-svcs.

Convert TIM start flag to uppercase for consistency and improve logging to dynamically include the expected start flag value. This enhances clarity and ensures robust matching during header stripping.

Consolidated signing logic into `depositToTimCertExpirationTopic` and removed redundant `signTimWithExpiration` method. Added utility for obtaining hex-encoded signed messages in `SignatureResultModel` to streamline encoding operations.

If we fail to sign the message, we should fail to process the message further. We don't want any unsigned messages in our system (if signing is enabled)

…/spring-kafka/asn1-encoded-router

…aVersion 8

mcook42 added 30 commits December 17, 2024 16:35

style: reformat Asn1EncodedDataRouter and Asn1CommandManager

35f386b

chore: delete disabled tests from Asn1EncodedDataRouterTest to start …

f7000c9

…from clean slate

refactor: extract nested logic to named methods

62b34b3

test: add asn1-encoder-output-tim.xml

d04010d

Generated by sending tim to deposit tim endpoint and grabbing value produced to the `topic.Asn1EncoderOutput` topic via kafka-ui

refactor: encapsulate date setting logic for code clarity

b51b14a

chore: rename asn1-encoder-output-tim.xml to asn1-encoder-output-unsi…

23821e4

…gned-tim.xml

refactor: pull object creation up to allow mocking of dependencies

1cb9d97

test: initial processSNMPDeposit test setup. encoder not consuming

f4147e4

test: processEncodedTimUnsecured_depositsToSdxTopic

eed0a82

test: processEncodedTimUnsecured_depositsToSdxTopic -> processEncoded…

c43c3e8

…TimUnsecured_depositsToSdxTopicAndTimTmcFiltered

refactor: pass asn1CommandManager into Asn1EncodedDataRouter to impro…

785795a

…ve DI and testability

refactor: use ISecurityServicesClient in Asn1EncodedDataRouter

1011688

This moves the calls to security services from Asn1CommandManager to SecurityServicesClientImpl which allows for more testability, flexibility (mocking client calls), and modularity

test: processSNMPDepositOnly

6a85fbf

style: update license header to not be javadoc

3dce2a4

test: update tests in Asn1EncodedDataRouterTest to enable running tog…

3898ad9

…ether (make data dynamic)

test: processEncodedTimUnsecured in Asn1EncodedDataRouterTest

225960c

test: make consumer groups unique between tests to keep them isolated

1aba9e0

test: tweak securityServices settings to get code to run through proc…

e7e7bf1

…essSignedMessage

test: disable all Asn1CommandManagerTest tests as they don't work wit…

7aa4920

…h refactor nor do they confirm behavior

style: added missing Javadocs and reorganized Asn1CommandManager

ab2a633

refactor: don't swallow exception on startup of rsuDepositor in Asn1C…

b4473a3

…ommandManager

refactor: ResponseEvent to use generic Address type

c1f1178

Updated method signatures and instance usage to specify ResponseEvent with Address as the generic type. This change improves type safety and aligns with best practices for clearer code readability and maintainability.

style: reformat and add missing Javadocs SnmpSession

79e0e0c

chore: delete unused AsnCodecRouterServiceController.java

24fbba7

refactor: pull Asn1CommandManager functionality into Asn1EncodedDataR…

c99bbe8

…outer

refactor: replace SDXDepositorTopics.java reference with Value refere…

cdb14a2

…nce for topic

refactor: introduce Config beans for Asn1EncodedDataRouter dependencies

da1b379

refactor: convert Asn1EncodedDataRouter to KafkaListener and update t…

1c0e260

…ests - not working well with other tests

mcook42 added 9 commits December 26, 2024 11:20

refactor: rename SecurityServicesClientImpl.java to SecurityServicesC…

bcb6901

…lient

chore: remove SecurityServicesConfig and related dependencies.

44b73cc

chore: correct import paths in Asn1EncodedDataRouter and test

2246c2d

test: handle null responses and improve error handling in signing.

b920745

Added `RestClientException` to `signMessage` for better error handling. Implemented a new test case to verify behavior when the signing service returns a null response.

refactor: use autowired props to drive URL in SecurityServicesClientTest

5c5ba73

refactor: use MockSecurityServicesClient in Asn1EncodedDataRouterTest

411c562

refactor: service request extraction to use metadata JSON.

6d60ed2

Simplified `getServiceRequest` method by directly passing the metadata JSON object instead of the full consumed object. Updated relevant calls to align with this change, improving clarity and reducing redundant operations.

docs: simplify comments and remove redundant JavaDoc.

0d30254

Removed verbose and redundant JavaDoc comments, replacing them with concise inline comments to improve code readability. This change ensures that key processing logic remains documented without unnecessary detail duplication.

mcook42 force-pushed the mcook42/spring-kafka/asn1-encoded-router branch from 0bc948a to 0d30254 Compare December 26, 2024 23:56

mcook42 added 5 commits December 26, 2024 16:58

docs: remove dead comments

69ac11e

chore: simplify RSU message handling logic.

1e7ffbc

Removed redundant null checks before calling `sendToRsus` in the main logic. The null checks are properly handled within the `sendToRsus` method, ensuring clearer and more maintainable code.

chore: simplify error message for SDX deposit failure

7a58bce

refactor: don't use TimTransmorgrifier to buildEncodings

fd715ab

Standardizing on the ObjectMapper provided in the app context will reduce variability and decrease risk of unintended differences in serdes behavior

chore: rename buildJsonTimFromPacket to buildTimFromPacket

bcfcfc6

This name better represents the behavior of the method. The Javadocs were updated to be accurate as well

mcook42 commented Dec 27, 2024

View reviewed changes

mcook42 added 14 commits December 27, 2024 10:21

style: reorganize methods in Asn1EncodedDataRouterTest

a765be7

chore: correct TIM header processing in Asn1EncodedDataRouter

dc48020

Convert TIM start flag to uppercase for consistency and improve logging to dynamically include the expected start flag value. This enhances clarity and ensures robust matching during header stripping.

refactor: TIM signing exception handling

52209b4

If we fail to sign the message, we should fail to process the message further. We don't want any unsigned messages in our system (if signing is enabled)

test: add better test failure messaging

789acc9

test: add better test failure messaging

c63b92d

test: add better test failure messaging (actually stringify iterable)

c8055ad

test: simplify testing assertions for better failure reporting

bd9f820

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

5b86319

…/spring-kafka/asn1-encoded-router

test: update expected-asn1-encoded-router-snmp-deposit.xml with schem…

95f154f

…aVersion 8

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Migrate Asn1EncodedDataRouter to use Spring Kafka #138

Migrate Asn1EncodedDataRouter to use Spring Kafka #138

mcook42 commented Dec 24, 2024 •

edited

Loading

mcook42 left a comment

mcook42 Dec 26, 2024

mcook42 Dec 26, 2024

mcook42 Dec 26, 2024

mcook42 Dec 26, 2024

mcook42 Dec 27, 2024

mcook42 Dec 27, 2024

mcook42 Dec 27, 2024

		return streams.state().isRunningOrRebalancing();
		return streams.state().equals(KafkaStreams.State.RUNNING);

		kafkaTemplate.send(jsonTopics.getTim(), consumerRecord.key(), odeTimData);
		kafkaTemplate.send(jsonTopics.getTim(), streamId, odeTimData);

Migrate Asn1EncodedDataRouter to use Spring Kafka #138

Are you sure you want to change the base?

Migrate Asn1EncodedDataRouter to use Spring Kafka #138

Conversation

mcook42 commented Dec 24, 2024 • edited Loading

PR Details

Description

Related Issue

Motivation and Context

How Has This Been Tested?

Types of changes

Checklist:

Helpful Documentation

mcook42 left a comment

Choose a reason for hiding this comment

mcook42 Dec 26, 2024

Choose a reason for hiding this comment

mcook42 Dec 26, 2024

Choose a reason for hiding this comment

mcook42 Dec 26, 2024

Choose a reason for hiding this comment

mcook42 Dec 26, 2024

Choose a reason for hiding this comment

mcook42 Dec 27, 2024

Choose a reason for hiding this comment

mcook42 Dec 27, 2024

Choose a reason for hiding this comment

mcook42 Dec 27, 2024

Choose a reason for hiding this comment

mcook42 commented Dec 24, 2024 •

edited

Loading