Struct Log 

pub struct Log { /* private fields */ }

A Kafka-format log: a sorted collection of Segments plus a single active segment that accepts appends.

Log has one writer and many concurrent readers. Mutation takes &mut self; read, log_start_offset, and the other read methods take &self. Construct one with Log::open.

Implementations

impl Log

pub fn active_producers(&self) -> Vec<ActiveProducer>

Every producer that this log holds state for, in producer id order.

This is Kafka’s UnifiedLog.activeProducers. An append from a client, a replicated append and a reopen all update the same state, so a leader and its followers give the same answer at the same log end.

pub fn remove_expired_producers(&mut self, now_ms: i64, expiration_ms: i64)

Remove every producer that has no open transaction and whose last timestamp is expiration_ms or more before now_ms. Also remove the transaction verification state that is that old.

This is Kafka’s ProducerStateManager.removeExpiredProducers with isProducerExpired, for producer.id.expiration.ms. The next producer-state snapshot does not hold a removed producer.

impl Log

pub fn append( &mut self, batch: &mut RecordBatch, ) -> Result<(Offset, Option<i64>), LogError>

Append a RecordBatch and return the assigned base_offset together with the log-append stamp the batch carries away from this call.

The log overwrites the batch’s base_offset with the next assigned offset. last_offset_delta sets how many absolute offsets this batch consumes.

The second element is Some(ms) exactly when the partition’s message.timestamp.type is LogAppendTime, and it is the broker clock reading this append stamped into the batch. It is Kafka’s LogAppendInfo.logAppendTime, which ProduceResponse.logAppendTimeMs reports; None is a CreateTime partition, which Kafka answers with -1.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

pub fn append_with_commit_stamp( &mut self, batch: &mut RecordBatch, stamp: u64, ) -> Result<Offset, LogError>

Append a commit-marker batch with a coordinator-supplied transaction stamp and return its assigned base offset.

The stamp is internal metadata. It is not encoded into the marker or any client-facing bytes. The marker must be a COMMIT control batch and a crate::StampSource must already be installed.

Errors

Returns an error for a non-commit batch, a missing stamp source, or any log/index I/O failure.

pub fn restarts_at_batch_base( &self, batch_base: Offset, batch_last: Offset, ) -> bool

Whether this follower log must restart at batch_base before it can take the leader batch [batch_base, batch_last]: Kafka’s Partition.appendRecordsToFollowerOrFutureReplica recovery from UnexpectedAppendOffsetException.

A leader serves the whole batch that holds the fetch offset, so a follower that restarted at a leader log start inside a batch – DeleteRecords, retention, or a diskless flusher’s trim put it there – receives a batch that begins below its own log end. When the log holds nothing (log_start == log_end) and the batch spans its end, restarting it at the batch’s base loses nothing, and the batch then appends where it begins. The follower raises its log start to the leader’s again from a later fetch response. A log that holds records keeps the Log::append_at refusal, as Kafka rethrows there.

pub fn append_at( &mut self, batch: &mut RecordBatch, offset: Offset, ) -> Result<(), LogError>

Append a RecordBatch whose base_offset the caller sets.

Unlike Log::append, this method does NOT overwrite batch.base_offset. The broker’s replicator uses it to keep the leader-assigned offset on the follower’s local log.

offset must not be below the log’s current Log::log_end_offset (or its reconciled frontier). If it is, this method returns LogError::OffsetMismatch. An offset above the end is a hole in the offsets, as a compacted leader’s log has, and is appended as it is. On success the log sets batch.base_offset to offset, which should already match, before it writes the batch.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

pub fn append_at_with_commit_stamp( &mut self, batch: &mut RecordBatch, offset: Offset, stamp: u64, ) -> Result<(), LogError>

Append a replicated commit marker at offset with its internal commit stamp. This is the follower counterpart of Log::append_with_commit_stamp.

Errors

Returns an error for an offset mismatch, a non-commit batch, a missing stamp source, or any log/index I/O failure.

impl Log

pub fn compaction_due( &self, now: SystemTime, last_stable_offset: Offset, ) -> bool

Whether a compaction pass over this partition is due, as Kafka’s LogCleanerManager decides it: the cleanable range has to be a large enough share of the log (min.cleanable.dirty.ratio), unless the dirty region has been dirty so long that max.compaction.lag.ms forces a pass regardless. min.compaction.lag.ms withholds the young tail from that range rather than holding the pass back.

The cleanup policy itself is not read here. That is the caller’s test, because a partition the policy excludes is one the broker’s cleaner skips before it opens the log at all.

pub fn compact(&mut self, ctx: &CompactionContext) -> Result<(), LogError>

Run one compaction pass over the cleanable sealed segments, holding the log throughout: Self::prepare_compaction, then CompactionJob::build, then Self::finish_compaction. A caller that shares the log between threads runs the three itself and builds without the log’s lock.

The pass never touches the active segment, and it stops short of the sealed segments min.compaction.lag.ms or ctx.last_stable_offset withholds – the same range Self::compaction_due measures, so the decision and the pass agree on what one pass would do. The output is one or more new sealed segments, grouped so none exceeds segment.bytes, starting at the lowest input base offset; together they replace the sealed segments this pass consumed. Any withheld sealed segments stay where they are and become cleanable once their records outlive the lag.

