Skip to main content

Consumer

partitionline::consumer

Struct Consumer 

Source 
pub struct Consumer { /* private fields */ }
Expand description

Manual-assignment fetch client.

Implementations§

Source§

impl Consumer

Source

pub async fn connect(bootstrap: impl Into<String>) -> Result<Self>

Connect with default config to one bootstrap server.

Examples found in repository?
examples/consume.rs (line 9)
6async fn main() -> partitionline::Result<()> {
7    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
8    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
9    let mut consumer = Consumer::connect(bootstrap).await?;
10    consumer.assign_topic(topic, 0).await?;
11    loop {
12        for rec in consumer.fetch().await? {
13            println!(
14                "{}-{}@{} bytes={}",
15                rec.topic,
16                rec.partition,
17                rec.offset,
18                rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
19            );
20        }
21    }
22}
More examples
Hide additional examples
examples/pause.rs (line 11)
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 mut consumer = Consumer::connect(bootstrap).await?;
12    consumer.assign_topic(topic, 0).await?;
13    let assigned = consumer.assignment();
14    println!("assigned {assigned:?}");
15    if let Some(tp) = assigned.first().cloned() {
16        consumer.pause([tp.clone()]);
17        println!("paused {:?}", consumer.paused());
18        let recs = consumer.fetch().await?;
19        println!("while paused: {} records", recs.len());
20        consumer.resume([tp]);
21    }
22    let recs = consumer.fetch().await?;
23    println!("after resume: {} records", recs.len());
24    consumer.close().await?;
25    Ok(())
26}
examples/wakeup.rs (line 13)
10async fn main() -> partitionline::Result<()> {
11    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
12    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
13    let mut consumer = Consumer::connect(bootstrap).await?;
14    consumer.assign_topic(topic, 0).await?;
15    let wakeup = consumer.wakeup_handle();
16    drop(tokio::spawn(async move {
17        tokio::signal::ctrl_c().await.unwrap_or(());
18        wakeup.wakeup();
19    }));
20    loop {
21        match consumer.fetch().await {
22            Ok(recs) => {
23                for rec in recs {
24                    println!("{}-{}@{}", rec.topic, rec.partition, rec.offset);
25                }
26            }
27            Err(partitionline::Error::Wakeup) => break,
28            Err(e) => return Err(e),
29        }
30    }
31    consumer.close().await?;
32    Ok(())
33}
examples/offsets.rs (line 25)
11async fn main() -> partitionline::Result<()> {
12    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
13    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
14
15    let producer = Producer::new(
16        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
17    )
18    .await?;
19    let md = producer
20        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
21        .await?;
22    println!("produced {}-{}@{}", md.topic, md.partition, md.offset);
23    producer.close().await?;
24
25    let mut consumer = Consumer::connect(bootstrap.clone()).await?;
26    consumer.assign(&topic, 0, 0).await?;
27    let begin = consumer.beginning_offsets([(topic.as_str(), 0)]).await?;
28    let end = consumer.end_offsets([(topic.as_str(), 0)]).await?;
29    let lag = consumer.current_lag((topic.as_str(), 0)).await?;
30    println!("begin={begin:?} end={end:?} lag={lag:?}");
31    consumer.close().await?;
32
33    let mut group = ConsumerGroup::join(
34        ConsumerConfig::bootstrap([bootstrap]).max_wait_ms(500),
35        "partitionline-offsets",
36        topic.clone(),
37    )
38    .await?;
39    let recs = group.poll().await?;
40    if let Some(rec) = recs.first() {
41        let tp = TopicPartition::new(&rec.topic, rec.partition);
42        group
43            .commit_with_metadata([(
44                tp,
45                OffsetAndMetadata::with_metadata(rec.offset + 1, "example"),
46            )])
47            .await?;
48    }
49    for (tp, md) in group.committed().await? {
50        println!("{tp} committed={} meta={}", md.offset, md.metadata);
51    }
52    group.leave().await?;
53    Ok(())
54}
examples/metrics.rs (line 50)
9async fn main() -> partitionline::Result<()> {
10    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
11    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
12    let format = std::env::var("FORMAT").unwrap_or_else(|_| "log".into());
13
14    let producer = Producer::new(
15        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
16    )
17    .await?;
18    let md = producer
19        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
20        .await?;
21    let pm = producer.metrics();
22
23    if format == "prom" {
24        println!("# HELP partitionline_produce_records_acked Produce records acked");
25        println!("# TYPE partitionline_produce_records_acked counter");
26        println!("partitionline_produce_records_acked {}", pm.records_acked);
27        println!("# HELP partitionline_produce_ack_p99_seconds Produce ack p99 latency");
28        println!("# TYPE partitionline_produce_ack_p99_seconds gauge");
29        println!(
30            "partitionline_produce_ack_p99_seconds {}",
31            (pm.ack_latency.p99_nanos as f64) / 1_000_000_000.0
32        );
33    } else {
34        println!(
35            "produced {}-{}@{} queued={} acked={} bytes={} ack_us={} p50_us={} p99_us={} topics={}",
36            md.topic,
37            md.partition,
38            md.offset,
39            pm.records_queued,
40            pm.records_acked,
41            pm.bytes_queued,
42            pm.ack_latency.mean_nanos().unwrap_or(0) / 1000,
43            pm.ack_latency.p50_nanos / 1000,
44            pm.ack_latency.p99_nanos / 1000,
45            pm.topics.len()
46        );
47    }
48    producer.close().await?;
49
50    let mut consumer = Consumer::connect(bootstrap).await?;
51    consumer.assign(&topic, 0, 0).await?;
52    let recs = consumer.fetch().await?;
53    let cm = consumer.metrics();
54    if format == "prom" {
55        println!("# HELP partitionline_fetch_rounds Fetch rounds completed");
56        println!("# TYPE partitionline_fetch_rounds counter");
57        println!("partitionline_fetch_rounds {}", cm.fetch_rounds);
58        println!("# HELP partitionline_fetch_p99_seconds Fetch round p99 latency");
59        println!("# TYPE partitionline_fetch_p99_seconds gauge");
60        println!(
61            "partitionline_fetch_p99_seconds {}",
62            (cm.fetch_latency.p99_nanos as f64) / 1_000_000_000.0
63        );
64    } else {
65        println!(
66            "fetched {} records rounds={} bytes={} errors={} fetch_us={} p50_us={} p99_us={} topics={}",
67            recs.len(),
68            cm.fetch_rounds,
69            cm.bytes_fetched,
70            cm.fetch_errors,
71            cm.fetch_latency.mean_nanos().unwrap_or(0) / 1000,
72            cm.fetch_latency.p50_nanos / 1000,
73            cm.fetch_latency.p99_nanos / 1000,
74            cm.topics.len()
75        );
76    }
77    consumer.close().await?;
78    Ok(())
79}
Source

