Summary
DataHubEventSource — the datahub-cloud event source at datahub-actions/src/datahub_actions/plugin/source/acryl/datahub_cloud_event_source.py — writes its consumer offset in exactly one place: close(), at :292. There is no periodic checkpoint in the poll loop, so the durable position only advances when the pipeline shuts down.
KafkaEventSource behaves differently. With async_commit_enabled (default true), librdkafka auto-commits acked offsets every async_commit_interval ms — 10s by default. With it set to false, ack() commits synchronously per event. Either way the stored position tracks progress within seconds.
Impact
On the REST source the replay window grows for as long as the process lives. A pipeline running for a week that then exits ungracefully resumes from wherever it last stopped cleanly and re-delivers everything since. At-least-once delivery is the contract and actions should be idempotent, but the window here is unbounded rather than bounded by a commit interval.
Two details make an ungraceful exit more common than it looks:
run_pipeline calls pipeline.stop() only on PipelineException (datahub_actions/pipeline/pipeline_manager.py:40-48), while the poll loop swallows every exception into running = False (datahub_cloud_event_source.py:247-249). An in-loop error therefore ends the pipeline without ever reaching close().kill_after_idle_timeout: true exits by returning from the generator, so run() returns normally — again no PipelineException, again no stop(), again no commit. The pipeline thread ends quietly while the CLI keeps running.
The signal-handling half of this is addressed separately in #19402.
Proposed change
Commit the safe-to-ack watermark from inside the poll loop, on a configurable throttle.
The right insertion point already exists. At datahub_cloud_event_source.py:204 the loop sets safe_to_ack_offsets[topic] = consumer.offset_id, immediately after the ack-drain gate has confirmed every previously yielded event was acked — which is exactly the precondition a durable commit needs. It is also the same value close() commits. A throttled commit_offsets(offset_id=safe_to_ack_offsets[topic]) there needs no other change to the loop.
Two constraints worth stating up front:
- Pass the watermark explicitly.
consumer.offset_id advances on poll (datahub_cloud_events_consumer.py:160), before any event is processed, so the no-argument form of commit_offsets() would commit past work still in flight — the opposite of the intended fix.
- Throttle it. Each commit is a synchronous
EmitMode.SYNC_PRIMARY MCP writing a PlatformResource (datahub_cloud_events_consumer_offsets_store.py:88-95) — a server round-trip, not a local write. It also emits an MCL, which is harmless on the default topics: PlatformEvent_v1 but a feedback loop for any consumer subscribed to MCL, so the state URN should be excluded. Cadence should be configurable (every N batches, or every N seconds).
Related notes
EventSource does not declare commit_offsets, so there is no supported way for an action to checkpoint itself in the meantime — it is reachable only by downcasting to the concrete source and reading private attributes.- On this path, acking is not committing.
act() returning False plus a later source.ack(...) is the documented batching API, but it only feeds AckManager, which gates the poll loop; it does not move the stored offset.
Summary
DataHubEventSource— thedatahub-cloudevent source atdatahub-actions/src/datahub_actions/plugin/source/acryl/datahub_cloud_event_source.py— writes its consumer offset in exactly one place:close(), at:292. There is no periodic checkpoint in the poll loop, so the durable position only advances when the pipeline shuts down.KafkaEventSourcebehaves differently. Withasync_commit_enabled(defaulttrue), librdkafka auto-commits acked offsets everyasync_commit_intervalms — 10s by default. With it set tofalse,ack()commits synchronously per event. Either way the stored position tracks progress within seconds.Impact
On the REST source the replay window grows for as long as the process lives. A pipeline running for a week that then exits ungracefully resumes from wherever it last stopped cleanly and re-delivers everything since. At-least-once delivery is the contract and actions should be idempotent, but the window here is unbounded rather than bounded by a commit interval.
Two details make an ungraceful exit more common than it looks:
run_pipelinecallspipeline.stop()only onPipelineException(datahub_actions/pipeline/pipeline_manager.py:40-48), while the poll loop swallows every exception intorunning = False(datahub_cloud_event_source.py:247-249). An in-loop error therefore ends the pipeline without ever reachingclose().kill_after_idle_timeout: trueexits byreturning from the generator, sorun()returns normally — again noPipelineException, again nostop(), again no commit. The pipeline thread ends quietly while the CLI keeps running.The signal-handling half of this is addressed separately in #19402.
Proposed change
Commit the safe-to-ack watermark from inside the poll loop, on a configurable throttle.
The right insertion point already exists. At
datahub_cloud_event_source.py:204the loop setssafe_to_ack_offsets[topic] = consumer.offset_id, immediately after the ack-drain gate has confirmed every previously yielded event was acked — which is exactly the precondition a durable commit needs. It is also the same valueclose()commits. A throttledcommit_offsets(offset_id=safe_to_ack_offsets[topic])there needs no other change to the loop.Two constraints worth stating up front:
consumer.offset_idadvances on poll (datahub_cloud_events_consumer.py:160), before any event is processed, so the no-argument form ofcommit_offsets()would commit past work still in flight — the opposite of the intended fix.EmitMode.SYNC_PRIMARYMCP writing aPlatformResource(datahub_cloud_events_consumer_offsets_store.py:88-95) — a server round-trip, not a local write. It also emits an MCL, which is harmless on the defaulttopics: PlatformEvent_v1but a feedback loop for any consumer subscribed to MCL, so the state URN should be excluded. Cadence should be configurable (every N batches, or every N seconds).Related notes
EventSourcedoes not declarecommit_offsets, so there is no supported way for an action to checkpoint itself in the meantime — it is reachable only by downcasting to the concrete source and reading private attributes.act()returningFalseplus a latersource.ack(...)is the documented batching API, but it only feedsAckManager, which gates the poll loop; it does not move the stored offset.