Migrate Asn1EncodedDataRouter to use Spring Kafka #138
Conversation
Generated by sending tim to deposit tim endpoint and grabbing value produced to the `topic.Asn1EncoderOutput` topic via kafka-ui
…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)
…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
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.
…ests - not working well with other tests
Renamed methods and variables for clarity in message encoding workflows. Updated references in test cases and adjusted test resources to align with naming changes. This improves code readability and better reflects the encoding stages.
| } | ||
|
|
||
| /** | ||
| * Sign the encoded TIM message and write to Kafka with an expiration time. | ||
| * Constructs an XML representation of an Advisory Situation Data (ASD) message containing a | ||
| * signed Traveler Information Message (TIM). Processes the provided service request and signed |
There was a problem hiding this comment.
suggestion: the ASD doesn't always have a signed TIM. We should update documentation to reflect it's just a TIM, and ensure our code paths are capable of delivering unsigned
There was a problem hiding this comment.
updated the javadoc comment to remove the "signed" portion 3a07370. I also read through the execution path and can confirm that the documentation was inaccurate & the code supports both signed and unsigned TIMs.
| } | ||
| private String packageSignedTimIntoAsd(ServiceRequest request, String signedMsg) throws JsonProcessingException, ParseException { | ||
| SDW sdw = request.getSdw(); | ||
| SNMP snmp = request.getSnmp(); |
There was a problem hiding this comment.
suggestion: We're only packing into ASD if we're depositing into the SDX, so we should ignore the SNMP portion here and just take the SDW.
| if (dataSigningEnabledSDW && request.getSdw() != null) { | ||
| hexEncodedTim = signTIMAndProduceToExpireTopic(hexEncodedTim, consumedObj); | ||
| } | ||
| // no SDW in metadata (SNMP deposit only) -> sign MF -> send to RSU |
There was a problem hiding this comment.
suggestion: this comment is a little misleading, we certainly could have SDW in metadata here and frequently do.
There was a problem hiding this comment.
Good callout. I've removed the comment in favor of "self-documenting" code 20a7b0a
|
|
||
| hexEncodedTim = mfObj.getString(BYTES); | ||
| JSONObject deposit = new JSONObject(); |
There was a problem hiding this comment.
question(non-blocking): would it be worth making a POJO for this object that gets sent to the SdwDepositor and on to the SDX?
There was a problem hiding this comment.
yeah, it was simple enough to implement so I defined one in 7c03cce
…/spring-kafka/asn1-encoded-router
…sn1EncodedDataRouter
…eSignedTimIntoAsd
There was a problem hiding this comment.
Everything passes tests and works for the most part. The one thing that seems to no longer be occurring is that TIM messages are not being written to the topic.OdeTimJsonTMCFiltered topic. This seems to occur no matter how you configure the jpo-ode to sign TIMs. This seems incorrect. Am I missing something?
There was a problem hiding this comment.
Re-tested with the security module running and everything is working correctly. My mistake on not running the sec module! Having both RSU and SDW set to unsigned still results in this behavior but this is apparently how the existing functionality works too.
Everything looks to be good to go.
…h from Asn1EncodedDataRouter" This reverts commit a00ab06.
…sitTim when dataSigningEnabledSDW is false Eliminated processing logic and tests related to unsigned TIM messages, as well as the conditional ASD element addition when dataSigningEnabledSDW is false. This streamlines the codebase by ensuring all TIM messages follow the secured flow, improving maintainability and alignment with production recommendations.
Added/replaced logic to handle RSU XML-to-JSON array normalization in `processRsusIfPresent`. Refactored metadata handling in `processUnsignedMessage` and enhanced test cases to include RSU-related input/output scenarios. This ensures proper handling of RSU data while improving test robustness.
There was a problem hiding this comment.
Looks really good! I like the updates here. One other question I had is that I had to update the jpo-security-svcs module reference in the .gitmodules file. Should this PR modify that reference so it is in line with what the current ODE expects?
Passed Checks:
- tests pass
- log files process properly
- All services start up through docker-compose
- REST created TIMs for SDW deposit functions properly
- REST created TIMs for SNMP deposit functions properly
…/spring-kafka/asn1-encoded-router
Co-authored-by: Michael7371 <[email protected]>
PR Details
Description
Dateclass with usages ofLocalDateTime(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)NOTE: The interactions with the security service rely on the changes contained within: CDOT-CV/jpo-security-svcs#16. The jpo-security-svcs submodule points to the CDOT-CV dev branch in this PR
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?
For the live testing, I set the following environment variables on my system to enable signing and confirm all pieces played well together
To test the signing flow, I produced the test XML to the
topic.Asn1EncoderOutputtopic via the kafka-ui.Snippet from the locals during local testing:
2025-01-06 21:57:56 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - ServiceRequest: {"ode":{"verb":"POST","version":"3"},"sdw":{"recordId":"6B573067","serviceRegion":{"nwCorner":{"latitude":"38.98721843900006","longitude":"-104.76767069499999"},"seCorner":{"latitude":"38.96666515900006","longitude":"-104.74048299899994"}},"ttl":"oneday"}} 2025-01-06 21:57:56 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - Mapped to object ServiceRequest: {"ode":{"version":3,"verb":"POST"},"sdw":{"serviceRegion":{"nwCorner":{"latitude":38.98721843900006,"longitude":-104.76767069499999},"seCorner":{"latitude":38.96666515900006,"longitude":-104.74048299899994}},"ttl":"oneday","recordId":"6B573067"}} 2025-01-06 21:57:56 [Asn1EncodedDataRouter-0-C-1] INFO Asn1EncodedDataRouter - Processing unsigned message. 2025-01-06 21:57:56 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - Signing encoded TIM message... 2025-01-06 21:57:56 [Asn1EncodedDataRouter-0-C-1] INFO SecurityServicesClient - Sending data to security services module at http://172.21.103.243:8090/sign with validity override 17820000 to be signed 2025-01-06 21:57:56 [Asn1EncodedDataRouter-0-C-1] DEBUG SecurityServicesClient - Data to be signed: <SignatureRequestModel(message=AB+AhHARTzcDsCfPIHEyjhIPd12bAwHCZw83QWa8MCf/+T9AvmKBKaAH+Tk34c9ayY36cGpfbQmrdmwbN4AAAAAJnDzdBZrwwITiMCAAhEAzYRQHQBczzcLWID/aF1Y5g1jtJ1V8aFb7qsHDoPjZwiHdhYpg4K2llhJFZ/KxggAQAgCe7rs2AA==, sigValidityOverride=17820000),[Content-Type:"application/json"]> 2025-01-06 21:58:00 [Asn1EncodedDataRouter-0-C-1] DEBUG SecurityServicesClient - Security services module response: <200 OK OK,SignatureResultModel(messageSigned=AB+AhHARTzcDsCfPIHEyjhIPd12bAwHCZw83QWa8MCf/+T9AvmKBKaAH+Tk34c9ayY36cGpfbQmrdmwbN4AAAAAJnDzdBZrwwITiMCAAhEAzYRQHQBczzcLWID/aF1Y5g1jtJ1V8aFb7qsHDoPjZwiHdhYpg4K2llhJFZ/KxggAQAgCe7rs2AA==c26d50a4, messageExpiry=17820000),[Content-Type:"application/json", Transfer-Encoding:"chunked", Date:"Mon, 06 Jan 2025 21:58:00 GMT", Keep-Alive:"timeout=60", Connection:"keep-alive"]> 2025-01-06 21:58:00 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - Encoded message - phase 1: 001F808470114F3703B027CF2071328E120F775D9B0301C2670F374166BC3027FFF93F40BE628129A007F93937E1CF5AC98DFA706A5F6D09AB766C1B3780000000099C3CDD059AF0C084E2302000844033611407401733CDC2D6203FDA1756398358ED27557C6856FBAAC1C3A0F8D9C221DD858A60E0ADA596124567F2B182001002009EEEBB3600736E9DE746B8 2025-01-06 21:58:00 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - No RSUs or SNMP provided. Not sending to RSUs. 2025-01-06 21:58:00 [kafka-producer-network-thread | producer-5] DEBUG LoggingProducerListener - Successfully produced to topic topic.OdeTIMCertExpirationTimeJson with key null and value {"packetID":"03B027CF2071328E12","startDateTime":"2024-03-08T16:37:05.414Z","requiredExpirationDate":"2024-03-08T21:34:05.414Z","expirationDate":"1970-07-26T06:00:00.000Z"} 2025-01-06 21:58:00 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - TIM not found in k-table. Skipping deposit to TMC-filtered topic. 2025-01-06 21:58:00 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - Publishing message for round 2 encoding 2025-01-06 21:58:00 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - Fully crafted ASD to be encoded: <OdeAsn1Data><metadata><payloadType>us.dot.its.jpo.ode.model.OdeAsdPayload</payloadType><serialId><streamId>05ce8f1c-9c30-4ba1-afb3-5cac3d681ce9</streamId><bundleSize>1</bundleSize><bundleId>0</bundleId><recordId>0</recordId><serialNumber>0</serialNumber></serialId><odeReceivedAt>2025-01-06T21:58:00.453Z</odeReceivedAt><schemaVersion>8</schemaVersion><maxDurationTime>0</maxDurationTime><sanitized>false</sanitized><request><ode><version>3</version><verb>POST</verb></ode><sdw><serviceRegion><nwCorner><latitude>38.98721843900006</latitude><longitude>-104.76767069499999</longitude></nwCorner><seCorner><latitude>38.96666515900006</latitude><longitude>-104.74048299899994</longitude></seCorner></serviceRegion><ttl>oneday</ttl><recordId>6B573067</recordId></sdw></request><encodings><encodings><elementName>AdvisorySituationData</elementName><elementType>AdvisorySituationData</elementType><encodingRule>UPER</encodingRule></encodings></encodings></metadata><payload><dataType>us.dot.its.jpo.ode.plugin.j2735.DdsAdvisorySituationData</dataType><data><AdvisorySituationData><dialogID>156</dialogID><seqID>5</seqID><groupID>00000000</groupID><requestID>0CDD17A4</requestID><recordID>6B573067</recordID><timeToLive>2</timeToLive><serviceRegion><nwCorner><lat>389872184</lat><long>-1047676707</long></nwCorner><seCorner><lat>389666652</lat><long>-1047404830</long></seCorner></serviceRegion><asdmDetails><asdmID>0CDD17A4</asdmID><asdmType>2</asdmType><distType>02</distType><startTime><year>0</year><month>0</month><day>0</day><hour>31</hour><minute>60</minute></startTime><stopTime><year>0</year><month>0</month><day>0</day><hour>31</hour><minute>60</minute></stopTime><advisoryMessage>001F808470114F3703B027CF2071328E120F775D9B0301C2670F374166BC3027FFF93F40BE628129A007F93937E1CF5AC98DFA706A5F6D09AB766C1B3780000000099C3CDD059AF0C084E2302000844033611407401733CDC2D6203FDA1756398358ED27557C6856FBAAC1C3A0F8D9C221DD858A60E0ADA596124567F2B182001002009EEEBB3600736E9DE746B8</advisoryMessage></asdmDetails></AdvisorySituationData></data></payload></OdeAsn1Data> 2025-01-06 21:58:00 [kafka-producer-network-thread | producer-5] DEBUG LoggingProducerListener - Successfully produced to topic topic.Asn1EncoderInput with key null and value <OdeAsn1Data><metadata><payloadType>us.dot.its.jpo.ode.model.OdeAsdPayload</payloadType><serialId><streamId>05ce8f1c-9c30-4ba1-afb3-5cac3d681ce9</streamId><bundleSize>1</bundleSize><bundleId>0</bundleId><recordId>0</recordId><serialNumber>0</serialNumber></serialId><odeReceivedAt>2025-01-06T21:58:00.453Z</odeReceivedAt><schemaVersion>8</schemaVersion><maxDurationTime>0</maxDurationTime><sanitized>false</sanitized><request><ode><version>3</version><verb>POST</verb></ode><sdw><serviceRegion><nwCorner><latitude>38.98721843900006</latitude><longitude>-104.76767069499999</longitude></nwCorner><seCorner><latitude>38.96666515900006</latitude><longitude>-104.74048299899994</longitude></seCorner></serviceRegion><ttl>oneday</ttl><recordId>6B573067</recordId></sdw></request><encodings><encodings><elementName>AdvisorySituationData</elementName><elementType>AdvisorySituationData</elementType><encodingRule>UPER</encodingRule></encodings></encodings></metadata><payload><dataType>us.dot.its.jpo.ode.plugin.j2735.DdsAdvisorySituationData</dataType><data><AdvisorySituationData><dialogID>156</dialogID><seqID>5</seqID><groupID>00000000</groupID><requestID>0CDD17A4</requestID><recordID>6B573067</recordID><timeToLive>2</timeToLive><serviceRegion><nwCorner><lat>389872184</lat><long>-1047676707</long></nwCorner><seCorner><lat>389666652</lat><long>-1047404830</long></seCorner></serviceRegion><asdmDetails><asdmID>0CDD17A4</asdmID><asdmType>2</asdmType><distType>02</distType><startTime><year>0</year><month>0</month><day>0</day><hour>31</hour><minute>60</minute></startTime><stopTime><year>0</year><month>0</month><day>0</day><hour>31</hour><minute>60</minute></stopTime><advisoryMessage>001F808470114F3703B027CF2071328E120F775D9B0301C2670F374166BC3027FFF93F40BE628129A007F93937E1CF5AC98DFA706A5F6D09AB766C1B3780000000099C3CDD059AF0C084E2302000844033611407401733CDC2D6203FDA1756398358ED27557C6856FBAAC1C3A0F8D9C221DD858A60E0ADA596124567F2B182001002009EEEBB3600736E9DE746B8</advisoryMessage></asdmDetails></AdvisorySituationData></data></payload></OdeAsn1Data> 2025-01-06 21:58:00 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - ServiceRequest: {"ode":{"verb":"POST","version":"3"},"sdw":{"recordId":"6B573067","serviceRegion":{"nwCorner":{"latitude":"38.98721843900006","longitude":"-104.76767069499999"},"seCorner":{"latitude":"38.96666515900006","longitude":"-104.74048299899994"}},"ttl":"oneday"}} 2025-01-06 21:58:00 [Asn1EncodedDataRouter-0-C-1] DEBUG Asn1EncodedDataRouter - Mapped to object ServiceRequest: {"ode":{"version":3,"verb":"POST"},"sdw":{"serviceRegion":{"nwCorner":{"latitude":38.98721843900006,"longitude":-104.76767069499999},"seCorner":{"latitude":38.96666515900006,"longitude":-104.74048299899994}},"ttl":"oneday","recordId":"6B573067"}} 2025-01-06 21:58:00 [kafka-producer-network-thread | producer-5] DEBUG LoggingProducerListener - Successfully produced to topic topic.SDWDepositorInput with key null and value {"encodedMsg":"C44000000000CDD17A46B57306742670F19C166BC56E099BD80B859B761C3866E8BD2202000007FC000007FC11C003F0108E0229E6E07604F9E40E2651C241EEEBB36060384CE1E6E82CD78604FFFF27E817CC50253400FF2726FC39EB5931BF4E0D4BEDA1356ECD8366F00000000133879BA0B35E18109C4604001088066C2280E802E679B85AC407FB42EAC7306B1DA4EAAF8D0ADF755838741F1B38443BB0B14C1C15B4B2C248ACFE56304002004013DDD766C00E6DD3BCE8D700"}Types of changes
Checklist:
ODE Contributing Guide
Helpful Documentation