pub async fn new(cfg: ConsumerConfig) -> Result<Self>

Connect using cfg. Negotiates ApiVersions and optional SASL/TLS.

Examples found in repository?
examples/consume_intercept.rs (line 24)
20async fn main() -> partitionline::Result<()> {
21    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
22    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
23    let mut consumer =
24        Consumer::new(ConsumerConfig::bootstrap([bootstrap]).interceptor(SkipHeartbeats)).await?;
25    consumer.assign_topic(topic, 0).await?;
26    loop {
27        for rec in consumer.fetch().await? {
28            println!(
29                "{}-{}@{} bytes={}",
30                rec.topic,
31                rec.partition,
32                rec.offset,
33                rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
34            );
35        }
36    }
37}
More examples
Hide additional examples
examples/roundtrip.rs (line 23)
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/bench_latency.rs (line 158)
100async fn main() -> partitionline::Result<()> {
101    let bootstrap = env_or("KAFKA_BOOTSTRAP", "127.0.0.1:9092");
102    let topic = env_or("KAFKA_TOPIC", "pllat");
103    let payload = env_parse("PAYLOAD_BYTES", 100usize);
104    let warmup = env_parse("WARMUP", 1_000u64);
105    let count = env_parse("COUNT", 10_000u64);
106    let linger_ms = env_parse("LINGER_MS", 0u64);
107    let acks = env_parse("ACKS", 1i16);
108    let max_wait_ms = env_parse("MAX_WAIT_MS", 100i32);
109    let max_bytes = env_parse("MAX_BYTES", 4_096i32);
110    let min_bytes = env_parse("MIN_BYTES", 1i32);
111    let mode = env_or("MODE", "both");
112
113    let mut pcfg = ProducerConfig::bootstrap([bootstrap.clone()]);
114    pcfg.linger = Duration::from_millis(linger_ms);
115    pcfg.acks = acks;
116    pcfg.batch_records = 1;
117    pcfg.batch_bytes = payload.saturating_add(256).max(1);
118    pcfg.connections = 1;
119    pcfg.max_in_flight = 1;
120    let producer = Producer::new(pcfg).await?;
121    let value = Bytes::from(vec![b'x'; payload]);
122
123    if warmup > 0 {
124        let _ = produce_ack(&producer, &topic, &value, warmup).await?;
125    }
126
127    if mode == "produce" || mode == "both" {
128        let samples = produce_ack(&producer, &topic, &value, count).await?;
129        print_latency(
130            "produce_ack",
131            samples,
132            &format!(
133                ",\"payload_bytes\":{payload},\"acks\":{acks},\"linger_ms\":{linger_ms},\"client\":\"partitionline\""
134            ),
135        )?;
136    }
137    producer.close().await?;
138
139    if mode == "fetch" || mode == "both" {
140        if mode == "fetch" {
141            let producer = Producer::new({
142                let mut cfg = ProducerConfig::bootstrap([bootstrap.clone()]);
143                cfg.linger = Duration::ZERO;
144                cfg.acks = acks;
145                cfg.batch_records = 1;
146                cfg.connections = 1;
147                cfg.max_in_flight = 1;
148                cfg
149            })
150            .await?;
151            let _ = produce_ack(&producer, &topic, &value, count).await?;
152            producer.close().await?;
153        }
154        let mut ccfg = ConsumerConfig::bootstrap([bootstrap]);
155        ccfg.max_wait_ms = max_wait_ms;
156        ccfg.max_bytes = max_bytes;
157        ccfg.min_bytes = min_bytes;
158        let mut consumer = Consumer::new(ccfg).await?;
159        consumer.assign(&topic, 0, 0).await?;
160        let (samples, got) = fetch_rpc(&mut consumer, count).await?;
161        print_latency(
162            "fetch_rpc",
163            samples,
164            &format!(
165                ",\"consumed\":{got},\"max_wait_ms\":{max_wait_ms},\"max_bytes\":{max_bytes},\"min_bytes\":{min_bytes},\"client\":\"partitionline\""
166            ),
167        )?;
168    }
169    Ok(())
170}
examples/bench_fetch.rs (line 66)
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}
Source

pub async fn assign( &mut self, topic: impl Into<String>, partition: i32, offset: i64, ) -> Result<()>

Assign one partition at offset. Replaces a previous offset for the same pair.

Java assign calls crate::protocol::group::Topic::validate on the topic name.

