Reading records

Two APIs answering two different questions. Neither is a consumer — see The read path.

Both take a &Cluster. Get one from Cluster::connect, or from admin.cluster().

The tail — "what just happened"

The most-used view in any Kafka UI, and it is a backward walk rather than a forward read.

use kafka_read::TailSpec;

async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
// The last 500 records of the topic, across all partitions.
let tails = kafka_read::tail(cluster, &TailSpec::new("orders", 500)).await?;

for tail in &tails {
    println!(
        "partition {}: {} records ({}..{}), {} malformed batches, {} fetches, more below: {}",
        tail.partition,
        tail.records.len(),
        tail.log_start,
        tail.log_end,
        tail.malformed,
        tail.fetches,
        !tail.reached_log_start,
    );
}

// Or narrow it.
let spec = TailSpec::new("orders", 100).partitions([0, 3]);
let tails = kafka_read::tail(cluster, &spec).await?;
Ok(())
}

This returns a Vec rather than a stream, deliberately: you asked for a bounded number of records, and the implementation reads roughly that many bytes rather than the whole partition. On a compacted topic with large offset gaps it still converges — the step grows when a chunk yields fewer records than its offset span suggested.

limit is a topic-wide target, not a per-partition ration: partitions that hold nothing are excluded from the division before anything is read, and a partition that runs out of records hands its unspent share to the ones that have not — so "the last 500" of a topic whose records all sit in one partition is 500 records, not ⌈500 ÷ partitions⌉. Each PartitionTail also says why its walk stopped: reached_log_start is true when the oldest record returned is the oldest the partition retains, and false when more records lie below — anchor the next page at records[0].offset - 1 to keep walking. The log_start/log_end bounds the walk measured ride along, so "how much does this partition hold" costs no extra ListOffsets.

The scan — "show me this topic"

use futures::StreamExt;
use kafka_read::{ScanEvent, ScanSpec, StartPosition};

async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
let spec = ScanSpec::new("orders")
    .from(StartPosition::Earliest)
    .partitions([0, 1, 2])
    .limit(10_000);

let mut stream = Box::pin(kafka_read::scan(cluster, spec).await?);
while let Some(event) = stream.next().await {
    match event? {
        ScanEvent::Record(record) => {
            println!("{}:{} {:?}", record.partition, record.offset, record.value);
        }
        ScanEvent::Progress(progress) => {
            if let Some(fraction) = progress.fraction() {
                println!("{:.0}%", fraction * 100.0);
            }
        }
        ScanEvent::Malformed { offset, last_offset, reason, .. } => {
            eprintln!("offsets {offset}..={last_offset:?} did not decode: {reason}");
        }
        _ => {}
    }
}
Ok(())
}

Box::pin because the returned stream is not Unpin.

Start positions:

Handle Malformed, do not ignore it

This is the point of the whole design. A batch that will not decode becomes an event carrying the offsets it covered and the raw bytes, and the scan continues.

use kafka_read::ScanEvent;
fn example(event: ScanEvent) {
match event {
    ScanEvent::Malformed { offset, last_offset, raw, reason } => {
        // Render "offsets 4102–4530 would not decode", offer the hex.
        // Do NOT abort the scan, and do NOT treat this as a transport error.
    }
    _ => {}
}
}

Granularity is a batch, not a record — a corrupt record takes its batch with it, bounded by max.message.bytes. That was a deliberate choice, not a limitation discovered late; see Tolerant decoding.

What you will not see as Malformed: a batch truncated by max_bytes (normal on every fetch), control batches (transaction markers), or aborted records under CommittedOnly. Those are filtered silently, because reporting them means crying wolf on every fetch of every healthy cluster.

Where the scan actually started

A start position can name something the log no longer holds: an offset below the retained range, an offset the partition has not reached, or a timestamp that resolves to no offset. The scan substitutes the nearest position it can honour — right for browsing, quietly wrong for "did my record land at 900001" — and says so: the first events of every scan are one ScanEvent::PartitionStarted per partition, carrying the offset reading actually begins at and, when the requested position could not be honoured, a StartSubstitution naming what was asked and what was done instead. For a timestamp start, start_offset is what the instant resolved to on that partition, so an empty window can be rendered as "14:30 resolved to no offset" without a second ListOffsets.

Transactions and visibility

use kafka_read::{ScanSpec, Visibility};

fn example() {
// Default: read_uncommitted. Aborted records are visible.
let all = ScanSpec::new("orders").visibility(Visibility::All);

// read_committed. Aborted records filtered client-side.
let committed = ScanSpec::new("orders").visibility(Visibility::CommittedOnly);
}

read_committed does not mean the broker filters for you — it sends the records plus an AbortedTransactions list, and the client does the work. That is Kafka's design, not a shortcut here.

Filtering

use kafka_read::{RecordFilter, ScanSpec};
fn example(filter: RecordFilter) {
let spec = ScanSpec::new("orders").filter(filter);
}

RecordFilter runs client-side, after decoding — Kafka has no server-side filtering, so a filter reduces what you iterate, not what crosses the network. Use partitions and limit to reduce bytes; use the filter to reduce noise.

Cancelling

Drop the stream. That is the whole protocol.

Dropping mid-scan releases the buffer, drops the in-flight fetch futures, and leaves every connection consistent — no half-read responses, nothing to unwind. See Cancel safety.

use futures::StreamExt;
async fn example(cluster: &kafka_meta::Cluster) -> kafka_read::Result<()> {
let spec = kafka_read::ScanSpec::new("orders");
let mut stream = Box::pin(kafka_read::scan(cluster, spec).await?);
while let Some(event) = stream.next().await {
    // …stop whenever you like; just drop it
    break;
}
Ok(())
}

Memory

Bounded by ScanSpec::max_buffered_records (default 10,000) across the whole scan, not per partition. Scanning a thousand-partition topic uses the same budget as scanning one — which is the difference between a UI backend that survives a large cluster and one that does not.

Lowering it tightens memory at the cost of cross-partition ordering: a smaller buffer forces more emits before every partition is represented, which widens the bounded reorder window that ScanEvent::Progress reports.