kacrab 0.1.1

A Kafka client for Rust, built from the protocol up.
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

//! Compression ratio estimates used to mirror Kafka producer batch splitting behavior.

use ahash::AHashMap;
use kacrab_protocol::compression::Compression;

const INITIAL_COMPRESSION_RATIO: f32 = 1.0;
#[cfg(test)]
const COMPRESSION_RATIO_IMPROVING_STEP: f32 = 0.005;
#[cfg(test)]
const COMPRESSION_RATIO_DETERIORATE_STEP: f32 = 0.05;

#[derive(Debug, Default, Clone)]
pub(crate) struct CompressionRatioEstimator {
    estimates: AHashMap<CompressionRatioKey, f32>,
}

impl CompressionRatioEstimator {
    pub(crate) fn estimation(&self, topic: &str, codec: Compression) -> f32 {
        self.estimates
            .get(&CompressionRatioKey::new(topic, codec))
            .copied()
            .unwrap_or(INITIAL_COMPRESSION_RATIO)
    }

    #[cfg(test)]
    pub(crate) fn update_estimation(
        &mut self,
        topic: &str,
        codec: Compression,
        observed_ratio: f32,
    ) -> f32 {
        let current = self.estimation(topic, codec);
        let next = if observed_ratio > current {
            (current + COMPRESSION_RATIO_DETERIORATE_STEP).max(observed_ratio)
        } else if observed_ratio < current {
            (current - COMPRESSION_RATIO_IMPROVING_STEP).max(observed_ratio)
        } else {
            current
        };
        self.set_estimation(topic, codec, next);
        next
    }

    pub(crate) fn reset_after_split(
        &mut self,
        topic: &str,
        codec: Compression,
        observed_ratio: f32,
    ) {
        self.set_estimation(topic, codec, observed_ratio.max(INITIAL_COMPRESSION_RATIO));
    }

    pub(crate) fn set_estimation(&mut self, topic: &str, codec: Compression, ratio: f32) {
        let _old = self
            .estimates
            .insert(CompressionRatioKey::new(topic, codec), ratio);
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct CompressionRatioKey {
    topic: String,
    codec: i16,
}

impl CompressionRatioKey {
    fn new(topic: &str, codec: Compression) -> Self {
        Self {
            topic: topic.to_owned(),
            codec: codec as i16,
        }
    }
}