SIRI-ET updater via MQTT #6851

jessicaKoehnke · 2025-09-09T16:58:49Z

Summary

This PR adds the support to import SIRI-ET realtime updates via MQTT. It's implemented as a sandbox feature. When no SIRI MQTT updater is configured in the router-config.json the sandbox code is not executed.

Issue

Closes #6639

Unit tests

Without an MQTT this is hard to test. Existing tests all run successfully and changes to non-sandbox code are minimal.

Documentation

Documentation has been updated.

Changelog

Added to changelog

Bumping the serialization version id

Not necessary.

…ogging the password

# Conflicts: # application/src/main/java/org/opentripplanner/updater/trip/siri/ModifiedTripBuilder.java

This reverts commit e647b5c.

jessicaKoehnke · 2025-09-24T14:28:45Z

I've done some performance tests.
Max receiving rate (Paho): 1900 messages per second
Max receiving rate (HiveMQ): 4500 messages per second

Bottleneck is however the XML Parsing with about 1000 messages per second on my local machine and 420 messages per second on our cloud machine.

The reason to change the library to HiveMQ would be that it's a newer library that still receives updates. It is faster, but right now we would not be able to profit from that.

jessicaKoehnke · 2025-09-25T14:01:03Z

We decided in the dev meeting to go forward with the HiveMq library. I will check, if all current MQTT implementations work with HiveMq, and then substitute Paho with HiveMQ.

codecov · 2025-10-16T09:29:30Z

Codecov Report

❌ Patch coverage is 24.14966% with 223 lines in your changes missing coverage. Please review.
✅ Project coverage is 71.98%. Comparing base (1278304) to head (4ba95fe).
⚠️ Report is 20 commits behind head on dev-2.x.

Files with missing lines	Patch %	Lines
...iri/updater/mqtt/MqttEstimatedTimetableSource.java	0.00%	189 Missing ⚠️
...anner/ext/siri/updater/mqtt/SiriETMqttUpdater.java	0.00%	17 Missing ⚠️
...siri/updater/mqtt/MqttSiriETUpdaterParameters.java	52.00%	12 Missing ⚠️
...planner/updater/configure/UpdaterConfigurator.java	0.00%	2 Missing and 1 partial ⚠️
...routerconfig/updaters/SiriETMqttUpdaterConfig.java	98.21%	1 Missing ⚠️
...ip/siri/updater/AsyncEstimatedTimetableSource.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             dev-2.x    #6851      +/-   ##
=============================================
- Coverage      72.14%   71.98%   -0.17%     
- Complexity     19772    19789      +17     
=============================================
  Files           2151     2156       +5     
  Lines          79955    80274     +319     
  Branches        8058     8073      +15     
=============================================
+ Hits           57687    57786      +99     
- Misses         19423    19643     +220     
  Partials        2845     2845

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

leonardehrenfried · 2025-10-16T17:25:40Z

doc/user/sandbox/siri/SiriMqttUpdater.md