Examples found in repository?
examples/roundtrip.rs (line 24)
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}
More examples
Hide additional examples
examples/offsets.rs (line 26)
11async fn main() -> partitionline::Result<()> {
12    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
13    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
14
15    let producer = Producer::new(
16        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
17    )
18    .await?;
19    let md = producer
20        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
21        .await?;
22    println!("produced {}-{}@{}", md.topic, md.partition, md.offset);
23    producer.close().await?;
24
25    let mut consumer = Consumer::connect(bootstrap.clone()).await?;
26    consumer.assign(&topic, 0, 0).await?;
27    let begin = consumer.beginning_offsets([(topic.as_str(), 0)]).await?;
28    let end = consumer.end_offsets([(topic.as_str(), 0)]).await?;
29    let lag = consumer.current_lag((topic.as_str(), 0)).await?;
30    println!("begin={begin:?} end={end:?} lag={lag:?}");
31    consumer.close().await?;
32
33    let mut group = ConsumerGroup::join(
34        ConsumerConfig::bootstrap([bootstrap]).max_wait_ms(500),
35        "partitionline-offsets",
36        topic.clone(),
37    )
38    .await?;
39    let recs = group.poll().await?;
40    if let Some(rec) = recs.first() {
41        let tp = TopicPartition::new(&rec.topic, rec.partition);
42        group
43            .commit_with_metadata([(
44                tp,
45                OffsetAndMetadata::with_metadata(rec.offset + 1, "example"),
46            )])
47            .await?;
48    }
49    for (tp, md) in group.committed().await? {
50        println!("{tp} committed={} meta={}", md.offset, md.metadata);
51    }
52    group.leave().await?;
53    Ok(())
54}
examples/bench_latency.rs (line 159)
100async fn main() -> partitionline::Result<()> {
101    let bootstrap = env_or("KAFKA_BOOTSTRAP", "127.0.0.1:9092");
102    let topic = env_or("KAFKA_TOPIC", "pllat");
103    let payload = env_parse("PAYLOAD_BYTES", 100usize);
104    let warmup = env_parse("WARMUP", 1_000u64);
105    let count = env_parse("COUNT", 10_000u64);
106    let linger_ms = env_parse("LINGER_MS", 0u64);
107    let acks = env_parse("ACKS", 1i16);
108    let max_wait_ms = env_parse("MAX_WAIT_MS", 100i32);
109    let max_bytes = env_parse("MAX_BYTES", 4_096i32);
110    let min_bytes = env_parse("MIN_BYTES", 1i32);
111    let mode = env_or("MODE", "both");
112
113    let mut pcfg = ProducerConfig::bootstrap([bootstrap.clone()]);
114    pcfg.linger = Duration::from_millis(linger_ms);
115    pcfg.acks = acks;
116    pcfg.batch_records = 1;
117    pcfg.batch_bytes = payload.saturating_add(256).max(1);
118    pcfg.connections = 1;
119    pcfg.max_in_flight = 1;
120    let producer = Producer::new(pcfg).await?;
121    let value = Bytes::from(vec![b'x'; payload]);
122
123    if warmup > 0 {
124        let _ = produce_ack(&producer, &topic, &value, warmup).await?;
125    }
126
127    if mode == "produce" || mode == "both" {
128        let samples = produce_ack(&producer, &topic, &value, count).await?;
129        print_latency(
130            "produce_ack",
131            samples,
132            &format!(
133                ",\"payload_bytes\":{payload},\"acks\":{acks},\"linger_ms\":{linger_ms},\"client\":\"partitionline\""
134            ),
135        )?;
136    }
137    producer.close().await?;
138
139    if mode == "fetch" || mode == "both" {
140        if mode == "fetch" {
141            let producer = Producer::new({
142                let mut cfg = ProducerConfig::bootstrap([bootstrap.clone()]);
143                cfg.linger = Duration::ZERO;
144                cfg.acks = acks;
145                cfg.batch_records = 1;
146                cfg.connections = 1;
147                cfg.max_in_flight = 1;
148                cfg
149            })
150            .await?;
151            let _ = produce_ack(&producer, &topic, &value, count).await?;
152            producer.close().await?;
153        }
154        let mut ccfg = ConsumerConfig::bootstrap([bootstrap]);
155        ccfg.max_wait_ms = max_wait_ms;
156        ccfg.max_bytes = max_bytes;
157        ccfg.min_bytes = min_bytes;
158        let mut consumer = Consumer::new(ccfg).await?;
159        consumer.assign(&topic, 0, 0).await?;
160        let (samples, got) = fetch_rpc(&mut consumer, count).await?;
161        print_latency(
162            "fetch_rpc",
163            samples,
164            &format!(
165                ",\"consumed\":{got},\"max_wait_ms\":{max_wait_ms},\"max_bytes\":{max_bytes},\"min_bytes\":{min_bytes},\"client\":\"partitionline\""
166            ),
167        )?;
168    }
169    Ok(())
170}
examples/metrics.rs (line 51)
9async fn main() -> partitionline::Result<()> {
10    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
11    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
12    let format = std::env::var("FORMAT").unwrap_or_else(|_| "log".into());
13
14    let producer = Producer::new(
15        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
16    )
17    .await?;
18    let md = producer
19        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
20        .await?;
21    let pm = producer.metrics();
22
23    if format == "prom" {
24        println!("# HELP partitionline_produce_records_acked Produce records acked");
25        println!("# TYPE partitionline_produce_records_acked counter");
26        println!("partitionline_produce_records_acked {}", pm.records_acked);
27        println!("# HELP partitionline_produce_ack_p99_seconds Produce ack p99 latency");
28        println!("# TYPE partitionline_produce_ack_p99_seconds gauge");
29        println!(
30            "partitionline_produce_ack_p99_seconds {}",
31            (pm.ack_latency.p99_nanos as f64) / 1_000_000_000.0
32        );
33    } else {
34        println!(
35            "produced {}-{}@{} queued={} acked={} bytes={} ack_us={} p50_us={} p99_us={} topics={}",
36            md.topic,
37            md.partition,
38            md.offset,
39            pm.records_queued,
40            pm.records_acked,
41            pm.bytes_queued,
42            pm.ack_latency.mean_nanos().unwrap_or(0) / 1000,
43            pm.ack_latency.p50_nanos / 1000,
44            pm.ack_latency.p99_nanos / 1000,
45            pm.topics.len()
46        );
47    }
48    producer.close().await?;
49
50    let mut consumer = Consumer::connect(bootstrap).await?;
51    consumer.assign(&topic, 0, 0).await?;
52    let recs = consumer.fetch().await?;
53    let cm = consumer.metrics();
54    if format == "prom" {
55        println!("# HELP partitionline_fetch_rounds Fetch rounds completed");
56        println!("# TYPE partitionline_fetch_rounds counter");
57        println!("partitionline_fetch_rounds {}", cm.fetch_rounds);
58        println!("# HELP partitionline_fetch_p99_seconds Fetch round p99 latency");
59        println!("# TYPE partitionline_fetch_p99_seconds gauge");
60        println!(
61            "partitionline_fetch_p99_seconds {}",
62            (cm.fetch_latency.p99_nanos as f64) / 1_000_000_000.0
63        );
64    } else {
65        println!(
66            "fetched {} records rounds={} bytes={} errors={} fetch_us={} p50_us={} p99_us={} topics={}",
67            recs.len(),
68            cm.fetch_rounds,
69            cm.bytes_fetched,
70            cm.fetch_errors,
71            cm.fetch_latency.mean_nanos().unwrap_or(0) / 1000,
72            cm.fetch_latency.p50_nanos / 1000,
73            cm.fetch_latency.p99_nanos / 1000,
74            cm.topics.len()
75        );
76    }
77    consumer.close().await?;
78    Ok(())
79}
Source

pub async fn assign_many( &mut self, starts: impl IntoIterator<Item = (impl Into<TopicPartition>, i64)>, ) -> Result<()>

Replace the assignment with these (partition, offset) pairs.

Source

pub async fn assign_partitions( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, ) -> Result<()>

Replace the assignment (Java assign(Collection)).

Offsets come from ConsumerConfig::auto_offset_reset via ListOffsets (earliest or latest). crate::AutoOffsetReset::None is an error (a manual consumer has no committed offsets). An empty list drops the assignment (Self::unassign). Each topic name is checked with crate::protocol::group::Topic::validate.

Waits up to ConsumerConfig::request_timeout. For a one-shot timeout, use Self::assign_partitions_timeout.

Source

pub async fn assign_partitions_timeout( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, timeout: Duration, ) -> Result<()>

Self::assign_partitions with a one-shot timeout for ListOffsets.

Source

pub fn unassign(&mut self)

Drop every assigned partition (Java unsubscribe for a manual consumer).

Source

pub async fn assign_topic( &mut self, topic: impl Into<String>, offset: i64, ) -> Result<()>

Assign every partition of topic at offset (from metadata).

Java assign calls crate::protocol::group::Topic::validate on the topic name.

