EosReplies

EosReplies joins every reply to the pipeline's open window, paired with its delivery's consumed offset.

It is a publish transform for the reply position, written after the policy: .out_reply(EosPublish::new(id)).transform(EosReplies). A Commit::Transactional subscription marks each delivery with its source coordinates, and the transform carries them onto the reply. The pipeline reads them, pairs the reply with that offset, and drops them before the record is produced, so they never reach the topic. The handler only returns its value: it names no publisher and no offset. Without the transform, a reply carries no coordinates and its publish fails with KafkaError::InvalidOptions, so nothing leaves outside the window. A reply from a subscription in another commit mode fails the same way.

A handler that settles without a reply, a drop or an Err(outcome), publishes nothing, and its offset still commits with the window.

Example

A payment is booked on a pool of four lanes; a payment of zero is dropped.

Macros
Manual

examples/kafka_eos_replies.rs
/// Books a payment; an empty one is dropped, and its offset still commits with the window.
#[subscriber(
    KafkaTopic::new("payments").commit(Commit::Transactional("ledger-1".into())),
    reply,
    workers(4, by_key)
)]
async fn book(payment: &Payment) -> Result<LedgerEntry, HandlerOutcome> {
    if payment.cents == 0 {
        return Err(HandlerOutcome::drop());
    }
    Ok(LedgerEntry {
        payment_id: payment.id,
        cents: payment.cents,
    })
}

examples/manual/kafka_eos_replies.rs
/// Books a payment; an empty one is dropped, and its offset still commits with the window.
struct Book;

impl Handle<Payment, LedgerEntry> for Book {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<LedgerEntry, HandlerOutcome>> {
        if payment.cents == 0 {
            return ready(Err(HandlerOutcome::drop()));
        }
        ready(Ok(LedgerEntry {
            payment_id: payment.id,
            cents: payment.cents,
        }))
    }
}

Macros
Manual

examples/kafka_eos_replies.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("ledger-svc"),
        |b| {
            // The transform carries each delivery's coordinates onto its reply, where the pipeline
            // pairs them.
            b.include(book)
                .out_reply(EosPublish::new("ledger-1"))
                .transform(EosReplies);
        },
    )
}

examples/manual/kafka_eos_replies.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(
        KafkaBroker::new(["localhost:9092"]).default_group("ledger-svc"),
        |b| {
            let payments = subscriber("payments", Book)
                .map_source(|_| {
                    KafkaTopic::new("payments").commit(Commit::Transactional("ledger-1".into()))
                })
                .reply()
                .workers_by_key(nonzero!(4));
            // The transform carries each delivery's coordinates onto its reply, where the pipeline
            // pairs them.
            b.include(payments.build())
                .out_reply(EosPublish::new("ledger-1"))
                .transform(EosReplies);
        },
    )
}

Example​

Example