Skip to main content

Error

partitionline::error

Enum Error 

Source 
pub enum Error {
    Io(Error),
    Protocol(String),
    Broker {
        code: i16,
        message: String,
    },
    UnknownTopic(String),
    NoLeader {
        topic: String,
        partition: i32,
    },
    Unsupported(String),
    Closed,
    Timeout,
    QueueFull,
    RecordTooLarge {
        size: u64,
        max: u64,
        config: &'static str,
    },
    MaxPollInterval,
    Wakeup,
}
Expand description

Client, protocol, or broker failure.

Variants§

§

Io(Error)

I/O or TLS failure.

§

Protocol(String)

Client-side protocol or usage error.

§

Broker

Broker error_code plus a short context string.

Fields

§code: i16

Kafka error_code.

§message: String

Api name or topic-partition.

§

UnknownTopic(String)

Metadata did not list this topic.

§

NoLeader

Metadata has no leader for this partition.

Fields

§topic: String

Topic name.

§partition: i32

Partition index.

§

Unsupported(String)

Broker does not support a required API version.

§

Closed

The producer (or connection) is shut down.

§

Timeout

A request exceeded crate::ProducerConfig::request_timeout or similar.

§

QueueFull

try_send could not queue (metadata, connection, or buffer.memory).

§

RecordTooLarge

Serialized record size exceeds crate::ProducerConfig::max_request_size or crate::ProducerConfig::buffer_memory.

Java KafkaProducer.ensureValidRecordSize checks max.request.size first, then buffer.memory. std::fmt::Display is Java RecordTooLargeException. For Self::MAX_REQUEST_SIZE_CONFIG: The message is {size} bytes when serialized which is larger than {max}, which is the value of the max.request.size configuration. For Self::BUFFER_MEMORY_CONFIG: The message is {size} bytes when serialized which is larger than the total memory buffer you have configured with the buffer.memory configuration. size is Java AbstractRecords.estimateSizeInBytesUpperBound.

Fields

§size: u64

Java AbstractRecords.estimateSizeInBytesUpperBound of the record.

§max: u64

Configured cap that was exceeded (max.request.size or buffer.memory).

§config: &'static str

Java config name: Self::MAX_REQUEST_SIZE_CONFIG or Self::BUFFER_MEMORY_CONFIG.

§

MaxPollInterval

crate::ConsumerGroup::poll was not called within max.poll.interval.ms.

§

Wakeup

crate::Consumer::wakeup interrupted fetch or poll.

Implementations§

Source§

impl Error

Source

pub const MAX_REQUEST_SIZE_CONFIG: &str = "max.request.size"

Java ProducerConfig.MAX_REQUEST_SIZE_CONFIG.

Source

pub const BUFFER_MEMORY_CONFIG: &str = "buffer.memory"

Java ProducerConfig.BUFFER_MEMORY_CONFIG.

Source

pub fn protocol(msg: impl Into<String>) -> Self

Wrap a protocol / client-side failure.

