Eliaaazzz commented on PR #39461: URL: https://github.com/apache/beam/pull/39461#issuecomment-5291189876
Ran the Dataflow integration check you asked for, with three streaming jobs watching one GCS prefix at the same time. | mode | options | job | | --- | --- | --- | | plain | defaults | [`2026-08-13_09_44_09-493410287874925296`](https://console.cloud.google.com/dataflow/jobs/us-east1/2026-08-13_09_44_09-493410287874925296?project=apache-beam-testing) | | updated | `match_updated_files=True` | [`2026-08-13_09_44_27-14456010319325947680`](https://console.cloud.google.com/dataflow/jobs/us-east1/2026-08-13_09_44_27-14456010319325947680?project=apache-beam-testing) | | cursor | `timestamp_cursor=True` | [`2026-08-13_09_44_42-3459853442344162764`](https://console.cloud.google.com/dataflow/jobs/us-east1/2026-08-13_09_44_42-3459853442344162764?project=apache-beam-testing) | Project `apache-beam-testing`, region `us-east1`, Runner v2, streaming, poll interval 15s, one worker per job. SDK `2.77.0.dev` built from this branch and passed to every job with `--sdk_location`: source commit `626ee74c97ee9c225edd6463cfc56c5e5760bfc8` on a clean tree, sdist sha256 `a0c524840e2057ee6c5d087fbaad63e2eb125679ecf9a79f8d4f97d9fb2dded8`. ### Design I ran a first version that seeded two files, added two more, then overwrote one. It reproduced the expected 1/2/2 counts, but it added the newer files *before* the overwrite, so the cursor job's second match was consistent with key retirement and also with several other explanations. It also had no evidence that the plain job was still polling after the overwrite, so its non-match was not a clean negative. This run is built so the outcome isolates retirement. Only `f.txt` is seeded, so the cursor lands exactly on its last-modified time and its key stays retained rather than being retired by a newer file. | UTC | step | purpose | | --- | --- | --- | | 16:43:42 | seed `f.txt` | the only file, so its key stays alive | | 17:09:17 | all three jobs have matched `f.txt` | | | 17:09:42 | overwrite `f.txt` | key still retained, cursor must not match it again | | 17:10:47 | add `sentinelA.txt` | every job must match it, proving each polled past the overwrite, and it advances the cursor so `f.txt`'s key is retired | | 17:30:04 | overwrite `f.txt` again | key now retired, cursor must match it again | | 17:31:09 | add `sentinelB.txt` | every job must match it, proving each polled past the second overwrite | The two overwrites are the same operation on the same path. The only thing that differs is whether the key was still retained. ### Result | step | plain | updated | cursor | | --- | --- | --- | --- | | seed `f.txt` | match | match | match | | overwrite 1, key retained | no match | match | **no match** | | `sentinelA.txt` | match | match | match | | overwrite 2, key retired | no match | match | **match** | | `sentinelB.txt` | match | match | match | Totals: `f.txt` 1 / 3 / 2, `sentinelA.txt` 1 / 1 / 1, `sentinelB.txt` 1 / 1 / 1. - `plain` deduplicates by path, so neither overwrite is matched again. Both sentinels are matched after each overwrite, so the two absences come from deduplication rather than from a job that stopped polling. - `match_updated_files=True` deduplicates by path and last-modified time, so both overwrites are matched again. - `timestamp_cursor=True` deduplicates by path. It does not match the first overwrite, because the key is still retained, and does match the second, after `sentinelA.txt` moved the cursor past `f.txt`. That is the documented cost of bounding the state, and the first overwrite also rules out the cursor mode keying on the last-modified time behind the scenes. Raw markers, prefix stripped. Markers from one poll round share a millisecond, so their order inside a round is not meaningful. ``` --- plain --- WATCH_EMIT 1786639861234 f.txt Timestamp(1786639861.071971) 1786639422.667 WATCH_EMIT 1786641061995 sentinelA.txt Timestamp(1786641061.870721) 1786641046.536 WATCH_EMIT 1786642270977 sentinelB.txt Timestamp(1786642270.863813) 1786642269.562 --- updated --- WATCH_EMIT 1786639877550 f.txt Timestamp(1786639877.383040) 1786639422.667 WATCH_EMIT 1786640992063 f.txt Timestamp(1786640991.945338) 1786640981.762 WATCH_EMIT 1786641056832 sentinelA.txt Timestamp(1786641056.711839) 1786641046.536 WATCH_EMIT 1786642206994 f.txt Timestamp(1786642206.873078) 1786642204.316 WATCH_EMIT 1786642271846 sentinelB.txt Timestamp(1786642271.725531) 1786642269.562 --- cursor --- WATCH_EMIT 1786639943192 f.txt Timestamp(1786639422.667000) 1786639422.667 WATCH_EMIT 1786641055708 sentinelA.txt Timestamp(1786641046.536000) 1786641046.536 WATCH_EMIT 1786642204353 f.txt Timestamp(1786642204.316000) 1786642204.316 WATCH_EMIT 1786642284137 sentinelB.txt Timestamp(1786642269.562000) 1786642269.562 ``` ### Event times `plain` and `updated` stamp a match with the poll time; `cursor` stamps it with the file's last-modified time, visible above where `event_time` equals `mtime` exactly. ### Scope Each marker is written by a `DoFn` before it yields, and the pipeline has no durable sink, so these are counts of how many times each file was processed into a log marker. That is not a transactional acknowledgement and does not on its own establish exactly-once delivery. Nothing here exercises restart or pipeline update. -- 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]
