Merged MR to configure batch size by event count: https://gitlab.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-streaming-consumer/-/merge_requests/13

trueg opened https://gitlab.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-streaming-consumer/-/merge_requests/18

Fixed timeout handling on update batch polling.

trueg opened https://gitlab.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-streaming-consumer/-/merge_requests/20

Performance optimizations

trueg opened https://gitlab.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-streaming-consumer/-/merge_requests/21

Handle entity deletion and sub-sequent addition in the same patch.

trueg merged https://gitlab.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-streaming-consumer/-/merge_requests/20

Performance optimizations

The wdqs-streaming-consumer version 0.1.2 cannot properly handle big batch sizes. In a 1h run on wdqs1029 against Qlever 0.5.47 with a batchSize of 100k and a batchTimeout of 10min we only see the following collected batches:

09:00:15.165 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 2627 events and 16541 triples into update batch in 600000 seconds.
09:10:18.791 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 26937 events and 161272 triples into update batch in 600000 seconds.
09:20:31.594 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 3281 events and 20937 triples into update batch in 600000 seconds.
09:30:36.374 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 36340 events and 236779 triples into update batch in 600042 seconds.
09:40:53.409 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 24162 events and 174920 triples into update batch in 600000 seconds.
09:51:08.119 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 5809 events and 40000 triples into update batch in 600000 seconds.

The consumer is unable to collect enough events in 10 minutes. When examining the logs we see one recurring pattern:

08:50:16.434 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 424 messages from Kafka topic.
08:50:16.863 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 500 messages from Kafka topic.
08:50:17.531 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 402 messages from Kafka topic.
09:00:15.165 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 2627 events and 16541 triples into update batch in 600000 seconds.

The post-processing of collected events from Kafka takes up to 10 minutes. Further investigation showed that this is in fact the timeout value, and that the consumer simply idles in an endless loop. This was addressed in MR 18.

The same log excerpt shows another issue: the processing of about 500 events polled from Kafka takes around 500ms. When polling about 30k events for one batch, this already sums to 4 minutes.
Profiling of the 1h run shows the worst offenders in terms of CPU cycles:

Furthermore, once we look into the logs of the 1h run of the patched consumer, we see how many more events could be pushed in the same time:

10:14:12.055 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 754921 triples into update batch in 15183 ms.
10:15:12.066 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 786375 triples into update batch in 13767 ms.
10:16:20.042 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 683455 triples into update batch in 13860 ms.
10:17:24.473 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 648099 triples into update batch in 11734 ms.
10:18:39.874 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 684077 triples into update batch in 20859 ms.
10:19:59.415 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 695236 triples into update batch in 12798 ms.
10:21:23.535 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 877790 triples into update batch in 17350 ms.
10:25:01.065 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 1054958 triples into update batch in 134435 ms.
10:26:44.546 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 754946 triples into update batch in 16970 ms.
10:28:27.225 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 803248 triples into update batch in 16263 ms.
10:30:36.842 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 681064 triples into update batch in 15671 ms.
10:32:31.053 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 675556 triples into update batch in 18975 ms.
10:34:21.832 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 569968 triples into update batch in 12280 ms.
10:36:12.904 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 684216 triples into update batch in 14491 ms.
10:38:16.334 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 783096 triples into update batch in 17437 ms.
10:40:31.672 [main] INFO  o.w.q.r.u.c.KafkaStreamConsumer - Collected a total of 100000 events and 732342 triples into update batch in 17397 ms.

Not only was the patched consumer able to push almost 3 times as many updates to Qlever, each update also contained the requested 100k events.
The post-processing time of events after being polled from Kafka was reduced from the afore mentioned 500ms per 500 events to something between 30ms and 100ms:

