tkaymak commented on PR #39971: URL: https://github.com/apache/beam/pull/39971#issuecomment-5521528812
@Abacn thanks for the thorough review, all points addressed in 8af17d5. Checkpointing now follows Spark's lifecycle. Per source state lives under the location Spark passes to `toMicroBatchStream`, written through `CheckpointFileManager` with the session Hadoop conf broadcast to executors, marks coded with `getCheckpointMarkCoder()`. `commit(end)` purges marks below `end`, there is no fixed retention any more. Finalization moved out of `PartitionReader.close()`. A reader finalizes the mark taken at its start offset when the next batch for that split is scheduled. Spark only starts a batch at the initial offset or at the end offset of a batch already in the commit log (`MicroBatchExecution.populateStartOffsets`), so that point is always after the commit. Any mismatch (retry, killed attempt, executor change, restart) drops the reader without finalizing and restores it from the durable mark at the start offset. Finalizing decoded marks on the driver in `commit()` is not an option, `PubsubCheckpoint.finalizeCheckpoint` throws on a restored checkpoint and `KafkaCheckpointMark` is a no-op without the live reader, and DSv2 has no executor side commit hook. The remaining limit is `spark.speculation` with non deterministic sources, a losing attempt that completed before its kill arrived can be reused, DSv2 gives a task no way to learn it lost. On TransformWithState: it carries Beam state and timers in the next slices. Reading inside a stateful operator would be the legacy `mapWithState` design and it cannot own the watermark declaration the DSv2 source provides. On `spark/4`: `Dataset.ofRows` and `StreamingRelationV2` moved packages in 4.0, the module already forks `BoundedDatasetFactory` for the first, and the only consumer is the Spark 4 streaming translator. `CheckpointFileManager` moves to a `checkpointing` package in later 4.x, one import to adjust at the bump. Smaller items: the dataset is built like `BoundedDatasetFactory` (real objects in a `Table` wrapped in `StreamingRelationV2`, no string options, no Base64), splits are plain fields of the `InputPartition`, `maxRecordsPerBatch` is divided across splits like `MicrobatchSource.splitNumRecords`, `defaultParallelism` decides the split count, `CoderHelpers` is reused, `FluentBackoff` replaces the sleep, the offset serializes as the bare epoch like `LongOffset` with the base class equality, no class wide `SuppressWarnings`, the mark map is gone. One new option, `readerIdleTimeoutMillis`, replaces a hardcoded reader expiry. New tests drive the reader cache protocol directly and prove restart recovery, finalization only after commit and mark purging against Spark's real `offsets` and `commits` logs. The per test JUnit timeout is removed from the streaming test, its throwaway thread group poisoned Spark's static pools for later batch tests in the same forked JVM, which explains the flaky batch failures seen locally. -- 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]