ctx carries the wall clock, which drives the KIP-534 delete-horizon computation, and the last stable offset. The pass keeps the last record of each producer in the producer state of this log, and the last batch of the pass, as an empty batch (RETAIN_EMPTY) when compaction removes all of their records. The last record of a producer is its last data batch, or a marker at its epoch when it has no data batch at that epoch. This is Kafka’s Cleaner.cleanSegments, which reads UnifiedLog.lastRecordsOfActiveProducers on every replica. A leader and a follower update that state for each batch that they append, so they keep the same batches.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

pub fn prepare_compaction( &mut self, ctx: &CompactionContext, ) -> Result<Option<CompactionJob>, LogError>

Take a compaction pass over the cleanable sealed segments from the log, for CompactionJob::build to run without it. See Self::compact for what one pass does.

Returns None when there is nothing to do: the cleanup policy has no compact, no sealed segment is cleanable, the log is fenced, or an earlier job is still out. Only one job runs at a time; it ends when Self::finish_compaction takes it back, or when it is dropped.

The job holds what the pass reads from the log now: the consumed segments’ open files, the aborted transactions of their range, and the last record of each active producer. Kafka’s cleaner reads the same facts when a pass starts. An append after this call lands past the consumed range, so it changes none of them; a producer it moves on only leaves the pass keeping an empty batch the producer no longer needs.

Errors

Returns an error when the log’s rolled segments could not be flushed.

pub fn finish_compaction( &mut self, built: BuiltCompaction, ) -> Result<CompactionOutcome, LogError>

Install a built compaction pass: replace the segments it read with its output.

Nothing is installed when the log has dropped or replaced a segment the pass read since Self::prepare_compaction, by a truncation, retention, DeleteRecords or a reset, or when the log has been fenced. The pass’s files are deleted and CompactionOutcome::Stale reports it, whether or not the build succeeded: a build that failed because the log took its input away is no fault of the disk.

Errors

Returns the build’s error when the build failed on a log that still holds its input, and an error when the swap’s log I/O fails or the rewritten segment is corrupt.

impl Log

pub fn advance_delivery_watermark(&mut self, now_ms: i64) -> DeliveryAdvance

Recompute the deliver-at-time watermark for the clock reading now_ms, and report when the next scheduled batch comes due.

The watermark is the first offset that is not visible yet. A fetch caps its limit offset at it, so a batch whose activation time has not passed is never served. The value stays inside [log_start_offset(), log_end_offset()], and it only moves forward while the records under it stay: a truncation carries it back down with the records it cut away, because the offsets they held may be filled again by a batch that is not due.

A topic that does not schedule delivery answers log_end_offset() before any I/O: this sits on the fetch hot path and an ordinary topic must pay nothing for it. A scheduled topic walks batch headers forward from the cached watermark, and skips whole segments whose own maximum timestamp proves everything in them is active. A repeat call with no change of time and no new records answers from the cache. Increasing clock uncertainty or moving the clock backward does not retract the published prefix; current settings govern records at and above the cached watermark.

Nothing here is persisted. The schedule is in the record timestamps, so after a restart or a leader change the walk starts again from the log start using the current clock and bound. If uncertainty increased or the clock moved backward, a rebuilt watermark can be below the prefix this instance had already released.

pub fn delivery_watermark(&self) -> Offset

Deliver-at-time watermark as the last advance left it.

This method reads the cached value and walks nothing, so a reader that must not do I/O can use it. It answers log_end_offset() on a topic that does not schedule delivery, where every durable record is visible.

pub fn pending_activation_ranges( &self, start: Offset, end: Offset, now_ms: i64, ) -> Vec<(Offset, Offset)>

Inclusive, batch-aligned offset ranges inside [start, end] that are not visible yet.

A share consumer may skip a waiting record and come back to it later, so it needs every gap in the window and not only the leading prefix that Log::advance_delivery_watermark reports. Adjacent ranges are merged. Each range covers whole batches even where the window cuts through one, because the share reader fetches with Log::read_raw and that is batch-granular.

The result is empty on a topic that does not schedule delivery, and empty when the window holds no records. Records below the published delivery watermark stay visible even if clock uncertainty subsequently increases or the clock moves backward.

impl Log

pub fn maintain(&mut self, now: SystemTime) -> Result<(), LogError>

Flush dirty data, write a log start retention moved once its checkpoint is due, and reclaim retired files whose deadlines have passed.

Errors

Returns the first flush or file deletion failure.

pub fn maintain_deferring_deletions( &mut self, now: SystemTime, ) -> Result<DueDeletions, LogError>

Self::maintain, except that the retired files whose deadlines have passed are handed back rather than deleted under &mut self.

Errors

Returns the first flush or checkpoint failure. The retired files then stay with the log.

pub fn take_due_deletions(&mut self, now: SystemTime) -> DueDeletions

Take the retired files whose deadlines have passed, leaving the rest.

pub fn restore_deletions( &mut self, remaining: DueDeletions, not_before: SystemTime, )

Hand back the files a DueDeletions::delete did not delete. They become due again at not_before, so a caller retrying a failing disk can space its attempts.

pub fn maintenance_delay(&self, now: SystemTime) -> Option<Duration>

Time until the next flush, log-start checkpoint or retired-file deadline, or None when idle.

impl Log

pub fn open(dir: impl AsRef<Path>, config: LogConfig) -> Result<Self, LogError>