10:13:58.749 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 463 messages from Kafka topic.
10:13:58.845 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 500 messages from Kafka topic.
10:13:58.916 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 196 messages from Kafka topic.
10:13:58.944 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 500 messages from Kafka topic.
10:13:58.986 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 203 messages from Kafka topic.
10:13:59.015 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 500 messages from Kafka topic.
10:13:59.050 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 273 messages from Kafka topic.
10:13:59.083 [main] DEBUG o.w.q.r.u.c.KafkaStreamConsumer - Polled 500 messages from Kafka topic.

While the performance improvements presented above are substantial, we sill want to verify two things:

1. With the introduction of Qlever' SortedSequence we expect a much more stable update timing and memory usage. Thus, the above tests should be run again with the newer version of Qlever
2. Performance is one thing, correctness an even more important one. While most of the performance fixes are trivial, not all are. These need to be verified.

Performance

To have a good grasp on the performance improvements we should ideally first create a baseline of handled updates in a certain amount of time, depending on batchSize and batchTimeout.
Then we run the same data on the same state of Qlever through the patched version of the consumer, again with different values for batchSize and batchTimeout.
As metrics we should record:

• The total number of processed events
• The development of the number of processed events per time window (Using Qlever 0.5.47 update times gradually increased, meaning that the number of processed events gradually shrinked)
• The memory usage development of Qlever (See also T431271)

Correctness

In order to verify correctness (or at least parity with previous iterations of the consumer) we need to compare update results of the consumer v0.1.2 with the patched version. While unit tests can provide some verification, they are not exhaustive.
A possible approach would be the introduction of another parameter to the consumer, which allows to limit the number of events it processes.
Then we could run a fixed number of updates through both versions of the consumer, letting them apply the changes to an empty Qlever database.
Finally we compare the exports of both instances of Qlever. If they match, then the performance patches are correct.

trueg merged https://gitlab.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-streaming-consumer/-/merge_requests/18

Fixed timeout handling on update batch polling.

In T430782#12108321, @trueg wrote:

Thanks for this @trueg. Plan looks solid!
[...]

I do miss a definition of "target": that is, our budget for one week of backfills. WDQS will be depooled while Kafka catches up. How much slack do we allow? I would cap it at one hour max for 6MM offsets, which is well within the throughput we measured (though not really reported) with both the Python and Golang updaters (although they have simplified logic compared to the Java one).

One dimension I would add is which patches (behaviour) we are testing. I see three cases:

Control: qlever's own updater on about 6MM offsets. This IMHO should give a cleaner definition of target.
Baseline: current java updater with timeout fixes, small batches of 250 triples (not events). We know performance is good at steady state (compared to Blazegraph), but we need a better estimate of how fast backfills go. Blazegraph backfills 24 hours of backlog in about 45 minutes.
Test A: what are the throughput / memory allocation patterns with the patches that are safe, with varying batch size (incrementing, say on a log scale 1k, 5k, 10k, 50k, 100k, 500k)? Do they meet our target (~6MM offset backfilled in < 1 hour)? Where is the ROI point? Where do batch sizes start to starve?

At this stage, I would leave further optimization and correctness work aside, unless we see that Test A does not meet our criteria.

Does this track with you?

How / where do you plan to run the experiments? And how do you see these tests fitting in with T431271: [SPIKE] understand, document and tune QLever memory behaviour when real time updates are applied.?

This said, I to think we need a better test bed for measuring correctness of the updated and tracking possible regression on qlever (we work under the assumption that updates are idempotent). I'd scope and plan this work already this quarter, but maybe (depending on test) as essential an not KR (cc @BTracy-WMF @GTurkington-WMF re how we split work).

I do not see much values in running the qlever control updater ourselves, because its logic is different. If we want a baseline, we can simply ask Hannah about their metrics.
The only real "target" we have is the previous measurement of roughly 24 hours of backfill in 45 minutes.
That leaves us with the current consumer with small batches as a baseline, and several runs with increasing batch sizes.
As for T431271 - as I mentioned it would make sense to also monitor the resource usage of Qlever during all these tests.
I was thinking of running the tests on wdqs1029 or wdqs1030.

In T430782#12122602, @trueg wrote:

