tkaymak commented on PR #39971: URL: https://github.com/apache/beam/pull/39971#issuecomment-5539016833
@Abacn thanks, all addressed in 7376415. Size. Three main files now, `UnboundedSourceDataset` with the DataSourceV2 glue as nested classes like `BoundedDatasetFactory`, `BeamReaderCache`, `BeamSourceCheckpoint`. One test class over one test source. The PR is 5 files and 2337 lines against master, tests 44 percent. Locality. Rendezvous hashing and the `getPeers` call are gone, `preferredLocations()` is not overridden. Tests. The three synthetic sources are one `TestSource`. `BeamSourceCheckpointTest` is deleted. `BeamReaderCacheProtocolTest` is deleted except three cases folded into the live test class because a live query cannot provoke them: a retried batch restarts from the durable mark and finalizes nothing, a start offset above zero without a mark fails instead of restarting the source, a failed mark write fails the batch and the retry recreates the reader. Quotas. `splitQuotas` follows `MicrobatchSource.splitNumRecords`, a zero quota split emits nothing, below zero is unlimited. One addition, the remainder rotates by epoch so a limit below the split count does not starve the same splits forever. Verified live with a limit of 1 over 4 splits. Purge. Layout is `marks/<epoch>/<split>`, purging below the committed offset is one recursive directory delete per epoch, independent of the split count, still off the stream thread. The driver creates the epoch directory when it plans the batch. Finalization, both observations were correct. Idle close now finalizes the pending mark when Spark's commit log covers its epoch. The executor reads `<root>/commits` and this source's line in the matching `<root>/offsets/<id>`, which has no lag because Spark writes `commits/N` in `markMicroBatchEnd` while `commit(end)` reaches the source only when N+1 is constructed. A sweeper thread runs the idle check every 10 s, so after `query.stop()` the last completed batch is finalized within the idle timeout. Still unfinalized: an executor JVM that exits before the sweep, a batch Spark aborted mid flight, and async progress tracking where the commit log itself lags. DSv2 has no executor side stop or commit hook, in those cases the source redelivers. A wrong or racing read can only yield a lower epoch, so the failure direction is redelivery, never an extra finalization. -- 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]
