Struct EosPipeline 

pub struct EosPipeline { /* private fields */ }

An exactly-once pipeline over one transactional producer.

Wiring, all three naming the same id:

1. The publisher: broker.publisher().transactional_id("pipeline-1").
2. Each source subscription: Commit::Transactional("pipeline-1".into()) - its consumer stops committing offsets on its own and registers with the pipeline instead.
3. The pipeline: EosPipeline::new(publisher), held in the application state; handlers call publish with the delivery’s SourceOffset.

Every commit_interval the pipeline closes the window: it waits until every delivery that published into it has settled (the shared watermark reached the enrolled offsets), adds the settled source positions and their group metadata to the transaction, and commits. On any failure - a failed publish, a settle stall (a handler hanging or retry()-ing past the publisher’s transaction timeout), a rebalance revoking an enrolled partition, a commit error - the window aborts and the consumers seek back, so the whole window redelivers and republishes into a fresh transaction; committed output still never duplicates.

Works best over the default LaneKey::Partition worker lanes: a partition processes in order on one lane, so the settle condition follows the lane head and windows close promptly. Clones share the pipeline.

Implementations

impl EosPipeline

pub fn new(publisher: KafkaPublisher) -> Self

Builds the pipeline over publisher, which must carry a transactional_id - it doubles as the pipeline id that Commit::Transactional subscriptions register under. A publisher without one fails the first publish with a clear error.

pub fn commit_interval(self, interval: Duration) -> Self

How long a window stays open before committing; defaults to 100ms (the Kafka Streams exactly-once default). Longer intervals amortize the commit over more records at the cost of end-to-end latency (records become visible only at the commit). Configure before handing the pipeline out.

pub async fn publish( &self, source: &SourceOffset, msg: OutgoingMessage<'_>, ) -> Result<(), KafkaError>

Publishes msg into the pipeline’s open window on behalf of the delivery at source.

The record joins the window’s transaction and becomes visible at its commit, atomically with the source position. Publish, then return Ack: the settled watermark is what releases the window’s commit.

Errors

Returns KafkaError::InvalidOptions when the publisher carries no transactional id, KafkaError::NotConnected before Broker::connect, and KafkaError::Publish when opening the transaction or producing the record fails - the window aborts and redelivers, so failing the handler (retry()) is the right response.

Cancel safety

Not cancel safe: dropping the future may leave the record in the window’s transaction.

Panics

Panics when the internal window mutex is poisoned, which requires a prior panic inside the pipeline (an invariant violation, not an operational failure).

impl EosPipeline

pub fn replies( &self, ) -> TypedPublisher<Self, DefaultCodec, PublishTransformStack<PublishTransformIdentity, EosReplies>>

A reply publisher for #[subscriber(.., publish("replies"))] handlers: every reply joins the pipeline’s open window paired with its delivery’s consumed offset, making the publishing-handler form exactly-once end to end - the handler just returns the value.

Pairs only with subscriptions in Commit::Transactional mode naming this pipeline’s id (they stamp the source coordinates the reply path relays); a reply from any other subscription fails with a clear error. The retry_after deferred-republish fallback does not apply to these replies: a delayed copy would break the offset-record pairing.

Equivalent explicit form: TypedPublisher::new(pipeline.clone()).transform(EosReplies).

Examples

use ruststream_rdkafka::{EosPipeline, KafkaBroker};

let broker = KafkaBroker::new(["localhost:9092"]);
let pipeline = EosPipeline::new(broker.publisher().transactional_id("enrich-1"));
let replies = pipeline.replies();
// b.include_publishing(enrich, replies);

pub fn replies_with<C: Codec>( &self, codec: C, ) -> TypedPublisher<Self, C, PublishTransformStack<PublishTransformIdentity, EosReplies>>

Like replies, with an explicit codec instead of the default one.

Trait Implementations

impl Clone for EosPipeline

fn clone(&self) -> EosPipeline

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for EosPipeline

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Publisher for EosPipeline

async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error>

Publishes a reply into the pipeline’s open window, paired with the source coordinates the EOS_SOURCE_HEADER carries (stripped before the record is produced).

Errors

Returns KafkaError::InvalidOptions when the header is missing or malformed - the originating subscription is not in Commit::Transactional mode for this pipeline, or the reply publisher was wired without EosReplies (use replies); otherwise as EosPipeline::publish.

Cancel safety

Not cancel safe: dropping the future may leave the record in the window’s transaction.

type Error = KafkaError

The error type returned by publish.