+{
+  "updaters" : [
+    {
+      "type" : "siri-et-mqtt-updater",


I'm not a reviewer but I just want to drop this: I find the suffix -updater in these type values strange because they are all updaters. In the ones I have added I didn't use it.

habrahamsson-skanetrafiken

I added some comments you might want to consider.

habrahamsson-skanetrafiken · 2025-10-17T12:12:35Z

...ion/src/ext/java/org/opentripplanner/ext/siri/updater/mqtt/MqttEstimatedTimetableSource.java

+      parameters.user() == null ||
+      parameters.user().isBlank() ||
+      parameters.password() == null ||
+      parameters.password().isBlank()


There is a utility method for this at StringUtils.hasValue() that you can use if you want.

habrahamsson-skanetrafiken · 2025-10-20T08:23:35Z

...ion/src/ext/java/org/opentripplanner/ext/siri/updater/mqtt/MqttEstimatedTimetableSource.java

+      primingFutures.add(f);
+    }
+    LOG.info("Started {} priming workers", parameters.numberOfPrimingWorkers());
+    liveExecutor.submit(new LiveRunner());


Do you want to run this in parallel with your priming? This could cause a live ET message be overwritten by a retained message. In our implementation we apply all the full history before we start consuming live messages.

good point, thanks!

habrahamsson-skanetrafiken · 2025-10-20T08:28:09Z

...ion/src/ext/java/org/opentripplanner/ext/siri/updater/mqtt/MqttEstimatedTimetableSource.java

+  private void onMessage(Mqtt5Publish message) {
+    boolean offer;
+    if (message.isRetain() && !primed) {
+      offer = primingMessageQueue.offer(message.getPayloadAsBytes());


I think there could be a race condition here if a message is put on the primingMessageQueue at the same time as the last RetainRunner times out. Then this message won't be processed. That might not be a catastrophe, but worth to consider.

When a new client connects to the broker, only old messages will be marked as retained for that client. All messages that are processed immediately after they are sent to the broker will never have the retained flag. So the idea is that the fixed amount of retained messages get processed, and when the runners idle long enough (maxPrimingIdleTime), then it is assumed that all retained messages are processed so the runners can get shut down.

habrahamsson-skanetrafiken · 2025-10-20T08:36:08Z

...ion/src/ext/java/org/opentripplanner/ext/siri/updater/mqtt/MqttEstimatedTimetableSource.java

+    List<CompletableFuture<Void>> primingFutures = new ArrayList<>();
+
+    for (int i = 0; i < parameters.numberOfPrimingWorkers(); i++) {
+      CompletableFuture<Void> f = CompletableFuture.runAsync(new RetainRunner(i), primingExecutor);


The thing to consider about parallelizing your ET processing is that you might apply your ET messages out of order. If you have multiple messages for the same trip (for example a time update followed by a cancellation) then you will get a different state depending on the order that these are applied. If you don't have duplicate messages for the same trip in your retained messages this won't be a problem i think.

We only have one message per trip, so the order shouldn't matter:

One mqtt topic per trip, in every topic is always only one message

the only way to get 2 messages for a trip is if a new live message comes in (which is exactly the problem you mentioned in your other comment about the LiveRunner starting too early)

jessicaKoehnke and others added 30 commits May 16, 2025 18:10

establish mqtt connection and do message handling

002cfdc

establish mqtt connection and do message handling

2e42654

add configuration

78e8a95

Merge remote-tracking branch 'origin/dev-2.x' into siri-via-mqtt

09aad09

Merge remote-tracking branch 'origin/dev-2.x' into siri-via-mqtt

c009481

decrease number of realtime received log messages

18134b5

decrease number of realtime received log messages

501e03d

decrease number of realtime received log messages

48ef869

improve logging for failed realtime updates

6d4476d

enable debug logging for siri realtime updates

39be570

separate user and password from url parameter to avoid accidentally l…

dc23d4d

…ogging the password

set default values for user and password to null

7a04ddd

disable debug log

d69630f

try improving performance

216dc13

delete log

fcd9d03

revert performance optimization tries

8cedf63

Merge branch 'dev-2.x' into siri-via-mqtt

5967bbb

# Conflicts: # application/src/main/java/org/opentripplanner/updater/trip/siri/ModifiedTripBuilder.java

add cleanup method to fix flaky test

1fc9378

Merge branch 'dev-2.x' into siri-via-mqtt

294449f

Merge branch 'dev-2.x' into siri-via-mqtt

4405066

use virtual threads for faster initial siri message consumption

25be222

Merge remote-tracking branch 'upstream/dev-2.x' into siri-via-mqtt

bfe1722

disable flappy test

09e68dd

improve concurrent message handling

e406a7f

improve concurrent message handling

8e56921

adjust max limit of virtual threads

c07fbd5

improve for initial data delivery

23ab859

adjust initial data delivery handling

89bd333

adjust initial data delivery handling

2b04653

improve initial data delivery performance

ae53733

jessicaKoehnke added 9 commits September 2, 2025 17:24

handle mqtt connection without auth

ba6f633

make number of priming threads configurable, improve logging

cb9579d

make historicMessageAgeThreshold configurable

13b4945

use retained flag instead of message age

f457729

submit tasks to executor instead of multiple runners

e647b5c

Revert "submit tasks to executor instead of multiple runners"

b7a05b2

This reverts commit e647b5c.

add metrics

d141fb5

add metrics

f53e414

update paho implementation with improved logging for performance testing

7d8356d

t2gran added this to the 2.9 (next release) milestone Sep 10, 2025

jessicaKoehnke added !Improvement A functional improvement or micro feature +Sandbox This will be implemented as a Sandbox feature +Real-Time The issue/PR is related to RealTime updates HBT HBT (Hamburg) roadmap labels Sep 24, 2025

lmmhbt mentioned this pull request Sep 26, 2025

Replace MQTT library #6906

Merged

jessicaKoehnke added 4 commits October 15, 2025 15:34

Merge remote-tracking branch 'upstream/dev-2.x' into siri-via-mqtt

bbcc35e

delete paho implementation and add documentation

d2fb3e9

clean up

567363e

cleanup

4ba95fe

jessicaKoehnke marked this pull request as ready for review October 16, 2025 09:34

jessicaKoehnke requested a review from a team as a code owner October 16, 2025 09:34

habrahamsson-skanetrafiken requested review from habrahamsson-skanetrafiken and vpaturet October 16, 2025 13:51

leonardehrenfried changed the title ~~Add Siri via MQTT Realtime Updater~~ Add SIRI-ET via MQTT updater Oct 16, 2025

leonardehrenfried reviewed Oct 16, 2025

View reviewed changes

leonardehrenfried changed the title ~~Add SIRI-ET via MQTT updater~~ SIRI-ET updater via MQTT Oct 16, 2025

habrahamsson-skanetrafiken reviewed Oct 20, 2025

View reviewed changes

SIRI-ET updater via MQTT #6851

Are you sure you want to change the base?

SIRI-ET updater via MQTT #6851

Uh oh!

Conversation

jessicaKoehnke commented Sep 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Issue

Unit tests

Documentation

Changelog

Bumping the serialization version id

Uh oh!

jessicaKoehnke commented Sep 24, 2025

Uh oh!

jessicaKoehnke commented Sep 25, 2025

Uh oh!

codecov bot commented Oct 16, 2025

Codecov Report

Uh oh!

leonardehrenfried Oct 16, 2025 • 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!

habrahamsson-skanetrafiken left a comment

Choose a reason for hiding this comment

Uh oh!

habrahamsson-skanetrafiken Oct 17, 2025

Choose a reason for hiding this comment

Uh oh!

habrahamsson-skanetrafiken Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

jessicaKoehnke Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

habrahamsson-skanetrafiken Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

jessicaKoehnke Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

habrahamsson-skanetrafiken Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

jessicaKoehnke Oct 20, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

jessicaKoehnke commented Sep 9, 2025 •

edited

Loading

leonardehrenfried Oct 16, 2025 •

edited

Loading