I do not see much values in running the qlever control updater ourselves, because its logic is different. If we want a baseline, we can simply ask Hannah about their metrics.

The investigation started with observing their consumer's behaviour, and I just wanted to document its perf at 100k events somewhere (no need for extensive tests). This is interesting for folks that run their own wikidata qlever instances and might want to understand real-time update perf. We'll def chat with Hannah, but they run on different hardware config. It's good to compare notes :).

As for T431271 - as I mentioned it would make sense to also monitor the resource usage of Qlever during all these tests.
I was thinking of running the tests on wdqs1029 or wdqs1030.

Those hosts sound good, but beware they might have worse iops than the k8s nodes. Should be no issue in practice, as long
as you compare runs on the same hw config.

As discussed at collab; we have 0.50.x deployed on k8s, in parallel with the more precise tests you are running on
the eqiad test nodes, we can deploy the consumer with timeout fixes. We can see nice metrics in grafana right away: https://grafana.wikimedia.org/d/gmpdvcr/wdqs-v2-wip?orgId=1&from=now-24h&to=now&timezone=utc

Performance Experiments

The following experiments were run with

• on wfdqs1029.eqiad.wmnet
• Qlever v0.5.50 dockerized via wdqs-qlever
• with a consistent Kafka offset of 303244833 (reset for each run)
• The consumer was run for exactly 30 minutes on each run
• Most runs were performed with wdqs-streaming-consumer v0.2.0 which contains some timeout fixes as compared to the one in WDQSv1. The runs tagged patched were run with an additional optimization patch (as of this writing the correctness of that patch has not been confirmed yet. That is future work which is valuable as these tests will show).
• Qlever memory consumption was monitored via pmap $QLEVER_PID|grep total|awk '{print $NF}'
• Total events processed was calculated based on the initial Kafka offset (303244833) and the current offset after the consumer terminated.
• All other consumer metrics were extracted from the consumer debug logs using some crude bash-foo (Example for average events per batch: grep Collected consumer.log |cut -d " " -f11|awk '{ total += $1; count++ } END { print total/count }').

Docker Runs

Both Qlever and the consumer were run on docker images (Qlever v0.5.50 and a custom build of wdqs-streamin-consumer from v0.2.0).

$ docker run -d --rm -v /srv/wdqs/qlever/index:/srv/data --network host docker-registry.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-qlever:v0.5.50_2026-07-15T07-49-58Z_cba194e

$ timeout 30m docker run --rm -e "SPARQL_ENDPOINT=http://localhost:7001/sparql" -e "KAFKA_GROUP=wdqs1029test" -e "KAFKA_TOPIC=eqiad.rdf-streaming-updater.mutation-main" -e "KAFKA_BROKERS=kafka-main1006.eqiad.wmnet:9092" -e "LOG_LEVEL=DEBUG" -e "BATCH_SIZE=1000" -e "BUFFERED_INPUT_MESSAGES=250" -e "BATCH_TIMEOUT=3" -e "HEAP_SIZE=2g" --network=host wdqs-streaming-consumer

The Test Configurations

Test Results

Observations

• Qlever's memory increase is consistent over all experiments, increasing at a steady rate with each inserted triple.
• Qlever prefers large update batch sizes, bigger batches mean more throughput
• The batch time is only relevant at higher batch sizes, providing a higher timeout for backfull as compared to normal operation is only really relevant for final optimizations (compare T431577)
• The patched version's performance is considerably better at high batch sizes, meaning verification of its correctness is worth the time.
• Considering a rough estimate of WDQSv1's back-fill performance (24h worth of updates ingested in 45min - with an estimate of 5 Wikidata update events per second, this gives an events per minute value of 9600), even unpatched and with a low batch time, performance is much better already.
• Qlever's --persist-updates does not make much of a difference it seems.

With wdqs-streaming-consumer 0.2.0 this optimization task is finished. Additional work to verify the correctness of the mentioned performance patch will happen in T432437.