krafka 0.5.0

A pure Rust, async-native Apache Kafka client
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

//! Advanced consumer example.
//!
//! Demonstrates advanced consumer features:
//! - Pause and resume partition consumption
//! - Seek to specific offsets
//! - Manual offset commits
//! - Position tracking
//!
//! Run with:
//! ```
//! cargo run --example consumer_advanced
//! ```

use std::time::Duration;

use krafka::consumer::{AutoOffsetReset, Consumer, ConsumerRecord};
use krafka::error::Result;

#[tokio::main]
async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
    // Initialize tracing
    tracing_subscriber::fmt()
        .with_env_filter("krafka=debug")
        .init();

    // Create consumer with manual commit for precise offset control
    let consumer: Consumer = Consumer::builder()
        .bootstrap_servers("localhost:9092")
        .group_id("krafka-advanced-consumer")
        .auto_offset_reset(AutoOffsetReset::Earliest)
        .enable_auto_commit(false) // Manual commit for precise control
        .build()
        .await?;

    println!("Consumer connected!");

    // Subscribe to topics
    consumer.subscribe(&["test-topic"]).await?;

    // Get assigned partitions
    let assignments = consumer.assignment().await;
    println!("Assigned partitions: {:?}", assignments);

    let mut message_count = 0;
    let mut batch_count = 0;

    loop {
        let poll_result: Result<Vec<ConsumerRecord>> =
            consumer.poll(Duration::from_millis(500)).await;
        match poll_result {
            Ok(records) => {
                if records.is_empty() {
                    continue;
                }

                batch_count += 1;
                println!("\n--- Batch {} ---", batch_count);

                for record in &records {
                    message_count += 1;
                    println!(
                        "[{}] {} partition={} offset={}: {:?}",
                        message_count,
                        record.topic,
                        record.partition,
                        record.offset,
                        record.value_str().unwrap_or("<binary>"),
                    );

                    // Example: Pause partition after receiving 5 messages from it
                    if message_count == 5 {
                        println!(
                            "\n>>> Pausing partition {} for demonstration...",
                            record.partition
                        );
                        consumer.pause(&record.topic, &[record.partition]).await;

                        // Show paused partitions
                        let paused = consumer.paused_partitions().await;
                        println!(">>> Paused partitions: {:?}", paused);
                    }

                    // Example: Resume after 10 messages
                    if message_count == 10 {
                        println!("\n>>> Resuming all paused partitions...");
                        let paused = consumer.paused_partitions().await;
                        for (topic, partition) in paused {
                            consumer.resume(&topic, &[partition]).await;
                        }
                    }

                    // Example: Seek to beginning after 15 messages
                    if message_count == 15 {
                        println!(
                            "\n>>> Seeking to beginning of partition {}...",
                            record.partition
                        );
                        consumer
                            .seek_to_beginning(&record.topic, record.partition)
                            .await?;
                    }
                }

                // Manual commit after processing batch
                println!("Committing offsets for {} records...", records.len());
                consumer.commit().await?;

                // Show current positions
                for (topic, partitions) in consumer.assignment().await {
                    for partition in partitions {
                        if let Some(pos) = consumer.position(&topic, partition).await {
                            println!("  Position {}-{}: {}", topic, partition, pos);
                        }
                    }
                }

                // Stop after processing 20 messages for demo
                if message_count >= 20 {
                    println!("\n>>> Demo complete! Processed {} messages.", message_count);
                    break;
                }
            }
            Err(e) => {
                eprintln!("Error polling: {}", e);
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }

    // Clean shutdown
    consumer.commit().await?;
    consumer.close().await;
    println!("Consumer closed gracefully.");

    Ok(())
}