Errors

KafkaError is the one error type of the crate: connect, publish, consume, transactions, the registry and the wire format.

Each variant names the step that failed. The librdkafka error behind it is its source, so the client library's types stay out of the crate's API.

Variant	When
`Connect`	the client could not be created, a property was refused, or the cluster did not answer the startup probe
`Subscribe`	a consumer could not be created or subscribed, or its group did not assign it partitions in time
`Consume`	an open consumer failed to receive
`Publish`	the cluster refused a record or did not confirm it, or the shutdown flush ran out of time
`Closed`	a publisher or a transaction was used after the broker shut down
`InvalidOptions`	the settings cannot work together, such as a subscription with no consumer group
`TransactionBusy`, `NoTransaction`	a transaction was begun twice on one publisher, or ended with none open
`SchemaRegistry`, `SchemaRegistryTimeout`	a registry request failed or went unanswered (`schema-registry` feature)
`WireFormat`	a value did not match its Avro or Protobuf wire form (`schema-registry` feature)

A service meets the error in two places. A failed startup carries it as the source of the runtime's error, and the service exits before it reads anything. A publish from a handler returns it inside the core's PublishError, and the handler decides how the delivery settles. The enum is #[non_exhaustive], so a match on it ends with a wildcard arm.

Example

A shipment the cluster refuses is logged with its cause, and the order is settled as a retry.

Macros
Manual

examples/kafka_broker_errors.rs
#[subscriber("orders")]
async fn ship(order: &Order, Out(shipments): Out<impl Publisher, Shipments>) -> HandlerOutcome {
    let shipment = Shipment {
        order_id: order.id,
        labels: (1..=order.parcels)
            .map(|parcel| format!("{}-{parcel:05}", order.id))
            .collect(),
    };
    if let Err(err) = shipments.message(&shipment).publish().await {
        // The error names the topic and the cluster's reason.
        tracing::warn!(subscription = "orders", order.id, error = %err, "shipment not published");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kafka_broker_errors.rs
struct Ship;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Ship
where
    Entry: OutEntry<Shipments, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let shipment = Shipment {
            order_id: order.id,
            labels: (1..=order.parcels)
                .map(|parcel| format!("{}-{parcel:05}", order.id))
                .collect(),
        };
        if let Err(err) = outs.get(Shipments).message(&shipment).publish().await {
            // The error names the topic and the cluster's reason.
            tracing::warn!(subscription = "orders", order.id, error = %err, "shipment not published");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Example​

Example