Shared-stream triggers: add producer-side ack channel for non-idempotent upstreams

[Lee-W]

Body

Context

PR #66584 added SharedStreamManager and the opt-in BaseEventTrigger.{shared_stream_key,open_shared_stream,filter_shared_stream} hooks. Scope is idempotent / read-only / subscriber-side-effect upstreams (directory scans, polling APIs, Kafka with enable.auto.commit=true).

filter_shared_stream is one-way producer → subscriber.

What's missing

Upstreams whose consumption needs a producer-held handle to advance can't use shared-stream today:

• Kafka manual commit, SQS delete-on-process / visibility-extend, Pub/Sub ack_id, Service Bus peek-lock.

Each needs the subscriber's accept/reject decision to flow back to the producer's handle.

Design candidates (from Jason)

1. (event, ack) tuple — filter calls ack closure on accept. Simple, but silently lossy: first-ack moves the offset past events siblings never
   saw.
2. Producer waits for N acks before advancing — closes the at-least-once gap, but slow filters block the group; interacts with the shared_stream_subscriber_queue_size bound.
3. Watermark / commit-up-to-offset — matches real Kafka consumer-group semantics; meaningfully larger API change.

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.