krabka_log/log/
active_producers.rs

1//! The producer state that `DescribeProducers` reports and that log compaction
2//! keeps, and the expiry of the producers that stopped writing.
3//!
4//! Kafka keeps one producer state for each partition log. Every replica
5//! updates it for each batch that it appends: a leader for a client batch, and
6//! a follower for a batch that it replicates. `ReplicaManager
7//! .activeProducerState` answers `DescribeProducers` from that state on every
8//! replica that hosts the partition, through `Partition.activeProducerState`
9//! and `UnifiedLog.activeProducers`. The log cleaner of every replica reads the
10//! same state: `Cleaner.cleanSegments` takes `UnifiedLog
11//! .lastRecordsOfActiveProducers`. `ProducerStateManager
12//! .removeExpiredProducers` removes a producer when `producer.id.expiration.ms`
13//! has passed since its last write and it has no open transaction.
14
15use {
16    super::Log,
17    crate::compact::ProducerLastRecord,
18    krabka_ids::{Offset, ProducerId},
19    std::collections::HashMap,
20};
21
22/// One producer as `DescribeProducers` reports it.
23///
24/// This is Kafka's `DescribeProducersResponseData.ProducerState`, which
25/// `UnifiedLog.activeProducers` fills from a `ProducerStateEntry`.
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub struct ActiveProducer {
28    pub producer_id: ProducerId,
29    pub producer_epoch: i16,
30    /// The last sequence of the last data batch of the producer at
31    /// `producer_epoch`, or `-1` when the log holds no such batch.
32    pub last_sequence: i32,
33    /// The max timestamp of the last data batch of the producer, or the
34    /// timestamp of the transaction marker that came after it.
35    pub last_timestamp: i64,
36    /// The coordinator epoch of the last transaction marker of the producer,
37    /// or `-1` before its first marker.
38    pub coordinator_epoch: i32,
39    /// The first offset of the open transaction of the producer, or `None`
40    /// when it has no open transaction.
41    pub current_txn_start_offset: Option<Offset>,
42}
43
44impl Log {
45    /// Every producer that this log holds state for, in producer id order.
46    ///
47    /// This is Kafka's `UnifiedLog.activeProducers`. An append from a client,
48    /// a replicated append and a reopen all update the same state, so a
49    /// leader and its followers give the same answer at the same log end.
50    #[must_use]
51    pub fn active_producers(&self) -> Vec<ActiveProducer> {
52        let mut producers: Vec<ActiveProducer> = self
53            .producer_state
54            .values()
55            .map(|entry| ActiveProducer {
56                producer_id: entry.producer_id,
57                producer_epoch: entry.producer_epoch,
58                last_sequence: entry.last_sequence,
59                last_timestamp: entry.timestamp,
60                coordinator_epoch: entry.coordinator_epoch,
61                current_txn_start_offset: entry.current_txn_first_offset,
62            })
63            .collect();
64        producers.sort_unstable_by_key(|producer| producer.producer_id);
65        producers
66    }
67
68    /// The last record of every producer that this log holds state for. A
69    /// compaction pass keeps that record, so that the producer state
70    /// survives the pass.
71    ///
72    /// This is Kafka's `UnifiedLog.lastRecordsOfActiveProducers`: the last
73    /// offset of the last data batch of the producer, and its epoch. A
74    /// producer has no last data offset when a transaction marker at a new
75    /// epoch has cleared its batches (transaction version 2), or when it has
76    /// written only markers. Every append path updates this state, so a
77    /// leader and a follower give the same answer at the same log end.
78    pub(crate) fn last_records_of_active_producers(
79        &self,
80    ) -> HashMap<ProducerId, ProducerLastRecord> {
81        self.producer_state
82            .values()
83            .map(|entry| {
84                let last_record = ProducerLastRecord {
85                    last_data_offset: entry.last_batch().map(|batch| batch.last_offset),
86                    producer_epoch: entry.producer_epoch,
87                };
88                (entry.producer_id, last_record)
89            })
90            .collect()
91    }
92
93    /// Remove every producer that has no open transaction and whose last
94    /// timestamp is `expiration_ms` or more before `now_ms`. Also remove the
95    /// transaction verification state that is that old.
96    ///
97    /// This is Kafka's `ProducerStateManager.removeExpiredProducers` with
98    /// `isProducerExpired`, for `producer.id.expiration.ms`. The next
99    /// producer-state snapshot does not hold a removed producer.
100    pub fn remove_expired_producers(&mut self, now_ms: i64, expiration_ms: i64) {
101        let expired: Vec<ProducerId> = self
102            .producer_state
103            .values()
104            .filter(|entry| {
105                entry.current_txn_first_offset.is_none()
106                    && now_ms.saturating_sub(entry.timestamp) >= expiration_ms
107            })
108            .map(|entry| entry.producer_id)
109            .collect();
110        for producer_id in expired {
111            self.producer_state.remove(&producer_id);
112            self.earlier_batches.remove(&producer_id);
113            // Kafka keeps the coordinator epoch in the `ProducerStateEntry`,
114            // so it goes with the expired producer.
115            self.coordinator_epochs.remove(&producer_id);
116        }
117        self.verification_states
118            .retain(|_, state| now_ms.saturating_sub(state.created_ms) < expiration_ms);
119    }
120}
121
122#[cfg(test)]
123mod tests;
krabka_log/log/active_producers.rs

krabka_log/log/
active_producers.rs