Open or create a Log at dir.

This method finds existing segments by .log filename and marks all but the latest as sealed. If the directory is empty, it creates a fresh active segment at offset 0.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

impl Log

pub fn plan_raw_read( &self, fetch_offset: Offset, limit_offset: Offset, max_size: ByteSize, ) -> Result<RawReadPlan, LogError>

Plan a raw read from fetch_offset, to run once the lock on this log is released.

A fetch of cold bytes walks batch headers, asks the kernel to read ahead and copies or describes megabytes, and all of it waits on the disk or costs system calls: milliseconds per fetch of a consumer that is catching up. Under the log lock those milliseconds hold up every append of the partition, which a producer sees as its publish p99 while a backlog drains. The plan takes what the read needs of the log in a few Arc clones and index lookups, and the read then runs on it without the log: RawReadPlan::read_bytes and RawReadPlan::read_regions give what Log::read_raw and Log::read_raw_desc would, as of the moment of the plan.

The plan views each segment the read may walk, from the one that holds fetch_offset on, as RawReadPlan walks them: a later segment is read from the offset after the one before it, once that one is read to its end, so every view’s offset is known now. Planning stops once the views can fill max_size, counting each from its index floor, which is at or before where its read starts.

Errors

Returns LogError::OffsetTooLow below the log start and LogError::OffsetBelowLocalStart below the local one, as the reads do, and an error for an offset too far past a segment for its index.

pub fn read_raw_desc( &self, fetch_offset: Offset, limit_offset: Offset, max_size: ByteSize, ) -> Result<RawReadDesc, LogError>

Descriptor variant of Log::read_raw for the zero-copy sendfile fetch path.

This method walks sealed segments and then the active segment exactly as read_raw does. It collects one krabka_protocol::records::FileRegion for each contributing segment through Segment::read_raw_desc, instead of owned Bytes. Multi-segment fetches are not coalesced. Each region goes out in its own sendfile call, so the cross-segment copy disappears.

The selected byte ranges are byte-identical to what read_raw would have returned for the same (fetch_offset, limit_offset, max_size). It plans and runs under the caller’s lock: see Log::plan_raw_read for the fetch path’s split.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

pub fn read( &self, offset: Offset, max_size: ByteSize, ) -> Result<ReadOutput, LogError>

Read batches from offset and return up to max_size of .log data.

The read walks sealed segments first, then the active segment, so a read can span segment boundaries. It goes on into a segment only after it reads the segment before it to the end.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

pub fn read_raw( &self, fetch_offset: Offset, limit_offset: Offset, max_size: ByteSize, ) -> Result<RawRead, LogError>

Like Log::read, but returns verbatim wire bytes with no decode.

The read walks sealed segments and then the active segment. It goes on into a segment only after it reads the segment before it to the end. It includes only batches with base_offset < limit_offset, up to max_size, and always at least one batch. The first batch is included whole, even when it is larger than max_size. It plans and runs under the caller’s lock: see Log::plan_raw_read for the fetch path’s split.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

impl Log

pub fn set_stamp_source( &mut self, source: Arc<dyn StampSource>, ) -> Result<(), LogError>

Inject the StampSource that folds the additional internal stamp coordinate into this partition.

This method opens or recovers every local segment’s .stampindex. From this point each durable non-transactional batch is stamped at append, while transactional batches are stamped together when their COMMIT marker lands. Aborted and still-open transactions stay unstamped.

This is the only switch that enables the feature. With no source injected the log stamps nothing and its bytes are identical to those of an unstamped log. It changes no wire-facing state: not the .log bytes, not offset assignment, not the LSO, and not the high-watermark.

Errors

Returns an error when opening the .stampindex sidecar fails.

Panics

Panics only if there is no active segment, which cannot happen after Log::open.

pub fn stamp_source(&self) -> Option<Arc<dyn StampSource>>

Clone the currently installed internal stamp source.

Log-directory moves use this to preserve stamping when they close and reopen a partition around the final filesystem swap.

pub fn stamp_for_offset(&self, offset: Offset) -> Option<u64>

The internal stamp that covers offset, or None when there is no such stamp.

The result is None when the partition is unstamped, that is, when no source is injected. It is also None when no stamped range covers offset. This method searches all local segment indexes because a transaction can commit after the segment holding its data has rolled. This is an internal, server-side query. No produce or fetch handler calls it, so the stamp can never reach a client-facing response.

impl Log

pub fn checkpoint_log_start_if_due(&mut self) -> Result<(), LogError>

Write the log start retention moved, once it has waited LogConfig::start_offset_checkpoint_interval. Log::maintain and Log::maintain_deferring_deletions call this.

Errors

Returns LogError::Io when the checkpoint cannot be written. The write stays owed.

pub fn flush_log_start_checkpoint(&mut self) -> Result<(), LogError>

Write an owed log start now, whatever its deadline. Log::sync calls this, and the broker calls it when a partition’s writer stops, as Kafka’s LogManager.shutdown checkpoints every log start. A fenced log is being removed, so it writes nothing.

Errors

Returns LogError::Io when the checkpoint cannot be written.

impl Log

pub fn dir(&self) -> &Path

Directory this log was opened against. The broker’s intra-broker log-dir reassignment (KIP-113) reads this to find the current owning log.dir of a partition. The broker does not have to repeat the directory-layout convention.

