Struct NthKeyBasedRetention 

pub struct NthKeyBasedRetention { /* private fields */ }

Similar to KeyBasedRetention, but instead of retaining the latest offset for a key. this strategy retains the oldest nth offset associated with a key.

Implementations

impl NthKeyBasedRetention

pub fn new(max_compaction_keys: usize, max_records_per_key: usize) -> Self

A max_compaction_keys parameter is used to limit the number of distinct topic/partition/keys processed in a single run of the compactor. The max_records_per_key is used to retain the nth oldest key.

pub fn with_qids( max_compaction_keys: usize, max_records_per_key: usize, qid_from_record: fn(&ConsumerRecord) -> Option<Qid>, ) -> Self

Similar to the above, but the qualified id of a record will be determined by a function. A max_compaction_keys parameter is used to limit the number of distinct topic/partition/keys processed in a single run of the compactor. The max_records_per_key is used to retain the nth oldest key.

Trait Implementations

impl CompactionStrategy for NthKeyBasedRetention

type S = (HashMap<u64, VecDeque<u64>>, usize, usize)

The state to manage throughout a compaction run.

type KS = Option<(HashMap<SmolStr, u64>, fn(&ConsumerRecord) -> Option<SmolStr>)>

The type of state required to manage keys. It may be that no state is required i.e. if the key field from the record is used.

fn key_init(&self) -> Self::KS

Produce the initial key state for the compaction.

fn key(key_state: &mut Self::KS, r: &ConsumerRecord) -> Option<Key>

The key function computes a key that will be used by the compactor for subsequent use. In simple scenarios, this key can be the key field from the topic’s record itself, which is the default assumption here. Returning None will result in this record being excluded from compaction.

fn init<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Self::S> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Produce the initial state for the reducer function.

fn reduce( state: &mut Self::S, key: Key, record: ConsumerRecord, ) -> MaxKeysReached

The reducer function receives a mutable state reference, a key and a consumer record, and returns a bool indicating whether the function has reached its maximum number of distinct keys. If it has then compaction may occur again once the current consumption of records has finished. The goal is to avoid needing to process every type of topic/partition/key in one go i.e. a subset can be processed, and then another subset etc. This strategy helps to manage memory.

fn collect(state: Self::S) -> CompactionMap

The collect function is responsible for mapping the state into a map of keys and their minimum offsets. The compactor will filter out the first n keys in a subsequent run where n is the number of keys found on the first run. This permits memory to be controlled given a large number of distinct keys. The cost of the re-run strategy is that additional compaction scans will be required, resulting in more I/O.