Struct mz_kafka_util::client::MzClientContext

pub struct MzClientContext {
    error_tx: Sender<MzKafkaError>,
    statistics_tx: Sender<Statistics>,
}

A ClientContext implementation that uses tracing instead of log macros.

All code in Materialize that constructs Kafka clients should use this context or a custom context that delegates the log and error methods to this implementation.

Fields

error_tx: Sender<MzKafkaError>

The last observed error log, if any.

statistics_tx: Sender<Statistics>

A tokio watch that retains the last statistics received by rdkafka and provides async notifications to anyone interested in subscribing.

Implementations

impl MzClientContext

pub fn with_errors() -> (Self, Receiver<MzKafkaError>)

Constructs a new client context and returns an mpsc Receiver that can be used to learn about librdkafka errors.

pub fn subscribe_statistics(&self) -> Receiver<Statistics>

Creates a tokio Watch subscription for statistics reported by librdkafka. It is necessary that the statistics.ms.interval is set for this stream to contain any values.

fn record_error(&self, msg: &str)

Trait Implementations

impl ClientContext for MzClientContext

fn log(&self, level: RDKafkaLogLevel, fac: &str, log_message: &str)

Receives log lines from librdkafka.

fn stats(&self, statistics: Statistics)

Receives the decoded statistics of the librdkafka client. To enable, the statistics.interval.ms configuration parameter must be specified.

fn error(&self, error: KafkaError, reason: &str)

Receives global errors from the librdkafka client.

const ENABLE_REFRESH_OAUTH_TOKEN: bool = false

Whether to periodically refresh the SASL OAUTHBEARER token by calling ClientContext::generate_oauth_token.

fn stats_raw(&self, statistics: &[u8])

Receives the JSON-encoded statistics of the librdkafka client. To enable, the statistics.interval.ms configuration parameter must be specified.

fn resolve_broker_addr( &self, host: &str, port: u16, ) -> Result<Vec<SocketAddr>, Error>

Performs DNS resolution on a broker address.

fn generate_oauth_token( &self, _oauthbearer_config: Option<&str>, ) -> Result<OAuthToken, Box<dyn Error>>

Generates an OAuth token from the provided configuration.

impl ConsumerContext for MzClientContext

fn rebalance( &self, native_client: &NativeClient, err: rd_kafka_resp_err_t, tpl: &mut TopicPartitionList, )

Implements the default rebalancing strategy and calls the pre_rebalance and post_rebalance methods. If this method is overridden, it will be responsibility of the user to call them if needed.

fn pre_rebalance<'a>(&self, rebalance: &Rebalance<'a>)

Pre-rebalance callback. This method will run before the rebalance and should terminate its execution quickly.

fn post_rebalance<'a>(&self, rebalance: &Rebalance<'a>)

Post-rebalance callback. This method will run after the rebalance and should terminate its execution quickly.

fn commit_callback( &self, result: Result<(), KafkaError>, offsets: &TopicPartitionList, )

Post commit callback. This method will run after a group of offsets was committed to the offset store.

fn main_queue_min_poll_interval(&self) -> Timeout

Returns the minimum interval at which to poll the main queue, which services the logging, stats, and error callbacks.

impl Default for MzClientContext

fn default() -> Self

Returns the “default value” for a type.

impl ProducerContext for MzClientContext

type DeliveryOpaque = <DefaultProducerContext as ProducerContext>::DeliveryOpaque

A DeliveryOpaque is a user-defined structure that will be passed to the producer when producing a message, and returned to the delivery method once the message has been delivered, or failed to.

fn delivery( &self, delivery_result: &DeliveryResult<'_>, delivery_opaque: Self::DeliveryOpaque, )

This method will be called once the message has been delivered (or failed to). The DeliveryOpaque will be the one provided by the user when calling send.