Examples found in repository?
examples/bench_latency.rs (line 25)
23fn percentile_us(sorted: &[u64], p: u32) -> partitionline::Result<u64> {
24    if sorted.is_empty() {
25        return Err(partitionline::Error::protocol("no latency samples"));
26    }
27    let n = sorted.len();
28    let rank = n
29        .saturating_mul(p as usize)
30        .div_ceil(100)
31        .saturating_sub(1)
32        .min(n.saturating_sub(1));
33    sorted
34        .get(rank)
35        .copied()
36        .ok_or_else(|| partitionline::Error::protocol("percentile index"))
37}
38
39fn print_latency(kind: &str, mut samples: Vec<u64>, extra: &str) -> partitionline::Result<()> {
40    if samples.is_empty() {
41        return Err(partitionline::Error::protocol(format!(
42            "{kind}: no latency samples"
43        )));
44    }
45    samples.sort_unstable();
46    let n = samples.len();
47    let min_us = samples.first().copied().unwrap_or(0);
48    let max_us = samples.last().copied().unwrap_or(0);
49    let sum: u128 = samples.iter().map(|v| u128::from(*v)).sum();
50    let mean_us = u64::try_from(sum / u128::from(n as u64))
51        .map_err(|_| partitionline::Error::protocol("mean overflow"))?;
52    let p50_us = percentile_us(&samples, 50)?;
53    let p99_us = percentile_us(&samples, 99)?;
54    println!(
55        "{{\"kind\":\"{kind}\",\"samples\":{n},\"p50_us\":{p50_us},\"p99_us\":{p99_us},\"min_us\":{min_us},\"max_us\":{max_us},\"mean_us\":{mean_us}{extra}}}"
56    );
57    Ok(())
58}
More examples
Hide additional examples
examples/roundtrip.rs (line 29)
8async fn main() -> partitionline::Result<()> {
9    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
10    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
11    let payload = b"live-roundtrip";
12
13    let mut pcfg = ProducerConfig::bootstrap([bootstrap.clone()]);
14    pcfg.linger = Duration::ZERO;
15    let producer = Producer::new(pcfg).await?;
16    let md = producer
17        .send(ProduceRecord::to(topic.clone()).value(&payload[..]))
18        .await?;
19    producer.close().await?;
20
21    let mut ccfg = ConsumerConfig::bootstrap([bootstrap]);
22    ccfg.max_wait_ms = 1000;
23    let mut consumer = Consumer::new(ccfg).await?;
24    consumer.assign(topic, md.partition, md.offset).await?;
25    let recs = consumer.fetch().await?;
26    let rec = recs
27        .iter()
28        .find(|r| r.offset == md.offset)
29        .ok_or_else(|| partitionline::Error::protocol("record not fetched"))?;
30    if rec.value.as_deref() != Some(&payload[..]) {
31        return Err(partitionline::Error::protocol("payload mismatch"));
32    }
33    println!(
34        "ok {}-{}@{} bytes={}",
35        rec.topic,
36        rec.partition,
37        rec.offset,
38        rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
39    );
40    Ok(())
41}
examples/oauth.rs (line 44)
25async fn main() -> partitionline::Result<()> {
26    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
27    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
28
29    let sasl = if let (Ok(url), Ok(id), Ok(secret)) = (
30        std::env::var("OIDC_TOKEN_URL"),
31        std::env::var("OIDC_CLIENT_ID"),
32        std::env::var("OIDC_CLIENT_SECRET"),
33    ) {
34        Sasl::oidc(OidcConfig::new(url, id, secret))
35    } else {
36        let principal = std::env::var("SASL_OAUTH_PRINCIPAL").unwrap_or_else(|_| "alice".into());
37        Sasl::oauthbearer(principal)
38    };
39
40    let mut cfg = ProducerConfig::bootstrap([bootstrap]).sasl(sasl);
41    if let Ok(ca_path) = std::env::var("TLS_CA_PEM") {
42        let mut tls =
43            TlsConfig::default().ca_pem(tokio::fs::read(&ca_path).await.map_err(|e| {
44                partitionline::Error::protocol(format!("read TLS_CA_PEM {ca_path}: {e}"))
45            })?);
46        if let Ok(name) = std::env::var("TLS_SERVER_NAME") {
47            if !name.is_empty() {
48                tls = tls.server_name(name);
49            }
50        }
51        cfg = cfg.tls(tls);
52    }
53
54    let producer = Producer::new(cfg).await?;
55    let md = producer
56        .send(ProduceRecord::to(topic).value(&b"hello over oauthbearer"[..]))
57        .await?;
58    println!("{}-{}@{}", md.topic, md.partition, md.offset);
59    producer.close().await?;
60    Ok(())
61}
examples/tls.rs (line 20)
14async fn main() -> partitionline::Result<()> {
15    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
16    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
17    let mut tls = TlsConfig::default();
18    if let Ok(ca_path) = std::env::var("TLS_CA_PEM") {
19        tls = tls.ca_pem(tokio::fs::read(&ca_path).await.map_err(|e| {
20            partitionline::Error::protocol(format!("read TLS_CA_PEM {ca_path}: {e}"))
21        })?);
22    }
23    if let Ok(name) = std::env::var("TLS_SERVER_NAME") {
24        if !name.is_empty() {
25            tls = tls.server_name(name);
26        }
27    }
28    if let (Ok(cert_path), Ok(key_path)) = (
29        std::env::var("TLS_CLIENT_CERT_PEM"),
30        std::env::var("TLS_CLIENT_KEY_PEM"),
31    ) {
32        let cert = tokio::fs::read(&cert_path).await.map_err(|e| {
33            partitionline::Error::protocol(format!("read TLS_CLIENT_CERT_PEM {cert_path}: {e}"))
34        })?;
35        let key = tokio::fs::read(&key_path).await.map_err(|e| {
36            partitionline::Error::protocol(format!("read TLS_CLIENT_KEY_PEM {key_path}: {e}"))
37        })?;
38        tls = tls.client_identity(cert, key);
39    }
40
41    let producer = Producer::new(ProducerConfig::bootstrap([bootstrap]).tls(tls)).await?;
42    let md = producer
43        .send(ProduceRecord::to(topic).value(&b"hello over tls"[..]))
44        .await?;
45    println!("{}-{}@{}", md.topic, md.partition, md.offset);
46    producer.close().await?;
47    Ok(())
48}
examples/sasl.rs (lines 26-28)
15async fn main() -> partitionline::Result<()> {
16    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
17    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
18    let username = std::env::var("KAFKA_USERNAME").unwrap_or_else(|_| "alice".into());
19    let password = std::env::var("KAFKA_PASSWORD").unwrap_or_else(|_| "secret".into());
20    let mechanism = std::env::var("SASL_MECHANISM").unwrap_or_else(|_| "SCRAM-SHA-256".into());
21    let sasl = match mechanism.as_str() {
22        "PLAIN" => Sasl::plain(username, password),
23        "SCRAM-SHA-256" => Sasl::scram_sha256(username, password),
24        "SCRAM-SHA-512" => Sasl::scram_sha512(username, password),
25        other => {
26            return Err(partitionline::Error::protocol(format!(
27                "unsupported SASL_MECHANISM {other}; use PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512"
28            )));
29        }
30    };
31
32    let mut cfg = ProducerConfig::bootstrap([bootstrap]).sasl(sasl);
33    if let Ok(ca_path) = std::env::var("TLS_CA_PEM") {
34        let mut tls =
35            TlsConfig::default().ca_pem(tokio::fs::read(&ca_path).await.map_err(|e| {
36                partitionline::Error::protocol(format!("read TLS_CA_PEM {ca_path}: {e}"))
37            })?);
38        if let Ok(name) = std::env::var("TLS_SERVER_NAME") {
39            if !name.is_empty() {
40                tls = tls.server_name(name);
41            }
42        }
43        cfg = cfg.tls(tls);
44    }
45
46    let producer = Producer::new(cfg).await?;
47    let md = producer
48        .send(ProduceRecord::to(topic).value(&b"hello over sasl"[..]))
49        .await?;
50    println!("{}-{}@{}", md.topic, md.partition, md.offset);
51    producer.close().await?;
52    Ok(())
53}
examples/bench_fetch.rs (line 35)
8async fn main() -> partitionline::Result<()> {
9    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
10    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "plbench".into());
11    let count: u64 = std::env::var("COUNT")
12        .ok()
13        .and_then(|s| s.parse().ok())
14        .unwrap_or(8_000_000);
15    let max_wait_ms = std::env::var("MAX_WAIT_MS")
16        .ok()
17        .and_then(|s| s.parse().ok())
18        .unwrap_or(100i32);
19    let max_bytes = std::env::var("MAX_BYTES")
20        .ok()
21        .and_then(|s| s.parse().ok())
22        .unwrap_or(16_777_216i32);
23    let min_bytes = std::env::var("MIN_BYTES")
24        .ok()
25        .and_then(|s| s.parse().ok())
26        .unwrap_or(1i32);
27
28    let mut cfg = ConsumerConfig::bootstrap([bootstrap]);
29    cfg.max_wait_ms = max_wait_ms;
30    cfg.max_bytes = max_bytes;
31    cfg.min_bytes = min_bytes;
32    if let Ok(ca_path) = std::env::var("TLS_CA_PEM") {
33        let mut tls = TlsConfig {
34            ca_pem: Some(tokio::fs::read(&ca_path).await.map_err(|e| {
35                partitionline::Error::protocol(format!("read TLS_CA_PEM {ca_path}: {e}"))
36            })?),
37            ..TlsConfig::default()
38        };
39        if let Ok(name) = std::env::var("TLS_SERVER_NAME") {
40            if !name.is_empty() {
41                tls.server_name = Some(name);
42            }
43        }
44        cfg.tls = Some(tls);
45    }
46    let mech = std::env::var("SASL_MECHANISM").unwrap_or_else(|_| "PLAIN".into());
47    if mech == "OAUTHBEARER" {
48        let principal = std::env::var("SASL_OAUTH_PRINCIPAL").unwrap_or_else(|_| "alice".into());
49        cfg.sasl_oauthbearer = Some(principal);
50    } else if let (Ok(user), Ok(pass)) = (
51        std::env::var("SASL_USERNAME"),
52        std::env::var("SASL_PASSWORD"),
53    ) {
54        match mech.as_str() {
55            "SCRAM-SHA-256" => cfg.sasl_scram = Some((user, pass)),
56            "SCRAM-SHA-512" => cfg.sasl_scram_sha512 = Some((user, pass)),
57            "PLAIN" => cfg.sasl_plain = Some((user, pass)),
58            other => {
59                return Err(partitionline::Error::protocol(format!(
60                    "unknown SASL_MECHANISM {other}"
61                )));
62            }
63        }
64    }
65
66    let mut consumer = Consumer::new(cfg).await?;
67    consumer.assign_topic(&topic, 0).await?;
68    let assigned = consumer.assignment().len();
69    let start = Instant::now();
70    let mut got = 0u64;
71    let mut empty = 0u32;
72    while got < count {
73        let recs = consumer.fetch().await?;
74        if recs.is_empty() {
75            empty += 1;
76            if empty > 600 {
77                return Err(partitionline::Error::Timeout);
78            }
79            continue;
80        }
81        empty = 0;
82        got += recs.len() as u64;
83    }
84    let elapsed = start.elapsed().as_secs_f64();
85    let rec_s = got as f64 / elapsed.max(1e-9);
86    println!(
87        "{{\"consumed\":{got},\"elapsed_s\":{elapsed:.6},\"consumed_rec_s\":{rec_s:.3},\"partitions\":{assigned},\"max_wait_ms\":{max_wait_ms},\"max_bytes\":{max_bytes}}}"
88    );
89    Ok(())
90}
Additional examples can be found in:
Source