Examples found in repository?
examples/consume.rs (line 10)
6async fn main() -> partitionline::Result<()> {
7    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
8    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
9    let mut consumer = Consumer::connect(bootstrap).await?;
10    consumer.assign_topic(topic, 0).await?;
11    loop {
12        for rec in consumer.fetch().await? {
13            println!(
14                "{}-{}@{} bytes={}",
15                rec.topic,
16                rec.partition,
17                rec.offset,
18                rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
19            );
20        }
21    }
22}
More examples
Hide additional examples
examples/consume_intercept.rs (line 25)
20async fn main() -> partitionline::Result<()> {
21    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
22    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
23    let mut consumer =
24        Consumer::new(ConsumerConfig::bootstrap([bootstrap]).interceptor(SkipHeartbeats)).await?;
25    consumer.assign_topic(topic, 0).await?;
26    loop {
27        for rec in consumer.fetch().await? {
28            println!(
29                "{}-{}@{} bytes={}",
30                rec.topic,
31                rec.partition,
32                rec.offset,
33                rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
34            );
35        }
36    }
37}
examples/pause.rs (line 12)
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 mut consumer = Consumer::connect(bootstrap).await?;
12    consumer.assign_topic(topic, 0).await?;
13    let assigned = consumer.assignment();
14    println!("assigned {assigned:?}");
15    if let Some(tp) = assigned.first().cloned() {
16        consumer.pause([tp.clone()]);
17        println!("paused {:?}", consumer.paused());
18        let recs = consumer.fetch().await?;
19        println!("while paused: {} records", recs.len());
20        consumer.resume([tp]);
21    }
22    let recs = consumer.fetch().await?;
23    println!("after resume: {} records", recs.len());
24    consumer.close().await?;
25    Ok(())
26}
examples/wakeup.rs (line 14)
10async fn main() -> partitionline::Result<()> {
11    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
12    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
13    let mut consumer = Consumer::connect(bootstrap).await?;
14    consumer.assign_topic(topic, 0).await?;
15    let wakeup = consumer.wakeup_handle();
16    drop(tokio::spawn(async move {
17        tokio::signal::ctrl_c().await.unwrap_or(());
18        wakeup.wakeup();
19    }));
20    loop {
21        match consumer.fetch().await {
22            Ok(recs) => {
23                for rec in recs {
24                    println!("{}-{}@{}", rec.topic, rec.partition, rec.offset);
25                }
26            }
27            Err(partitionline::Error::Wakeup) => break,
28            Err(e) => return Err(e),
29        }
30    }
31    consumer.close().await?;
32    Ok(())
33}
examples/bench_fetch.rs (line 67)
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}
Source

pub fn assignment(&self) -> Vec<TopicPartition>

Assigned partitions (Java assignment). Offsets are Self::positions.

Examples found in repository?
examples/pause.rs (line 13)
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 mut consumer = Consumer::connect(bootstrap).await?;
12    consumer.assign_topic(topic, 0).await?;
13    let assigned = consumer.assignment();
14    println!("assigned {assigned:?}");
15    if let Some(tp) = assigned.first().cloned() {
16        consumer.pause([tp.clone()]);
17        println!("paused {:?}", consumer.paused());
18        let recs = consumer.fetch().await?;
19        println!("while paused: {} records", recs.len());
20        consumer.resume([tp]);
21    }
22    let recs = consumer.fetch().await?;
23    println!("after resume: {} records", recs.len());
24    consumer.close().await?;
25    Ok(())
26}
More examples
Hide additional examples
examples/bench_fetch.rs (line 68)
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}
Source

pub fn assigned_partitions(&self) -> Vec<TopicPartition>

Same as Self::assignment.

Source

pub fn positions(&self) -> Vec<(TopicPartition, i64)>

Assigned partitions with their next fetch offsets.

Source

pub fn position(&self, topic: &str, partition: i32) -> Result<i64>

Next fetch offset for an assigned partition.

An unassigned partition is Java IllegalStateException (You can only check the position for partitions assigned to this consumer.).

Source

pub fn position_of(&self, partition: impl Into<TopicPartition>) -> Result<i64>

Self::position for a TopicPartition.

Source

pub fn pause( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, )

Stop fetching these assigned partitions until resume.

Pause is stored on the consumer, so it survives group rebalance. Fetch skips a partition only while it is both assigned and paused. Records already buffered for a paused partition are held until resume.

Examples found in repository?
examples/pause.rs (line 16)
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 mut consumer = Consumer::connect(bootstrap).await?;
12    consumer.assign_topic(topic, 0).await?;
13    let assigned = consumer.assignment();
14    println!("assigned {assigned:?}");
15    if let Some(tp) = assigned.first().cloned() {
16        consumer.pause([tp.clone()]);
17        println!("paused {:?}", consumer.paused());
18        let recs = consumer.fetch().await?;
19        println!("while paused: {} records", recs.len());
20        consumer.resume([tp]);
21    }
22    let recs = consumer.fetch().await?;
23    println!("after resume: {} records", recs.len());
24    consumer.close().await?;
25    Ok(())
26}
Source

pub fn resume( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, )

Undo pause for these partitions.

Examples found in repository?
examples/pause.rs (line 20)
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 mut consumer = Consumer::connect(bootstrap).await?;
12    consumer.assign_topic(topic, 0).await?;
13    let assigned = consumer.assignment();
14    println!("assigned {assigned:?}");
15    if let Some(tp) = assigned.first().cloned() {
16        consumer.pause([tp.clone()]);
17        println!("paused {:?}", consumer.paused());
18        let recs = consumer.fetch().await?;
19        println!("while paused: {} records", recs.len());
20        consumer.resume([tp]);
21    }
22    let recs = consumer.fetch().await?;
23    println!("after resume: {} records", recs.len());
24    consumer.close().await?;
25    Ok(())
26}
Source

pub fn paused(&self) -> Vec<TopicPartition>

Assigned partitions that fetch currently skips.

Examples found in repository?
examples/pause.rs (line 17)
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 mut consumer = Consumer::connect(bootstrap).await?;
12    consumer.assign_topic(topic, 0).await?;
13    let assigned = consumer.assignment();
14    println!("assigned {assigned:?}");
15    if let Some(tp) = assigned.first().cloned() {
16        consumer.pause([tp.clone()]);
17        println!("paused {:?}", consumer.paused());
18        let recs = consumer.fetch().await?;
19        println!("while paused: {} records", recs.len());
20        consumer.resume([tp]);
21    }
22    let recs = consumer.fetch().await?;
23    println!("after resume: {} records", recs.len());
24    consumer.close().await?;
25    Ok(())
26}
Source

pub async fn fetch(&mut self) -> Result<ConsumerRecords>

Fetch one round from every assigned partition that is not paused.

Returns ConsumerRecords, which indexes like a slice of FetchedRecord. Empty when every assigned partition is paused. Nothing assigned is Java IllegalStateException (Consumer is not subscribed to any topics or assigned any partitions). Partitions that share a leader go in one Fetch. Distinct leaders are fetched at the same time.

When ConsumerConfig::max_poll_records is set, extra records from the Fetch stay buffered and are returned on the next call.