pub fn log_start_offset(&self) -> Offset

First absolute offset any reader may ask for, wherever the records for it live: Kafka’s global logStartOffset.

On a tiered topic (KIP-405) this can sit below Log::local_log_start_offset, and the offsets between the two are the ones the remote tier serves. A fetch below this floor is OFFSET_OUT_OF_RANGE and no tier answers it.

pub fn established_log_start(&self) -> Option<Offset>

The global floor, when this process is the one that moved it.

None on a log whose floor was only inferred from the segments present at Log::open. A caller that would refuse or delete on the strength of the floor must use this rather than Log::log_start_offset: on a tiered partition whose local segments were evicted, the inferred floor sits above everything the archive holds, so refusing below it hides readable records and deleting below it destroys them. The log-start checkpoint carries an established floor across a restart, so a reopened log answers None only when nobody has moved it.

pub fn reported_log_start(&self) -> Offset

The global floor as a leader reports it to the replicas that follow it: Kafka’s UnifiedLog.logStartOffset in a Fetch response.

This is Log::log_start_offset except on a remote-storage-enabled log whose floor nobody established. Such a log reopened over the segments local retention left, so its inferred start sits above everything the archive holds. Kafka’s LogLoader.load never infers a remote-storage-enabled log’s start from its segments: it takes the log-start checkpoint, which is 0 when nobody moved it. A follower raises its own start to what its leader reports, and a follower’s start is a floor it refuses reads below and deletes the archive below once it leads, so reporting the inferred start would hide and then destroy every archived record under it.

pub fn log_start_epoch(&self) -> Option<LeaderEpoch>

The epoch of the earliest leader-epoch entry, when a log start that somebody established truncated the cache to it: Kafka’s leaderEpochCache.earliestEntry after truncateFromStart(logStartOffset).

Only then does the entry say something about the remote tier. Kafka’s RemoteLogManager deletes a remote segment whose epochs all lie below it (deleteLogSegmentsDueToLeaderEpochCacheTruncation), which is right for the epoch that owns the log start, because every offset from there on has that epoch or a later one. An earliest entry that no advance of the log start put there (a cache that never held the older epochs, or one cleared by Log::reset_to and refilled from the fetched records) says nothing about them, so this answers None. The epoch cache is cut from the start only by Log::set_log_start_offset and by the log-start checkpoint Log::open restores, never by the segments a tiered partition still holds locally, so a restart does not move it.

pub fn first_segment_base(&self) -> Option<Offset>

The base offset of the log’s first segment on disk, sealed or active, or None when it holds none: Kafka’s logSegments().get(0).baseOffset().

pub fn set_log_start_offset( &mut self, new_start: Offset, ) -> Result<(), LogError>

Move the global log start up to new_start.

The pointer only ever moves forward, the way Kafka’s maybeIncrementLogStartOffset does: a request that names an offset at or below the current floor is a no-op. trim_to_offset uses this method for the active-segment case, the broker’s DeleteRecords handler uses it, and the RemoteLogManager uses it after a remote segment is deleted. This method does NOT truncate on-disk segments. It only moves the start pointer.

The new value is checkpointed to log-start-offset-checkpoint before this method returns, so a start no segment name witnesses – a trim that landed inside a segment, or a tiered floor below every segment still on disk – survives a reopen. Without that, Log::open derives the start from the first surviving base offset and serves records that a DeleteRecords already deleted. Retention moves the start to a segment boundary instead and defers the write; see Log::checkpoint_log_start_if_due.

When the start does move up, the rest of the log follows it the way Kafka’s maybeIncrementLogStartOffset makes it follow:

• the leader-epoch cache drops what lies below the new start (leaderEpochCache.truncateFromStart, see LeaderEpochCheckpoint::truncate_from_start);
• a complete transaction whose marker lies below the new start stops holding the last stable offset (ProducerStateManager.onLogStartOffsetIncremented → removeUnreplicatedTransactions), because nothing a leader change could truncate is left of it;
• the first unstable offset is recomputed (maybeIncrementFirstUnstableOffset).

new_start must be non-negative.

Errors

Returns LogError::InvalidArgument if new_start is negative, and LogError::Io if the log-start or leader-epoch checkpoint cannot be written.

pub fn differs_from_leader_log_start(&self, leader_start: Offset) -> bool

Whether Log::follow_leader_log_start would change anything for a leader that reports leader_start. A follower answers this on every fetch response, so it only reads.

pub fn follow_leader_log_start( &mut self, leader_start: Offset, ) -> Result<Offset, LogError>

Take a leader’s global log start as this follower’s own, and return the resulting log start.

A log with no remote tier only ever raises its start, as Kafka’s ReplicaFetcherThread.processPartitionData → maybeIncrementLogStartOffset does, through Log::trim_to_offset.

A remote-storage-enabled log takes the leader’s start in either direction and establishes it. The leader is the replica that owns the archive and moves its floor, so a follower start above the leader’s is one no acknowledged deletion put there:

• A broker without the fix for #1444 reported a restarted leader’s inferred start, the base of its first local segment, and its followers established it. Such a floor hides the archive below it, and remote retention deletes against it once the follower leads.
• Log::reset_to rebuilds a follower at its leader’s local start, and Kafka’s truncateFullyAndStartAt(localLogStartOffset, leaderLogStartOffset) gives the rebuilt log the leader’s global start beneath it.

