Update Consumer Data Extraction #29

pmonington · 2026-01-23T17:46:27Z

PR Details

Description

With the update in the ODE, the format in which the Logger Kafka Consumer is consuming has changed. This causes all TIMs going through the TIM Manager to be unable to be inserted into the Active TIMs Table when they should be. The solution is to adjust the format in which the kafka Logger Consumer is navigating to get data processed and into the active TIMs table.

How Has This Been Tested?

All unit tests were verified to pass. End-to-End testing was preformed to ensure a local deployment is easily spined up and inserted TIMs made their way into the Active TIM Table. Vsl, rw, rc, cc, parking, and incident TIMs were inserted and verified to make their way into the Active TIM table.
Example TIMs sent through:
{ "timParkingList": [ { "direction": "I", "startPoint": { "latitude": 39.613221750299076, "longitude": -104.89570514902218, "valid": true }, "endPoint": { "latitude": 40.73654117648727, "longitude": -104.99349024587299, "valid": true }, "route": "025A_DEC", "roadCode": "1", "advisory": [ "3" ], "segment": "2", "clientId": "testid", "itisCodes": [ "12559" ], "mileMarker": 1.0, "availability": 1, "exit": "12559" } ] }

{ "timIncidentList": [ { "direction": "I", "startPoint": { "latitude": 39.613221750299076, "longitude": -104.89570514902218, "valid": true }, "route": "025A_DEC", "roadCode": "1", "advisory": [ "3" ], "segment": "2", "clientId": "testid", "itisCodes": [ "12559" ], "problem": "mudslide", "effect": "test", "action": "stop", "incidentId": "test", "highway": "025A_DEC", "pk": 1 } ] }

Types of changes

[ X] 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:

My changes require new environment variables:
- I have updated the docker-compose, K8s YAML, and all dependent deployment configuration files.
My changes require updates to the documentation:
- I have updated the documentation accordingly.
[ X] My changes require updates and/or additions to the unit tests:
- [X ] I have modified/added tests to cover my changes.
[ X] All existing tests pass.

… new nested `TravelerInformation` node

…ate management

…consumer-data-extraction

… data to match OdeTimJson new format

…son data formats

…or TIM payload conversion

…ta and metadata formats

pmonington · 2026-01-23T17:48:21Z

...mer/src/main/java/com/trihydro/loggerkafkaconsumer/app/services/ActiveTimHoldingService.java

    }

-    public boolean updateTimExpiration(String packetID, String expDate) {
+    public boolean updateTimExpiration(String packetID, Instant expDate) {


Note: these changes are due to the database being unable to read the expiration date properly. In order for the prepared statement to work, it needed to be converted into a time stamp for the database.

pmonington · 2026-01-23T17:55:23Z

cv-data-service-library/src/main/java/com/trihydro/library/helpers/JsonToJavaConverter.java

        return region;
    }

-    public OdeTimPayload convertTimPayloadJsonToJava(String value) {


Note: the removal of this class is due to the fact that with the changes to update how the string value from the Kafka topic caused it to match up almost exactly with the other method. The only difference had been that this method had not set the StartDateTime, DurationTime, Priority, or FrameType, however, these are required fields so should be set anyway.

dmccoystephenson

This looks good to me! Great job!

Verifications made:

Successful compilation
Unit tests pass
Logger Kafka Consumer successfully processes TIM generated using "Incident" TIM body provided in PR description with the following output:

logger-kafka-consumer-1  | 2026-01-23 19:13:13 [main] [,] INFO  LoggerKafkaConsumer - Polling found 1 new record(s) to process 
logger-kafka-consumer-1  | 2026-01-23 19:13:13 [main] [,] DEBUG LoggerKafkaConsumer - Processing record from partition=0, offset=15 
logger-kafka-consumer-1  | 2026-01-23 19:13:13 [main] [,] INFO  LoggerKafkaConsumer - Processing message for topic: topic.OdeTimJson 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] DEBUG LoggerKafkaConsumer - Processing TIM generated by TMC 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] INFO  TimService - Processing active TIM to add to database 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] DEBUG TimService - Extracting TIM information from region name: 'I_025A_DEC_SAT-9E70E3AD_I_test-I-12559_1' 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] INFO  DbInteractions - Creating connection pool with the following configuration: 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] INFO  DbInteractions - driverClassName: org.postgresql.Driver 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] INFO  DbInteractions - dbUrl: jdbc:postgresql://postgres:5432/postgres?user=username 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] INFO  DbInteractions - dbUsername: username 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] INFO  DbInteractions - connectionTimeout: 300000 
logger-kafka-consumer-1  | 2026-01-23 19:13:14 [main] [,] INFO  DbInteractions - maximumPoolSize: 13 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] INFO  DbInteractions - Successfully initialized connection pool 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG TimService - Extracted information - Route: 025A, Direction: I, ClientId: I, TimType: null 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG TimService - Checking if TIM already exists with packet ID: 1B28432C09D015FEC7 and timestamp: 2026-01-23 19:13:00.0 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] INFO  TimService - TIM not found in database, adding new TIM with packet ID: 1B28432C09D015FEC7 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG TimService - Successfully added TIM with ID: 161245 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG TimService - Set satellite record ID from metadata: 9E70E3AD for TIM ID: 161245 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG TimService - Using metadata start time: 2026-01-23T19:13:09.956Z for TIM ID: 161245 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG TimService - Processing SDX TIM with satellite record ID: 9E70E3AD for TIM ID: 161245 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] WARN  TimService - No active TIM holding found for SAT TIM with client ID: 'I', direction: 'I', satellite record ID: '9E70E3AD' 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG TimService - TIM ID: 161245 has indefinite duration (32000 minutes) 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG TimService - No desired end time found in active TIM holding for TIM ID: 161245, setting to null 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] INFO  TimService - Inserting new active TIM for TIM ID: 161245 (not from WYDOT application - no TimType found) 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG ActiveTimService - Inserting active_tim record with client_id = I and sat_record_id = 9E70E3AD 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] INFO  ActiveTimService - Converting 2026-01-23T19:13:09.956Z for TIM_START value 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] INFO  TimService - Successfully processed active TIM with ID: 161245 (packet ID: 1B28432C09D015FEC7) 
logger-kafka-consumer-1  | 2026-01-23 19:13:15 [main] [,] DEBUG LoggerKafkaConsumer - Successfully added active TIM to database