Examples found in repository?
examples/consume.rs (line 12)
6async fn main() -> partitionline::Result<()> {
7    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
8    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
9    let mut consumer = Consumer::connect(bootstrap).await?;
10    consumer.assign_topic(topic, 0).await?;
11    loop {
12        for rec in consumer.fetch().await? {
13            println!(
14                "{}-{}@{} bytes={}",
15                rec.topic,
16                rec.partition,
17                rec.offset,
18                rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
19            );
20        }
21    }
22}
More examples
Hide additional examples
examples/bench_latency.rs (line 83)
77async fn fetch_rpc(consumer: &mut Consumer, count: u64) -> partitionline::Result<(Vec<u64>, u64)> {
78    let mut samples = Vec::new();
79    let mut got = 0u64;
80    let mut empty = 0u32;
81    while got < count {
82        let start = Instant::now();
83        let recs = consumer.fetch().await?;
84        let elapsed = u64::try_from(start.elapsed().as_micros()).unwrap_or(u64::MAX);
85        if recs.is_empty() {
86            empty += 1;
87            if empty > 600 {
88                return Err(partitionline::Error::Timeout);
89            }
90            continue;
91        }
92        empty = 0;
93        samples.push(elapsed);
94        got += recs.len() as u64;
95    }
96    Ok((samples, got))
97}
examples/consume_intercept.rs (line 27)
20async fn main() -> partitionline::Result<()> {
21    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
22    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
23    let mut consumer =
24        Consumer::new(ConsumerConfig::bootstrap([bootstrap]).interceptor(SkipHeartbeats)).await?;
25    consumer.assign_topic(topic, 0).await?;
26    loop {
27        for rec in consumer.fetch().await? {
28            println!(
29                "{}-{}@{} bytes={}",
30                rec.topic,
31                rec.partition,
32                rec.offset,
33                rec.value.as_ref().map(|v| v.len()).unwrap_or(0)
34            );
35        }
36    }
37}
examples/pause.rs (line 18)
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 mut consumer = Consumer::connect(bootstrap).await?;
12    consumer.assign_topic(topic, 0).await?;
13    let assigned = consumer.assignment();
14    println!("assigned {assigned:?}");
15    if let Some(tp) = assigned.first().cloned() {
16        consumer.pause([tp.clone()]);
17        println!("paused {:?}", consumer.paused());
18        let recs = consumer.fetch().await?;
19        println!("while paused: {} records", recs.len());
20        consumer.resume([tp]);
21    }
22    let recs = consumer.fetch().await?;
23    println!("after resume: {} records", recs.len());
24    consumer.close().await?;
25    Ok(())
26}
examples/wakeup.rs (line 21)
10async fn main() -> partitionline::Result<()> {
11    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
12    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
13    let mut consumer = Consumer::connect(bootstrap).await?;
14    consumer.assign_topic(topic, 0).await?;
15    let wakeup = consumer.wakeup_handle();
16    drop(tokio::spawn(async move {
17        tokio::signal::ctrl_c().await.unwrap_or(());
18        wakeup.wakeup();
19    }));
20    loop {
21        match consumer.fetch().await {
22            Ok(recs) => {
23                for rec in recs {
24                    println!("{}-{}@{}", rec.topic, rec.partition, rec.offset);
25                }
26            }
27            Err(partitionline::Error::Wakeup) => break,
28            Err(e) => return Err(e),
29        }
30    }
31    consumer.close().await?;
32    Ok(())
33}
examples/roundtrip.rs (line 25)
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}
Additional examples can be found in:
Source

pub async fn fetch_timeout( &mut self, timeout: Duration, ) -> Result<ConsumerRecords>

Fetch with a one-shot fetch.max.wait.ms (Java poll(Duration)).

ConsumerConfig::max_wait_ms is restored afterwards. Nothing assigned is the same Java IllegalStateException as Self::fetch.

Source

pub fn metrics(&self) -> ConsumerMetrics

Fetch counters and round latency since connect (min/mean/max and p50/p99).

crate::ConsumerMetrics::topics is one row per topic that returned at least one record.

Examples found in repository?
examples/metrics.rs (line 53)
9async fn main() -> partitionline::Result<()> {
10    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
11    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
12    let format = std::env::var("FORMAT").unwrap_or_else(|_| "log".into());
13
14    let producer = Producer::new(
15        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
16    )
17    .await?;
18    let md = producer
19        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
20        .await?;
21    let pm = producer.metrics();
22
23    if format == "prom" {
24        println!("# HELP partitionline_produce_records_acked Produce records acked");
25        println!("# TYPE partitionline_produce_records_acked counter");
26        println!("partitionline_produce_records_acked {}", pm.records_acked);
27        println!("# HELP partitionline_produce_ack_p99_seconds Produce ack p99 latency");
28        println!("# TYPE partitionline_produce_ack_p99_seconds gauge");
29        println!(
30            "partitionline_produce_ack_p99_seconds {}",
31            (pm.ack_latency.p99_nanos as f64) / 1_000_000_000.0
32        );
33    } else {
34        println!(
35            "produced {}-{}@{} queued={} acked={} bytes={} ack_us={} p50_us={} p99_us={} topics={}",
36            md.topic,
37            md.partition,
38            md.offset,
39            pm.records_queued,
40            pm.records_acked,
41            pm.bytes_queued,
42            pm.ack_latency.mean_nanos().unwrap_or(0) / 1000,
43            pm.ack_latency.p50_nanos / 1000,
44            pm.ack_latency.p99_nanos / 1000,
45            pm.topics.len()
46        );
47    }
48    producer.close().await?;
49
50    let mut consumer = Consumer::connect(bootstrap).await?;
51    consumer.assign(&topic, 0, 0).await?;
52    let recs = consumer.fetch().await?;
53    let cm = consumer.metrics();
54    if format == "prom" {
55        println!("# HELP partitionline_fetch_rounds Fetch rounds completed");
56        println!("# TYPE partitionline_fetch_rounds counter");
57        println!("partitionline_fetch_rounds {}", cm.fetch_rounds);
58        println!("# HELP partitionline_fetch_p99_seconds Fetch round p99 latency");
59        println!("# TYPE partitionline_fetch_p99_seconds gauge");
60        println!(
61            "partitionline_fetch_p99_seconds {}",
62            (cm.fetch_latency.p99_nanos as f64) / 1_000_000_000.0
63        );
64    } else {
65        println!(
66            "fetched {} records rounds={} bytes={} errors={} fetch_us={} p50_us={} p99_us={} topics={}",
67            recs.len(),
68            cm.fetch_rounds,
69            cm.bytes_fetched,
70            cm.fetch_errors,
71            cm.fetch_latency.mean_nanos().unwrap_or(0) / 1000,
72            cm.fetch_latency.p50_nanos / 1000,
73            cm.fetch_latency.p99_nanos / 1000,
74            cm.topics.len()
75        );
76    }
77    consumer.close().await?;
78    Ok(())
79}
Source

