datum-mq 0.10.1

Kafka sources and sinks for Datum streams, backed by rdkafka
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
137
138
139
140
141
142
143
144
145
146
147

use std::sync::{
    Arc,
    atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering},
};

/// Cheap cloneable metrics handle for Kafka sources/sinks.
#[derive(Debug, Clone, Default)]
pub struct KafkaMetrics {
    inner: Arc<KafkaMetricsInner>,
}

#[derive(Debug, Default)]
struct KafkaMetricsInner {
    assigned_partitions: AtomicU64,
    revoked_partitions: AtomicU64,
    lost_partitions: AtomicU64,
    rebalances: AtomicU64,
    emitted: AtomicU64,
    committed_offsets: AtomicU64,
    commit_failures: AtomicU64,
    paused: AtomicBool,
    outstanding: AtomicU64,
    producer_in_flight: AtomicU64,
    produced: AtomicU64,
    delivery_failures: AtomicU64,
    queue_full: AtomicU64,
    high_watermark: AtomicI64,
    committed_watermark: AtomicI64,
}

/// Point-in-time metrics snapshot.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct KafkaMetricsSnapshot {
    pub assigned_partitions: u64,
    pub revoked_partitions: u64,
    pub lost_partitions: u64,
    pub rebalances: u64,
    pub emitted: u64,
    pub committed_offsets: u64,
    pub commit_failures: u64,
    pub paused: bool,
    pub outstanding: u64,
    pub producer_in_flight: u64,
    pub produced: u64,
    pub delivery_failures: u64,
    pub queue_full: u64,
    pub high_watermark: i64,
    pub committed_watermark: i64,
}

impl KafkaMetrics {
    #[must_use]
    pub fn snapshot(&self) -> KafkaMetricsSnapshot {
        KafkaMetricsSnapshot {
            assigned_partitions: self.inner.assigned_partitions.load(Ordering::Relaxed),
            revoked_partitions: self.inner.revoked_partitions.load(Ordering::Relaxed),
            lost_partitions: self.inner.lost_partitions.load(Ordering::Relaxed),
            rebalances: self.inner.rebalances.load(Ordering::Relaxed),
            emitted: self.inner.emitted.load(Ordering::Relaxed),
            committed_offsets: self.inner.committed_offsets.load(Ordering::Relaxed),
            commit_failures: self.inner.commit_failures.load(Ordering::Relaxed),
            paused: self.inner.paused.load(Ordering::Relaxed),
            outstanding: self.inner.outstanding.load(Ordering::Relaxed),
            producer_in_flight: self.inner.producer_in_flight.load(Ordering::Relaxed),
            produced: self.inner.produced.load(Ordering::Relaxed),
            delivery_failures: self.inner.delivery_failures.load(Ordering::Relaxed),
            queue_full: self.inner.queue_full.load(Ordering::Relaxed),
            high_watermark: self.inner.high_watermark.load(Ordering::Relaxed),
            committed_watermark: self.inner.committed_watermark.load(Ordering::Relaxed),
        }
    }

    pub(crate) fn add_assigned(&self, count: u64) {
        self.inner
            .assigned_partitions
            .fetch_add(count, Ordering::Relaxed);
        self.inner.rebalances.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn add_revoked(&self, count: u64) {
        self.inner
            .revoked_partitions
            .fetch_add(count, Ordering::Relaxed);
        self.inner.rebalances.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn add_lost(&self, count: u64) {
        self.inner
            .lost_partitions
            .fetch_add(count, Ordering::Relaxed);
        self.inner.rebalances.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn emitted(&self, outstanding: u64, high_watermark: i64) {
        self.inner.emitted.fetch_add(1, Ordering::Relaxed);
        self.inner.outstanding.store(outstanding, Ordering::Relaxed);
        self.inner
            .high_watermark
            .store(high_watermark, Ordering::Relaxed);
    }

    pub(crate) fn emitted_batch(&self, count: u64, outstanding: u64, high_watermark: i64) {
        self.inner.emitted.fetch_add(count, Ordering::Relaxed);
        self.inner.outstanding.store(outstanding, Ordering::Relaxed);
        self.inner
            .high_watermark
            .store(high_watermark, Ordering::Relaxed);
    }

    pub(crate) fn committed(&self, outstanding: u64, committed_watermark: i64) {
        self.inner.committed_offsets.fetch_add(1, Ordering::Relaxed);
        self.inner.outstanding.store(outstanding, Ordering::Relaxed);
        self.inner
            .committed_watermark
            .store(committed_watermark, Ordering::Relaxed);
    }

    pub(crate) fn set_outstanding(&self, outstanding: u64) {
        self.inner.outstanding.store(outstanding, Ordering::Relaxed);
    }

    pub(crate) fn commit_failed(&self) {
        self.inner.commit_failures.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn set_paused(&self, paused: bool) {
        self.inner.paused.store(paused, Ordering::Relaxed);
    }

    pub(crate) fn producer_in_flight(&self, in_flight: u64) {
        self.inner
            .producer_in_flight
            .store(in_flight, Ordering::Relaxed);
    }

    pub(crate) fn produced(&self) {
        self.inner.produced.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn delivery_failed(&self) {
        self.inner.delivery_failures.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn queue_full(&self) {
        self.inner.queue_full.fetch_add(1, Ordering::Relaxed);
    }
}