talos_certifier_adapters 0.2.44

Adapters used in Talos Certifier
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

use futures_executor::block_on;
use log::{error, info};
use rdkafka::{
    consumer::{ConsumerContext, Rebalance},
    ClientContext,
};
use talos_certifier::{model::CandidateMessage, ChannelMessage};
use tokio::sync::mpsc;

/// Certifier's consumer context used to detect a rebalance and send a `ChannelMessage::Reset` to reset the suffix in `certifier service`.
pub struct CertifierConsumerContext {
    pub topic: String,
    pub message_channel_tx: mpsc::Sender<ChannelMessage<CandidateMessage>>,
}

impl ClientContext for CertifierConsumerContext {}
impl ConsumerContext for CertifierConsumerContext {
    fn post_rebalance(&self, rebalance: &rdkafka::consumer::Rebalance<'_>) {
        match rebalance {
            Rebalance::Assign(partitions) => {
                let k = partitions.elements_for_topic(&self.topic);
                if !k.is_empty() {
                    info!("Assigned to partition of certification topic: {:?}", self.topic);
                }
            }
            Rebalance::Revoke(partitions) => {
                let k = partitions.elements_for_topic(&self.topic);
                if !k.is_empty() {
                    info!("Sending reset message due to partition revoked on certification topic {}", self.topic);
                    if let Err(_error) = block_on(self.message_channel_tx.send(ChannelMessage::Reset)) {
                        error!("Failed to send reset message");
                    };
                }
            }
            Rebalance::Error(e) => error!("Rebalance error: {}", e),
        }
    }
}