pub async fn client_instance_id(&mut self) -> Result<Uuid>

Java clientInstanceId (KIP-714 GetTelemetrySubscriptions).

Returns crate::Uuid (Java Uuid). The first call sends a zero UUID; the broker assigns one. Later calls return the cached id without another round-trip. Waits up to ConsumerConfig::request_timeout. For a one-shot timeout, use Self::client_instance_id_timeout.

Source

pub async fn client_instance_id_timeout( &mut self, timeout: Duration, ) -> Result<Uuid>

Self::client_instance_id with a one-shot timeout (Java clientInstanceId(Duration)).

timeout is the GetTelemetrySubscriptions RPC deadline. Cached after the first successful call; later calls ignore timeout.

Source

pub fn wakeup(&self)

Interrupt Self::fetch (and group poll that calls it).

Safe to call while fetch is running on this task. From another task, use Self::wakeup_handle.

Source

pub fn wakeup_handle(&self) -> WakeupHandle

Cloneable handle for Self::wakeup from another task.

Examples found in repository?
examples/wakeup.rs (line 15)
10async fn main() -> partitionline::Result<()> {
11    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
12    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
13    let mut consumer = Consumer::connect(bootstrap).await?;
14    consumer.assign_topic(topic, 0).await?;
15    let wakeup = consumer.wakeup_handle();
16    drop(tokio::spawn(async move {
17        tokio::signal::ctrl_c().await.unwrap_or(());
18        wakeup.wakeup();
19    }));
20    loop {
21        match consumer.fetch().await {
22            Ok(recs) => {
23                for rec in recs {
24                    println!("{}-{}@{}", rec.topic, rec.partition, rec.offset);
25                }
26            }
27            Err(partitionline::Error::Wakeup) => break,
28            Err(e) => return Err(e),
29        }
30    }
31    consumer.close().await?;
32    Ok(())
33}
Source

pub async fn close(self) -> Result<()>

Drop fetch connections. The consumer is then gone (same as Producer::close).

Examples found in repository?
examples/pause.rs (line 24)
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 mut consumer = Consumer::connect(bootstrap).await?;
12    consumer.assign_topic(topic, 0).await?;
13    let assigned = consumer.assignment();
14    println!("assigned {assigned:?}");
15    if let Some(tp) = assigned.first().cloned() {
16        consumer.pause([tp.clone()]);
17        println!("paused {:?}", consumer.paused());
18        let recs = consumer.fetch().await?;
19        println!("while paused: {} records", recs.len());
20        consumer.resume([tp]);
21    }
22    let recs = consumer.fetch().await?;
23    println!("after resume: {} records", recs.len());
24    consumer.close().await?;
25    Ok(())
26}
More examples
Hide additional examples
examples/wakeup.rs (line 31)
10async fn main() -> partitionline::Result<()> {
11    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
12    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
13    let mut consumer = Consumer::connect(bootstrap).await?;
14    consumer.assign_topic(topic, 0).await?;
15    let wakeup = consumer.wakeup_handle();
16    drop(tokio::spawn(async move {
17        tokio::signal::ctrl_c().await.unwrap_or(());
18        wakeup.wakeup();
19    }));
20    loop {
21        match consumer.fetch().await {
22            Ok(recs) => {
23                for rec in recs {
24                    println!("{}-{}@{}", rec.topic, rec.partition, rec.offset);
25                }
26            }
27            Err(partitionline::Error::Wakeup) => break,
28            Err(e) => return Err(e),
29        }
30    }
31    consumer.close().await?;
32    Ok(())
33}
examples/offsets.rs (line 31)
11async fn main() -> partitionline::Result<()> {
12    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
13    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
14
15    let producer = Producer::new(
16        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
17    )
18    .await?;
19    let md = producer
20        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
21        .await?;
22    println!("produced {}-{}@{}", md.topic, md.partition, md.offset);
23    producer.close().await?;
24
25    let mut consumer = Consumer::connect(bootstrap.clone()).await?;
26    consumer.assign(&topic, 0, 0).await?;
27    let begin = consumer.beginning_offsets([(topic.as_str(), 0)]).await?;
28    let end = consumer.end_offsets([(topic.as_str(), 0)]).await?;
29    let lag = consumer.current_lag((topic.as_str(), 0)).await?;
30    println!("begin={begin:?} end={end:?} lag={lag:?}");
31    consumer.close().await?;
32
33    let mut group = ConsumerGroup::join(
34        ConsumerConfig::bootstrap([bootstrap]).max_wait_ms(500),
35        "partitionline-offsets",
36        topic.clone(),
37    )
38    .await?;
39    let recs = group.poll().await?;
40    if let Some(rec) = recs.first() {
41        let tp = TopicPartition::new(&rec.topic, rec.partition);
42        group
43            .commit_with_metadata([(
44                tp,
45                OffsetAndMetadata::with_metadata(rec.offset + 1, "example"),
46            )])
47            .await?;
48    }
49    for (tp, md) in group.committed().await? {
50        println!("{tp} committed={} meta={}", md.offset, md.metadata);
51    }
52    group.leave().await?;
53    Ok(())
54}
examples/metrics.rs (line 77)
9async fn main() -> partitionline::Result<()> {
10    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
11    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
12    let format = std::env::var("FORMAT").unwrap_or_else(|_| "log".into());
13
14    let producer = Producer::new(
15        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
16    )
17    .await?;
18    let md = producer
19        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
20        .await?;
21    let pm = producer.metrics();
22
23    if format == "prom" {
24        println!("# HELP partitionline_produce_records_acked Produce records acked");
25        println!("# TYPE partitionline_produce_records_acked counter");
26        println!("partitionline_produce_records_acked {}", pm.records_acked);
27        println!("# HELP partitionline_produce_ack_p99_seconds Produce ack p99 latency");
28        println!("# TYPE partitionline_produce_ack_p99_seconds gauge");
29        println!(
30            "partitionline_produce_ack_p99_seconds {}",
31            (pm.ack_latency.p99_nanos as f64) / 1_000_000_000.0
32        );
33    } else {
34        println!(
35            "produced {}-{}@{} queued={} acked={} bytes={} ack_us={} p50_us={} p99_us={} topics={}",
36            md.topic,
37            md.partition,
38            md.offset,
39            pm.records_queued,
40            pm.records_acked,
41            pm.bytes_queued,
42            pm.ack_latency.mean_nanos().unwrap_or(0) / 1000,
43            pm.ack_latency.p50_nanos / 1000,
44            pm.ack_latency.p99_nanos / 1000,
45            pm.topics.len()
46        );
47    }
48    producer.close().await?;
49
50    let mut consumer = Consumer::connect(bootstrap).await?;
51    consumer.assign(&topic, 0, 0).await?;
52    let recs = consumer.fetch().await?;
53    let cm = consumer.metrics();
54    if format == "prom" {
55        println!("# HELP partitionline_fetch_rounds Fetch rounds completed");
56        println!("# TYPE partitionline_fetch_rounds counter");
57        println!("partitionline_fetch_rounds {}", cm.fetch_rounds);
58        println!("# HELP partitionline_fetch_p99_seconds Fetch round p99 latency");
59        println!("# TYPE partitionline_fetch_p99_seconds gauge");
60        println!(
61            "partitionline_fetch_p99_seconds {}",
62            (cm.fetch_latency.p99_nanos as f64) / 1_000_000_000.0
63        );
64    } else {
65        println!(
66            "fetched {} records rounds={} bytes={} errors={} fetch_us={} p50_us={} p99_us={} topics={}",
67            recs.len(),
68            cm.fetch_rounds,
69            cm.bytes_fetched,
70            cm.fetch_errors,
71            cm.fetch_latency.mean_nanos().unwrap_or(0) / 1000,
72            cm.fetch_latency.p50_nanos / 1000,
73            cm.fetch_latency.p99_nanos / 1000,
74            cm.topics.len()
75        );
76    }
77    consumer.close().await?;
78    Ok(())
79}
Source