New active_tim record shows up as expected:

161245,,,I,2026-01-23 19:13:09.956000,,,025A,I,9E70E3AD,,11404,,,,,,,false

payneBrandon · 2026-01-26T19:50:06Z

cv-data-service-library/src/test/java/com/trihydro/library/helpers/JsonToJavaConverterTest.java

-    }
-
-    @Test
-    public void TestConvertTimPayloadJsonToJava_Path_MultipleRegions() throws IOException, URISyntaxException {


question: the multiple region capacity is an important scenario to show support for - is there a reason to remove the test?

we have other tests that tests for multiple regions so I didn't think there was a need for this test as well.

referring to these tests:

TIM-Manager/cv-data-service-library/src/test/java/com/trihydro/library/helpers/JsonToJavaConverterTest.java

Line 224 in 8437882

public void TestConvertTimTopicJsonToJava_HandlesVslContentType_MultipleRegions() throws IOException {

,

TIM-Manager/cv-data-service-library/src/test/java/com/trihydro/library/helpers/JsonToJavaConverterTest.java

Line 260 in 8437882

public void TestConvertTimTopicJsonToJava_HandlesParkingContentType_MultipleRegions() throws IOException {

, and

TIM-Manager/cv-data-service-library/src/test/java/com/trihydro/library/helpers/JsonToJavaConverterTest.java

Line 297 in 8437882

public void TestConvertTimTopicJsonToJava_HandlesConstructionContentType_MultipleRegions() throws IOException {

payneBrandon · 2026-01-26T19:50:38Z

cv-data-service-library/src/test/java/com/trihydro/library/helpers/JsonToJavaConverterTest.java

-    }
-
-    @Test
-    public void TestConvertTimPayloadJsonToJava_SpeedLimit() throws IOException, URISyntaxException {


question: speed limit is a very common TIM we need to support, is there a reason to remove the tests?

the speed limit type is very similar to the VSL type, so I figured having tests covering the VSL type would suffice.

referring to

TIM-Manager/cv-data-service-library/src/test/java/com/trihydro/library/helpers/JsonToJavaConverterTest.java

Line 206 in 8437882

public void TestConvertTimTopicJsonToJava_HandlesVslContentType() throws IOException {

, and

TIM-Manager/cv-data-service-library/src/test/java/com/trihydro/library/helpers/JsonToJavaConverterTest.java

Line 224 in 8437882

public void TestConvertTimTopicJsonToJava_HandlesVslContentType_MultipleRegions() throws IOException {

.

payneBrandon · 2026-01-26T19:55:56Z

cv-data-service-library/src/test/java/com/trihydro/library/helpers/JsonToJavaConverterTest.java

-        geometry.setDirection("1010101010101010");
-        geometry.setExtent(1);// this is an enum
-        geometry.setLaneWidth(BigDecimal.valueOf(33));
+        geometry.setDirection("F0F0");


question: direction is a HeadingSlice (which we're moving away from using in favor of directionality), what does "F0F0" represent here?

Is this just the hex representation of the heading slice? It doesnt' seem to be the same as previous direction, are we just adding new data for all the tests?

I had to add new data for all the tests since the data structure changed so much. The direction "F0F0" was what came out of the ODE.

payneBrandon · 2026-01-26T20:19:57Z

...r-kafka-consumer/src/main/java/com/trihydro/loggerkafkaconsumer/app/LoggerKafkaConsumer.java

                if (ath != null) {
                    log.debug("Found record in holding table, updating expiration");
-                    success = activeTimHoldingService.updateTimExpiration(certExpirationModel.getPacketID(), certExpirationModel.getExpirationDate());
+                    success = activeTimHoldingService.updateTimExpiration(certExpirationModel.getPacketID(), Instant.parse(certExpirationModel.getExpirationDate()));


question: will we ever see a null/empty certExpirationModel.expirationDate and what's the consequence of that?

good catch. I adjusted it so that, in the case of a null expiration date, it will query for the null value like it used to rather then throwing the null pointer exception.

payneBrandon · 2026-01-26T20:22:05Z

...umer/src/main/java/com/trihydro/loggerkafkaconsumer/app/dataConverters/TimDataConverter.java

@@ -58,7 +57,7 @@ public OdeData processTimJson(String value) {
    public OdeData translateTimJson(String value) {


suggestion: references to the OdeData now show: OdeData is a raw type. References to generic type OdeData<MetadataT,PayloadT> should be parameterized Java(16777788)

payneBrandon

couple questions, overall looking good!

…prevent null pointer exception

…traction

…OdeData object type

pmonington added 7 commits January 13, 2026 08:27

chore: update JSON traversal logic in JsonToJavaConverter to handle…

8a82cbd

… new nested `TravelerInformation` node

chore: update updateTimExpiration to use Instant for expiration d…

8dfd257

…ate management

Merge branch 'feat/add-cdot-tim-pathing' into pmonington/feat/update-…

0063cb4

…consumer-data-extraction

test(TimDataConverter): update VSL, Parking, and construction testing…

6365cd7

… data to match OdeTimJson new format

test(TimDataConverter): update testing data to align with new OdeTimJ…

e9bce98

…son data formats

chore(JsonToJavaConverter): remove deplicate and outdated test data f…

47038cb

…or TIM payload conversion

test(TimDataConverter): update test cases with revised TIM payload da…

858f979

…ta and metadata formats

pmonington requested review from dmccoystephenson, mcook42 and payneBrandon January 23, 2026 17:46

pmonington commented Jan 23, 2026

View reviewed changes

dmccoystephenson approved these changes Jan 23, 2026

View reviewed changes

payneBrandon reviewed Jan 26, 2026

View reviewed changes

payneBrandon requested changes Jan 26, 2026

View reviewed changes

pmonington added 2 commits February 2, 2026 13:40

chore: update OdeData type definitions

32956a8

chore: handle null expiration dates case in updateTimExpiration to …

e12cd72

…prevent null pointer exception

pmonington requested a review from payneBrandon February 2, 2026 22:10

pmonington added 3 commits February 4, 2026 05:27

Merge branch 'origindev' into pmonington/feat/update-consumer-data-ex…

a34123e

…traction

test(TimServiceTest): standardize usage of variable types and update …

2b2959e

…OdeData object type

test(TimServiceTest): remove unused imports and redundant casting

8437882

		@@ -58,7 +57,7 @@ public OdeData processTimJson(String value) {
		public OdeData translateTimJson(String value) {

Update Consumer Data Extraction #29

Are you sure you want to change the base?

Update Consumer Data Extraction #29

Uh oh!

Conversation

pmonington commented Jan 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

PR Details

Description

How Has This Been Tested?

Types of changes

Checklist:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dmccoystephenson left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

payneBrandon Jan 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

payneBrandon left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

pmonington commented Jan 23, 2026 •

edited

Loading

payneBrandon Jan 26, 2026 •

edited

Loading