Support for custom Python UnboundedSource

[kennknowles]

Imported from Jira BEAM-6119. Original Jira may contain additional context.
Reported by: hroussel.

[shaheeramjad]

.take-issue

[Eliaaazzz]

For additional context, I’ve been exploring this in parallel, and my main takeaway so far is that this problem extends beyond foundational API definitions.

In a separate public PoC (https://github.com/Eliaaazzz/beam-native-streaming-poc), I’ve been validating an end-to-end SDF-style UnboundedSource path, including restriction/tracker/coder handling, checkpoint persistence and finalization, watermark restore/progression, deferred resumption, and a standalone reader transform. The UnboundedSource PoC currently has 77 passing tests.

What became clear from implementing it is that the core difficulty here is not the class surface itself, but the execution contract and runtime semantics: checkpoint-based splitting, primary/residual representation, watermark behavior, idle resume behavior, Read(UnboundedSource) integration, and alignment with Beam’s existing Java/Python model all need to be pinned down carefully.

I wanted to share the PoC here because I think the discussion will benefit from staying grounded in those end-to-end semantic/runtime requirements.