pub async fn close_timeout(self, _timeout: Duration) -> Result<()>

Drop fetch connections (Java close(Duration)).

A manual consumer has no LeaveGroup RPC; this is the same as Self::close. Group and share members use crate::ConsumerGroup::close_timeout / crate::ShareGroup::close_timeout.

Source

pub fn versions(&self) -> &HashMap<i16, ApiVersion>

Negotiated ApiVersions for this connection.

Source

pub async fn list_offsets( &mut self, topic: impl Into<String>, partition: i32, timestamp: i64, ) -> Result<i64>

ListOffsets timestamp: crate::EARLIEST_TIMESTAMP (-2), crate::LATEST_TIMESTAMP (-1), crate::MAX_TIMESTAMP (-3), crate::EARLIEST_LOCAL_TIMESTAMP (-4), crate::LATEST_TIERED_TIMESTAMP (-5), or milliseconds.

Negotiates ListOffsets v1–v10 (v6–v10 flexible; v10 TimeoutMs). Waits up to ConsumerConfig::request_timeout. For a one-shot timeout, use Self::list_offsets_timeout.

Source

pub async fn list_offsets_timeout( &mut self, topic: impl Into<String>, partition: i32, timestamp: i64, timeout: Duration, ) -> Result<i64>

Self::list_offsets with a one-shot timeout.

Source

pub async fn list_offset( &mut self, partition: impl Into<TopicPartition>, timestamp: i64, ) -> Result<i64>

Self::list_offsets for a TopicPartition.

Source

pub async fn list_offset_timeout( &mut self, partition: impl Into<TopicPartition>, timestamp: i64, timeout: Duration, ) -> Result<i64>

Self::list_offset with a one-shot timeout.

Source

pub fn seek(&mut self, topic: &str, partition: i32, offset: i64) -> Result<()>

Set the next fetch offset for an assigned partition (Java seek(TopicPartition, long)).

A negative offset is Java IllegalArgumentException (seek offset must not be a negative number). An unassigned partition is Java IllegalStateException (No current assignment for partition). Clears Fetch LastFetchedEpoch (KIP-320). To keep a leader epoch, use Self::seek_with_metadata.

Source

pub fn seek_to( &mut self, partition: impl Into<TopicPartition>, offset: i64, ) -> Result<()>

Self::seek for a TopicPartition.

Source

pub fn seek_with_metadata( &mut self, partition: impl Into<TopicPartition>, offset: impl Into<OffsetAndMetadata>, ) -> Result<()>

Seek using OffsetAndMetadata (Java seek(TopicPartition, OffsetAndMetadata)).

The offset is the next fetch position. A negative offset and an unassigned partition use the same Java messages as Self::seek. The leader epoch is sent as Fetch LastFetchedEpoch (KIP-320). Unknown epoch (None) clears it, matching Java Optional.empty(). The metadata string is ignored (Java does the same).

Source

pub async fn seek_to_beginning(&mut self) -> Result<()>

Seek every assigned partition to the log start (ListOffsets earliest).

Source

pub async fn seek_to_beginning_of( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, ) -> Result<()>

Seek these assigned partitions to the log start (Java seekToBeginning).

Source

pub async fn seek_to_end(&mut self) -> Result<()>

Seek every assigned partition to the high watermark (ListOffsets latest).

Source

pub async fn seek_to_end_of( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, ) -> Result<()>

Seek these assigned partitions to the high watermark (Java seekToEnd).

Source

pub async fn partitions_for( &mut self, topic: impl Into<String>, ) -> Result<Vec<PartitionInfo>>

Partition metadata for topic (Java partitionsFor: leader, replicas, ISR, offline replicas, leader epoch).

Waits up to ConsumerConfig::request_timeout. For a one-shot timeout, use Self::partitions_for_timeout.

Source

pub async fn partitions_for_timeout( &mut self, topic: impl Into<String>, timeout: Duration, ) -> Result<Vec<PartitionInfo>>

Self::partitions_for with a one-shot timeout (Java partitionsFor(String, Duration)).

Source

pub async fn list_topics(&mut self) -> Result<Vec<PartitionInfo>>

Cluster Metadata for every topic (Java listTopics).

Waits up to ConsumerConfig::request_timeout. For a one-shot timeout, use Self::list_topics_timeout.

Source

pub async fn list_topics_timeout( &mut self, timeout: Duration, ) -> Result<Vec<PartitionInfo>>

Self::list_topics with a one-shot timeout (Java listTopics(Duration)).

Source

pub async fn beginning_offsets( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, ) -> Result<Vec<(TopicPartition, i64)>>

Log-start offset for each partition (ListOffsets earliest).

Waits up to ConsumerConfig::request_timeout. For a one-shot timeout, use Self::beginning_offsets_timeout.

Examples found in repository?
examples/offsets.rs (line 27)
11async fn main() -> partitionline::Result<()> {
12    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
13    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
14
15    let producer = Producer::new(
16        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
17    )
18    .await?;
19    let md = producer
20        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
21        .await?;
22    println!("produced {}-{}@{}", md.topic, md.partition, md.offset);
23    producer.close().await?;
24
25    let mut consumer = Consumer::connect(bootstrap.clone()).await?;
26    consumer.assign(&topic, 0, 0).await?;
27    let begin = consumer.beginning_offsets([(topic.as_str(), 0)]).await?;
28    let end = consumer.end_offsets([(topic.as_str(), 0)]).await?;
29    let lag = consumer.current_lag((topic.as_str(), 0)).await?;
30    println!("begin={begin:?} end={end:?} lag={lag:?}");
31    consumer.close().await?;
32
33    let mut group = ConsumerGroup::join(
34        ConsumerConfig::bootstrap([bootstrap]).max_wait_ms(500),
35        "partitionline-offsets",
36        topic.clone(),
37    )
38    .await?;
39    let recs = group.poll().await?;
40    if let Some(rec) = recs.first() {
41        let tp = TopicPartition::new(&rec.topic, rec.partition);
42        group
43            .commit_with_metadata([(
44                tp,
45                OffsetAndMetadata::with_metadata(rec.offset + 1, "example"),
46            )])
47            .await?;
48    }
49    for (tp, md) in group.committed().await? {
50        println!("{tp} committed={} meta={}", md.offset, md.metadata);
51    }
52    group.leave().await?;
53    Ok(())
54}
Source