pub fn broker(code: i16, message: impl Into<String>) -> Self

Broker error_code plus a short context string (api or topic-partition).

Source

pub fn broker_code(&self) -> Option<i16>

Kafka error code when this is a broker error.

Source

pub fn is_retriable(&self) -> bool

Kafka transient errors (NOT_LEADER, coordinator move, timeout) plus I/O.

Trait Implementations§

Source§

impl Clone for Error

Source§

fn clone(&self) -> Self

Returns a duplicate of the value. Read more
1.0.0 (const: unstable) · Source§

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source. Read more
Source§

impl Debug for Error

Source§

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

Formats the value using the given formatter. Read more
Source§

impl Display for Error

Source§

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

Formats the value using the given formatter. Read more
Source§

impl Error for Error

Source§

fn source(&self) -> Option<&(dyn Error + 'static)>

Returns the lower-level source of this error, if any. Read more
1.0.0 · Source§

fn description(&self) -> &str

👎Deprecated since 1.42.0:

use the Display impl or to_string()

Read more
1.0.0 · Source§

fn cause(&self) -> Option<&dyn Error>

👎Deprecated since 1.33.0:

replaced by Error::source, which can support downcasting

Source§

fn provide<'a>(&'a self, request: &mut Request<'a>)

🔬This is a nightly-only experimental API. (error_generic_member_access)
Provides type-based access to context intended for error reports. Read more
Source§

impl From<Error> for Error

Source§

fn from(value: Error) -> Self

Converts to this type from the input type.

Auto Trait Implementations§

§

impl !RefUnwindSafe for Error

§

impl !UnwindSafe for Error

§

impl Freeze for Error

§

impl Send for Error

§

impl Sync for Error

§

impl Unpin for Error

§

impl UnsafeUnpin for Error

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<T> CloneToUninit for T
where T: Clone,

Source§

unsafe fn clone_to_uninit(&self, dest: *mut u8)

🔬This is a nightly-only experimental API. (clone_to_uninit)
Performs copy-assignment from self to dest. Read more
Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T> Instrument for T

Source§

fn instrument(self, span: Span) -> Instrumented<Self> ⓘ

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
Source§

fn in_current_span(self) -> Instrumented<Self> ⓘ

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<T> Same for T

Source§

type Output = T

Should always be Self
Source§

impl<T> ToOwned for T
where T: Clone,

Source§

type Owned = T

The resulting type after obtaining ownership.
Source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
Source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
Source§

impl<T> ToString for T
where T: Display + ?Sized,

Source§

fn to_string(&self) -> String

Converts the given value to a String. Read more
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = !

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, !>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
Source§

impl<T> WithSubscriber for T

Source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self> ⓘ
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
Source§

fn with_current_subscriber(self) -> WithDispatch<Self> ⓘ

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more