krabka_log/log/
control.rs

1//! Classification of Kafka control batches and of the krabka barrier
2//! marker, which the append path and both recovery walks share.
3//!
4//! The control-record type lives in the key of a control batch's first
5//! record, so every caller that must tell a transaction end marker from a
6//! barrier marker reads it through this module.
7
8use {crate::error::LogError, krabka_protocol::records::RecordBatch};
9
10/// Control-record type of a krabka barrier marker.
11///
12/// Kafka assigns the control-record types 0 to 6: `ABORT`, `COMMIT`,
13/// `LEADER_CHANGE`, `SNAPSHOT_HEADER`, `SNAPSHOT_FOOTER`, `KRAFT_VERSION`, and
14/// `VOTERS`. The value 1000 starts a krabka-private range that Kafka cannot
15/// reach by normal growth. `krabka-raft` uses the same convention for its
16/// private api keys at 1003 and 1004.
17///
18/// A barrier marker is a control batch with one record. The record key holds
19/// this type. The batch sets `producer_id` to -1, it sets `producer_epoch` to
20/// -1, and it clears the transactional attribute bit. The log keeps no
21/// transaction state and no producer state for such a batch. Kafka's
22/// `ControlRecordType.parse` reports an unknown type for the value 1000 and
23/// skips the batch, so a JVM consumer never sees the record.
24///
25/// The broker builds the marker batch and the log classifies it. Both read
26/// this one constant.
27pub const BARRIER_CONTROL_TYPE: i16 = 1000;
28
29/// Control-record type of Kafka's ABORT end marker.
30pub const ABORT_CONTROL_TYPE: i16 = 0;
31
32/// Control-record type of Kafka's COMMIT end marker.
33pub const COMMIT_CONTROL_TYPE: i16 = 1;
34
35/// Whether the first control key closes a transaction as an abort or commit,
36/// as [`batch_control_marker_type`] reads it.
37///
38/// # Errors
39/// Returns the error of [`batch_control_marker_type`].
40pub(super) fn transaction_marker_flags(
41    batch: &RecordBatch,
42) -> Result<TransactionMarkerFlags, LogError> {
43    let marker = batch_control_marker_type(batch)?;
44    Ok(TransactionMarkerFlags {
45        is_abort: marker == Some(ABORT_CONTROL_TYPE),
46        is_commit: marker == Some(COMMIT_CONTROL_TYPE),
47    })
48}
49
50/// Whether a control batch closes a transaction as an abort or as a commit.
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
52pub(super) struct TransactionMarkerFlags {
53    pub(super) is_abort: bool,
54    pub(super) is_commit: bool,
55}
56
57/// The control-record key version this build writes and parses, Kafka's
58/// `ControlRecordType.CURRENT_CONTROL_RECORD_KEY_VERSION`.
59///
60/// It is part of the 1.x on-disk contract because control batches are in the
61/// `.log`. Kafka owns the number; krabka follows it.
62pub const CONTROL_RECORD_KEY_VERSION: i16 = 0;
63
64/// The `EndTxnMarker` value version this build writes and parses, Kafka's
65/// `EndTxnMarker.HIGHEST_SUPPORTED_VERSION`.
66///
67/// It is part of the 1.x on-disk contract because end markers are in the
68/// `.log`. Kafka owns the number; krabka follows it.
69pub const END_TXN_MARKER_VALUE_VERSION: i16 = 0;
70
71/// What the log does with one control batch.
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub enum ControlBatchKind {
74    /// A transaction end marker. The log closes the producer's open
75    /// transaction, it stamps the committed data, and it records an aborted
76    /// range. Every control type outside the krabka-private range lands here,
77    /// because Kafka writes no other control type into a data partition.
78    Transaction,
79    /// A krabka barrier marker, control type [`BARRIER_CONTROL_TYPE`]. The log
80    /// leaves `pending`, `pending_stamp_ranges`, `coordinator_epochs`, the
81    /// active `.txnindex`, and the producer state unchanged for it.
82    Barrier,
83}
84
85/// Classify one batch for the append path and for the two recovery paths.
86///
87/// Returns `None` for a data batch. The control-record type comes from the key
88/// of the batch's first record, in the layout that
89/// [`parse_control_marker_type`] reads. A control batch with no records is a
90/// [`ControlBatchKind::Transaction`], which is the classification such a batch
91/// had before the barrier type existed. Compaction emits one for the
92/// `RETAIN_EMPTY` rule.
93pub fn control_batch_kind(batch: &RecordBatch) -> Option<ControlBatchKind> {
94    use krabka_verified::transaction::{LogBatchKind, log_batch_kind};
95    let key = batch
96        .records
97        .first()
98        .and_then(|record| record.key.as_deref())
99        .unwrap_or(&[]);
100    match log_batch_kind(batch.attributes.is_control_batch(), key) {
101        LogBatchKind::Data => None,
102        LogBatchKind::Barrier => Some(ControlBatchKind::Barrier),
103        LogBatchKind::Abort | LogBatchKind::Commit | LogBatchKind::OtherControl => {
104            Some(ControlBatchKind::Transaction)
105        }
106    }
107}
108
109/// The length of a control-record key: `(version: i16, type: i16)`, Kafka's
110/// `ControlRecordType.CURRENT_CONTROL_RECORD_KEY_SIZE`.
111const CONTROL_RECORD_KEY_BYTES: usize = 4;
112
113/// The length of an end-marker value: `(version: i16, coordinator_epoch:
114/// i32)`, the bytes Kafka's `EndTxnMarker` version 0 holds.
115const END_TXN_MARKER_VALUE_BYTES: usize = 6;
116
117/// Parse the control-marker type from the key of a control record. The key
118/// encodes `(version: i16, type: i16)` in big-endian. Returns
119/// [`ABORT_CONTROL_TYPE`] for ABORT, [`COMMIT_CONTROL_TYPE`] for COMMIT, and
120/// [`BARRIER_CONTROL_TYPE`] for a krabka barrier marker. A missing key reads
121/// as an empty one.
122///
123/// This is Kafka's `ControlRecordType.parseTypeId`. A key shorter than four
124/// bytes, or one with a negative version, is corrupt data. A version above
125/// [`CONTROL_RECORD_KEY_VERSION`] parses as that version: Kafka's schema for
126/// the key may only grow compatibly, so it reads the type a later version
127/// carries in the same place. That forward tolerance is Kafka's, and krabka
128/// keeps it rather than refuse a marker that a JVM broker accepts.
129///
130/// # Errors
131/// Returns [`LogError::InvalidControlRecordSize`] for a key shorter than four
132/// bytes and [`LogError::InvalidControlRecordVersion`] for a negative
133/// version.
134pub fn parse_control_marker_type(key: Option<&[u8]>) -> Result<i16, LogError> {
135    let key = key.unwrap_or_default();
136    let Some((&[v0, v1, t0, t1], _)) = key.split_first_chunk::<CONTROL_RECORD_KEY_BYTES>() else {
137        return Err(LogError::InvalidControlRecordSize {
138            record: "end control record key",
139            needed: CONTROL_RECORD_KEY_BYTES,
140            found: key.len(),
141        });
142    };
143    let version = i16::from_be_bytes([v0, v1]);
144    if version < CONTROL_RECORD_KEY_VERSION {
145        return Err(LogError::InvalidControlRecordVersion {
146            record: "control record",
147            version,
148        });
149    }
150    Ok(i16::from_be_bytes([t0, t1]))
151}
152
153/// Parse the transaction coordinator epoch from an end-marker value, which
154/// Kafka encodes as `(version: i16, coordinator_epoch: i32)` in big-endian.
155/// A missing value reads as an empty one.
156///
157/// This is Kafka's `EndTransactionMarker.deserializeValue`. A value shorter
158/// than six bytes, or one with a negative version, is corrupt data. A version
159/// above [`END_TXN_MARKER_VALUE_VERSION`] parses as that version, Kafka's
160/// forward tolerance for a marker that a later broker wrote: the fields a
161/// later version adds follow the coordinator epoch.
162///
163/// # Errors
164/// Returns [`LogError::InvalidControlRecordSize`] for a value shorter than
165/// six bytes and [`LogError::InvalidControlRecordVersion`] for a negative
166/// version.
167pub fn parse_control_marker_coordinator_epoch(value: Option<&[u8]>) -> Result<i32, LogError> {
168    let value = value.unwrap_or_default();
169    let Some((&[v0, v1, e0, e1, e2, e3], _)) =
170        value.split_first_chunk::<END_TXN_MARKER_VALUE_BYTES>()
171    else {
172        return Err(LogError::InvalidControlRecordSize {
173            record: "end transaction marker value",
174            needed: END_TXN_MARKER_VALUE_BYTES,
175            found: value.len(),
176        });
177    };
178    let version = i16::from_be_bytes([v0, v1]);
179    if version < END_TXN_MARKER_VALUE_VERSION {
180        return Err(LogError::InvalidControlRecordVersion {
181            record: "end transaction marker",
182            version,
183        });
184    }
185    Ok(i32::from_be_bytes([e0, e1, e2, e3]))
186}
187
188/// The control-record type of a batch's first record, as
189/// [`parse_control_marker_type`] reads it. Returns `None` for a batch with
190/// no record, which an earlier compaction pass emptied; Kafka skips such a
191/// batch without parsing it.
192///
193/// # Errors
194/// Returns the error of [`parse_control_marker_type`].
195pub fn batch_control_marker_type(batch: &RecordBatch) -> Result<Option<i16>, LogError> {
196    batch
197        .records
198        .first()
199        .map(|record| parse_control_marker_type(record.key.as_deref()))
200        .transpose()
201}
202
203/// The coordinator epoch of a batch's first record, as
204/// [`parse_control_marker_coordinator_epoch`] reads it. Returns `None` for a
205/// batch with no record.
206///
207/// # Errors
208/// Returns the error of [`parse_control_marker_coordinator_epoch`].
209pub fn batch_control_marker_coordinator_epoch(
210    batch: &RecordBatch,
211) -> Result<Option<i32>, LogError> {
212    batch
213        .records
214        .first()
215        .map(|record| parse_control_marker_coordinator_epoch(record.value.as_deref()))
216        .transpose()
217}
218
219/// Check a control batch's marker before the batch is written, as Kafka's
220/// `ProducerAppendInfo` deserializes an end marker before the segment write:
221/// the key of any control batch, and the value of an ABORT or COMMIT marker.
222/// A data batch passes unchecked.
223///
224/// # Errors
225/// Returns [`LogError::InvalidControlRecordSize`] or
226/// [`LogError::InvalidControlRecordVersion`] for a marker Kafka refuses.
227pub fn check_control_record_versions(batch: &RecordBatch) -> Result<(), LogError> {
228    if !batch.attributes.is_control_batch() {
229        return Ok(());
230    }
231    if matches!(
232        batch_control_marker_type(batch)?,
233        Some(ABORT_CONTROL_TYPE | COMMIT_CONTROL_TYPE)
234    ) {
235        batch_control_marker_coordinator_epoch(batch)?;
236    }
237    Ok(())
238}
239
240#[cfg(test)]
241mod tests {
242    use {
243        super::*,
244        crate::{
245            compact::build_offset_map,
246            log::{
247                Log,
248                test_support::{
249                    BarrierEpoch, BarrierMarkerSetup, KeyedRecord, PartitionState,
250                    ProducerTransaction, TempLog, TransactionFields, abort_marker, barrier_marker,
251                    barrier_marker_from_producer, commit_marker, compact_test_log, compaction_ctx,
252                    control_key, control_value, install_stamps, keyed_batch, partition_state,
253                    sample_batch, stamped_test_log, test_log, transactional_batch,
254                },
255            },
256            producer_snapshot::ProducerSnapshotEntry,
257            segment::Segment,
258            test_support::{aborted_txn, open_log},
259            txn_index::AbortedTxn,
260        },
261        assert2::check,
262        bytes::Bytes,
263        krabka_ids::{Offset, ProducerId},
264        krabka_protocol::records::Record,
265        krabka_units::prelude::mebibytes,
266        tempfile::tempdir,
267    };
268
269    /// What one parse came to, with a refusal reduced to the fields that say
270    /// which rule refused it.
271    #[derive(Debug, PartialEq, Eq)]
272    enum Parsed<T> {
273        Value(T),
274        TooShort { record: &'static str, found: usize },
275        BadVersion { record: &'static str, version: i16 },
276        OtherError(String),
277    }
278
279    impl<T> From<Result<T, LogError>> for Parsed<T> {
280        fn from(result: Result<T, LogError>) -> Self {
281            match result {
282                Ok(value) => Self::Value(value),
283                Err(LogError::InvalidControlRecordSize { record, found, .. }) => {
284                    Self::TooShort { record, found }
285                }
286                Err(LogError::InvalidControlRecordVersion { record, version }) => {
287                    Self::BadVersion { record, version }
288                }
289                Err(other) => Self::OtherError(other.to_string()),
290            }
291        }
292    }
293
294    /// A record's key and value, as a compaction pass kept them.
295    #[derive(Debug, PartialEq, Eq)]
296    struct KeptRecord {
297        key: Bytes,
298        value: Bytes,
299    }
300
301    /// One parse case: its name, the bytes, and what the parse comes to.
302    struct ParseCase<T> {
303        name: &'static str,
304        bytes: Option<&'static [u8]>,
305        expected: Parsed<T>,
306    }
307
308    /// One append case: its name, the marker's key and value, and what the
309    /// append comes to.
310    struct AppendCase {
311        name: &'static str,
312        key: Option<Bytes>,
313        value: Option<Bytes>,
314        expected: Parsed<()>,
315    }
316
317    const KEY: &str = "end control record key";
318    const VALUE: &str = "end transaction marker value";
319
320    /// The golden bytes of the two control-record shapes Kafka fixes, as a
321    /// JVM broker writes them: an end-marker key `(version 0, type)` and an
322    /// end-marker value `(version 0, coordinator epoch)`.
323    #[test]
324    fn end_marker_golden_bytes_decode() {
325        check!(
326            Parsed::from(parse_control_marker_type(Some(&[0x00, 0x00, 0x00, 0x01])))
327                == Parsed::Value(1)
328        );
329        check!(
330            Parsed::from(parse_control_marker_coordinator_epoch(Some(&[
331                0x00, 0x00, 0x00, 0x00, 0x00, 0x11
332            ]))) == Parsed::Value(17)
333        );
334    }
335
336    /// Kafka's `ControlRecordType.parseTypeId`: the type is bytes 2..4 of the
337    /// key; a short or missing key and a negative version are corrupt, and a
338    /// later version parses as version 0.
339    #[test]
340    fn a_control_key_parses_as_kafka_parses_it() {
341        let cases: [ParseCase<i16>; 12] = [
342            ParseCase {
343                name: "ABORT",
344                bytes: Some(&[0, 0, 0, 0]),
345                expected: Parsed::Value(0),
346            },
347            ParseCase {
348                name: "COMMIT",
349                bytes: Some(&[0, 0, 0, 1]),
350                expected: Parsed::Value(1),
351            },
352            ParseCase {
353                name: "barrier",
354                bytes: Some(&[0, 0, 0x03, 0xe8]),
355                expected: Parsed::Value(1000),
356            },
357            ParseCase {
358                name: "a longer key",
359                bytes: Some(&[0, 0, 0, 1, 9, 9]),
360                expected: Parsed::Value(1),
361            },
362            ParseCase {
363                name: "a later version",
364                bytes: Some(&[0, 1, 0, 1]),
365                expected: Parsed::Value(1),
366            },
367            ParseCase {
368                name: "the largest version",
369                bytes: Some(&[0x7f, 0xff, 0, 1]),
370                expected: Parsed::Value(1),
371            },
372            ParseCase {
373                name: "minus one",
374                bytes: Some(&[0xff, 0xff, 0, 1]),
375                expected: Parsed::BadVersion {
376                    record: "control record",
377                    version: -1,
378                },
379            },
380            ParseCase {
381                name: "the smallest version",
382                bytes: Some(&[0x80, 0x00, 0, 1]),
383                expected: Parsed::BadVersion {
384                    record: "control record",
385                    version: i16::MIN,
386                },
387            },
388            ParseCase {
389                name: "no key",
390                bytes: None,
391                expected: Parsed::TooShort {
392                    record: KEY,
393                    found: 0,
394                },
395            },
396            ParseCase {
397                name: "one byte",
398                bytes: Some(&[0]),
399                expected: Parsed::TooShort {
400                    record: KEY,
401                    found: 1,
402                },
403            },
404            ParseCase {
405                name: "two bytes",
406                bytes: Some(&[0, 0]),
407                expected: Parsed::TooShort {
408                    record: KEY,
409                    found: 2,
410                },
411            },
412            ParseCase {
413                name: "three bytes",
414                bytes: Some(&[0, 0, 0]),
415                expected: Parsed::TooShort {
416                    record: KEY,
417                    found: 3,
418                },
419            },
420        ];
421        for ParseCase {
422            name,
423            bytes,
424            expected,
425        } in cases
426        {
427            check!(
428                Parsed::from(parse_control_marker_type(bytes)) == expected,
429                "case {name}"
430            );
431        }
432    }
433
434    /// Kafka's `EndTransactionMarker.deserializeValue`: the coordinator epoch
435    /// is bytes 2..6 of the value; a short or missing value and a negative
436    /// version are corrupt, and a later version parses as version 0.
437    #[test]
438    fn an_end_marker_value_parses_as_kafka_parses_it() {
439        let cases: [ParseCase<i32>; 10] = [
440            ParseCase {
441                name: "current",
442                bytes: Some(&[0, 0, 0, 0, 0, 7]),
443                expected: Parsed::Value(7),
444            },
445            ParseCase {
446                name: "a longer value",
447                bytes: Some(&[0, 0, 0, 0, 0, 7, 9]),
448                expected: Parsed::Value(7),
449            },
450            ParseCase {
451                name: "a later version",
452                bytes: Some(&[0, 1, 0, 0, 0, 7, 9]),
453                expected: Parsed::Value(7),
454            },
455            ParseCase {
456                name: "the largest version",
457                bytes: Some(&[0x7f, 0xff, 0, 0, 0, 7]),
458                expected: Parsed::Value(7),
459            },
460            ParseCase {
461                name: "minus one",
462                bytes: Some(&[0xff, 0xff, 0, 0, 0, 7]),
463                expected: Parsed::BadVersion {
464                    record: "end transaction marker",
465                    version: -1,
466                },
467            },
468            ParseCase {
469                name: "the smallest version",
470                bytes: Some(&[0x80, 0x00, 0, 0, 0, 7]),
471                expected: Parsed::BadVersion {
472                    record: "end transaction marker",
473                    version: i16::MIN,
474                },
475            },
476            ParseCase {
477                name: "no value",
478                bytes: None,
479                expected: Parsed::TooShort {
480                    record: VALUE,
481                    found: 0,
482                },
483            },
484            ParseCase {
485                name: "version only",
486                bytes: Some(&[0, 0]),
487                expected: Parsed::TooShort {
488                    record: VALUE,
489                    found: 2,
490                },
491            },
492            ParseCase {
493                name: "four bytes",
494                bytes: Some(&[0, 0, 0, 0]),
495                expected: Parsed::TooShort {
496                    record: VALUE,
497                    found: 4,
498                },
499            },
500            ParseCase {
501                name: "five bytes",
502                bytes: Some(&[0, 0, 0, 0, 0]),
503                expected: Parsed::TooShort {
504                    record: VALUE,
505                    found: 5,
506                },
507            },
508        ];
509        for ParseCase {
510            name,
511            bytes,
512            expected,
513        } in cases
514        {
515            check!(
516                Parsed::from(parse_control_marker_coordinator_epoch(bytes)) == expected,
517                "case {name}"
518            );
519        }
520    }
521
522    fn append_open_barrier_transaction(log: &mut Log) {
523        let mut data = transactional_batch(1000, 2, &["a", "b"]);
524        data.base_sequence = 0;
525        log.append(&mut data).unwrap(); // offsets 0 and 1
526    }
527
528    /// An end marker Kafka would refuse stops the append, as Kafka's
529    /// `ProducerAppendInfo` does when `EndTransactionMarker.deserialize`
530    /// throws, and the log is unchanged.
531    #[test]
532    fn an_end_marker_kafka_refuses_is_not_appended() {
533        let cases: [AppendCase; 4] = [
534            AppendCase {
535                name: "a negative key version",
536                key: Some(Bytes::from_static(&[0xff, 0xff, 0x00, 0x01])),
537                value: Some(control_value(17)),
538                expected: Parsed::BadVersion {
539                    record: "control record",
540                    version: -1,
541                },
542            },
543            AppendCase {
544                name: "a short key",
545                key: Some(Bytes::from_static(&[0x00, 0x00, 0x00])),
546                value: Some(control_value(17)),
547                expected: Parsed::TooShort {
548                    record: KEY,
549                    found: 3,
550                },
551            },
552            AppendCase {
553                name: "a negative value version",
554                key: Some(control_key(COMMIT_CONTROL_TYPE)),
555                value: Some(Bytes::from_static(&[0xff, 0xfe, 0, 0, 0, 17])),
556                expected: Parsed::BadVersion {
557                    record: "end transaction marker",
558                    version: -2,
559                },
560            },
561            AppendCase {
562                name: "no value",
563                key: Some(control_key(COMMIT_CONTROL_TYPE)),
564                value: None,
565                expected: Parsed::TooShort {
566                    record: VALUE,
567                    found: 0,
568                },
569            },
570        ];
571        for AppendCase {
572            name,
573            key,
574            value,
575            expected,
576        } in cases
577        {
578            let TempLog { dir: _dir, mut log } = test_log();
579            let mut data = transactional_batch(1000, 2, &["a"]);
580            data.base_sequence = 0;
581            log.append(&mut data).unwrap();
582            let before = partition_state(&log, &[1000]);
583
584            let mut marker = commit_marker(1000, 2);
585            marker.records[0].key = key;
586            marker.records[0].value = value;
587            let got = Parsed::from(log.append(&mut marker).map(|_| ()));
588
589            check!(got == expected, "case {name}");
590            check!(partition_state(&log, &[1000]) == before, "case {name}");
591            check!(log.log_end_offset() == Offset(1), "case {name}");
592        }
593    }
594
595    // ---- barrier-marker tests ----
596
597    /// A barrier marker changes no transaction state and no producer state
598    /// while a transaction is open.
599    ///
600    /// The log routes a control batch by its control-record type, not by its
601    /// producer id, so the second case carries the open transaction's
602    /// producer id and still closes nothing.
603    #[yare::parameterized(
604        no_producer_id = { "no producer id", barrier_marker().epoch(7).call() },
605        with_producer_id = {
606            "with a producer id",
607            barrier_marker_from_producer()
608                .marker(BarrierMarkerSetup {
609                    epoch: BarrierEpoch(7),
610                    ..Default::default()
611                })
612                .call(),
613        },
614    )]
615    fn a_barrier_marker_leaves_an_open_transaction_untouched(
616        name: &'static str,
617        barrier: RecordBatch,
618    ) {
619        let mut barrier = barrier;
620        let TempLog { dir: _dir, mut log } = stamped_test_log(40, 1);
621
622        append_open_barrier_transaction(&mut log);
623        let before = partition_state(&log, &[1000, -1]);
624
625        log.append(&mut barrier).unwrap(); // offset 2
626
627        check!(partition_state(&log, &[1000, -1]) == before, "case {name}");
628        // The marker still takes an offset of its own, and it carries no
629        // stamp, because a control batch is never stamped.
630        check!(log.log_end_offset() == Offset(3), "case {name}");
631        check!(log.stamp_for_offset(Offset(2)) == None, "case {name}");
632    }
633
634    #[test]
635    fn barrier_marker_identity_fields_write_no_producer_tail() {
636        // The marker's identity fields give no producer tail, so the append
637        // writes no producer entry for it.
638        check!(Log::data_producer_tail(ProducerId(-1), -1, 0, Offset(2)).unwrap() == None);
639        check!(Log::data_producer_tail(ProducerId(1000), -1, 0, Offset(2)).unwrap() == None);
640    }
641
642    /// A barrier marker moves the last-stable-offset exactly as an ordinary
643    /// non-transactional data batch moves it.
644    #[test]
645    fn a_barrier_marker_moves_the_lso_like_a_data_batch() {
646        for (name, open_transaction, want_lso, want_log_end) in [
647            ("no open transaction", false, Offset(1), Offset(1)),
648            ("open transaction", true, Offset(0), Offset(2)),
649        ] {
650            for (kind, is_barrier) in [("data batch", false), ("barrier marker", true)] {
651                let TempLog { dir: _dir, mut log } = test_log();
652                if open_transaction {
653                    let mut data = transactional_batch(1000, 0, &["a"]);
654                    data.base_sequence = 0;
655                    log.append(&mut data).unwrap();
656                }
657                let mut batch = if is_barrier {
658                    barrier_marker().call()
659                } else {
660                    sample_batch(1)
661                };
662                log.append(&mut batch).unwrap();
663                check!(log.lso() == want_lso, "case {name} / {kind}");
664                check!(log.log_end_offset() == want_log_end, "case {name} / {kind}");
665            }
666        }
667    }
668
669    /// A barrier marker between a transaction's data and its end marker does
670    /// not disturb what the end marker does.
671    #[yare::parameterized(
672        commit = {
673            "commit",
674            commit_marker(1000, 2),
675            Vec::new(),
676            vec![Some(40), Some(40), None, None]
677        },
678        abort = {
679            "abort",
680            abort_marker(1000, 2),
681            vec![aborted_txn().call()],
682            vec![None, None, None, None]
683        },
684    )]
685    fn a_barrier_marker_does_not_disturb_a_following_end_marker(
686        name: &'static str,
687        marker: RecordBatch,
688        want_aborted: Vec<AbortedTxn>,
689        want_stamps: Vec<Option<u64>>,
690    ) {
691        let mut marker = marker;
692        let committed = ProducerSnapshotEntry {
693            producer_id: ProducerId(1000),
694            producer_epoch: 2,
695            last_sequence: 1,
696            last_offset: Offset(1),
697            offset_delta: 1,
698            timestamp: 0,
699            coordinator_epoch: 17,
700            current_txn_first_offset: None,
701        };
702        let TempLog { dir: _dir, mut log } = stamped_test_log(40, 1);
703
704        append_open_barrier_transaction(&mut log);
705        log.append(&mut barrier_marker().epoch(9).call()).unwrap(); // offset 2
706        check!(
707            log.lso() == Offset(0),
708            "case {name}: the barrier holds the LSO"
709        );
710        log.append(&mut marker).unwrap(); // offset 3
711        // A high watermark past the marker releases the transaction.
712        log.release_replicated_transactions(Offset(4));
713
714        check!(
715            partition_state(&log, &[1000])
716                == PartitionState {
717                    lso: Offset(4),
718                    transactions: vec![ProducerTransaction {
719                        producer_id: 1000,
720                        fields: TransactionFields {
721                            coordinator_epoch: 17,
722                            first_offset: None,
723                        },
724                    }],
725                    aborted: want_aborted,
726                    producers: vec![committed],
727                },
728            "case {name}"
729        );
730        let stamps: Vec<Option<u64>> = (0..4).map(|o| log.stamp_for_offset(Offset(o))).collect();
731        check!(stamps == want_stamps, "case {name}");
732    }
733
734    /// Recovery mirrors the append path. A reopened log rebuilds the same
735    /// producer state and the same transaction state across barrier markers.
736    #[test]
737    fn barrier_markers_rebuild_identical_state_after_reopen() {
738        let dir = tempdir().unwrap();
739        let ids = [1000, 2000, 3000, -1];
740        let before = {
741            let mut log = open_log(dir.path());
742            log.append(&mut barrier_marker().call()).unwrap(); // 0
743
744            let mut committed = transactional_batch(1000, 2, &["a"]);
745            committed.base_sequence = 0;
746            log.append(&mut committed).unwrap(); // 1
747            log.append(&mut barrier_marker().epoch(2).call()).unwrap(); // 2
748            log.append(&mut commit_marker(1000, 2)).unwrap(); // 3
749
750            let mut rolled_back = transactional_batch(2000, 5, &["b", "c"]);
751            rolled_back.base_sequence = 0;
752            log.append(&mut rolled_back).unwrap(); // 4 and 5
753            log.append(&mut barrier_marker().epoch(3).call()).unwrap(); // 6
754            log.append(&mut abort_marker(2000, 5)).unwrap(); // 7
755
756            let mut still_open = transactional_batch(3000, 1, &["d"]);
757            still_open.base_sequence = 0;
758            log.append(&mut still_open).unwrap(); // 8
759            // A marker that carries the open transaction's producer id closes
760            // nothing on the append path, and recovery reaches the same
761            // result.
762            log.append(
763                &mut barrier_marker_from_producer()
764                    .marker(BarrierMarkerSetup {
765                        epoch: BarrierEpoch(4),
766                        ..Default::default()
767                    })
768                    .producer(3000)
769                    .producer_epoch(1)
770                    .call(),
771            )
772            .unwrap(); // 9
773
774            partition_state(&log, &ids)
775        };
776        // No high watermark has passed a marker yet, so the committed
777        // transaction of producer 1000 still holds the LSO at its first offset.
778        check!(before.lso == Offset(1));
779
780        let mut reopened = open_log(dir.path());
781        check!(partition_state(&reopened, &ids) == before);
782        check!(reopened.log_end_offset() == Offset(10));
783        // Once the high watermark passes both markers, the open transaction of
784        // producer 3000 holds the LSO at its first offset, and the barrier that
785        // follows does not release it.
786        check!(reopened.last_stable_offset(Offset(10)) == Offset(8));
787    }
788
789    /// Recovery rebuilds a transaction's stamp ranges across a barrier
790    /// marker, so a commit that lands after a restart still stamps the
791    /// transaction's data.
792    #[test]
793    fn a_barrier_marker_keeps_stamp_ranges_across_a_restart() {
794        let dir = tempdir().unwrap();
795        {
796            let mut log = open_log(dir.path());
797            append_open_barrier_transaction(&mut log);
798            // A marker that carries the open transaction's producer id clears
799            // no stamp range, on the append path or on the recovery path.
800            log.append(
801                &mut barrier_marker_from_producer()
802                    .marker(BarrierMarkerSetup {
803                        epoch: BarrierEpoch(5),
804                        ..Default::default()
805                    })
806                    .call(),
807            )
808            .unwrap(); // offset 2
809        }
810
811        let mut reopened = open_log(dir.path());
812        install_stamps(&mut reopened, 40, 1);
813        check!(reopened.lso() == Offset(0));
814
815        reopened.append(&mut commit_marker(1000, 2)).unwrap(); // offset 3
816
817        check!(reopened.lso() == Offset(0));
818        check!(reopened.last_stable_offset(Offset(4)) == Offset(4));
819        let stamps: Vec<Option<u64>> = (0..4)
820            .map(|offset| reopened.stamp_for_offset(Offset(offset)))
821            .collect();
822        check!(stamps == vec![Some(40), Some(40), None, None]);
823    }
824
825    /// Compaction keeps every barrier marker, and the marker key never enters
826    /// the dedup map.
827    #[test]
828    fn compaction_keeps_barrier_markers_and_never_indexes_their_key() {
829        let TempLog { dir: _dir, mut log } = compact_test_log();
830        log.append(&mut keyed_batch(0, &[KeyedRecord {
831            offset_delta: 0,
832            key: b"k1",
833            value: b"v0",
834        }]))
835        .unwrap(); // 0
836        log.append(&mut barrier_marker().call()).unwrap(); // 1
837        log.append(&mut keyed_batch(0, &[KeyedRecord {
838            offset_delta: 0,
839            key: b"k1",
840            value: b"v1",
841        }]))
842        .unwrap(); // 2
843        log.append(&mut barrier_marker().epoch(2).call()).unwrap(); // 3
844        log.append(&mut keyed_batch(0, &[KeyedRecord {
845            offset_delta: 0,
846            key: b"tail",
847            value: b"t",
848        }]))
849        .unwrap(); // 4, active
850
851        // The dedup map holds the data key only. `should_index_key` keeps the
852        // marker key out of it, so no barrier can shadow another.
853        let sealed: Vec<_> = log.segments.iter().map(Segment::sealed_input).collect();
854        let mut indexed: Vec<Bytes> = build_offset_map(&sealed, vec![], None)
855            .unwrap()
856            .into_keys()
857            .collect();
858        indexed.sort();
859        check!(indexed == vec![Bytes::from_static(b"k1")]);
860
861        log.compact(&compaction_ctx()).unwrap();
862
863        let out = log.read(Offset(0), mebibytes(1)).unwrap();
864        // Both markers survive the pass, unchanged.
865        let kept_markers: Vec<Record> = out
866            .batches
867            .iter()
868            .filter(|batch| batch.attributes.is_control_batch())
869            .flat_map(|batch| batch.records.iter().cloned())
870            .collect();
871        let want_markers: Vec<Record> = [1, 2]
872            .into_iter()
873            .flat_map(|epoch| barrier_marker().epoch(epoch).call().records)
874            .collect();
875        check!(kept_markers == want_markers);
876        // The data still dedups newest-wins around them.
877        let kept_data: Vec<KeptRecord> = out
878            .batches
879            .iter()
880            .filter(|batch| !batch.attributes.is_control_batch())
881            .flat_map(|batch| batch.records.iter())
882            .map(|record| KeptRecord {
883                key: record.key.clone().unwrap(),
884                value: record.value.clone().unwrap(),
885            })
886            .collect();
887        check!(
888            kept_data
889                == vec![
890                    KeptRecord {
891                        key: Bytes::from_static(b"k1"),
892                        value: Bytes::from_static(b"v1")
893                    },
894                    KeptRecord {
895                        key: Bytes::from_static(b"tail"),
896                        value: Bytes::from_static(b"t")
897                    },
898                ]
899        );
900    }
901}
krabka_log/log/control.rs

krabka_log/log/
control.rs