Lowering the start leaves the local segments and the leader-epoch cache alone, so Log::log_start_epoch answers None until a later raise cuts the cache again.

Errors

Returns LogError::InvalidArgument if leader_start is negative, and LogError::Io if the log-start checkpoint cannot be written or a raise fails.

pub fn reset_to(&mut self, new_base: Offset) -> Result<(), LogError>

Reset the log to be empty at new_base.

This method drops every segment and every on-disk file, then creates a fresh active segment at new_base. The replicator’s OFFSET_OUT_OF_RANGE recovery path uses it when the follower has fallen behind the leader’s log_start. truncate_to cannot help there, because log_start must move forward past the point where no local data exists.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

pub fn take_producer_snapshot(&mut self) -> Result<(), LogError>

Kafka’s ProducerStateManager.takeSnapshot: write the producer state at the log end offset, so a reopen restores it from the snapshot and replays nothing. A snapshot already at that offset is kept, and none is written at or below the log start, which Kafka takes as the last snapshot offset when no snapshot survives a reload.

Errors

Returns an error when the snapshot cannot be written.

pub fn fence(&mut self)

Refuses every append from now on with LogError::Fenced, and stops any compaction job.

The broker fences a partition’s log, under the log’s mutex, before it removes the partition’s directory. An append that holds the mutex first finishes before the directory goes; any later one, from whatever task, is refused rather than written into a log that is about to be deleted. Reads are unaffected.

A compaction job builds without the mutex, writing .cleaned files into the directory. The fence stops it at its next batch and returns only once it writes nothing more, so the directory can go as soon as the fence returns. Log::finish_compaction then installs nothing.

pub fn read_may_block(&self, fetch_offset: Offset) -> bool

Whether a raw read from fetch_offset may need bytes outside memory.

Buffered reads of this process’s active-segment appends use the page cache. Direct reads can run in memory when the tail cache covers the sparse-index floor and every byte after it. Older segments, reopened bytes, and evicted direct tails need the blocking path. The caller must retain the log lock through the read so a concurrent append cannot evict the checked bytes.

pub fn append_may_block( &mut self, incoming_size: ByteSize, incoming_max_timestamp: i64, ) -> bool

Whether appending incoming_size bytes whose newest record is stamped incoming_max_timestamp may wait on more than the page cache.

It is true when the log’s flush policy can make an append fsync (flush.messages, flush.ms, or a flush on every append), and when the append would roll the active segment, which creates files and can wait for an earlier roll’s flush budget, and for every append to a log with preallocate: that append can reserve the segment’s blocks, open its O_DIRECT handle, or wait for room in its direct write queue. It errs towards true: a caller that sees false may run the append where a wait would stall others.

pub fn verbatim_append_may_block(&mut self, batch: &VerbatimBatch) -> bool

Whether appending this one verbatim batch can wait on storage. An initialized direct writer with queue space can accept ordinary batches without a worker handoff. Flushes, epoch changes, segment setup, and queue backpressure still require blocking placement. The caller must hold the log lock through the append.

pub fn log_end_offset(&self) -> Offset

Next offset that append will assign.

pub fn size(&self) -> ByteSize

Total .log size across sealed and active segments.

The value comes from the segments’ tracked logical size, not from a filesystem stat. It therefore shows buffered appends immediately and in the same way on every platform. On some operating systems a directory stat can lag an open, unflushed write handle.

pub fn lso(&self) -> Offset

First unstable offset: the first offset of the earliest transaction that is open, or complete with a marker the high watermark has not passed. It is the log end offset when there is no such transaction.

This value does not know the high watermark. A reader must use Log::last_stable_offset, which releases replicated transactions and caps the answer at the high watermark.

pub fn pending_transaction_start( &self, producer_id: ProducerId, ) -> Option<Offset>

First offset of producer_id’s currently open transaction on this partition, or None when no transaction from that producer is pending.

pub fn transaction_marker_state( &self, producer_id: ProducerId, ) -> (i16, i32, bool)

Producer and coordinator generations used to admit one transaction marker, plus whether that producer currently has an open transaction. Missing generations use Kafka’s -1 sentinel.

pub fn producer_state_entry( &self, producer_id: ProducerId, ) -> Option<ProducerSnapshotEntry>

The producer state this log holds for producer_id, if any.

Append and recovery update the same entry with the same rule, so the value after a live append equals the value that a reopen rebuilds.

pub fn producer_state_snapshot(&self) -> Vec<ProducerSnapshotEntry>

Producer state restored from the newest valid Kafka-compatible snapshot and the uncovered local log tail.

pub fn recovered_producers(&self) -> Vec<RecoveredProducer>

Every producer’s state with the batches it appended before its last one: Self::producer_state_snapshot plus the retained batches that Kafka’s ProducerStateEntry keeps for duplicate detection. After a reopen they are the batches the replay past the loaded snapshot rebuilt.

pub fn close(self)

Close all segments, first taking a producer-state snapshot at the log end as Kafka’s UnifiedLog.close does. A snapshot that cannot be written is not an error: the next open replays the log instead.

A running compaction build stops first, as on Log::fence, and the call returns once it writes nothing more.

pub fn set_config(&self, new: LogConfig)

Atomically swap the active LogConfig.

