dropped messages in TUMBLINGWINDOW from FILE source

[cormac-raftery-gea]

Hi,
I am experiencing non deterministic behaviour in my eKuiper source & Rule.
My source is a json .LINES file that I upload to eKuiper. The first 3 lines of the file are as follows

{"EnqueuedTimeUtc":"2025-06-04T08:53:30.7510000Z","Properties":{},"SystemProperties":{"correlationId":"825fc18b-97fe-4310-940d-979a0a6c3f32","connectionDeviceId":"sim000001","connectionAuthMethod":"{"scope":"hub","type":"sas","issuer":"iothub"}","connectionDeviceGenerationId":"638284825497147007","contentType":"application/json","contentEncoding":"utf-8","enqueuedTime":"2025-06-04T08:53:30.7510000Z"},"Body":{"deviceId": "sim000001", "time": "2025-06-04T08:53:36.6120938Z", "counter": 81}}
{"EnqueuedTimeUtc":"2025-06-04T08:53:31.7510000Z","Properties":{},"SystemProperties":{"correlationId":"93fd00a8-19c3-4810-bd6d-b50f57e5d20a","connectionDeviceId":"sim000001","connectionAuthMethod":"{"scope":"hub","type":"sas","issuer":"iothub"}","connectionDeviceGenerationId":"638284825497147007","contentType":"application/json","contentEncoding":"utf-8","enqueuedTime":"2025-06-04T08:53:31.7510000Z"},"Body":{"deviceId": "sim000001", "time": "2025-06-04T08:53:37.6125181Z", "counter": 82}}
{"EnqueuedTimeUtc":"2025-06-04T08:53:32.7510000Z","Properties":{},"SystemProperties":{"correlationId":"a0cf87de-3a49-459f-a2b4-79afbd2cec11","connectionDeviceId":"sim000001","connectionAuthMethod":"{"scope":"hub","type":"sas","issuer":"iothub"}","connectionDeviceGenerationId":"638284825497147007","contentType":"application/json","contentEncoding":"utf-8","enqueuedTime":"2025-06-04T08:53:32.7510000Z"},"Body":{"deviceId": "sim000001", "time": "2025-06-04T08:53:38.6120443Z", "counter": 83}}
....

There are 120 lines in the file. The lines represent test iot data, 1 event sent every second for 120 seconds,
I intended using the EnqueuedTimeUtc as the event time.
The earliest EnqueuedTimeUtc is 2025-06-04T08:53:30.7510000Z and the latest is 2025-06-04T08:55:29.7420000Z.
With a tumbling window of 10 seconds over events that span 120 seconds, is I would have expected to 12 windows.
My query is a simple count(*) query:

select count(*) from playback group by SystemProperties.connectionDeviceId, TUMBLINGWINDOW(ss, 10)

I expected 12 events with a payload of 10.

What actually happens is

1. I often get an inconsistent count
2. I don't get 12 windowed events

I few other things to add:

• I have setup the Timestamp field to EnqueuedTimeUtc in the stream, which is called playback. The EnqueuedTimeUtc is being respected as it "often" does place them in the right window.

{
"Name": "playback",
"StreamFields": null,
"Options": {
"datasource": "custom2.line",
"format": "json",
"confKey": "playbackfile",
"type": "file",
"timestamp": "EnqueuedTimeUtc"
},
"StreamType": 0,
"Statement": null
}

Here is the rule, and I have ensured that isEventTime is true.

{
"triggered": true,
"id": "playback",
"name": "playback",
"sql": "select count() from playback\r\ngroup by SystemProperties.connectionDeviceId, TUMBLINGWINDOW(ss, 10)",
"actions": [
{
"mqtt": {
"bufferLength": 1024,
"enableCache": false,
"format": "json",
"insecureSkipVerify": false,
"omitIfEmpty": false,
"password": "!!",
"protocolVersion": "3.1.1",
"qos": 0,
"resourceId": "testOutput",
"retained": false,
"runAsync": false,
"sendSingle": true,
"server": "tcp://emqx:1883",
"topic": "testOutput",
"username": ""
}
}
],
"options": {
"debug": false,
"isEventTime": true,
"lateTolerance": "2s",
"concurrency": 1,
"bufferLength": 1024,
"sendMetaToSink": false,
"sendNilField": false,
"sendError": false,
"checkpointInterval": "5m0s",
"restartStrategy": {
"delay": "1s",
"multiplier": 2,
"maxDelay": "30s",
"jitterFactor": 0.1
}
}
}

I have tried to play around with the lateTolerance setting but it does not have any effect. Besides, the lines in the file are already sorted in ascending order, each 1 second apart.

When I look at the topology flow I see the culprit seems to be the op_4_watermark function (screenshot below). This time it goes from 120 message to 40. If I run it numerous times this number will differ each time. This number does match up with the events I get out in the end. So, I presume that the events are being dropped. I just don't know why.

Here is the zip of the input file that contains 120 messages
custom2.zip

4b7ce00c8)

Hope someone can shed some light on this.

[Yisaer]

Hi @cormac-raftery-gea

Please provide the content of the key playbackfile so that we can reproduce your issue.

[cormac-raftery-gea]

Hi Yisaer,
Thanks for quick reply. I included the original playback file in custom2.zip link above. Could you not access it?

[Yisaer]

Hi @cormac-raftery-gea

Thanks for reporting. The issue is caused by the bug when parsing the timestamp(EnqueuedTimeUtc) to the wrong value.

The bug issue is fixed in #3735.

For your current ekuiper version, please make the length of fraction number in your timestamp as 3/6/9 to avoid the bug.

For example, making "EnqueuedTimeUtc":"2025-06-04T08:53:32.7510000Z" as "EnqueuedTimeUtc":"2025-06-04T08:53:32.751000000Z" or "EnqueuedTimeUtc":"2025-06-04T08:53:32.751000Z" which let the fraction number become from 7 to 6/9