Struct KafkaConsumer

pub struct KafkaConsumer {
    inner: StreamConsumer<TracingContext>,
}

Thin wrapper around [StreamConsumer].

Arguments

• inner - The inner consumer

Returns

• KafkaConsumer - The Kafka consumer

Fields

inner: StreamConsumer<TracingContext>

Implementations

impl KafkaConsumer

pub fn get_default_config( bootstrap_servers: &str, group_id: &str, credentials: &ClientCredentials, ) -> ClientConfig

Get the default configuration.

Arguments

• bootstrap_servers - The bootstrap servers to use for the consumer
• group_id - The group id to use for the consumer
• credentials - The credentials to use for authentication

Returns

• ClientConfig - The configured client config

pub fn new(topics: &[Topic], config: ClientConfig) -> Result<Self>

Explicit configuration.

Arguments

• topics - The topics to subscribe to
• config - The consumer configuration

Returns

• Result<Self, SDKError> - The result of the operation

Errors

• SDKError::Consumer - If the consumer fails to create.

pub fn default( bootstrap_servers: &str, topics: &[Topic], group_id: &str, credentials: &ClientCredentials, ) -> Result<Self>

Default configuration.

Arguments

• topics - The topics to subscribe to
• group_id - The group id to use for the consumer
• credentials - The credentials to use for authentication

Returns

• Result<Self, SDKError> - The result of the operation

Errors

• SDKError::Consumer - If the consumer fails to create.

pub fn default_with_strings( bootstrap_servers: &str, topics: &[&str], group_id: &str, credentials: &ClientCredentials, ) -> Result<Self>

Default configuration with string topics.

Arguments

• topics - The topics to subscribe to
• group_id - The group id to use for the consumer
• bootstrap_servers - The bootstrap servers to use for the consumer
• credentials - The credentials to use for authentication

Returns

• Result<Self, SDKError> - The result of the operation

Errors

• SDKError::Consumer - If the consumer fails to create.

pub fn stream( &self, ) -> impl Stream<Item = KafkaResult<BorrowedMessage<'_>>> + '_

Async stream of messages (KafkaResult<BorrowedMessage>).

Returns

• impl Stream<Item = KafkaResult<BorrowedMessage<'_>>> + '_ - The stream of messages

pub fn event_stream( &self, ) -> impl Stream<Item = Result<EventStream, SDKError>> + '_

Async stream of deserialized EventStream messages.

This method provides a higher-level API that automatically deserializes Kafka messages into EventStream objects. This eliminates the need for manual deserialization in your application code.

Returns

• impl Stream<Item = Result<EventStream, SDKError>> + '_ - The stream of deserialized events

Errors

• SDKError::Consumer - If the consumer fails to deserialize the message.

pub fn inner(&self) -> &StreamConsumer<TracingContext>

Get the underlying consumer for manual polling.

Returns

• &StreamConsumer<TracingContext> - The underlying consumer

pub fn commit_message(&self, msg: &BorrowedMessage<'_>, mode: CommitMode)

Commits the offset of a message.

Arguments

• msg - The message to commit
• mode - The commit mode

Returns

• () - Always succeeds, logs any errors

pub fn deserialize_message( &self, msg: &BorrowedMessage<'_>, ) -> Result<EventStream>

Deserializes a Kafka message into an EventStream<T>.

• T must implement serde::Serialize and serde::de::DeserializeOwned.

Arguments

• msg - The message to deserialize

Returns

• Result<EventStream, SDKError> - The result of the operation

Errors

• SDKError::Consumer - If the consumer fails to deserialize the message.