Eliaaazzz commented on PR #39461: URL: https://github.com/apache/beam/pull/39461#issuecomment-5299500123
Good catch, and it was worse than an edge case. Fixed in `7669d4abbe9` and re-validated below. **The job is running**, so you can inspect it live. ### The bug The cursor keyed on the path alone: ```python output_key_fn=(_file_path_and_mtime_key if self.match_upd else _file_path_key), ``` So an update was matched or not depending on whether the cursor had happened to retire the key yet. While the key was retained the update was dropped, which is what you saw. Once a newer file moved the cursor past it, the same update went through. The outcome depended on cursor timing rather than on anything the user asked for. ### The fix The cursor now keys on the path and the last-modified time, so an update is a new key and is always matched. It also removes the weakness I had documented as the cost of bounding the state. A retired key can no longer produce a duplicate: the only file that would recreate key `(path, m)` carries last-modified time `m`, and `m` is below the retention floor precisely when that key was retired, so the same mark that retired it also skips the file. Retirement and the floor use one comparison, so they cannot disagree. What remains is the intended limitation: a file whose last-modified time is already below the floor when it first appears is skipped, which is the backfill and copy-preserving-mtime case, and is what `allowed_lateness` widens. One consequence worth your call: `timestamp_cursor=True` now implies the `match_updated_files` key, so `match_updated_files=False` no longer changes behaviour under the cursor. I documented that on the option rather than raising on the combination. Happy to reject it loudly instead if you would prefer. ### Re-validation Job: [`2026-08-14_16_59_38-13437278132120143817`](https://console.cloud.google.com/dataflow/jobs/us-east1/2026-08-14_16_59_38-13437278132120143817?project=apache-beam-testing) Markers: [open in Cloud Logging](https://console.cloud.google.com/logs/query;query=resource.labels.job_id%3D%222026-08-14_16_59_38-13437278132120143817%22%0AjsonPayload.message%3A%22WATCH_EMIT%22?project=apache-beam-testing), or paste ``` resource.labels.job_id="2026-08-14_16_59_38-13437278132120143817" jsonPayload.message:"WATCH_EMIT" ``` Watched prefix `gs://io-performance-temp/watch-fix-0815-095913/`, still polled every 15s, so dropping or overwriting a file there shows all three branches reacting within a poll or two. Three modes run as three roots in one pipeline, each with its own `Watch` SDF and restriction, on one worker and one clock. Project `apache-beam-testing`, `us-east1`, Runner v2, streaming, poll interval 15s, one worker. SDK `2.77.0.dev` from commit `7669d4abbe9`, clean tree, sdist sha256 `808a0667972f761ec907812cd0ab754cdc73b81ea27184d5ab6b443f9b374580`. Same sequence as before. Only `f.txt` is seeded so the cursor sits on its last-modified time and its key stays retained. It is overwritten once while nothing newer exists, then `sentinelA.txt` advances the cursor and retires the key, then it is overwritten again, then `sentinelB.txt`. | step | defaults | updated | cursor | | --- | --- | --- | --- | | seed `f.txt` | match | match | match | | overwrite 1, key retained | no match | match | **match** | | `sentinelA.txt` | match | match | match | | overwrite 2, key retired | no match | match | **match** | | `sentinelB.txt` | match | match | match | Totals: `f.txt` 1 / 3 / 3, `sentinelA.txt` 1 / 1 / 1, `sentinelB.txt` 1 / 1 / 1. The cursor column now equals the `match_updated_files` column, which is the point: the cursor changes what the state costs, not which files match. Both sentinels are matched after their overwrite, so the `defaults` non-matches come from deduplication and not from a branch that stopped polling. Branch spread per file was 218ms, 103ms, 182ms, 25ms and 280ms. ``` --- defaults --- WATCH_EMIT plain 1786752410976 f.txt Timestamp(1786752410.468923) 1786751959.742 WATCH_EMIT plain 1786753058345 sentinelA.txt Timestamp(1786753058.181599) 1786753044.357 WATCH_EMIT plain 1786753398308 sentinelB.txt Timestamp(1786753398.175815) 1786753396.126 --- updated --- WATCH_EMIT updated 1786752410979 f.txt Timestamp(1786752410.676979) 1786751959.742 WATCH_EMIT updated 1786752993941 f.txt Timestamp(1786752993.611642) 1786752978.849 WATCH_EMIT updated 1786753058419 sentinelA.txt Timestamp(1786753058.174470) 1786753044.357 WATCH_EMIT updated 1786753334718 f.txt Timestamp(1786753334.489870) 1786753330.68 WATCH_EMIT updated 1786753398579 sentinelB.txt Timestamp(1786753398.365537) 1786753396.126 --- cursor --- WATCH_EMIT cursor 1786752410761 f.txt Timestamp(1786751959.742000) 1786751959.742 WATCH_EMIT cursor 1786752993838 f.txt Timestamp(1786752978.849000) 1786752978.849 WATCH_EMIT cursor 1786753058237 sentinelA.txt Timestamp(1786753044.357000) 1786753044.357 WATCH_EMIT cursor 1786753334693 f.txt Timestamp(1786753330.680000) 1786753330.68 WATCH_EMIT cursor 1786753398588 sentinelB.txt Timestamp(1786753396.126000) 1786753396.126 ``` Defaults and `match_updated_files` stamp a match with the poll time; the cursor stamps it with the last-modified time, visible where `event_time` equals `mtime`. Unit coverage: `test_timestamp_cursor_emits_an_updated_file` replaces the test that pinned the old behaviour. It fails on the previous key function with one match instead of two. `fileio_test.py` and `watch_test.py` are 97 passed, 2 skipped. As before, each marker is a `DoFn` log line with no durable sink, so these are processing counts rather than an exactly-once proof, and nothing here exercises restart or pipeline update. The earlier job showing the old behaviour is cancelled so it cannot be mistaken for current. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