The next retention or roll check reads the new value. In-flight append calls hold the lock for very short windows and will not see a half-applied config.

Callers can use this method through &self. The Arc<RwLock<…>> wrapper permits mutation of the inner value without an exclusive borrow on the Log.

Delivery settings apply to subsequent watermark advances. Changing clock uncertainty does not retract the already published delivery prefix; see LogConfig::delivery_clock_uncertainty.

pub fn config_snapshot(&self) -> LogConfig

Snapshot the current config. This allocates a clone, which is cheap because LogConfig is small and Clone.

pub fn aborted_in_range(&self, start: Offset, end: Offset) -> Vec<AbortedTxn>

Return all aborted transactions whose offset range overlaps [start, end), including entries in sealed segments.

Kafka’s UnifiedLog.collectAbortedTransactions: the search starts at the first segment that reaches start and stops at the first index that reports it complete (see TxnIndex::collect_aborted).

pub fn epoch_checkpoint(&self) -> &LeaderEpochCheckpoint

Access the per-partition leader-epoch checkpoint.

pub fn assign_epoch_start_offset( &mut self, epoch: LeaderEpoch, start_offset: Offset, ) -> Result<(), LogError>

Kafka’s UnifiedLog.assignEpochStartOffset: record that epoch starts at start_offset through LeaderEpochCheckpoint::assign. Partition.makeLeader calls it with the log end offset when a replica takes a new leader epoch, so the leader can place a follower’s last_fetched_epoch before it has written anything in that epoch.

Errors

Returns an error for a negative epoch or offset, or when the checkpoint cannot be persisted.

pub fn reconcile_next_offset(&mut self, frontier: Offset)

Reconcile append-at offset assignment to an external next-offset frontier.

Diskless partitions use the KRaft metadata log as the offset authority. After a crash, KRaft may have committed a next-offset that is ahead of the recovered local WAL tail. In that case the gap is intentional: the caller sets this frontier and the next append-at must use it instead of the local LEO. Classic logs never call this method and keep the default frontier 0.

pub fn clear_reconciled_next_offset(&mut self)

Drop the frontier Log::reconcile_next_offset set, so append-at bases are held only to the log end offset again.

The frontier binds the broker that assigns this log’s offsets. A follower assigns none: it appends its leader’s batches at the leader’s offsets, and Kafka’s UnifiedLog.appendAsFollower refuses only a batch below the log end. A diskless replica that reopened behind the committed frontier – it was down while its leader appended – must take the batches between its log end and that frontier from the leader, and the frontier would refuse every one of them.

impl Log

pub fn sync(&mut self) -> Result<(), LogError>

Drain rollover work and fsync every segment to stable storage, independent of crate::LogConfig::flush_on_append.

This method also fsyncs the log directory after it creates a new segment file. The segment therefore stays reachable after a crash on filesystems that require a parent-directory fsync. A log start that retention moved and has not yet checkpointed is written too.

Errors

Returns a LogError if the underlying segment or directory flush fails.

impl Log

pub fn tick( &mut self, now: SystemTime, high_watermark: Offset, ) -> Result<(), LogError>

Periodic maintenance: apply time- and size-based retention when cleanup.policy holds delete, and delete segments below the log start offset under every policy. The passes follow Kafka’s UnifiedLog.deleteOldSegments: size retention deletes an oldest segment only while the bytes over retention.bytes still cover the whole segment, and time retention ages a segment by Kafka’s largestTimestamp() – its newest record timestamp, or its .log file’s modification time when no record carries one. The walk covers the active segment too; when every segment goes, the active one is rolled first, as Kafka’s deleteSegments does, and the fresh empty segment stays.

Like Kafka’s deletableSegments, the walk stops at the first segment the partition’s high_watermark has not passed: a segment whose upper bound (the next segment’s base, or the log end for the active one) is above it still holds records followers may not have replicated.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

impl Log

pub fn local_log_start_offset(&self) -> Offset

First absolute offset still readable from this broker’s local disk (KIP-405): Kafka’s localLogStartOffset.

It is the base offset of the oldest segment on disk, raised to the global Log::log_start_offset when a DeleteRecords floor sits above the files that are still there. Dropping a local segment whose copy is in the remote tier moves this pointer and leaves the global one where it was, so the offsets in [log_start_offset(), local_log_start_offset()) are exactly the ones the remote tier serves.

pub fn delete_local_segments_through( &mut self, target: Offset, ) -> Result<usize, LogError>

Delete every sealed segment whose last_offset < target from disk (KIP-405).

Only Log::local_log_start_offset moves. The global Log::log_start_offset stays where it was, because the records are still in the remote tier and a fetch for them must reach it rather than answer OFFSET_OUT_OF_RANGE. Kafka splits the two floors the same way: local.retention.* moves localLogStartOffset and leaves logStartOffset to DeleteRecords and to remote-segment deletion.

This method never touches the active segment. The producer snapshot at each removed segment’s base goes with it. It returns the number of segments removed. It does nothing and returns Ok(0) when target <= local_log_start_offset().

The caller must confirm that these segments are safely in the remote tier (CopySegmentFinished) before it calls this method. Log enforces no tiered-storage invariants. See crates/broker/src/remote_log_manager.rs for the production caller.

Errors

Returns LogError::InvalidArgument if target is negative.

