Struct KafkaMessage 

pub struct KafkaMessage { /* private fields */ }

One Kafka delivery: an owned snapshot of the record plus its settlement handle.

Settlement mapping depends on the Commit mode of the subscription:

Under Commit::Auto (the default) librdkafka owns the committed position - it is stored the moment a message is handed to the application - so ack and nack(false) are advisory no-ops, and nack(true) reports AckError::Unsupported: nothing brings the record back.

Under Commit::Tracked:

• ack settles the offset and advances the stored position across everything settled below it.
• nack(false) drops the message: the offset settles so the position can move past it.
• nack(true) hands the delivery back to its subscription, which delivers it again ahead of what it fetches next. Its offset stays unsettled meanwhile, so the committed position stays below it: were the process to stop first, Kafka redelivers the record from there to whoever reads the partition next. A rebalance or a seek that moves the partition before the delivery comes back drops it for the same reason. Until it settles, the unsettled offset also blocks the position, keeping every later ack uncommitted.

Under Commit::Transactional the settlements report to the exactly-once pipeline instead, and nack(true) keeps the open window from committing: it aborts, and its replay delivers the record again with the rest of the window.

Wire headers map name for name; a null-valued Kafka header arrives with an empty value (presence preserved).

Implementations

impl KafkaMessage

pub fn topic(&self) -> &str

The topic this record was consumed from.

pub fn partition(&self) -> i32

The partition this record was consumed from.

pub fn offset(&self) -> i64

The record’s offset within its partition.

pub fn timestamp_millis(&self) -> Option<i64>

The record’s timestamp in milliseconds since the epoch, when the broker provided one.

Read from the record on the ask rather than carried by the delivery: librdkafka answers it out of the fetch buffer this delivery holds open, and most deliveries are never asked.

pub fn key(&self) -> Option<&[u8]>

The record key, surfaced from Kafka’s native key (see PARTITION_KEY_HEADER).

Trait Implementations

impl BuildBatchContext<KafkaMessage> for KafkaBatchContext

fn build(first: &KafkaMessage) -> Self

Builds the context value by reading subscription-scoped fields out of the batch’s first delivery, or out of the batch itself where it lends its values as one slice.

impl BuildContext<KafkaMessage> for KafkaContext

fn build(msg: &KafkaMessage) -> Self

Builds the context value by reading fields out of msg.

impl Debug for KafkaMessage

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

Formats the value using the given formatter.

impl IncomingMessage for KafkaMessage

fn ack(self) -> impl Future<Output = Result<(), AckError>>

Marks the offset processed (see the type-level settlement mapping).

Errors

Returns AckError::Broker when librdkafka refuses the stored position, which is a client-state failure rather than a network one - the partition is no longer this consumer’s, say. A descriptor passthrough that would take the offset store back is not one of those cases and never reaches here: librdkafka would accept the store and keep owning the position, so Commit::Tracked and Commit::Transactional refuse such a subscription at startup instead.

Cancel safety

Cancel safe: the watermark update is synchronous, so the future either completed or did nothing.

fn nack(self, requeue: bool) -> impl Future<Output = Result<(), AckError>>

Settles negatively. requeue = false drops the delivery: the offset settles so the committed position can move past it. requeue = true hands the delivery back to its subscription, which delivers it again; its offset stays unsettled until then. Under Commit::Transactional it aborts the open window, whose replay delivers it again. Under Commit::Auto requeue = false is an advisory no-op and requeue = true is unsupported (see the type-level settlement mapping).

Kafka counts no deliveries of its own, so a registration that caps its attempts with max_attempts(..) gets the framework’s count instead: the runtime republishes the delivery through the registration’s retry publisher so the count travels with it.

Errors

Returns AckError::Unsupported for requeue = true under Commit::Auto, where the position was stored when the record was handed over, and AckError::Broker under the same conditions as ack.

Cancel safety

Cancel safe: the watermark update is synchronous, so the future either completed or did nothing.

fn partition_key(&self) -> Option<&[u8]>

The keyed-lane key, so keyed worker lanes see it without a Partitioned bound: the source partition (the default), or the record key under LaneKey::RecordKey.

fn payload(&self) -> &[u8]

Returns the raw payload of the message.

fn headers(&self) -> &HeaderMap

Returns the headers attached to the message.

fn redelivery_count(&self) -> Option<u64>

How many times the broker has delivered this message, counting this delivery. None where the transport counts nothing.

fn decode_error(&self) -> Option<&CodecError>

The error the broker met while reading this delivery, or None when it read the delivery whole.

fn supports_nack_after(&self) -> bool

Reports whether this transport can honor nack_after natively.

fn nack_after( self, delay: Duration, ) -> impl Future<Output = Result<(), AckError>> + Send

where Self: Sized,

Negatively acknowledges the message, asking the broker to redeliver it no sooner than delay from now.

impl Partitioned for KafkaMessage

fn partition_key(&self) -> Option<&[u8]>

The keyed-lane key (see IncomingMessage::partition_key on this type): the source partition (the default), or the record key under LaneKey::RecordKey.

impl Positioned for KafkaMessage

fn position(&self) -> Self::Position

This delivery’s own coordinates: seeking to them redelivers exactly this record (and the ordered suffix behind it on the partition).

type Position = KafkaPosition

The position type, matching the subscription’s Seeker::Position.