pub async fn beginning_offsets_timeout( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, timeout: Duration, ) -> Result<Vec<(TopicPartition, i64)>>

Self::beginning_offsets with a one-shot timeout (Java beginningOffsets(Collection, Duration)).

Source

pub async fn end_offsets( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, ) -> Result<Vec<(TopicPartition, i64)>>

High-watermark offset for each partition (ListOffsets latest).

Waits up to ConsumerConfig::request_timeout. For a one-shot timeout, use Self::end_offsets_timeout.

Examples found in repository?
examples/offsets.rs (line 28)
11async fn main() -> partitionline::Result<()> {
12    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
13    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
14
15    let producer = Producer::new(
16        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
17    )
18    .await?;
19    let md = producer
20        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
21        .await?;
22    println!("produced {}-{}@{}", md.topic, md.partition, md.offset);
23    producer.close().await?;
24
25    let mut consumer = Consumer::connect(bootstrap.clone()).await?;
26    consumer.assign(&topic, 0, 0).await?;
27    let begin = consumer.beginning_offsets([(topic.as_str(), 0)]).await?;
28    let end = consumer.end_offsets([(topic.as_str(), 0)]).await?;
29    let lag = consumer.current_lag((topic.as_str(), 0)).await?;
30    println!("begin={begin:?} end={end:?} lag={lag:?}");
31    consumer.close().await?;
32
33    let mut group = ConsumerGroup::join(
34        ConsumerConfig::bootstrap([bootstrap]).max_wait_ms(500),
35        "partitionline-offsets",
36        topic.clone(),
37    )
38    .await?;
39    let recs = group.poll().await?;
40    if let Some(rec) = recs.first() {
41        let tp = TopicPartition::new(&rec.topic, rec.partition);
42        group
43            .commit_with_metadata([(
44                tp,
45                OffsetAndMetadata::with_metadata(rec.offset + 1, "example"),
46            )])
47            .await?;
48    }
49    for (tp, md) in group.committed().await? {
50        println!("{tp} committed={} meta={}", md.offset, md.metadata);
51    }
52    group.leave().await?;
53    Ok(())
54}
Source

pub async fn end_offsets_timeout( &mut self, partitions: impl IntoIterator<Item = impl Into<TopicPartition>>, timeout: Duration, ) -> Result<Vec<(TopicPartition, i64)>>

Self::end_offsets with a one-shot timeout (Java endOffsets(Collection, Duration)).

Source

pub async fn current_lag( &mut self, partition: impl Into<TopicPartition>, ) -> Result<Option<i64>>

High watermark minus position (Java currentLag).

None when the high watermark is unknown (-1). An unassigned partition is Java IllegalStateException (No current assignment for partition).

Examples found in repository?
examples/offsets.rs (line 29)
11async fn main() -> partitionline::Result<()> {
12    let bootstrap = std::env::var("KAFKA_BOOTSTRAP").unwrap_or_else(|_| "127.0.0.1:9092".into());
13    let topic = std::env::var("KAFKA_TOPIC").unwrap_or_else(|_| "partitionline".into());
14
15    let producer = Producer::new(
16        ProducerConfig::bootstrap([bootstrap.clone()]).linger(std::time::Duration::ZERO),
17    )
18    .await?;
19    let md = producer
20        .send(ProduceRecord::to(topic.clone()).value(&b"hello"[..]))
21        .await?;
22    println!("produced {}-{}@{}", md.topic, md.partition, md.offset);
23    producer.close().await?;
24
25    let mut consumer = Consumer::connect(bootstrap.clone()).await?;
26    consumer.assign(&topic, 0, 0).await?;
27    let begin = consumer.beginning_offsets([(topic.as_str(), 0)]).await?;
28    let end = consumer.end_offsets([(topic.as_str(), 0)]).await?;
29    let lag = consumer.current_lag((topic.as_str(), 0)).await?;
30    println!("begin={begin:?} end={end:?} lag={lag:?}");
31    consumer.close().await?;
32
33    let mut group = ConsumerGroup::join(
34        ConsumerConfig::bootstrap([bootstrap]).max_wait_ms(500),
35        "partitionline-offsets",
36        topic.clone(),
37    )
38    .await?;
39    let recs = group.poll().await?;
40    if let Some(rec) = recs.first() {
41        let tp = TopicPartition::new(&rec.topic, rec.partition);
42        group
43            .commit_with_metadata([(
44                tp,
45                OffsetAndMetadata::with_metadata(rec.offset + 1, "example"),
46            )])
47            .await?;
48    }
49    for (tp, md) in group.committed().await? {
50        println!("{tp} committed={} meta={}", md.offset, md.metadata);
51    }
52    group.leave().await?;
53    Ok(())
54}
Source

pub async fn current_lag_timeout( &mut self, partition: impl Into<TopicPartition>, timeout: Duration, ) -> Result<Option<i64>>

Self::current_lag with a one-shot timeout for the ListOffsets RPC.

An unassigned partition is the same Java IllegalStateException as Self::current_lag.

Source

pub async fn offsets_for_times( &mut self, queries: impl IntoIterator<Item = (impl Into<TopicPartition>, i64)>, ) -> Result<Vec<(TopicPartition, Option<OffsetAndTimestamp>)>>

First offset at or after each timestamp (Java offsetsForTimes).

A negative timestamp is Java IllegalArgumentException (The target time cannot be negative). Use Self::beginning_offsets / Self::end_offsets (or Self::list_offsets with crate::EARLIEST_TIMESTAMP / crate::LATEST_TIMESTAMP) for those sentinels. Partitions with no matching record return None. OffsetAndTimestamp::leader_epoch is Java getLeaderEpoch. Waits up to ConsumerConfig::request_timeout. For a one-shot timeout, use Self::offsets_for_times_timeout.

Source

pub async fn offsets_for_times_timeout( &mut self, queries: impl IntoIterator<Item = (impl Into<TopicPartition>, i64)>, timeout: Duration, ) -> Result<Vec<(TopicPartition, Option<OffsetAndTimestamp>)>>

Self::offsets_for_times with a one-shot timeout (Java offsetsForTimes(Map, Duration)).

Auto Trait Implementations§

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> 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, 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