pub fn active_segment_export(&self) -> Option<ActiveSegmentExport>

Describe the active segment for tiered local retention (KIP-405), or return None when the log has no active segment.

max_timestamp is Kafka’s largestTimestamp(), so an empty segment and a segment whose records carry no timestamp both report the .log file’s modification time.

pub fn roll(&mut self) -> Result<bool, LogError>

Kafka’s UnifiedLog.roll(), as tiered local retention calls it: seal the active segment and open an empty one at the log end.

Kafka’s deletableSegments rolls a tiered log’s active segment once it breaches local.retention.ms or local.retention.bytes. The remote tier never holds the active segment, so without the roll an idle partition keeps its newest records on local disk until segment.ms or segment.bytes rolls the segment.

Returns true when the log rolled. It does nothing and returns false when there is no active segment or the active segment is empty: an empty segment holds nothing to copy, and the new segment would open at the same base offset.

Errors

Returns an error when the flush, the producer snapshot, the seal, or the new segment’s files fail.

pub fn tierable_segments(&self) -> Vec<SegmentExport>

Describe every sealed segment for tiered-storage offload (KIP-405).

The result includes sealed segments after their rollover flush has published the boundary snapshot. It never includes the active segment.

last_offset comes from the next segment’s base_offset. For the most-recent sealed segment it comes from the active segment’s base. The value is therefore correct even for segments loaded from disk without a tail scan. max_timestamp is Kafka’s largestTimestamp(), so a segment with no record timestamp reports its file’s modification time rather than an unknown.

impl Log

pub fn legacy_offsets_before( &self, timestamp: i64, max_num_offsets: usize, ) -> Result<Vec<Offset>, LogError>

Legacy ListOffsets v0 segment boundaries at or before timestamp, newest first and capped by max_num_offsets.

Version 0 predates record timestamps. Its timestamp lookup uses each segment file’s modification time and returns segment base offsets (plus the log end for a non-empty active segment), matching Kafka’s legacyFetchOffsetsBefore behavior.

Errors

Returns an I/O error when a segment’s modification time cannot be read.

pub fn offset_for_timestamp(&self, target_ts: i64) -> Option<(Offset, i64)>

Earliest local (offset, record_timestamp) whose record timestamp is >= target_ts, excluding records below the logical log start.

The search reads sealed segments oldest-first and then the active segment, and goes past a segment whose max_timestamp >= target_ts comes only from records below the log start. The per-segment helper does the index lookup and the forward scan. The result is None when no retained local record qualifies, including the case of an empty log.

This is the lookup the log’s own bookkeeping uses, and it applies no max.decompressed.message.bytes. A ListOffsets answer goes through Self::offset_for_timestamp_checked, which does, and which stops at the first candidate segment as Kafka does.

pub fn offset_for_timestamp_checked( &self, target_ts: i64, ) -> Result<Option<(Offset, i64)>, LogError>

Self::offset_for_timestamp under the log’s max_decompressed_record, as Kafka trunk’s UnifiedLog.fetchOffsetByTimestamp answers it.

Kafka’s scan decompresses a compressed batch whose max timestamp reaches the target, up to the record it returns, and fails on a record above the limit. A lookup that never reads the oversized record is unaffected.

The answer comes from the first segment whose largest timestamp reaches target_ts (Kafka’s LogSegment.largestTimestamp, which falls back to the file’s modification time for a segment with no record timestamp). When that segment holds no retained match the answer is None, even if a later segment holds one: Kafka 4.3.1’s UnifiedLog.searchOffsetInLocalLog returns segment.findOffsetByTimestamp(target, startOffset) from that first segment and reads no other.

Errors

Returns LogError::RecordTooLarge for such a record.

pub fn offset_for_timestamp_checked_from( &self, target_ts: i64, minimum: Offset, ) -> Result<Option<(Offset, i64)>, LogError>

Self::offset_for_timestamp_checked restricted to offsets at or above minimum, also respecting this log’s own start offset.

A caller with an authoritative deletion floor from another storage projection can search the remaining records exactly without changing this log’s state or clamping an earlier hit to a different record.

Errors

Returns LogError::RecordTooLarge when a compressed record the scan reads exceeds the configured decompressed-record limit.

pub fn max_timestamp_offset_and_ts( &self, ) -> Result<Option<(Offset, i64)>, LogError>

Offset and timestamp of the record that carries the partition’s largest timestamp (KIP-734 MAX_TIMESTAMP), under the log’s max_decompressed_record, as Kafka trunk’s UnifiedLog.fetchOffsetByTimestamp(MAX_TIMESTAMP) answers it: Kafka decompresses only the batch that holds the log’s maximum, up to the record it returns.

The scan reads sealed segments and then the active segment. Ties resolve to the earliest offset: the first segment wins, and the first record within it wins. The result is None when the log holds no records.

Errors

Returns LogError::RecordTooLarge for a record of that batch above the limit.

impl Log

pub fn last_stable_offset(&mut self, high_watermark: Offset) -> Offset

The last stable offset that a reader sees at high_watermark.

The call first releases every complete transaction whose marker offset is below high_watermark, as Kafka’s ProducerStateManager.onHighWatermarkUpdated does. It then answers min(first unstable offset, high_watermark), which is Kafka’s UnifiedLog.lastStableOffset. A transaction whose marker the high watermark has not passed still holds the answer at its first offset, because a leader change can truncate that marker away.

