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

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 22.46835% with 245 lines in your changes missing coverage. Please review.
✅ Project coverage is 72.03%. Comparing base (1278304) to head (243923e).
⚠️ Report is 220 commits behind head on dev-2.x.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
...iri/updater/mqtt/MqttEstimatedTimetableSource.java	0.00%	211 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%   72.03%   -0.12%     
- Complexity     19772    19918     +146     
=============================================
  Files           2151     2166      +15     
  Lines          79955    80535     +580     
  Branches        8058     8111      +53     
=============================================
+ Hits           57687    58015     +328     
- Misses         19423    19662     +239     
- Partials        2845     2858      +13

☔ 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.

Thanks for the explanation! If there is network congestion or some other circumstances I guess this could still happen in theory. In practice it will be uncommon and won't have very bad consequences. If you really wanted to protect against this eventuality you could consume any remaining messages from the primingMessageQueue after you set primed = true. But it is up to you if you think it's worth it since it's your updater.

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)

vpaturet · 2025-10-22T07:29:48Z

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

+
+  @Override
+  public void teardown() {
+    client.disconnect();


should you also shutdown the executors while tearing down?

@Override public void teardown() { liveExecutor.shutdownNow(); primingExecutor.shutdownNow(); client.disconnect(); }

vpaturet · 2025-10-22T07:36:52Z

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

+
+  public MqttEstimatedTimetableSource(MqttSiriETUpdaterParameters parameters) {
+    this.parameters = parameters;
+    this.primingExecutor = Executors.newFixedThreadPool(parameters.numberOfPrimingWorkers());


Configuring the naming of the threads may help when debugging, see
https://github.com/OpenTripPlanner/OpenTripPlanner/blob/5198c5ffef3d2db4f78f36778501333d02ec8444/application/src/main/java/org/opentripplanner/updater/GraphUpdaterManager.java#L79

vpaturet · 2025-10-22T08:01:28Z

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

+            continue;
+          }
+          var serviceDelivery = optionalServiceDelivery.get();
+          serviceDeliveryConsumer.apply(serviceDelivery);


Here the service delivery is sent to the graph writer thread without waiting for it to be applied.
In the priming logic of the GooglePubSub updater, we have a blocking wait: future.get();
https://github.com/OpenTripPlanner/OpenTripPlanner/blob/5198c5ffef3d2db4f78f36778501333d02ec8444/application/src/main/java/org/opentripplanner/updater/GraphUpdaterManager.java#L79
so that we can claim that the updater is primed when all initial data is applied to the transit model.

If you want to make sure that all priming runners are done AND all priming messages are applied to the transit model, you would have to collect all these futures and wait for them to be completed.

vpaturet · 2025-10-22T08:19:19Z

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

+      .addDisconnectedListener(this::onDisconnect)
+      .buildAsync();
+
+    client.connectWith().keepAlive(30).cleanStart(false).send().join();


Just out of curiosity: how does cleanStart=false work together with a random client ID and automatic reconnect?

as far as I understand it, the client will reconnect with the same client ID as before and then only receive the retained messages that it didn't yet get. So you avoid processing the same message twice.

vpaturet · 2025-10-22T08:20:37Z

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

+  private Mqtt5AsyncClient client;
+  private Function<ServiceDelivery, Future<?>> serviceDeliveryConsumer;
+
+  private final BlockingQueue<byte[]> liveMessageQueue = new LinkedBlockingQueue<>();


The queues are unbounded, is there a risk of OOM?

For the priming queue it could happen if there are more retained messages in the broker than can be hold in memory. I think this is a very specific number for different deployments. I could make the upper bound configurable, something like maxNumberOfRetainedMessages?

For the live queue the rate of the incoming messages would need to be higher than the processing rate. At the moment we are factor 10 to 20 away from that, even at absolute peak times. I could set an upper limit and then log warnings when the limit is reached. The queue needs to be able to hold all live messages that come in during priming however, so again, it's very deployment specific. I don't want to end up in configuration hell, so I am not quite sure. What do you think?

I'm not sure if this is a real problem. You could just release a first version of this updater without any additional configuration, and if needed come back to it later to put in place some back-pressure/throttling.

Ok great. We will have monitoring on this, so I don't think it will be an issue for us. If it does become one, we can come back to this.

vpaturet · 2025-10-22T08:23:17Z

...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());


offer is always true on unbounded queues.

True. I mainly did it that to get rid of the IntelliJ warning of an unused return value tbh. It would get relevant if the queue gets bounded in the future however.

vpaturet · 2025-10-22T08:40:10Z

...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());


You submit a single task that runs "forever" and catches only InterruptedException. If that task throws an exception, the single thread will die and the pool will create a new one. But no task will be submitted automatically to this new thread and the updater will stop processing updates.
You should probably make LiveRunner.run() resilient to other types of exception.

Good point, thanks!

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

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

vpaturet requested changes Oct 22, 2025

View reviewed changes

jessicaKoehnke added 3 commits October 22, 2025 14:56

rename updater

d264b61

review comments

a782cf4

improved error handling

243923e

jessicaKoehnke requested review from habrahamsson-skanetrafiken and vpaturet October 24, 2025 12:23

jessicaKoehnke assigned vpaturet and habrahamsson-skanetrafiken Oct 24, 2025

SIRI-ET updater via MQTT #6851

Are you sure you want to change the base?

SIRI-ET updater via MQTT #6851

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 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

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!

habrahamsson-skanetrafiken Oct 22, 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!

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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

jessicaKoehnke commented Sep 9, 2025 •

edited

Loading

codecov bot commented Oct 16, 2025 •

edited

Loading

leonardehrenfried Oct 16, 2025 •

edited

Loading

habrahamsson-skanetrafiken Oct 22, 2025 •

edited

Loading