crabka-replicator 0.3.8

Cross-cluster geo-replication service for Crabka (MirrorMaker-2 equivalent)
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

//! In-memory source->target offset translation built from MM2 offset-sync records.
//! Translation is "at-or-before": find the latest sync with upstream <= committed,
//! then downstream + (committed - upstream). Never maps past un-replicated data.

use std::collections::BTreeMap;

use crate::mm2::OffsetSync;

/// Accumulates [`OffsetSync`] records and translates committed source offsets to
/// their corresponding target offsets.
#[derive(Default)]
pub struct OffsetSyncStore {
    /// Outer key: (topic, partition). Inner key: upstream offset → downstream offset.
    syncs: BTreeMap<(String, i32), BTreeMap<i64, i64>>,
}

impl OffsetSyncStore {
    /// Ingest one offset-sync record, replacing any prior entry for the same
    /// `(topic, partition, upstream)` triple.
    pub fn ingest(&mut self, s: OffsetSync) {
        self.syncs
            .entry((s.topic, s.partition))
            .or_default()
            .insert(s.upstream, s.downstream);
    }

    /// Translate a committed source offset to its target offset.
    ///
    /// Returns `None` when:
    /// - the `(topic, partition)` pair has no syncs, or
    /// - `committed` is below every known upstream offset (un-replicated data).
    #[must_use]
    pub fn translate(&self, topic: &str, partition: i32, committed: i64) -> Option<i64> {
        let m = self.syncs.get(&(topic.to_string(), partition))?;
        let (&up, &down) = m.range(..=committed).next_back()?;
        Some(down + (committed - up))
    }
}

#[cfg(test)]
mod tests {
    use assert2::assert;

    use super::*;
    use crate::mm2::OffsetSync;

    #[test]
    fn translates_at_or_before_nearest_sync() {
        let mut s = OffsetSyncStore::default();
        s.ingest(OffsetSync {
            topic: "orders".into(),
            partition: 0,
            upstream: 100,
            downstream: 70,
        });
        s.ingest(OffsetSync {
            topic: "orders".into(),
            partition: 0,
            upstream: 200,
            downstream: 165,
        });
        assert!(s.translate("orders", 0, 250) == Some(215)); // 165 + (250-200)
        assert!(s.translate("orders", 0, 150) == Some(120)); // 70 + (150-100)
        assert!(s.translate("orders", 0, 50) == None); // below first sync
        assert!(s.translate("orders", 9, 100) == None); // unknown partition
    }
}