Call it with the partition’s current high watermark. The high watermark only moves forward, so a released transaction never has to come back.

pub fn release_replicated_transactions(&mut self, high_watermark: Offset)

Release every complete transaction whose marker offset is below high_watermark, and move the first unstable offset forward.

The writer calls this before it appends a transaction marker, so the set of complete transactions stays bounded on a partition that no reader fetches from.

impl Log

pub fn truncate_to(&mut self, offset: Offset) -> Result<(), LogError>

Truncate the log so that no record at offset >= offset remains. Replication and leader election use this method.

The target can lie below the log start. This is Kafka’s UnifiedLog.truncateTo, which never refuses: when no local segment reaches down to offset, the log is reset empty at offset (Log::reset_to, Kafka’s truncateFullyAndStartAt); otherwise the log is truncated and the log start moves down to offset if it was above it. A follower whose start sits above the leader’s divergence point, after a long outage and retention or an unclean election, can therefore always truncate.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

pub fn trim_to_offset(&mut self, target: Offset) -> Result<Offset, LogError>

Trim from the start of the log and return the resulting log_start_offset.

This method drops every sealed segment whose last offset is < target. It advances log_start_offset when target falls inside the active segment. It never deletes the active segment.

target is clamped to [0, log_end_offset()]. A caller that asks for a trim past the LEO gets a trim to the LEO.

Errors

Returns LogError::InvalidArgument if target < 0.

impl Log

pub fn append_verbatim( &mut self, batch: &VerbatimBatch, ) -> Result<(Offset, Option<i64>), LogError>

Append a producer batch verbatim and return the assigned base_offset plus the log-append stamp, as Log::append does.

The log does not decode or re-encode the batch, and it takes base_offset from the log’s current end. This is the produce zero-copy passthrough path. The caller has already CRC-validated the bytes and read the header fields into VerbatimBatch. The log patches base_offset and partition_leader_epoch, which both sit outside the CRC region, then writes the bytes as they are. Offset assignment, segment roll, flush, LSO and transaction tracking, and the leader-epoch checkpoint behave exactly as in Log::append. The verbatim path and the owned path differ only in how the log gets the batch bytes, not in any log-level invariant.

Errors

Returns an error when log I/O fails, a record or index is corrupt, or the requested offset violates the segment state.

pub fn append_verbatim_at( &mut self, batch: &VerbatimBatch, base_offset: Offset, ) -> Result<Offset, LogError>

Append a producer batch verbatim at a caller-supplied base offset.

base_offset must not be below the log’s current Log::log_end_offset. If it is, this method returns LogError::OffsetMismatch and appends nothing. A base_offset above the end is a hole in the offsets, as a compacted leader’s log has, and is appended as it is. On success the log stamps the stored batch with base_offset and the batch’s leader epoch. Control batches decode the marker for shared bookkeeping; CRC-covered bytes are never re-encoded.

Errors

Returns LogError::OffsetMismatch when base_offset is below the log end offset. It also propagates segment and checkpoint I/O errors and validation errors.

impl Log

pub fn try_append_verbatim_group( &mut self, batches: &[&VerbatimBatch], ) -> Option<Result<Vec<Offset>, LogError>>

Try to append ordinary producer batches with one aligned Direct I/O write.

None leaves the batches unwritten. The caller then uses individual appends. Groups are at most 1 MiB and stay in one nonempty active segment. Transactions, scheduled delivery, stamps, append-time timestamps, and explicit flush policies use individual appends. The caller has already validated the batch CRCs.

A successful group returns each assigned base offset in input order.

Errors

A write or bookkeeping failure rolls the whole group back before returning the error.

impl Log

pub fn has_ongoing_transaction( &self, producer_id: ProducerId, producer_epoch: i16, ) -> bool

Whether producer_id has an open transaction on this partition at producer_epoch. Kafka’s UnifiedLog.hasOngoingTransaction.

pub fn maybe_start_transaction_verification( &mut self, batch: TransactionalBatch, supports_epoch_bump: bool, (now_ms, expiration_ms): (i64, i64), ) -> Result<VerificationGuard, TransactionAppendRefusal>

Start a verification for a transactional batch, and return its guard.

A producer with an open transaction at the batch epoch needs no verification, and gets VerificationGuard::SENTINEL. Otherwise the producer’s guard is created, or kept with its lowest sequence moved as Kafka’s maybeUpdateLowestSequenceAndEpoch moves it. The call also drops verification state older than expiration_ms, as Kafka’s ProducerStateManager.removeExpiredProducers does.

Errors

Returns TransactionAppendRefusal::StaleProducerEpoch when the batch epoch is below the producer’s epoch on this partition.

pub fn check_transactional_append( &self, batch: TransactionalBatch, presented: VerificationGuard, ) -> Result<(), TransactionAppendRefusal>

Check a batch against the producer’s transaction state just before it appends, under the same lock as the append.

A transactional data batch with no open transaction at its epoch must present the producer’s current guard. A non-transactional data batch from a producer with an open transaction is refused. This is Kafka’s check in UnifiedLog.analyzeAndValidateProducerState and in ProducerAppendInfo.appendDataBatch.

Errors

Returns the refusal that Kafka’s append throws.

Trait Implementations

impl Debug for Log

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.