mod builder;
mod config;
mod fetch_session;
mod group;
mod group_metadata;
mod lock_order;
mod offset;
mod record;
mod stream;
pub mod compacted;
pub use builder::ConsumerBuilder;
pub use compacted::{
CompactedEntry, CompactedTable, CompactedTableClearListener, CompactedTableSnapshot,
CompactedTopicConsumer, TableChange,
};
pub use config::{
AutoOffsetReset, ConsumerConfig, GroupProtocol, IsolationLevel, PartitionAssignmentStrategy,
};
use group::ErasedRebalanceListener;
pub use group::{
ConsumerGroup, ConsumerRebalanceListener, CooperativeStickyAssignor, GroupCoordinator,
GroupMember, GroupState, HeartbeatController, HeartbeatStatus, MemberAssignment,
NoOpRebalanceListener, PartitionAssignor, PendingRebalance, RangeAssignor, RoundRobinAssignor,
StickyAssignor,
};
pub use group_metadata::ConsumerGroupMetadata;
pub use offset::{OffsetAndMetadata, OffsetStore, ResetOffset};
pub use record::{ConsumerRecord, ConsumerRecords, TopicPartition};
pub use stream::ConsumerStream;
use ahash::{AHashMap as HashMap, AHashSet as HashSet};
use std::future::{Future, Ready, ready};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use parking_lot::Mutex as SyncMutex;
use tracing::{debug, error, info, trace, warn};
use lock_order::LeveledRwLock;
use crate::error::{KrafkaError, ProtocolErrorKind, RecvError, Result};
use crate::metadata::{BrokerInfo, ClusterMetadata, TopicInfo, broker_info_for_node};
use crate::metrics::{ConnectionMetrics, ConsumerMetrics};
use crate::network::{ConnectionConfig, ConnectionPool};
use crate::protocol::{
ApiKey, FetchPartitionRequest, FetchRequest, FetchResponse, FetchTopicRequest,
ListOffsetsRequest, ListOffsetsRequestPartition, ListOffsetsRequestTopic, ListOffsetsResponse,
RecordBatch, VersionedDecode, VersionedEncode, validate_topic_name, validate_topic_names,
versions,
};
use crate::{Offset, PartitionId};
use fetch_session::FetchSessionCache;
#[derive(Default)]
struct PartitionState {
high_watermark: Option<Offset>,
watermark_updated_at: Option<Instant>,
last_stable_offset: Option<Offset>,
log_start_offset: Option<Offset>,
preferred_replica: Option<(crate::BrokerId, Instant)>,
offset_retry_backoff: Option<(Instant, Duration)>,
last_fetched_epoch: Option<i32>,
position_validated: bool,
}
impl PartitionState {
#[inline]
fn readable_end_offset(&self, isolation_level: IsolationLevel) -> Option<Offset> {
match isolation_level {
IsolationLevel::ReadCommitted => self.last_stable_offset.or(self.high_watermark),
IsolationLevel::ReadUncommitted => self.high_watermark,
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct FetchMetadataResult {
pub brokers: Vec<BrokerInfo>,
pub topics: Vec<TopicInfo>,
}
#[non_exhaustive]
#[derive(Debug)]
pub enum BatchRecvOutcome {
Records(Vec<ConsumerRecord>),
TimedOut,
Closed,
EmptyRequest,
}
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct LagResult {
pub lag: HashMap<(String, PartitionId), u64>,
pub stale_partitions: Vec<(String, PartitionId)>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct CommitPosition {
pub offset: Offset,
pub leader_epoch: i32,
pub metadata: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct CommittedPosition {
pub offset: Offset,
pub leader_epoch: i32,
}
type CommitRequestOffsets = HashMap<(String, PartitionId), CommitPosition>;
#[must_use = "await the returned handle to observe async offset commit outcome"]
#[non_exhaustive]
pub enum OffsetCommitHandle {
Ready(Ready<Result<()>>),
Task(tokio::task::JoinHandle<Result<()>>),
}
impl OffsetCommitHandle {
fn ready(result: Result<()>) -> Self {
Self::Ready(ready(result))
}
}
impl Future for OffsetCommitHandle {
type Output = Result<()>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.get_mut() {
Self::Ready(fut) => Pin::new(fut).poll(cx),
Self::Task(handle) => match Pin::new(handle).poll(cx) {
Poll::Ready(Ok(result)) => Poll::Ready(result),
Poll::Ready(Err(error)) => Poll::Ready(Err(KrafkaError::invalid_state(format!(
"consumer commit task failed: {error}"
)))),
Poll::Pending => Poll::Pending,
},
}
}
}
pub struct Consumer {
config: ConsumerConfig,
metadata: Arc<ClusterMetadata>,
pool: Arc<ConnectionPool>,
pool_owned: bool,
subscriptions: LeveledRwLock<1, HashSet<String>>,
assignments: LeveledRwLock<2, HashMap<String, Vec<PartitionId>>>,
offsets: LeveledRwLock<3, HashMap<(String, PartitionId), Offset>>,
paused: LeveledRwLock<4, HashSet<(String, PartitionId)>>,
closed: std::sync::atomic::AtomicBool,
wakeup_flag: std::sync::atomic::AtomicBool,
wakeup_notify: tokio::sync::Notify,
group_coordinator: Option<Arc<GroupCoordinator>>,
metrics: Arc<ConsumerMetrics>,
rebalance_listener: Arc<dyn ErasedRebalanceListener>,
interceptor: Arc<dyn crate::interceptor::ConsumerInterceptor>,
last_auto_commit: SyncMutex<Instant>,
recv_buffer: SyncMutex<std::collections::VecDeque<ConsumerRecord>>,
fetch_rotation: std::sync::atomic::AtomicUsize,
fetch_sessions: SyncMutex<FetchSessionCache>,
partition_state: LeveledRwLock<5, HashMap<(String, PartitionId), PartitionState>>,
key_deserializer: Option<Arc<dyn crate::serdes::Deserializer>>,
value_deserializer: Option<Arc<dyn crate::serdes::Deserializer>>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct FetchOffsetUpdate {
requested: Offset,
next: Offset,
epoch: i32,
}
#[derive(Debug)]
struct PartitionFetchFault {
key: (String, PartitionId),
offset: Offset,
error: KrafkaError,
}
impl PartitionFetchFault {
fn into_error(self, total_faults: usize) -> KrafkaError {
let (topic, partition) = self.key;
let others = match total_faults {
0 | 1 => String::new(),
n => format!(" ({} partitions affected in this poll)", n),
};
KrafkaError::protocol_kind(
self.error
.protocol_error_kind()
.unwrap_or(ProtocolErrorKind::Malformed),
format!(
"undecodable record batch at {topic}-{partition} offset {}{others}; \
the partition cannot advance past it and re-fetching returns the same \
bytes. Either seek({topic}-{partition}) past the offset to skip the \
corrupt data, or pause({topic}-{partition}) to keep consuming every \
other partition while you investigate. Underlying error: {}",
self.offset, self.error
),
)
}
}
#[derive(Debug, Default)]
struct FetchOutcome {
records: Vec<ConsumerRecord>,
offset_updates: Vec<((String, PartitionId), FetchOffsetUpdate)>,
hw_updates: Vec<((String, PartitionId), Offset)>,
faults: Vec<PartitionFetchFault>,
}
fn apply_fetch_offset_updates(
offsets: &mut HashMap<(String, PartitionId), Offset>,
updates: Vec<((String, PartitionId), FetchOffsetUpdate)>,
) -> Vec<(String, PartitionId)> {
let mut discarded = Vec::new();
for (key, update) in updates {
match offsets.get(&key) {
Some(¤t) if current == update.requested => {
offsets.insert(key, update.next);
}
_ => discarded.push(key),
}
}
discarded
}
fn lowest_undelivered_offsets(
buffered: &std::collections::VecDeque<ConsumerRecord>,
) -> HashMap<(String, PartitionId), Offset> {
let mut lowest: HashMap<(&str, PartitionId), Offset> = HashMap::new();
for record in buffered {
lowest
.entry((record.topic.as_str(), record.partition))
.and_modify(|o| {
if record.offset < *o {
*o = record.offset;
}
})
.or_insert(record.offset);
}
lowest
.into_iter()
.map(|((topic, partition), offset)| ((topic.to_string(), partition), offset))
.collect()
}
fn committable_positions(
positions: &HashMap<(String, PartitionId), Offset>,
buffered: &std::collections::VecDeque<ConsumerRecord>,
) -> HashMap<(String, PartitionId), Offset> {
let lowest_undelivered = lowest_undelivered_offsets(buffered);
positions
.iter()
.map(|(key, &position)| {
let committable = match lowest_undelivered.get(key) {
Some(&first_undelivered) => position.min(first_undelivered),
None => position,
};
(key.clone(), committable)
})
.collect()
}
fn compute_aggregate_lag(
offsets: &HashMap<(String, PartitionId), Offset>,
partition_state: &HashMap<(String, PartitionId), PartitionState>,
undelivered: &HashMap<(String, PartitionId), Offset>,
isolation_level: IsolationLevel,
) -> (u64, u64) {
let mut total_lag: u64 = 0;
let mut max_lag: u64 = 0;
for (key, state) in partition_state {
if let (Some(end), Some(&fetch_position)) =
(state.readable_end_offset(isolation_level), offsets.get(key))
{
let position = match undelivered.get(key) {
Some(&first_undelivered) => fetch_position.min(first_undelivered),
None => fetch_position,
};
let partition_lag = (end - position).max(0) as u64;
total_lag = total_lag.saturating_add(partition_lag);
max_lag = max_lag.max(partition_lag);
}
}
(total_lag, max_lag)
}
fn seed_initial_offsets_for_assigned(
assigned: &HashMap<String, Vec<PartitionId>>,
initial_offsets: &HashMap<(String, PartitionId), Offset>,
stored_offsets: &mut HashMap<(String, PartitionId), Offset>,
) -> usize {
let mut inserted = 0;
for ((topic, partition), &initial) in initial_offsets {
if !assigned
.get(topic)
.is_some_and(|partitions| partitions.contains(partition))
{
continue;
}
let key = (topic.clone(), *partition);
if let std::collections::hash_map::Entry::Vacant(e) = stored_offsets.entry(key) {
e.insert(initial);
inserted += 1;
}
}
inserted
}
fn control_batch_is_abort(batch: &RecordBatch) -> bool {
const CONTROL_TYPE_ABORT: i16 = 0;
batch.records.first().is_some_and(|record| {
record.key.as_ref().is_some_and(|key| {
key.len() >= 4 && i16::from_be_bytes([key[2], key[3]]) == CONTROL_TYPE_ABORT
})
})
}
fn claim_record_budget(budget: &std::sync::atomic::AtomicUsize, want: usize) -> usize {
use std::sync::atomic::Ordering;
let mut granted = 0;
let _ = budget.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |remaining| {
granted = remaining.min(want);
if granted == 0 {
None
} else {
Some(remaining - granted)
}
});
granted
}
struct PartitionDecodeOutcome {
last_offset: Option<Offset>,
last_epoch: i32,
error: Option<KrafkaError>,
corrupt: bool,
}
fn advance_through_batch(
last_offset: &mut Option<Offset>,
last_epoch: &mut i32,
fetch_offset: Offset,
batch_end: Offset,
batch_epoch: i32,
) {
if batch_end < fetch_offset {
return;
}
if last_offset.is_none_or(|advanced| batch_end > advanced) {
*last_offset = Some(batch_end);
*last_epoch = batch_epoch;
}
}
#[allow(clippy::too_many_arguments)]
fn decode_partition_batches(
topic_name: &str,
partition: PartitionId,
mut batch_buf: bytes::Bytes,
partition_fetch_offset: Offset,
mut aborted_txns: Vec<crate::protocol::AbortedTransaction>,
record_budget: Option<&std::sync::atomic::AtomicUsize>,
max_decompressed_size: usize,
records: &mut Vec<ConsumerRecord>,
) -> PartitionDecodeOutcome {
let mut outcome = PartitionDecodeOutcome {
last_offset: None,
last_epoch: -1,
error: None,
corrupt: false,
};
aborted_txns.sort_unstable_by_key(|at| at.first_offset);
let mut aborted_txns_iter = aborted_txns.iter().peekable();
let mut aborted_producers: HashSet<i64> = HashSet::new();
let mut budget_exhausted = false;
while batch_buf.len() >= 12 {
if record_budget.is_some_and(|b| b.load(std::sync::atomic::Ordering::Relaxed) == 0) {
break;
}
match RecordBatch::decode_with_limit(&mut batch_buf, max_decompressed_size) {
Ok(batch) => {
let batch_epoch = batch.partition_leader_epoch;
let batch_end = batch
.base_offset
.saturating_add(batch.last_offset_delta as i64);
while aborted_txns_iter
.peek()
.is_some_and(|at| at.first_offset <= batch.base_offset)
{
if let Some(at) = aborted_txns_iter.next() {
aborted_producers.insert(at.producer_id);
}
}
if batch.attributes.is_control_batch {
if control_batch_is_abort(&batch) {
aborted_producers.remove(&batch.producer_id);
}
advance_through_batch(
&mut outcome.last_offset,
&mut outcome.last_epoch,
partition_fetch_offset,
batch_end,
batch_epoch,
);
continue;
}
if batch.attributes.is_transactional
&& aborted_producers.contains(&batch.producer_id)
{
advance_through_batch(
&mut outcome.last_offset,
&mut outcome.last_epoch,
partition_fetch_offset,
batch_end,
batch_epoch,
);
continue;
}
if batch.records.is_empty() {
advance_through_batch(
&mut outcome.last_offset,
&mut outcome.last_epoch,
partition_fetch_offset,
batch_end,
batch_epoch,
);
continue;
}
let claimed = match record_budget {
Some(budget) => claim_record_budget(budget, batch.records.len()),
None => batch.records.len(),
};
if claimed == 0 {
break;
}
let mut used = 0usize;
for record in batch.records.into_iter() {
let record_offset =
batch.base_offset.saturating_add(record.offset_delta as i64);
if record_offset < partition_fetch_offset {
continue;
}
if used == claimed {
budget_exhausted = true;
break;
}
used += 1;
records.push(ConsumerRecord {
topic: topic_name.to_string(),
partition,
offset: record_offset,
timestamp: batch.base_timestamp.saturating_add(record.timestamp_delta),
timestamp_type: batch.attributes.timestamp_type as i8,
key: record.key,
value: record.value,
headers: record
.headers
.into_iter()
.map(|h| (h.key, h.value))
.collect(),
leader_epoch: Some(batch_epoch),
delivery_count: None,
});
outcome.last_offset = Some(record_offset);
outcome.last_epoch = batch_epoch;
}
if let Some(budget) = record_budget
&& claimed > used
{
budget.fetch_add(claimed - used, std::sync::atomic::Ordering::Relaxed);
}
if budget_exhausted {
break;
}
advance_through_batch(
&mut outcome.last_offset,
&mut outcome.last_epoch,
partition_fetch_offset,
batch_end,
batch_epoch,
);
}
Err(e) => {
let made_progress = outcome.last_offset.is_some();
let truncated_tail =
e.protocol_error_kind() == Some(ProtocolErrorKind::TruncatedFrame);
if made_progress && truncated_tail {
trace!(
topic = %topic_name,
partition,
"Trailing record batch truncated by the fetch size limit"
);
break;
}
outcome.corrupt = true;
if made_progress {
warn!(
topic = %topic_name,
partition,
fetch_offset = partition_fetch_offset,
error = %e,
"Corrupt record batch after a decodable prefix; \
delivering the prefix and stopping here"
);
break;
}
warn!(
topic = %topic_name,
partition,
fetch_offset = partition_fetch_offset,
error = %e,
"Record batch at the fetch position could not be decoded; \
the partition cannot advance"
);
outcome.error = Some(e);
break;
}
}
}
outcome
}
fn purge_buffered_records(
buffer: &mut std::collections::VecDeque<ConsumerRecord>,
repositioned: &HashSet<(String, PartitionId)>,
) -> usize {
if repositioned.is_empty() || buffer.is_empty() {
return 0;
}
let before = buffer.len();
buffer.retain(|r| !contains_partition(repositioned, &r.topic, r.partition));
before - buffer.len()
}
#[inline]
fn contains_partition(
set: &HashSet<(String, PartitionId)>,
topic: &str,
partition: PartitionId,
) -> bool {
!set.is_empty()
&& set
.iter()
.any(|(t, p)| *p == partition && t.as_str() == topic)
}
fn invalidate_position_epoch(
partition_state: &mut HashMap<(String, PartitionId), PartitionState>,
key: &(String, PartitionId),
) {
let entry = partition_state.entry(key.clone()).or_default();
entry.last_fetched_epoch = None;
entry.position_validated = false;
}
fn apply_seek_many_offsets(
stored_offsets: &mut HashMap<(String, PartitionId), Offset>,
offsets: &HashMap<(String, PartitionId), Offset>,
) -> usize {
for ((topic, partition), offset) in offsets {
stored_offsets.insert((topic.clone(), *partition), *offset);
}
offsets.len()
}
fn apply_assignment_offset_precedence(
assigned: &HashMap<String, Vec<PartitionId>>,
committed: &HashMap<(String, PartitionId), Offset>,
initial_offsets: &HashMap<(String, PartitionId), Offset>,
stored_offsets: &mut HashMap<(String, PartitionId), Offset>,
) -> Vec<(String, PartitionId)> {
let mut need_reset: Vec<(String, PartitionId)> = Vec::new();
for (topic, partitions) in assigned {
for &partition in partitions {
let key = (topic.clone(), partition);
if stored_offsets.contains_key(&key) {
continue;
}
if let Some(&offset) = committed.get(&key)
&& offset >= 0
{
stored_offsets.insert(key, offset);
continue;
}
if let Some(&initial) = initial_offsets.get(&key) {
stored_offsets.insert(key, initial);
continue;
}
need_reset.push(key);
}
}
need_reset
}
fn drain_buffered_records(
buffer: &mut std::collections::VecDeque<ConsumerRecord>,
batch: &mut Vec<ConsumerRecord>,
max_records: usize,
paused: &HashSet<(String, PartitionId)>,
) {
if paused.is_empty() {
while batch.len() < max_records {
match buffer.pop_front() {
Some(r) => batch.push(r),
None => break,
}
}
return;
}
let mut index = 0;
while batch.len() < max_records && index < buffer.len() {
let is_paused = {
let record = &buffer[index];
contains_partition(paused, &record.topic, record.partition)
};
if is_paused {
index += 1;
continue;
}
if let Some(record) = buffer.remove(index) {
batch.push(record);
}
}
}
#[allow(clippy::too_many_arguments)]
async fn batch_recv_with<FClosed, FPoll, FPollFut, FSetBuffered, FPaused, FPausedFut>(
recv_buffer: &SyncMutex<std::collections::VecDeque<ConsumerRecord>>,
mut set_buffered_records: FSetBuffered,
max_records: usize,
timeout: Duration,
max_idle_backoff: Duration,
is_closed: FClosed,
mut paused_snapshot: FPaused,
mut poll: FPoll,
) -> Result<BatchRecvOutcome>
where
FClosed: Fn() -> bool,
FPoll: FnMut(Duration) -> FPollFut,
FPollFut: Future<Output = Result<Vec<ConsumerRecord>>>,
FSetBuffered: FnMut(u64),
FPaused: FnMut() -> FPausedFut,
FPausedFut: Future<Output = HashSet<(String, PartitionId)>>,
{
if max_records == 0 {
return Ok(BatchRecvOutcome::EmptyRequest);
}
let deadline = tokio::time::Instant::now() + timeout;
let mut batch = Vec::with_capacity(max_records);
loop {
let paused = paused_snapshot().await;
{
let mut buffer = recv_buffer.lock();
drain_buffered_records(&mut buffer, &mut batch, max_records, &paused);
set_buffered_records(buffer.len() as u64);
}
if batch.len() >= max_records {
return Ok(BatchRecvOutcome::Records(batch));
}
if is_closed() {
return if batch.is_empty() {
Ok(BatchRecvOutcome::Closed)
} else {
Ok(BatchRecvOutcome::Records(batch))
};
}
let now = tokio::time::Instant::now();
if now >= deadline {
return if batch.is_empty() {
Ok(BatchRecvOutcome::TimedOut)
} else {
Ok(BatchRecvOutcome::Records(batch))
};
}
let remaining = deadline - now;
match tokio::time::timeout(remaining, poll(remaining)).await {
Ok(Ok(records)) => {
let before_len = batch.len();
let mut iter = records.into_iter();
while batch.len() < max_records {
match iter.next() {
Some(r) => batch.push(r),
None => break,
}
}
let leftover: Vec<_> = iter.collect();
if !leftover.is_empty() {
let mut buffer = recv_buffer.lock();
for r in leftover.into_iter().rev() {
buffer.push_front(r);
}
set_buffered_records(buffer.len() as u64);
}
if batch.len() >= max_records {
return Ok(BatchRecvOutcome::Records(batch));
}
if batch.len() == before_len {
let now_after_poll = tokio::time::Instant::now();
if now_after_poll >= deadline {
return if batch.is_empty() {
Ok(BatchRecvOutcome::TimedOut)
} else {
Ok(BatchRecvOutcome::Records(batch))
};
}
let remaining_after_poll = deadline - now_after_poll;
let backoff = remaining_after_poll.min(max_idle_backoff);
tokio::time::sleep(backoff).await;
}
}
Ok(Err(_)) if is_closed() => {
return if batch.is_empty() {
Ok(BatchRecvOutcome::Closed)
} else {
Ok(BatchRecvOutcome::Records(batch))
};
}
Ok(Err(e)) => {
if !batch.is_empty() {
let mut buffer = recv_buffer.lock();
for record in batch.into_iter().rev() {
buffer.push_front(record);
}
set_buffered_records(buffer.len() as u64);
}
return Err(e);
}
Err(_elapsed) => {
return if batch.is_empty() {
Ok(BatchRecvOutcome::TimedOut)
} else {
Ok(BatchRecvOutcome::Records(batch))
};
}
}
}
}
struct FetchRoutingPlan {
partitions_by_broker: HashMap<crate::BrokerId, Vec<(String, PartitionId)>>,
expired_preferred: Vec<(String, PartitionId)>,
skipped: Vec<(String, PartitionId)>,
}
fn build_fetch_routing_plan(
non_paused_keys: Vec<(String, PartitionId)>,
partition_state: &HashMap<(String, PartitionId), PartitionState>,
leaders: &HashMap<(String, PartitionId), crate::BrokerId>,
now: Instant,
) -> FetchRoutingPlan {
let mut partitions_by_broker: HashMap<crate::BrokerId, Vec<(String, PartitionId)>> =
HashMap::new();
let mut expired_preferred: Vec<(String, PartitionId)> = Vec::new();
let mut skipped: Vec<(String, PartitionId)> = Vec::new();
for key in non_paused_keys {
let target_broker = match partition_state.get(&key).and_then(|s| s.preferred_replica) {
Some((replica_id, expiry)) if now < expiry => Some(replica_id),
Some(_) => {
expired_preferred.push(key.clone());
None
}
None => None,
};
let broker_id = match target_broker {
Some(id) => id,
None => match leaders.get(&key).copied() {
Some(leader_id) => leader_id,
None => {
skipped.push(key);
continue;
}
},
};
partitions_by_broker.entry(broker_id).or_default().push(key);
}
FetchRoutingPlan {
partitions_by_broker,
expired_preferred,
skipped,
}
}
fn group_topic_partitions(partitions: &[(&str, PartitionId)]) -> HashMap<String, Vec<PartitionId>> {
let mut grouped: HashMap<String, Vec<PartitionId>> = HashMap::new();
for &(topic, partition) in partitions {
if let Some(v) = grouped.get_mut(topic) {
v.push(partition);
} else {
grouped.insert(topic.to_string(), vec![partition]);
}
}
grouped
}
fn apply_list_offsets_response(
response: &ListOffsetsResponse,
result: &mut HashMap<(String, PartitionId), Result<Offset>>,
) -> Vec<String> {
let mut stale_epoch_topics: Vec<String> = Vec::new();
for topic_resp in &response.topics {
for part_resp in &topic_resp.partitions {
let key = (topic_resp.name.clone(), part_resp.partition_index);
if part_resp.error_code.is_ok() {
result.insert(key, Ok(part_resp.offset));
} else {
warn!(
"ListOffsets error for {}-{}: {:?}",
topic_resp.name, part_resp.partition_index, part_resp.error_code
);
if matches!(
part_resp.error_code,
crate::error::ErrorCode::FencedLeaderEpoch
| crate::error::ErrorCode::UnknownLeaderEpoch
) && !stale_epoch_topics.contains(&topic_resp.name)
{
stale_epoch_topics.push(topic_resp.name.clone());
}
result.insert(
key,
Err(KrafkaError::broker(
part_resp.error_code,
format!(
"ListOffsets error for {}-{}",
topic_resp.name, part_resp.partition_index
),
)),
);
}
}
}
stale_epoch_topics
}
fn revoked_partitions_diff(
old: &HashMap<String, Vec<PartitionId>>,
new: &HashMap<String, Vec<PartitionId>>,
) -> Vec<TopicPartition> {
let new_sets: HashMap<&String, HashSet<PartitionId>> = new
.iter()
.map(|(topic, partitions)| (topic, partitions.iter().copied().collect()))
.collect();
let mut result = Vec::new();
for (topic, partitions) in old {
let new_set = new_sets.get(topic);
for &partition in partitions {
let gone = new_set.is_none_or(|assigned| !assigned.contains(&partition));
if gone {
result.push(TopicPartition::new(topic, partition));
}
}
}
result
}
impl Consumer {
pub fn builder() -> ConsumerBuilder {
ConsumerBuilder::default()
}
async fn new(
config: ConsumerConfig,
shared: Option<(Arc<ConnectionPool>, Arc<ClusterMetadata>)>,
) -> Result<Self> {
let pool_owned = shared.is_none();
let (pool, metadata) = if let Some((pool, metadata)) = shared {
(pool, metadata)
} else {
let mut pool_config_builder = config.transport.apply(
ConnectionConfig::builder()
.client_id(&config.client_id)
.request_timeout(config.request_timeout)
.connect_timeout(config.connect_timeout),
);
if let Some(ref auth) = config.auth {
pool_config_builder = pool_config_builder.auth(auth.clone());
}
let mut pool_config = pool_config_builder.build()?;
pool_config.init_tls().await?;
let pool = config.transport.build_pool(pool_config);
let bootstrap_servers =
crate::util::parse_bootstrap_servers(&config.bootstrap_servers)?;
let metadata = Arc::new({
let mut meta =
ClusterMetadata::new(bootstrap_servers, pool.clone(), config.metadata_max_age)
.with_recovery_strategy(config.metadata_recovery_strategy)
.with_rebootstrap_trigger(config.metadata_recovery_rebootstrap_trigger);
if let Some(ttl) = config.metadata_topic_cache_ttl {
meta = meta.with_topic_cache_ttl(ttl);
} else {
meta = meta.with_topic_cache_ttl_disabled();
}
meta
});
metadata.refresh().await?;
(pool, metadata)
};
let group_coordinator = if let Some(ref group_id) = config.group_id {
Some(Arc::new(
GroupCoordinator::new(
group_id.clone(),
pool.clone(),
metadata.clone(),
config.session_timeout,
config.heartbeat_interval,
config.max_poll_interval, )
.with_assignor_strategies(config.partition_assignment_strategies.clone())
.with_group_instance_id(config.group_instance_id.clone())
.with_client_rack(config.client_rack.clone())
.with_isolation_level(config.isolation_level.to_i8())
.with_group_protocol(config.group_protocol),
))
} else {
None
};
let metrics = Arc::new(ConsumerMetrics::default());
info!(
"Consumer initialized with {} brokers{}",
metadata.brokers().len(),
if let Some(ref gid) = config.group_id {
format!(", group_id='{gid}'")
} else {
String::new()
}
);
Ok(Self {
config,
metadata,
pool,
pool_owned,
subscriptions: LeveledRwLock::new(HashSet::new()),
assignments: LeveledRwLock::new(HashMap::new()),
offsets: LeveledRwLock::new(HashMap::new()),
paused: LeveledRwLock::new(HashSet::new()),
closed: std::sync::atomic::AtomicBool::new(false),
wakeup_flag: std::sync::atomic::AtomicBool::new(false),
wakeup_notify: tokio::sync::Notify::new(),
group_coordinator,
metrics,
rebalance_listener: Arc::new(NoOpRebalanceListener),
interceptor: Arc::new(crate::interceptor::NoOpConsumerInterceptor),
last_auto_commit: SyncMutex::new(Instant::now()),
recv_buffer: SyncMutex::new(std::collections::VecDeque::new()),
fetch_rotation: std::sync::atomic::AtomicUsize::new(0),
fetch_sessions: SyncMutex::new(FetchSessionCache::new()),
partition_state: LeveledRwLock::new(HashMap::new()),
key_deserializer: None,
value_deserializer: None,
})
}
pub async fn subscribe(&self, topics: &[&str]) -> Result<()> {
validate_topic_names(topics.iter().copied())?;
if let Some(ref coordinator) = self.group_coordinator {
coordinator.reset_poll_tracking();
}
{
let mut subscriptions = self.subscriptions.write().await;
subscriptions.clear();
for topic in topics {
subscriptions.insert((*topic).to_string());
}
}
self.metadata.refresh_for_topics(Some(topics)).await?;
if let Some(ref coordinator) = self.group_coordinator {
let mut topics_sorted: Vec<String> = topics.iter().map(|s| s.to_string()).collect();
topics_sorted.sort();
if coordinator.is_consumer_protocol() {
{
let state = coordinator.state().await;
if state == GroupState::Stable {
let mut old_sorted = coordinator.subscribed_topics().await;
old_sorted.sort();
if old_sorted != topics_sorted {
coordinator.trigger_rejoin().await;
}
}
}
coordinator.set_subscribed_topics(topics_sorted).await;
} else if coordinator.is_cooperative() {
{
let state = coordinator.state().await;
if state == GroupState::Stable {
let mut old_sorted = coordinator.subscribed_topics().await;
old_sorted.sort();
if old_sorted != topics_sorted {
coordinator.set_preparing_rebalance().await;
}
}
}
coordinator.set_subscribed_topics(topics_sorted).await;
} else {
let old_assignments = self.assignments.read().await.clone();
let (assignment, joined) =
coordinator.ensure_active_membership(&topics_sorted).await?;
if joined {
if !old_assignments.is_empty() {
let revoked: Vec<TopicPartition> = old_assignments
.iter()
.flat_map(|(t, ps)| ps.iter().map(move |&p| TopicPartition::new(t, p)))
.collect();
self.safe_on_partitions_revoked(&revoked).await;
let revoked_tuples: Vec<(String, PartitionId)> =
revoked_partitions_diff(&old_assignments, &assignment.partitions)
.into_iter()
.map(|tp| (tp.topic, tp.partition))
.collect();
self.apply_partition_revocations(&revoked_tuples).await;
}
self.metrics.rebalances.inc();
}
{
let mut assignments = self.assignments.write().await;
assignments.clear();
for (topic, partitions) in &assignment.partitions {
assignments.insert(topic.clone(), partitions.clone());
}
}
if joined {
let assigned: Vec<TopicPartition> = assignment
.partitions
.iter()
.flat_map(|(t, ps)| ps.iter().map(move |&p| TopicPartition::new(t, p)))
.collect();
self.safe_on_partitions_assigned(&assigned).await;
self.metrics.assigned_partitions.set(assigned.len() as u64);
self.fetch_and_apply_committed_offsets(&assignment.partitions)
.await?;
}
}
debug!("Subscribed to topics via group coordinator: {:?}", topics);
} else {
let mut assignments = self.assignments.write().await;
for topic in topics {
if let Some(topic_info) = self.metadata.topic(topic) {
let partitions: Vec<_> = topic_info
.partitions
.values()
.map(|p| p.partition)
.collect();
assignments.insert((*topic).to_string(), partitions);
}
}
let assigned_snapshot = assignments.clone();
drop(assignments);
let count: usize = assigned_snapshot.values().map(|p| p.len()).sum();
self.metrics.assigned_partitions.set(count as u64);
self.apply_auto_offset_reset(&assigned_snapshot).await?;
debug!("Subscribed to topics: {:?}", topics);
}
Ok(())
}
async fn apply_partition_revocations(&self, revoked: &[(String, PartitionId)]) {
let revoked_by_topic: HashMap<&str, HashSet<PartitionId>> = {
let mut m: HashMap<&str, HashSet<PartitionId>> = HashMap::new();
for (topic, partition) in revoked {
m.entry(topic.as_str()).or_default().insert(*partition);
}
m
};
let revoked_keys: Vec<(String, PartitionId)> =
revoked.iter().map(|(t, p)| (t.clone(), *p)).collect();
{
let mut assignments = self.assignments.write().await;
for (topic, revoked_parts) in &revoked_by_topic {
if let Some(parts) = assignments.get_mut(*topic) {
parts.retain(|p| !revoked_parts.contains(p));
if parts.is_empty() {
assignments.remove(*topic);
}
}
}
}
{
let mut offsets = self.offsets.write().await;
for key in &revoked_keys {
offsets.remove(key);
}
}
{
let revoked_set: HashSet<(&str, PartitionId)> =
revoked_keys.iter().map(|(t, p)| (t.as_str(), *p)).collect();
let mut buf = self.recv_buffer.lock();
buf.retain(|r| !revoked_set.contains(&(r.topic.as_str(), r.partition)));
self.metrics.buffered_records.set(buf.len() as u64);
}
{
let mut paused = self.paused.write().await;
for key in &revoked_keys {
paused.remove(key);
}
self.metrics.paused_partitions.set(paused.len() as u64);
}
{
let mut partition_state = self.partition_state.write().await;
for key in &revoked_keys {
partition_state.remove(key);
}
}
{
let live_broker_ids: Vec<crate::BrokerId> =
self.metadata.brokers().iter().map(|b| b.id()).collect();
self.fetch_sessions.lock().retain_brokers(&live_broker_ids);
}
self.recompute_lag_metrics().await;
}
async fn finalize_cooperative_assignment(
&self,
coordinator: &GroupCoordinator,
assignment: &MemberAssignment,
old_assignments: &HashMap<String, Vec<PartitionId>>,
) -> Result<()> {
let old_sets: HashMap<&String, HashSet<PartitionId>> = old_assignments
.iter()
.map(|(t, ps)| (t, ps.iter().copied().collect()))
.collect();
let mut newly_assigned = Vec::new();
for (topic, partitions) in &assignment.partitions {
let old_set = old_sets.get(topic);
for &p in partitions {
let is_new = old_set.is_none_or(|os| !os.contains(&p));
if is_new {
newly_assigned.push(TopicPartition::new(topic, p));
}
}
}
{
let mut assignments = self.assignments.write().await;
assignments.clear();
for (topic, partitions) in &assignment.partitions {
assignments.insert(topic.clone(), partitions.clone());
}
}
self.safe_on_partitions_assigned(&newly_assigned).await;
let total_assigned: usize = assignment.partitions.values().map(|ps| ps.len()).sum();
self.metrics.assigned_partitions.set(total_assigned as u64);
if !newly_assigned.is_empty() {
let new_parts = Self::group_partitions_by_topic(&newly_assigned);
self.fetch_and_apply_committed_offsets(&new_parts).await?;
}
let member_id = coordinator.member_id().await;
coordinator.record_owned_partitions(&member_id, assignment);
Ok(())
}
async fn clear_partition_state(&self) {
self.close_fetch_sessions().await;
self.offsets.write().await.clear();
self.recv_buffer.lock().clear();
self.paused.write().await.clear();
self.partition_state.write().await.clear();
self.metrics.buffered_records.set(0);
self.metrics.paused_partitions.set(0);
self.metrics.lag.set(0);
self.metrics.lag_max.set(0);
}
async fn close_fetch_sessions(&self) {
let closes = self.fetch_sessions.lock().close_all();
for close in closes {
let Some(broker) = self.metadata.broker(close.broker_id) else {
continue;
};
let Ok(conn) = self
.pool
.get_connection_by_id(close.broker_id, broker.address())
.await
else {
continue;
};
let Some(version) = conn.negotiate_api_version(ApiKey::Fetch, versions::FETCH_MAX, 7)
else {
continue;
};
let request = FetchRequest {
replica_id: -1,
max_wait_ms: 0,
min_bytes: 0,
max_bytes: 0,
isolation_level: self.config.isolation_level.to_i8(),
session_id: close.session_id,
session_epoch: close.session_epoch,
topics: Vec::new(),
forgotten_topics: Vec::new(),
rack_id: self.config.client_rack.clone().unwrap_or_default(),
};
if let Err(e) = conn
.send_request(ApiKey::Fetch, version, |buf| {
request.encode_versioned(version, buf)
})
.await
{
debug!(
broker_id = close.broker_id,
session_id = close.session_id,
"Failed to close fetch session: {e}"
);
}
}
}
async fn safe_on_partitions_assigned(&self, partitions: &[TopicPartition]) {
self.rebalance_listener
.on_partitions_assigned_erased(partitions)
.await;
}
async fn safe_on_partitions_revoked(&self, partitions: &[TopicPartition]) {
let timeout = self.config.revocation_timeout();
if tokio::time::timeout(
timeout,
self.rebalance_listener
.on_partitions_revoked_erased(partitions),
)
.await
.is_err()
{
warn!(
timeout_secs = timeout.as_secs_f64(),
"on_partitions_revoked timed out; proceeding with revocation. \
A hung rebalance listener can cause group coordinator session expiry."
);
}
}
async fn safe_on_partitions_lost(&self, partitions: &[TopicPartition]) {
self.rebalance_listener
.on_partitions_lost_erased(partitions)
.await;
}
async fn recompute_lag_metrics(&self) {
let offsets = self.offsets.read().await;
let partition_state = self.partition_state.read().await;
let undelivered = lowest_undelivered_offsets(&self.recv_buffer.lock());
let (total_lag, max_lag) = compute_aggregate_lag(
&offsets,
&partition_state,
&undelivered,
self.config.isolation_level,
);
self.metrics.lag.set(total_lag);
self.metrics.lag_max.set(max_lag);
}
async fn apply_revocation_round(
&self,
coordinator: &Arc<GroupCoordinator>,
revoked_tuples: &[(String, PartitionId)],
revoked_tps: &[TopicPartition],
) -> Result<bool> {
match coordinator.send_heartbeat().await {
Ok(status) if coordinator.handle_inline_heartbeat_status(status).await => {
return Ok(true);
}
Err(e) => {
warn!("Pre-revocation heartbeat failed: {}", e);
}
_ => {}
}
if self.config.enable_auto_commit
&& let Err(e) = self.commit().await
{
if e.is_retriable() {
warn!(
"Auto-commit before cooperative revocation failed (retriable): {}",
e
);
} else {
error!(
"Auto-commit before cooperative revocation failed (fatal): {}",
e
);
}
}
self.safe_on_partitions_revoked(revoked_tps).await;
self.apply_partition_revocations(revoked_tuples).await;
let member_id = coordinator.member_id().await;
let current = self.assignments.read().await;
let count: usize = current.values().map(|ps| ps.len()).sum();
self.metrics.assigned_partitions.set(count as u64);
let owned = MemberAssignment {
partitions: current.clone(),
};
drop(current);
coordinator.record_owned_partitions(&member_id, &owned);
Ok(false)
}
async fn handle_group_rebalance(&self, timeout: Duration) -> Result<bool> {
let Some(ref coordinator) = self.group_coordinator else {
return Ok(false);
};
coordinator.await_rejoin(timeout).await;
if let Some(pending) = coordinator.take_pending_rebalance() {
if coordinator.is_cooperative() {
if self
.handle_cooperative_rebalance(coordinator, Some(pending))
.await?
{
return Ok(true);
}
} else {
let topics: Vec<String> = self.subscriptions.read().await.iter().cloned().collect();
self.handle_eager_rebalance(coordinator, &topics, Some(pending.assignment))
.await?;
}
} else if coordinator.rejoin_in_flight() {
debug!("Background rebalance still in flight; returning an empty poll");
return Ok(true);
} else if coordinator.needs_rejoin().await {
let topics: Vec<String> = self.subscriptions.read().await.iter().cloned().collect();
if !topics.is_empty() {
coordinator.set_subscribed_topics(topics.clone()).await;
if coordinator.is_consumer_protocol() {
coordinator.ensure_active_membership(&topics).await?;
self.handle_kip848_rebalance(coordinator).await?;
} else if coordinator.is_cooperative() {
if self.handle_cooperative_rebalance(coordinator, None).await? {
return Ok(true);
}
} else {
self.handle_eager_rebalance(coordinator, &topics, None)
.await?;
}
}
}
if !coordinator.is_consumer_protocol() && coordinator.is_heartbeat_overdue().await {
match coordinator.send_heartbeat().await {
Ok(status) if coordinator.handle_inline_heartbeat_status(status).await => {
debug!("Heartbeat indicated rejoin needed");
}
Err(e) => {
warn!("Inline heartbeat failed: {}", e);
}
_ => {}
}
}
Ok(false)
}
async fn handle_cooperative_rebalance(
&self,
coordinator: &Arc<GroupCoordinator>,
phase1: Option<PendingRebalance>,
) -> Result<bool> {
let (new_assignment, to_revoke) = match phase1 {
Some(p) => (p.assignment, p.to_revoke),
None => coordinator.perform_cooperative_join_and_sync().await?,
};
if !to_revoke.is_empty() {
let revoked: Vec<TopicPartition> = to_revoke
.iter()
.map(|(t, p)| TopicPartition::new(t, *p))
.collect();
if self
.apply_revocation_round(coordinator, &to_revoke, &revoked)
.await?
{
return Ok(true);
}
self.metrics.rebalances.inc();
coordinator.trigger_rejoin().await;
let mut final_assignment = MemberAssignment::empty();
for round in 0..self.config.max_cooperative_rebalance_rounds {
let (assignment, extra_revoke) =
coordinator.perform_cooperative_join_and_sync().await?;
final_assignment = assignment;
if extra_revoke.is_empty() {
break;
}
let extra_revoked: Vec<TopicPartition> = extra_revoke
.iter()
.map(|(t, p)| TopicPartition::new(t, *p))
.collect();
if self
.apply_revocation_round(coordinator, &extra_revoke, &extra_revoked)
.await?
{
return Ok(true);
}
if round == self.config.max_cooperative_rebalance_rounds - 1 {
warn!(
"Cooperative rebalance exceeded {} rounds with pending revocations; \
this may indicate cascading membership changes. \
Deferring assignment to next poll cycle.",
self.config.max_cooperative_rebalance_rounds
);
coordinator.start_heartbeat_task().await;
coordinator.set_preparing_rebalance().await;
return Ok(true);
}
coordinator.trigger_rejoin().await;
}
let old_assignments = self.assignments.read().await.clone();
self.finalize_cooperative_assignment(coordinator, &final_assignment, &old_assignments)
.await?;
} else {
let old_assignments = self.assignments.read().await.clone();
let new_sets: HashMap<&String, HashSet<PartitionId>> = new_assignment
.partitions
.iter()
.map(|(t, ps)| (t, ps.iter().copied().collect()))
.collect();
let mut revoked_parts: Vec<TopicPartition> = Vec::new();
for (topic, partitions) in &old_assignments {
let new_set = new_sets.get(topic);
for &p in partitions {
let gone = new_set.is_none_or(|ns| !ns.contains(&p));
if gone {
revoked_parts.push(TopicPartition::new(topic, p));
}
}
}
if !revoked_parts.is_empty() {
if self.config.enable_auto_commit
&& self.group_coordinator.is_some()
&& let Err(e) = self.commit().await
{
warn!("Commit before cooperative revocation failed: {e}");
}
self.safe_on_partitions_revoked(&revoked_parts).await;
let revoked_tuples: Vec<(String, PartitionId)> = revoked_parts
.iter()
.map(|tp| (tp.topic.clone(), tp.partition))
.collect();
self.apply_partition_revocations(&revoked_tuples).await;
}
self.metrics.rebalances.inc();
self.finalize_cooperative_assignment(coordinator, &new_assignment, &old_assignments)
.await?;
}
Ok(false)
}
async fn handle_kip848_rebalance(&self, coordinator: &Arc<GroupCoordinator>) -> Result<()> {
let new_assignment = coordinator.assignment().await;
let old_assignments = self.assignments.read().await.clone();
let old_sets: HashMap<&String, HashSet<PartitionId>> = old_assignments
.iter()
.map(|(t, ps)| (t, ps.iter().copied().collect()))
.collect();
let new_sets: HashMap<&String, HashSet<PartitionId>> = new_assignment
.partitions
.iter()
.map(|(t, ps)| (t, ps.iter().copied().collect()))
.collect();
let mut revoked: Vec<TopicPartition> = Vec::new();
for (topic, old_set) in &old_sets {
let new_set = new_sets.get(*topic);
for &p in old_set {
let retained = new_set.is_some_and(|ns| ns.contains(&p));
if !retained {
revoked.push(TopicPartition::new(*topic, p));
}
}
}
let mut assigned: Vec<TopicPartition> = Vec::new();
for (topic, new_set) in &new_sets {
let old_set = old_sets.get(*topic);
for &p in new_set {
let was_assigned = old_set.is_some_and(|os| os.contains(&p));
if !was_assigned {
assigned.push(TopicPartition::new(*topic, p));
}
}
}
if revoked.is_empty() && assigned.is_empty() {
if old_assignments.is_empty() && !new_assignment.partitions.is_empty() {
for (topic, parts) in &new_assignment.partitions {
for &p in parts {
assigned.push(TopicPartition::new(topic, p));
}
}
} else if !old_assignments.is_empty() {
return Ok(());
}
}
if !revoked.is_empty() {
if self.config.enable_auto_commit
&& let Err(e) = self.commit().await
{
if e.is_retriable() {
warn!(
"Auto-commit before KIP-848 revocation failed (retriable): {}",
e
);
} else {
error!(
"Auto-commit before KIP-848 revocation failed (fatal): {}",
e
);
}
}
self.safe_on_partitions_revoked(&revoked).await;
let revoked_tuples: Vec<(String, PartitionId)> = revoked
.iter()
.map(|tp| (tp.topic.clone(), tp.partition))
.collect();
self.apply_partition_revocations(&revoked_tuples).await;
}
{
let mut assignments = self.assignments.write().await;
assignments.clear();
for (topic, partitions) in &new_assignment.partitions {
assignments.insert(topic.clone(), partitions.clone());
}
}
self.metrics.rebalances.inc();
self.safe_on_partitions_assigned(&assigned).await;
let count: usize = new_assignment.partitions.values().map(|ps| ps.len()).sum();
self.metrics.assigned_partitions.set(count as u64);
if !assigned.is_empty() {
let new_parts = Self::group_partitions_by_topic(&assigned);
self.fetch_and_apply_committed_offsets(&new_parts).await?;
}
coordinator.acknowledge_revocation().await;
Ok(())
}
async fn handle_eager_rebalance(
&self,
coordinator: &Arc<GroupCoordinator>,
topics: &[String],
pending: Option<MemberAssignment>,
) -> Result<()> {
let old_assignments = self.assignments.read().await.clone();
if !old_assignments.is_empty() {
let revoked: Vec<TopicPartition> = old_assignments
.iter()
.flat_map(|(t, ps)| ps.iter().map(move |&p| TopicPartition::new(t, p)))
.collect();
if self.config.enable_auto_commit
&& let Err(e) = self.commit().await
{
if e.is_retriable() {
warn!(
"Auto-commit before eager revocation failed (retriable): {}",
e
);
} else {
error!("Auto-commit before eager revocation failed (fatal): {}", e);
}
}
self.safe_on_partitions_revoked(&revoked).await;
self.clear_partition_state().await;
self.assignments.write().await.clear();
self.metrics.assigned_partitions.set(0);
}
self.metrics.rebalances.inc();
let assignment = match pending {
Some(assignment) => assignment,
None => coordinator.ensure_active_membership(topics).await?.0,
};
let mut assignments = self.assignments.write().await;
assignments.clear();
for (topic, partitions) in &assignment.partitions {
assignments.insert(topic.clone(), partitions.clone());
}
drop(assignments);
let assigned: Vec<TopicPartition> = assignment
.partitions
.iter()
.flat_map(|(t, ps)| ps.iter().map(move |&p| TopicPartition::new(t, p)))
.collect();
self.safe_on_partitions_assigned(&assigned).await;
self.metrics.assigned_partitions.set(assigned.len() as u64);
self.fetch_and_apply_committed_offsets(&assignment.partitions)
.await?;
Ok(())
}
fn group_partitions_by_topic(
partitions: &[TopicPartition],
) -> HashMap<String, Vec<PartitionId>> {
let mut map: HashMap<String, Vec<PartitionId>> = HashMap::new();
for tp in partitions {
map.entry(tp.topic.clone()).or_default().push(tp.partition);
}
map
}
async fn fetch_and_apply_committed_offsets(
&self,
assigned: &HashMap<String, Vec<PartitionId>>,
) -> Result<()> {
let coordinator = match self.group_coordinator {
Some(ref c) => c,
None => return Ok(()),
};
let committed = coordinator.fetch_committed_offsets(assigned).await?;
{
let mut partition_state = self.partition_state.write().await;
for (key, position) in &committed {
if position.leader_epoch >= 0 {
partition_state
.entry(key.clone())
.or_default()
.last_fetched_epoch = Some(position.leader_epoch);
}
}
}
let committed: HashMap<(String, PartitionId), Offset> = committed
.into_iter()
.map(|(key, position)| (key, position.offset))
.collect();
let mut offsets = self.offsets.write().await;
debug!("fetch_and_apply: existing offsets: {:?}", *offsets);
let need_reset = apply_assignment_offset_precedence(
assigned,
&committed,
&self.config.initial_offsets,
&mut offsets,
);
if need_reset.is_empty() {
return Ok(());
}
if let Some(timestamp) = self.config.auto_offset_reset.to_offset() {
let mut reset_partitions: HashMap<String, Vec<PartitionId>> = HashMap::new();
for (topic, partition) in &need_reset {
reset_partitions
.entry(topic.clone())
.or_default()
.push(*partition);
}
let resolved = coordinator
.list_offsets(&reset_partitions, timestamp)
.await?;
for (key, offset) in &resolved {
offsets.insert(key.clone(), *offset);
}
for (topic, partition) in &need_reset {
let key = (topic.clone(), *partition);
if !resolved.contains_key(&key) && !offsets.contains_key(&key) {
debug!(
"Falling back to direct ListOffsets for {}-{} \
(coordinator path returned no result)",
topic, partition
);
drop(offsets);
match self.resolve_list_offset(topic, *partition, timestamp).await {
Ok(offset) => {
offsets = self.offsets.write().await;
offsets.insert(key, offset);
}
Err(e) => {
warn!(
"Fallback offset resolution failed for {}-{}: {}",
topic, partition, e
);
offsets = self.offsets.write().await;
}
}
}
}
} else {
let missing: Vec<String> = need_reset.iter().map(|(t, p)| format!("{t}-{p}")).collect();
return Err(KrafkaError::invalid_state(format!(
"no committed offset for partitions and auto.offset.reset=none: {}",
missing.join(", ")
)));
}
drop(offsets);
self.recompute_lag_metrics().await;
Ok(())
}
pub async fn assign(&self, topic: &str, partitions: Vec<PartitionId>) -> Result<()> {
validate_topic_name(topic)?;
if self.group_coordinator.is_some() {
return Err(KrafkaError::invalid_state(
"cannot use manual partition assignment with consumer group subscription",
));
}
self.metadata.refresh_for_topics(Some(&[topic])).await?;
let topic_owned = topic.to_string();
let dropped: Vec<(String, PartitionId)> = {
let assignments = self.assignments.read().await;
let keep: HashSet<PartitionId> = partitions.iter().copied().collect();
assignments
.get(&topic_owned)
.map(|previous| {
previous
.iter()
.filter(|p| !keep.contains(p))
.map(|p| (topic_owned.clone(), *p))
.collect()
})
.unwrap_or_default()
};
if !dropped.is_empty() {
debug!(
topic = %topic_owned,
partitions = dropped.len(),
"Dropping partitions removed by a narrower assign()"
);
self.apply_partition_revocations(&dropped).await;
}
let mut assignments = self.assignments.write().await;
assignments.insert(topic_owned.clone(), partitions.clone());
let mut subscriptions = self.subscriptions.write().await;
subscriptions.insert(topic_owned.clone());
drop(subscriptions);
drop(assignments);
let mut assigned = HashMap::new();
debug!("Assigned partitions for {}: {:?}", topic, partitions);
assigned.insert(topic_owned, partitions);
self.apply_auto_offset_reset(&assigned).await?;
Ok(())
}
async fn apply_auto_offset_reset(
&self,
assigned: &HashMap<String, Vec<PartitionId>>,
) -> Result<()> {
if !self.config.initial_offsets.is_empty() {
let mut offsets = self.offsets.write().await;
let inserted = seed_initial_offsets_for_assigned(
assigned,
&self.config.initial_offsets,
&mut offsets,
);
if inserted > 0 {
debug!("Applied {} assignment-time initial offsets", inserted);
}
}
let need_reset: Vec<(String, PartitionId)> = {
let offsets = self.offsets.read().await;
let mut need = Vec::new();
for (topic, partitions) in assigned {
for &p in partitions {
let key = (topic.clone(), p);
if !offsets.contains_key(&key) {
need.push(key);
}
}
}
need
};
if need_reset.is_empty() {
return Ok(());
}
if let Some(timestamp) = self.config.auto_offset_reset.to_offset() {
let reset_pairs: Vec<(&str, PartitionId)> =
need_reset.iter().map(|(t, p)| (t.as_str(), *p)).collect();
let batch = group_topic_partitions(&reset_pairs);
let resolved = self.resolve_list_offsets(&batch, timestamp).await;
let mut offsets = self.offsets.write().await;
for (key, result) in &resolved {
if let Ok(offset) = result {
offsets.insert(key.clone(), *offset);
}
}
drop(offsets);
for key in &need_reset {
if resolved.get(key).is_none_or(|r| r.is_err()) {
warn!(
"Failed to resolve offset for {}-{}, will retry on next poll",
key.0, key.1
);
}
}
} else {
let missing = need_reset
.iter()
.map(|(t, p)| format!("{t}-{p}"))
.collect::<Vec<_>>()
.join(", ");
return Err(KrafkaError::invalid_state(format!(
"no offset for partitions and auto.offset.reset=none: {missing}"
)));
}
self.recompute_lag_metrics().await;
Ok(())
}
fn discard_buffered_for(&self, repositioned: &HashSet<(String, PartitionId)>) {
let mut buffer = self.recv_buffer.lock();
let dropped = purge_buffered_records(&mut buffer, repositioned);
if dropped > 0 {
debug!(
dropped,
partitions = repositioned.len(),
"Discarded buffered records for repositioned partitions"
);
}
self.metrics.buffered_records.set(buffer.len() as u64);
}
pub async fn seek(&self, topic: &str, partition: PartitionId, offset: Offset) -> Result<()> {
{
let mut offsets = self.offsets.write().await;
offsets.insert((topic.to_string(), partition), offset);
let mut partition_state = self.partition_state.write().await;
invalidate_position_epoch(&mut partition_state, &(topic.to_string(), partition));
}
self.discard_buffered_for(&HashSet::from([(topic.to_string(), partition)]));
self.recompute_lag_metrics().await;
self.metrics.record_seek(1);
debug!("Seek to offset {} for {}-{}", offset, topic, partition);
Ok(())
}
pub async fn seek_many(&self, offsets: &HashMap<(String, PartitionId), Offset>) -> Result<()> {
if offsets.is_empty() {
return Ok(());
}
{
let mut stored = self.offsets.write().await;
apply_seek_many_offsets(&mut stored, offsets);
let mut partition_state = self.partition_state.write().await;
for key in offsets.keys() {
invalidate_position_epoch(&mut partition_state, key);
}
}
self.discard_buffered_for(&offsets.keys().cloned().collect());
self.recompute_lag_metrics().await;
self.metrics.record_seek(offsets.len() as u64);
debug!("Sought {} partitions via seek_many", offsets.len());
Ok(())
}
pub async fn seek_to_beginning(&self, topic: &str, partition: PartitionId) -> Result<()> {
self.seek(topic, partition, 0).await
}
pub async fn seek_to_end(&self, topic: &str, partition: PartitionId) -> Result<()> {
let offset = self.resolve_list_offset(topic, partition, -1).await?;
self.seek(topic, partition, offset).await
}
pub async fn seek_to_timestamp(
&self,
topic: &str,
partition: PartitionId,
timestamp_ms: i64,
) -> Result<()> {
let offset = self
.resolve_list_offset(topic, partition, timestamp_ms)
.await?;
if offset < 0 {
return Err(KrafkaError::invalid_state(format!(
"no message with timestamp >= {timestamp_ms} ms found in {topic}-{partition}"
)));
}
self.seek(topic, partition, offset).await
}
pub async fn offsets_for_times(
&self,
partitions: &[(&str, PartitionId)],
timestamp: i64,
) -> HashMap<(String, PartitionId), Result<Offset>> {
let mut result: HashMap<(String, PartitionId), Result<Offset>> = HashMap::new();
let mut valid: Vec<(&str, PartitionId)> = Vec::with_capacity(partitions.len());
for &(topic, partition) in partitions {
match validate_topic_name(topic) {
Ok(()) => valid.push((topic, partition)),
Err(e) => {
result.insert((topic.to_string(), partition), Err(e));
}
}
}
if !valid.is_empty() {
let grouped = group_topic_partitions(&valid);
result.extend(self.resolve_list_offsets(&grouped, timestamp).await);
}
result
}
pub async fn offsets_for_times_for_topic(
&self,
topic: &str,
timestamp: i64,
) -> Result<HashMap<PartitionId, Result<Offset>>> {
validate_topic_name(topic)?;
self.metadata.refresh_for_topics(Some(&[topic])).await?;
let info = self
.metadata
.topic(topic)
.ok_or_else(|| KrafkaError::invalid_state(format!("topic not found: {topic}")))?;
let mut grouped: HashMap<String, Vec<PartitionId>> = HashMap::new();
grouped.insert(
topic.to_string(),
info.partitions.values().map(|p| p.partition).collect(),
);
let results = self.resolve_list_offsets(&grouped, timestamp).await;
Ok(results
.into_iter()
.map(|((_, p), result)| (p, result))
.collect())
}
pub async fn fetch_watermarks(
&self,
topic: &str,
partition: PartitionId,
) -> Result<(Offset, Offset)> {
validate_topic_name(topic)?;
let (low, high) = tokio::join!(
self.resolve_list_offset(topic, partition, -2),
self.resolve_list_offset(topic, partition, -1),
);
Ok((low?, high?))
}
pub async fn fetch_metadata(&self, topic: Option<&str>) -> Result<FetchMetadataResult> {
if let Some(name) = topic {
validate_topic_name(name)?;
self.metadata.refresh_for_topics(Some(&[name])).await?;
}
let brokers = self.metadata.brokers();
let topics = match topic {
Some(name) => self
.metadata
.topic(name)
.map(|t| vec![t])
.unwrap_or_default(),
None => self.metadata.topics(),
};
Ok(FetchMetadataResult { brokers, topics })
}
async fn resolve_list_offset(
&self,
topic: &str,
partition: PartitionId,
timestamp: i64,
) -> Result<Offset> {
let mut partitions = HashMap::new();
let topic_owned = topic.to_string();
partitions.insert(topic_owned.clone(), vec![partition]);
let mut results = self.resolve_list_offsets(&partitions, timestamp).await;
results
.remove(&(topic_owned, partition))
.unwrap_or_else(|| {
Err(KrafkaError::protocol_kind(
ProtocolErrorKind::Malformed,
format!("no offset returned for {topic}-{partition}"),
))
})
}
async fn resolve_list_offsets(
&self,
partitions: &HashMap<String, Vec<PartitionId>>,
timestamp: i64,
) -> HashMap<(String, PartitionId), Result<Offset>> {
if partitions.is_empty() {
return HashMap::new();
}
let mut result: HashMap<(String, PartitionId), Result<Offset>> = partitions
.iter()
.flat_map(|(topic, parts)| {
let topic = topic.clone();
parts.iter().map(move |&p| {
let msg = format!("no leader found for {topic}-{p}");
((topic.clone(), p), Err(KrafkaError::invalid_state(msg)))
})
})
.collect();
let mut by_leader: HashMap<crate::BrokerId, Vec<(String, PartitionId)>> = HashMap::new();
let mut leaderless: Vec<(String, PartitionId)> = Vec::new();
for (topic, parts) in partitions {
for &p in parts {
if let Some(leader) = self.metadata.leader(topic, p) {
by_leader
.entry(leader)
.or_default()
.push((topic.clone(), p));
} else {
leaderless.push((topic.clone(), p));
}
}
}
if !leaderless.is_empty() {
let topic_set: HashSet<&str> = leaderless.iter().map(|(t, _)| t.as_str()).collect();
let topics: Vec<&str> = topic_set.into_iter().collect();
if let Err(err) = self.metadata.refresh_for_topics(Some(&topics)).await {
warn!(
"Failed to refresh metadata for leaderless topics {:?}: {}",
topics, err
);
}
for (topic, partition) in leaderless {
if let Some(leader) = self.metadata.leader(&topic, partition) {
by_leader
.entry(leader)
.or_default()
.push((topic, partition));
} else {
warn!(
"No leader for {}-{} after metadata refresh",
topic, partition
);
}
}
}
for (&leader_id, leader_partitions) in &by_leader {
let mut topics_map: HashMap<String, Vec<ListOffsetsRequestPartition>> = HashMap::new();
for (topic, partition) in leader_partitions {
topics_map
.entry(topic.clone())
.or_default()
.push(ListOffsetsRequestPartition {
partition_index: *partition,
current_leader_epoch: self
.metadata
.leader_epoch(topic, *partition)
.unwrap_or(-1),
timestamp,
});
}
let topics: Vec<ListOffsetsRequestTopic> = topics_map
.into_iter()
.map(|(name, parts)| ListOffsetsRequestTopic {
name,
partitions: parts,
})
.collect();
let request = ListOffsetsRequest {
replica_id: -1,
isolation_level: self.config.isolation_level.to_i8(),
topics,
timeout_ms: None,
};
let broker_info = match self.metadata.broker(leader_id) {
Some(b) => b,
None => {
warn!("Broker {} not found in metadata, skipping", leader_id);
let err = KrafkaError::invalid_state(format!(
"broker {leader_id} not found in metadata"
));
for (topic, partition) in leader_partitions {
result.insert((topic.clone(), *partition), Err(err.clone()));
}
continue;
}
};
let conn = match self
.pool
.get_connection_by_id(leader_id, broker_info.address())
.await
{
Ok(c) => c,
Err(e) => {
warn!("Failed to connect to broker {}: {}, skipping", leader_id, e);
for (topic, partition) in leader_partitions {
result.insert((topic.clone(), *partition), Err(e.clone()));
}
continue;
}
};
let list_version = match conn.negotiate_api_version(
ApiKey::ListOffsets,
versions::LIST_OFFSETS_MAX,
versions::LIST_OFFSETS_MIN,
) {
Some(v) => v,
None => {
let err = KrafkaError::protocol_kind(
ProtocolErrorKind::UnknownApiVersion,
format!(
"no mutually supported ListOffsets API version for broker {leader_id}"
),
);
warn!("{err}");
for (topic, partition) in leader_partitions {
result.insert((topic.clone(), *partition), Err(err.clone()));
}
continue;
}
};
let response = match conn
.send_request(ApiKey::ListOffsets, list_version, |buf| {
request.encode_versioned(list_version, buf)
})
.await
{
Ok(r) => r,
Err(e) => {
warn!(
"ListOffsets v{} request failed for broker {}: {}, skipping",
list_version, leader_id, e
);
for (topic, partition) in leader_partitions {
result.insert((topic.clone(), *partition), Err(e.clone()));
}
continue;
}
};
let mut buf = response;
let list_response = match ListOffsetsResponse::decode_versioned(list_version, &mut buf)
{
Ok(r) => r,
Err(e) => {
warn!(
"Failed to decode ListOffsets v{} response from broker {}: {}, skipping",
list_version, leader_id, e
);
let err = KrafkaError::protocol_kind(
ProtocolErrorKind::Malformed,
format!(
"failed to decode ListOffsets response from broker {leader_id}: {e}"
),
);
for (topic, partition) in leader_partitions {
result.insert((topic.clone(), *partition), Err(err.clone()));
}
continue;
}
};
let stale_epoch_topics = apply_list_offsets_response(&list_response, &mut result);
for topic in stale_epoch_topics {
if let Err(e) = self
.metadata
.refresh_for_topics_forced(Some(&[&topic]))
.await
{
debug!(
topic = %topic,
error = %e,
"metadata refresh after a fenced ListOffsets epoch failed"
);
}
}
}
result
}
pub async fn poll(&self, timeout: Duration) -> Result<Vec<ConsumerRecord>> {
lock_order::with_lock_tracking(self.poll_inner(timeout)).await
}
async fn poll_inner(&self, timeout: Duration) -> Result<Vec<ConsumerRecord>> {
if self.closed.load(std::sync::atomic::Ordering::SeqCst) {
return Err(KrafkaError::invalid_state("consumer is closed"));
}
if self
.wakeup_flag
.swap(false, std::sync::atomic::Ordering::AcqRel)
{
return Err(KrafkaError::invalid_state("wakeup() was called"));
}
let _poll_timer = self.metrics.poll_latency.start();
self.metrics.polls.inc();
if let Some(ref coordinator) = self.group_coordinator
&& coordinator.poll_interval_exceeded()
{
let lost: Vec<TopicPartition> = {
let assignments = self.assignments.read().await;
assignments
.iter()
.flat_map(|(t, ps)| ps.iter().map(move |&p| TopicPartition::new(t, p)))
.collect()
};
if !lost.is_empty() {
self.safe_on_partitions_lost(&lost).await;
}
if let Err(e) = coordinator.leave_group().await {
debug!("LeaveGroup after poll-interval expiry failed: {e}");
}
return Err(KrafkaError::invalid_state(format!(
"consumer exceeded max_poll_interval ({:?}) between poll() calls and was \
removed from group '{}'; its partitions have been reassigned. Process \
records faster, reduce max_poll_records, or raise max_poll_interval.",
coordinator.max_poll_interval(),
coordinator.group_id(),
)));
}
if let Some(ref coordinator) = self.group_coordinator
&& coordinator.take_membership_lost()
{
let lost: Vec<TopicPartition> = {
let assignments = self.assignments.read().await;
assignments
.iter()
.flat_map(|(t, ps)| ps.iter().map(move |&p| TopicPartition::new(t, p)))
.collect()
};
if !lost.is_empty() {
warn!(
group = coordinator.group_id(),
partitions = lost.len(),
"Member was fenced by the coordinator; dropping its partitions"
);
self.assignments.write().await.clear();
self.metrics.assigned_partitions.set(0);
self.safe_on_partitions_lost(&lost).await;
}
}
if let Some(ref coordinator) = self.group_coordinator
&& let Some(message) = coordinator.take_fatal_error()
{
return Err(KrafkaError::invalid_state(message));
}
if let Some(ref coordinator) = self.group_coordinator {
coordinator.note_poll();
}
if self.config.enable_auto_commit && self.group_coordinator.is_some() {
let should_commit = {
let last = self.last_auto_commit.lock();
last.elapsed() >= self.config.auto_commit_interval
};
if should_commit {
match self.commit().await {
Ok(()) => {
*self.last_auto_commit.lock() = Instant::now();
}
Err(e) => {
warn!("Auto-commit failed: {}", e);
}
}
}
}
if self.handle_group_rebalance(timeout).await? {
return Ok(vec![]);
}
let assignments = self.assignments.read().await;
if assignments.is_empty() {
self.metrics.empty_polls.inc();
return Ok(Vec::new());
}
let max_records = if self.config.max_poll_records > 0 {
self.config.max_poll_records as usize
} else {
usize::MAX
};
let mut prefetched: Vec<ConsumerRecord> = Vec::new();
{
let paused = self.paused.read().await.clone();
let mut buffer = self.recv_buffer.lock();
drain_buffered_records(&mut buffer, &mut prefetched, max_records, &paused);
self.metrics.buffered_records.set(buffer.len() as u64);
}
if !prefetched.is_empty() {
drop(assignments);
return self.finish_delivery(prefetched).await;
}
let buffered_after_drain = self.recv_buffer.lock().len();
if self.config.max_buffered_records > 0
&& buffered_after_drain >= self.config.max_buffered_records as usize
{
debug!(
buffered = buffered_after_drain,
max = self.config.max_buffered_records,
"Buffer cap reached, skipping fetch"
);
self.metrics.empty_polls.inc();
return Ok(Vec::new());
}
{
let now = Instant::now();
let missing: Vec<(String, PartitionId)> = {
let offsets = self.offsets.read().await;
let partition_state = self.partition_state.read().await;
assignments
.iter()
.flat_map(|(topic, partitions)| {
partitions.iter().filter_map(|&p| {
let key = (topic.clone(), p);
if offsets.contains_key(&key) {
return None;
}
match partition_state
.get(&key)
.and_then(|s| s.offset_retry_backoff)
{
None => Some(key),
Some((next_retry, _)) if now >= next_retry => Some(key),
_ => None,
}
})
})
.collect()
};
if !missing.is_empty() {
debug!(
"Retrying offset resolution for {} partition(s) without tracked offsets",
missing.len()
);
let mut still_missing = missing;
let mut committed_fetch_failed = false;
if let Some(ref coordinator) = self.group_coordinator {
let mut topic_map: HashMap<String, Vec<PartitionId>> = HashMap::new();
for (topic, partition) in &still_missing {
topic_map.entry(topic.clone()).or_default().push(*partition);
}
if !topic_map.is_empty() {
match coordinator.fetch_committed_offsets(&topic_map).await {
Ok(committed) => {
{
let mut partition_state = self.partition_state.write().await;
for (key, position) in &committed {
if position.leader_epoch >= 0 {
partition_state
.entry(key.clone())
.or_default()
.last_fetched_epoch = Some(position.leader_epoch);
}
}
}
let mut offsets = self.offsets.write().await;
still_missing.retain(|key| {
if let Some(position) = committed.get(key)
&& position.offset >= 0
{
offsets.insert(key.clone(), position.offset);
false
} else {
true
}
});
}
Err(e) => {
warn!(
"Retry committed-offset fetch failed; deferring offset resolution until retry: {}",
e
);
committed_fetch_failed = true;
}
}
}
}
if !committed_fetch_failed {
if !self.config.initial_offsets.is_empty() {
let mut offsets = self.offsets.write().await;
still_missing.retain(|key| {
if let Some(&initial) = self.config.initial_offsets.get(key) {
debug!(
"Poll retry: using initial_offsets {} for {}-{}",
initial, key.0, key.1
);
offsets.insert(key.clone(), initial);
false
} else {
true
}
});
}
if still_missing.is_empty() {
self.recompute_lag_metrics().await;
return Ok(Vec::new());
}
let mut reset_partitions: HashMap<String, Vec<PartitionId>> = HashMap::new();
for (topic, partition) in &still_missing {
reset_partitions
.entry(topic.clone())
.or_default()
.push(*partition);
}
if let Some(ref coordinator) = self.group_coordinator {
if let Some(timestamp) = self.config.auto_offset_reset.to_offset() {
match coordinator.list_offsets(&reset_partitions, timestamp).await {
Ok(resolved) => {
let mut offsets = self.offsets.write().await;
for (key, offset) in &resolved {
offsets.insert(key.clone(), *offset);
}
drop(offsets);
for (topic, partition) in &still_missing {
if !resolved.contains_key(&(topic.clone(), *partition)) {
debug!(
"Poll retry: falling back to direct ListOffsets for {}-{}",
topic, partition
);
if let Ok(offset) = self
.resolve_list_offset(topic, *partition, timestamp)
.await
{
let mut offsets = self.offsets.write().await;
offsets.insert((topic.clone(), *partition), offset);
}
}
}
}
Err(e) => {
warn!("Offset resolution retry via coordinator failed: {}", e);
for (topic, partition) in &still_missing {
if let Ok(offset) = self
.resolve_list_offset(topic, *partition, timestamp)
.await
{
let mut offsets = self.offsets.write().await;
offsets.insert((topic.clone(), *partition), offset);
}
}
}
}
}
} else if let Err(e) = self.apply_auto_offset_reset(&reset_partitions).await {
warn!("Auto-offset-reset failed for missing partitions: {e}");
}
}
self.recompute_lag_metrics().await;
{
let offsets = self.offsets.read().await;
let mut partition_state = self.partition_state.write().await;
for (topic, partition) in &still_missing {
let key = (topic.clone(), *partition);
if offsets.contains_key(&key) {
if let Some(state) = partition_state.get_mut(&key) {
state.offset_retry_backoff = None;
}
} else {
let base = Duration::from_millis(100);
let max = Duration::from_secs(30);
let entry = partition_state.entry(key).or_default();
let prev_wait = entry
.offset_retry_backoff
.map(|(_, d)| d)
.unwrap_or(Duration::ZERO);
let next_wait = (prev_wait * 2).max(base).min(max);
entry.offset_retry_backoff =
Some((Instant::now() + next_wait, next_wait));
}
}
}
}
}
let paused = self.paused.read().await;
let mut non_paused_keys: Vec<(String, PartitionId)> = Vec::new();
let mut leaders: HashMap<(String, PartitionId), crate::BrokerId> = HashMap::new();
for (topic, partitions) in assignments.iter() {
for &partition in partitions {
let key = (topic.clone(), partition);
if paused.contains(&key) {
continue;
}
if let Some(leader_id) = self.metadata.leader(topic, partition) {
leaders.insert(key.clone(), leader_id);
}
non_paused_keys.push(key);
}
}
non_paused_keys.sort_unstable();
if !non_paused_keys.is_empty() {
let turn = self
.fetch_rotation
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let offset = turn % non_paused_keys.len();
non_paused_keys.rotate_left(offset);
}
let now = Instant::now();
let partition_state_read = self.partition_state.read().await;
let plan = build_fetch_routing_plan(non_paused_keys, &partition_state_read, &leaders, now);
drop(partition_state_read);
for (topic, partition) in &plan.skipped {
warn!(
"No leader or preferred replica for {topic}-{partition}, skipping in batch fetch"
);
}
if !plan.expired_preferred.is_empty() {
let mut partition_state = self.partition_state.write().await;
for key in &plan.expired_preferred {
if let Some(state) = partition_state.get_mut(key) {
state.preferred_replica = None;
}
}
}
drop(paused);
drop(assignments);
{
let to_validate: Vec<(String, PartitionId)> = plan
.partitions_by_broker
.values()
.flat_map(|keys| keys.iter().cloned())
.collect();
self.validate_pending_positions(&to_validate).await;
}
let deadline = Instant::now() + timeout;
let mut all_records = Vec::new();
let mut all_offset_updates: Vec<((String, PartitionId), FetchOffsetUpdate)> = Vec::new();
let mut all_hw_updates: Vec<((String, PartitionId), Offset)> = Vec::new();
let mut all_faults: Vec<PartitionFetchFault> = Vec::new();
let record_budget: Option<Arc<std::sync::atomic::AtomicUsize>> =
if max_records == usize::MAX {
None
} else {
let prefetch_headroom = if self.config.max_buffered_records > 0 {
(self.config.max_buffered_records as usize).saturating_sub(buffered_after_drain)
} else {
max_records
};
Some(Arc::new(std::sync::atomic::AtomicUsize::new(
max_records.saturating_add(prefetch_headroom),
)))
};
loop {
let remaining = deadline.saturating_duration_since(Instant::now());
let max_wait = self.config.fetch_max_wait.min(remaining);
let budget = record_budget.as_deref();
let fetches =
plan.partitions_by_broker
.iter()
.map(|(broker_id, topic_partitions)| async move {
let result = self
.batch_fetch_from_broker(*broker_id, topic_partitions, max_wait, budget)
.await;
(*broker_id, topic_partitions, result)
});
let fetch_results = tokio::select! {
biased;
() = self.wakeup_notify.notified() => {
self.wakeup_flag
.store(false, std::sync::atomic::Ordering::Release);
if all_records.is_empty() {
return Err(KrafkaError::invalid_state("wakeup() was called"));
}
break;
}
results = futures::future::join_all(fetches) => results,
};
for (broker_id, topic_partitions, result) in fetch_results {
match result {
Ok(outcome) => {
all_records.extend(outcome.records);
all_offset_updates.extend(outcome.offset_updates);
all_hw_updates.extend(outcome.hw_updates);
all_faults.extend(outcome.faults);
}
Err(e) => {
self.metrics.record_error();
warn!("Batch fetch from broker {} failed: {}", broker_id, e);
let mut partition_state = self.partition_state.write().await;
for tp in topic_partitions {
if let Some(state) = partition_state.get_mut(tp) {
state.preferred_replica = None;
}
}
}
}
}
if !all_records.is_empty() || !all_faults.is_empty() || Instant::now() >= deadline {
break;
}
}
if !all_faults.is_empty() {
let total = all_faults.len();
self.metrics.record_error();
if let Some(fault) = all_faults.into_iter().next() {
return Err(fault.into_error(total));
}
}
let (mut delivered, held) = {
let paused = self.paused.read().await;
let mut delivered: Vec<ConsumerRecord> =
Vec::with_capacity(all_records.len().min(max_records));
let mut held: Vec<ConsumerRecord> = Vec::new();
for record in all_records {
let is_paused = contains_partition(&paused, &record.topic, record.partition);
if !is_paused && delivered.len() < max_records {
delivered.push(record);
} else {
held.push(record);
}
}
(delivered, held)
};
let hw_changed = !all_hw_updates.is_empty();
if hw_changed {
let now = Instant::now();
let mut partition_state = self.partition_state.write().await;
for (key, watermark) in all_hw_updates {
let s = partition_state.entry(key).or_default();
s.high_watermark = Some(watermark);
s.watermark_updated_at = Some(now);
}
}
if hw_changed {
self.recompute_lag_metrics().await;
}
if !all_offset_updates.is_empty() || !held.is_empty() {
let epochs: Vec<((String, PartitionId), i32)> = all_offset_updates
.iter()
.map(|(key, update)| (key.clone(), update.epoch))
.collect();
let mut offsets = self.offsets.write().await;
let mut partition_state = self.partition_state.write().await;
let mut buffer = self.recv_buffer.lock();
let discarded = apply_fetch_offset_updates(&mut offsets, all_offset_updates);
let discarded_set: HashSet<&(String, PartitionId)> = discarded.iter().collect();
for (key, epoch) in &epochs {
if discarded_set.contains(key) {
continue;
}
if *epoch >= 0 {
partition_state
.entry(key.clone())
.or_default()
.last_fetched_epoch = Some(*epoch);
}
}
let mut held = held;
if !discarded.is_empty() {
let stale: HashSet<(String, PartitionId)> =
discarded_set.iter().map(|k| (*k).clone()).collect();
for (topic, partition) in &stale {
debug!(
topic = %topic,
partition,
"Discarding fetch response: position changed while the fetch was in flight"
);
}
delivered.retain(|r| !contains_partition(&stale, &r.topic, r.partition));
held.retain(|r| !contains_partition(&stale, &r.topic, r.partition));
}
buffer.extend(held);
self.metrics.buffered_records.set(buffer.len() as u64);
}
self.finish_delivery(delivered).await
}
async fn finish_delivery(
&self,
mut records: Vec<ConsumerRecord>,
) -> Result<Vec<ConsumerRecord>> {
if records.is_empty() {
self.metrics.empty_polls.inc();
return Ok(records);
}
if self.key_deserializer.is_some() || self.value_deserializer.is_some() {
for index in 0..records.len() {
if let Err(error) = self.deserialize_in_place(&mut records[index]).await {
self.requeue_undelivered(records);
self.metrics.record_error();
return Err(error);
}
}
}
let bytes: u64 = records
.iter()
.map(|r| r.value.as_ref().map(|v| v.len() as u64).unwrap_or(0))
.sum();
self.metrics.record_receive(records.len() as u64, bytes);
crate::interceptor::safe_on_consume(&*self.interceptor, &records);
Ok(records)
}
async fn deserialize_in_place(&self, record: &mut ConsumerRecord) -> Result<()> {
if let (Some(decoder), Some(value)) = (&self.value_deserializer, record.value.as_ref()) {
let decoded = decoder
.deserialize(value.clone(), &record.topic, false)
.await
.map_err(|e| {
KrafkaError::record_deserialization(
&record.topic,
record.partition,
record.offset,
"value",
e.to_string(),
)
})?;
record.value = Some(decoded);
}
if let (Some(decoder), Some(key)) = (&self.key_deserializer, record.key.as_ref()) {
let decoded = decoder
.deserialize(key.clone(), &record.topic, true)
.await
.map_err(|e| {
KrafkaError::record_deserialization(
&record.topic,
record.partition,
record.offset,
"key",
e.to_string(),
)
})?;
record.key = Some(decoded);
}
Ok(())
}
fn requeue_undelivered(&self, records: Vec<ConsumerRecord>) {
if records.is_empty() {
return;
}
let mut buffer = self.recv_buffer.lock();
for record in records.into_iter().rev() {
buffer.push_front(record);
}
self.metrics.buffered_records.set(buffer.len() as u64);
}
async fn batch_fetch_from_broker(
&self,
broker_id: crate::BrokerId,
topic_partitions: &[(String, PartitionId)],
max_wait: Duration,
record_budget: Option<&std::sync::atomic::AtomicUsize>,
) -> Result<FetchOutcome> {
if topic_partitions.is_empty() {
return Ok(FetchOutcome::default());
}
self.metrics.record_fetch();
let _fetch_timer = self.metrics.fetch_latency.start();
let address = self
.broker_address(broker_id)
.ok_or_else(|| KrafkaError::invalid_state(format!("broker {} not found", broker_id)))?;
let conn = self.pool.get_connection_by_id(broker_id, &address).await?;
let mut topics_map: HashMap<String, Vec<PartitionId>> = HashMap::new();
for (topic, partition) in topic_partitions {
topics_map
.entry(topic.clone())
.or_default()
.push(*partition);
}
let offsets_snapshot = self.offsets.read().await;
let last_fetched_epochs: HashMap<(String, PartitionId), i32> = {
let partition_state = self.partition_state.read().await;
topic_partitions
.iter()
.filter_map(|key| {
partition_state
.get(key)
.and_then(|s| s.last_fetched_epoch)
.map(|epoch| (key.clone(), epoch))
})
.collect()
};
let mut requested_offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
let mut fetch_topics = Vec::with_capacity(topics_map.len());
for (topic, partitions) in &topics_map {
let mut fetch_partitions = Vec::with_capacity(partitions.len());
for &partition in partitions {
let offset = match offsets_snapshot.get(&(topic.clone(), partition)).copied() {
Some(o) => o,
None => {
warn!(
"No offset for {}-{}, skipping fetch (will retry offset resolution)",
topic, partition
);
continue;
}
};
requested_offsets.insert((topic.clone(), partition), offset);
let leader_epoch = self.metadata.leader_epoch(topic, partition).unwrap_or(-1);
fetch_partitions.push(FetchPartitionRequest {
partition,
current_leader_epoch: leader_epoch,
fetch_offset: offset,
last_fetched_epoch: last_fetched_epochs
.get(&(topic.clone(), partition))
.copied()
.unwrap_or(-1),
log_start_offset: -1,
partition_max_bytes: self
.config
.topic_fetch_max_bytes
.get(topic.as_str())
.copied()
.unwrap_or(self.config.max_partition_fetch_bytes),
replica_directory_id: None,
high_watermark: None,
});
}
fetch_topics.push(FetchTopicRequest {
topic: topic.clone(),
topic_id: None,
partitions: fetch_partitions,
});
}
drop(offsets_snapshot);
let mut fetch_version = conn
.negotiate_api_version(ApiKey::Fetch, versions::FETCH_MAX, 7)
.unwrap_or_else(|| {
debug!(
"No mutually supported Fetch v7+ for broker {broker_id}, falling back to v4"
);
4
});
if fetch_version >= 13 {
let all_resolved = fetch_topics.iter_mut().all(|t| {
if let Some(id) = self.metadata.topic_id_for_name(&t.topic) {
t.topic_id = Some(id);
true
} else {
false
}
});
if !all_resolved {
fetch_version = 12;
}
}
let (session_id, session_epoch, request_topics, forgotten_topics) = if fetch_version >= 7 {
let mut sessions = self.fetch_sessions.lock();
let session = sessions.get_or_create(broker_id);
let session_req = session.build_request(&fetch_topics);
if session_req.is_full_fetch {
debug!(
"Fetch broker {}: full fetch (session_id={}, epoch={})",
broker_id, session_req.session_id, session_req.session_epoch
);
} else {
debug!(
"Fetch broker {}: incremental (session_id={}, epoch={}, changed={}, forgotten={})",
broker_id,
session_req.session_id,
session_req.session_epoch,
session_req.topics.len(),
session_req.forgotten_topics.len()
);
}
(
session_req.session_id,
session_req.session_epoch,
{
let mut topics = session_req.topics;
if fetch_version >= 13 {
for t in &mut topics {
if t.topic_id.is_none() {
t.topic_id = self.metadata.topic_id_for_name(&t.topic);
}
}
}
topics
},
{
let mut forgotten = session_req.forgotten_topics;
if fetch_version >= 13 {
for t in &mut forgotten {
if t.topic_id.is_none() {
t.topic_id = self.metadata.topic_id_for_name(&t.topic);
}
}
}
forgotten
},
)
} else {
(0, -1, std::mem::take(&mut fetch_topics), Vec::new())
};
let request = FetchRequest {
replica_id: -1, max_wait_ms: crate::util::duration_to_millis_i32(max_wait),
min_bytes: self.config.fetch_min_bytes,
max_bytes: self.config.fetch_max_bytes,
isolation_level: self.config.isolation_level.to_i8(),
session_id,
session_epoch,
topics: request_topics,
forgotten_topics,
rack_id: self.config.client_rack.clone().unwrap_or_default(),
};
let response = match conn
.send_request(ApiKey::Fetch, fetch_version, |buf| {
request.encode_versioned(fetch_version, buf)
})
.await
{
Ok(r) => r,
Err(e) => {
if fetch_version >= 7 {
let mut sessions = self.fetch_sessions.lock();
sessions.reset_broker(broker_id);
}
return Err(e);
}
};
let mut buf = response;
let mut fetch_response = match FetchResponse::decode_versioned(fetch_version, &mut buf) {
Ok(r) => r,
Err(e) => {
if fetch_version >= 7 {
let mut sessions = self.fetch_sessions.lock();
sessions.reset_broker(broker_id);
}
return Err(e);
}
};
conn.notify_throttle(fetch_response.throttle_time_ms);
if fetch_version >= 13 {
fetch_response.responses.retain_mut(|topic_response| {
if !topic_response.topic.is_empty() {
return true;
}
let Some(id) = topic_response.topic_id else {
warn!("Received FetchResponse v13+ with neither a topic name nor a topic_id");
return false;
};
match self.metadata.topic_name_for_id(&id) {
Some(name) => {
topic_response.topic = name;
true
}
None => {
warn!(
"Received FetchResponse v13+ with unknown topic_id {:?}; \
discarding its partitions (metadata will refresh)",
id
);
false
}
}
});
}
if fetch_version >= 7 {
if fetch_response.error_code == crate::error::ErrorCode::FetchSessionIdNotFound
|| fetch_response.error_code == crate::error::ErrorCode::InvalidFetchSessionEpoch
{
warn!(
"Fetch session error for broker {}: {:?}, resetting session",
broker_id, fetch_response.error_code
);
let mut sessions = self.fetch_sessions.lock();
sessions.reset_broker(broker_id);
return Ok(FetchOutcome::default());
}
let mut sessions = self.fetch_sessions.lock();
let session = sessions.get_or_create(broker_id);
session.update_from_response(fetch_response.session_id, &fetch_topics);
}
let mut records = Vec::new();
let mut offset_updates: Vec<((String, PartitionId), FetchOffsetUpdate)> = Vec::new();
let mut hw_updates: Vec<((String, PartitionId), Offset)> = Vec::new();
let mut log_start_updates: Vec<((String, PartitionId), Offset)> = Vec::new();
let mut stable_updates: Vec<((String, PartitionId), Offset)> = Vec::new();
let mut faults: Vec<PartitionFetchFault> = Vec::new();
let mut pref_updates: Vec<((String, PartitionId), Option<crate::BrokerId>)> = Vec::new();
let node_endpoints = std::mem::take(&mut fetch_response.node_endpoints);
for topic_response in fetch_response.responses {
let topic_name = &topic_response.topic;
for partition_response in topic_response.partitions {
let partition = partition_response.partition;
let key = (topic_name.clone(), partition);
if partition_response.high_watermark >= 0 {
hw_updates.push((key.clone(), partition_response.high_watermark));
}
if partition_response.log_start_offset >= 0 {
log_start_updates.push((key.clone(), partition_response.log_start_offset));
}
if partition_response.last_stable_offset >= 0 {
stable_updates.push((key.clone(), partition_response.last_stable_offset));
}
if fetch_version >= 11 {
if partition_response.preferred_read_replica >= 0 {
pref_updates
.push((key.clone(), Some(partition_response.preferred_read_replica)));
} else {
pref_updates.push((key.clone(), None));
}
}
if let Some(diverging) = partition_response.diverging_epoch {
self.truncate_to_diverging_offset(topic_name, partition, diverging)
.await;
continue;
}
if !partition_response.error_code.is_ok() {
let is_leader = self
.metadata
.leader(topic_name, partition)
.is_some_and(|leader_id| leader_id == broker_id);
if !is_leader {
debug!(
"Error from non-leader broker {} for {}-{}: {:?}, clearing preferred replica",
broker_id, topic_name, partition, partition_response.error_code
);
pref_updates.push((key.clone(), None));
}
if matches!(
partition_response.error_code,
crate::error::ErrorCode::NotLeaderForPartition
| crate::error::ErrorCode::FencedLeaderEpoch
) && let Some(leader) = partition_response.current_leader
{
debug!(
"Broker {} reports {}-{} now led by node {} (epoch {})",
broker_id, topic_name, partition, leader.leader_id, leader.leader_epoch
);
self.metadata.apply_leader_hint(
topic_name,
partition,
leader.leader_id,
leader.leader_epoch,
broker_info_for_node(&node_endpoints, leader.leader_id),
);
}
if partition_response.error_code == crate::error::ErrorCode::FencedLeaderEpoch
|| partition_response.error_code
== crate::error::ErrorCode::UnknownLeaderEpoch
{
warn!(
"Leader epoch error for {}-{}: {:?}, validating offset via OffsetForLeaderEpoch",
topic_name, partition, partition_response.error_code
);
if let Err(e) = self
.validate_offset_for_leader_epoch(topic_name, partition)
.await
{
warn!(
"OffsetForLeaderEpoch validation failed for {}-{}: {}, \
falling back to auto_offset_reset",
topic_name, partition, e
);
self.handle_offset_out_of_range(topic_name, partition).await;
}
} else if partition_response.error_code
== crate::error::ErrorCode::OffsetOutOfRange
{
warn!(
"OffsetOutOfRange for {}-{}, applying auto_offset_reset",
topic_name, partition
);
self.handle_offset_out_of_range(topic_name, partition).await;
} else {
warn!(
"Fetch error for {}-{}: {:?}",
topic_name, partition, partition_response.error_code
);
}
continue; }
if let Some(record_bytes) = partition_response.records {
let partition_fetch_offset = match requested_offsets.get(&key).copied() {
Some(offset) => offset,
None => {
debug!(
topic = %topic_name,
partition,
"Received records for a partition that was not requested, ignoring"
);
continue;
}
};
if record_budget
.is_some_and(|b| b.load(std::sync::atomic::Ordering::Relaxed) == 0)
{
continue;
}
let outcome = decode_partition_batches(
topic_name,
partition,
record_bytes,
partition_fetch_offset,
partition_response.aborted_transactions,
record_budget,
self.config.max_decompressed_size,
&mut records,
);
if outcome.corrupt {
self.metrics.record_batch_decode_error();
}
if let Some(error) = outcome.error {
faults.push(PartitionFetchFault {
key: key.clone(),
offset: partition_fetch_offset,
error,
});
}
if let Some(last_offset) = outcome.last_offset {
offset_updates.push((
key,
FetchOffsetUpdate {
requested: partition_fetch_offset,
next: last_offset.saturating_add(1),
epoch: outcome.last_epoch,
},
));
}
}
}
}
if !log_start_updates.is_empty() || !pref_updates.is_empty() || !stable_updates.is_empty() {
let expiry = Instant::now() + self.config.metadata_max_age;
let mut partition_state = self.partition_state.write().await;
for (key, offset) in log_start_updates {
partition_state.entry(key).or_default().log_start_offset = Some(offset);
}
for (key, offset) in stable_updates {
partition_state.entry(key).or_default().last_stable_offset = Some(offset);
}
for (key, value) in pref_updates {
match value {
Some(replica_id) => {
partition_state.entry(key).or_default().preferred_replica =
Some((replica_id, expiry));
}
None => {
if let Some(state) = partition_state.get_mut(&key) {
state.preferred_replica = None;
}
}
}
}
}
Ok(FetchOutcome {
records,
offset_updates,
hw_updates,
faults,
})
}
fn broker_address(&self, broker_id: crate::BrokerId) -> Option<String> {
self.metadata
.broker(broker_id)
.map(|broker| broker.address().to_string())
}
async fn truncate_to_diverging_offset(
&self,
topic: &str,
partition: PartitionId,
diverging: crate::protocol::DivergingEpoch,
) {
let key = (topic.to_string(), partition);
let new_position = diverging.end_offset;
let mut offsets = self.offsets.write().await;
let old_position = offsets.get(&key).copied();
offsets.insert(key.clone(), new_position);
let mut partition_state = self.partition_state.write().await;
let entry = partition_state.entry(key).or_default();
entry.last_fetched_epoch = None;
entry.position_validated = true;
drop(partition_state);
drop(offsets);
let dropped = {
let mut buffer = self.recv_buffer.lock();
let before = buffer.len();
buffer.retain(|record| {
record.topic != topic
|| record.partition != partition
|| record.offset < new_position
});
let after = buffer.len();
self.metrics.buffered_records.set(after as u64);
before - after
};
warn!(
topic = %topic,
partition,
old_position = ?old_position,
new_position,
diverging_epoch = diverging.epoch,
dropped_buffered_records = dropped,
"Log truncation detected: the partition's log diverged from the leader's; \
rewinding the fetch position"
);
self.metrics.record_seek(1);
self.recompute_lag_metrics().await;
}
async fn validate_pending_positions(&self, keys: &[(String, PartitionId)]) {
let pending: Vec<(String, PartitionId)> = {
let offsets = self.offsets.read().await;
let partition_state = self.partition_state.read().await;
keys.iter()
.filter(|key| offsets.contains_key(*key))
.filter(|key| {
partition_state
.get(*key)
.is_none_or(|state| !state.position_validated)
})
.cloned()
.collect()
};
for (topic, partition) in pending {
match self
.validate_offset_for_leader_epoch_inner(&topic, partition, false)
.await
{
Ok(()) => {
let mut partition_state = self.partition_state.write().await;
partition_state
.entry((topic, partition))
.or_default()
.position_validated = true;
}
Err(e) => {
debug!(
topic = %topic,
partition,
error = %e,
"Offset validation failed; retrying on the next poll"
);
}
}
}
}
async fn handle_offset_out_of_range(&self, topic: &str, partition: PartitionId) {
let Some(target) = self.config.auto_offset_reset.to_offset() else {
return;
};
let key = (topic.to_string(), partition);
let resolved = if let Some(ref gc) = self.group_coordinator {
let mut part_map = HashMap::new();
part_map.insert(key.0.clone(), vec![partition]);
match gc.list_offsets(&part_map, target).await {
Ok(offsets) => offsets.get(&key).copied(),
Err(e) => {
warn!(
"Coordinator list_offsets failed for {}-{}: {}, falling back to direct",
topic, partition, e
);
None
}
}
} else {
None
};
let offset = match resolved {
Some(o) => Some(o),
None => self
.resolve_list_offset(topic, partition, target)
.await
.map_err(|e| {
warn!("Direct list_offset failed for {topic}-{partition}: {e}");
e
})
.ok(),
};
if let Some(new_offset) = offset {
let mut offsets = self.offsets.write().await;
offsets.insert(key.clone(), new_offset);
let mut partition_state = self.partition_state.write().await;
invalidate_position_epoch(&mut partition_state, &key);
drop(partition_state);
drop(offsets);
self.discard_buffered_for(&HashSet::from([key]));
self.recompute_lag_metrics().await;
}
}
async fn validate_offset_for_leader_epoch(
&self,
topic: &str,
partition: PartitionId,
) -> Result<()> {
self.validate_offset_for_leader_epoch_inner(topic, partition, true)
.await
}
async fn validate_offset_for_leader_epoch_inner(
&self,
topic: &str,
partition: PartitionId,
refresh_metadata: bool,
) -> Result<()> {
use crate::protocol::OffsetForLeaderEpochPartition;
use crate::protocol::OffsetForLeaderEpochRequest;
use crate::protocol::OffsetForLeaderEpochResponse;
use crate::protocol::OffsetForLeaderEpochTopic;
if refresh_metadata && let Err(e) = self.metadata.refresh_for_topics(Some(&[topic])).await {
warn!(
"Metadata refresh failed for {}: {}, using cached metadata",
topic, e
);
}
let leader_epoch = self.metadata.leader_epoch(topic, partition).unwrap_or(-1);
if leader_epoch < 0 {
return Ok(());
}
let position_epoch = {
let partition_state = self.partition_state.read().await;
partition_state
.get(&(topic.to_string(), partition))
.and_then(|state| state.last_fetched_epoch)
.filter(|&epoch| epoch >= 0)
.unwrap_or(leader_epoch)
};
let leader_id = self.metadata.leader(topic, partition).ok_or_else(|| {
KrafkaError::invalid_state(format!("no leader for {topic}-{partition}"))
})?;
let broker = self
.metadata
.broker(leader_id)
.ok_or_else(|| KrafkaError::invalid_state(format!("broker {} not found", leader_id)))?;
let conn = self
.pool
.get_connection_by_id(leader_id, broker.address())
.await?;
let request = OffsetForLeaderEpochRequest {
replica_id: -1, topics: vec![OffsetForLeaderEpochTopic {
topic: topic.to_string(),
partitions: vec![OffsetForLeaderEpochPartition {
partition,
current_leader_epoch: leader_epoch,
leader_epoch: position_epoch,
}],
}],
};
let version = conn
.negotiate_api_version(
ApiKey::OffsetForLeaderEpoch,
versions::OFFSET_FOR_LEADER_EPOCH_MAX,
versions::OFFSET_FOR_LEADER_EPOCH_MIN,
)
.ok_or_else(|| {
KrafkaError::protocol_kind(
ProtocolErrorKind::UnknownApiVersion,
"no mutually supported OffsetForLeaderEpoch API version",
)
})?;
let response_bytes = conn
.send_request(ApiKey::OffsetForLeaderEpoch, version, |buf| {
request.encode_versioned(version, buf)
})
.await?;
let mut buf = response_bytes;
let response = OffsetForLeaderEpochResponse::decode_versioned(version, &mut buf)?;
let key = (topic.to_string(), partition);
let mut offset_changed = false;
for topic_result in response.topics {
for partition_result in topic_result.partitions {
if partition_result.partition != partition {
continue;
}
if partition_result.error_code.is_ok() && partition_result.end_offset >= 0 {
let current_offset = {
let offsets = self.offsets.read().await;
match offsets.get(&key).copied() {
Some(offset) => offset,
None => {
debug!(
topic = %topic,
partition,
"No tracked offset for partition during epoch validation"
);
0
}
}
};
if current_offset > partition_result.end_offset {
warn!(
topic = %topic,
partition,
old_position = current_offset,
new_position = partition_result.end_offset,
"Log truncation detected: the position is past the end of its \
leader epoch; rewinding the fetch position"
);
let mut offsets = self.offsets.write().await;
offsets.insert(key.clone(), partition_result.end_offset);
let mut partition_state = self.partition_state.write().await;
let entry = partition_state.entry(key.clone()).or_default();
entry.last_fetched_epoch = None;
drop(partition_state);
drop(offsets);
let mut buffer = self.recv_buffer.lock();
buffer.retain(|record| {
record.topic != topic
|| record.partition != partition
|| record.offset < partition_result.end_offset
});
self.metrics.buffered_records.set(buffer.len() as u64);
drop(buffer);
offset_changed = true;
}
}
}
}
if offset_changed {
self.recompute_lag_metrics().await;
}
Ok(())
}
pub async fn recv(&self) -> std::result::Result<ConsumerRecord, RecvError> {
loop {
{
let paused = self.paused.read().await.clone();
let mut buffer = self.recv_buffer.lock();
let mut one = Vec::with_capacity(1);
drain_buffered_records(&mut buffer, &mut one, 1, &paused);
self.metrics.buffered_records.set(buffer.len() as u64);
if let Some(record) = one.pop() {
return Ok(record);
}
}
if self.closed.load(std::sync::atomic::Ordering::SeqCst) {
return Err(RecvError::Closed);
}
match self.poll(Duration::from_secs(1)).await {
Ok(mut records) if !records.is_empty() => {
let rest = records.split_off(1);
self.requeue_undelivered(rest);
let Some(first) = records.pop() else {
unreachable!("non-empty ConsumerRecords yields at least one element");
};
return Ok(first);
}
Ok(_) => continue,
Err(e) => {
if self.closed.load(std::sync::atomic::Ordering::SeqCst) {
return Err(RecvError::Closed);
}
return Err(RecvError::Error(e));
}
}
}
}
pub async fn batch_recv(
&self,
max_records: usize,
timeout: Duration,
) -> Result<BatchRecvOutcome> {
batch_recv_with(
&self.recv_buffer,
|len| self.metrics.buffered_records.set(len),
max_records,
timeout,
self.config.idle_poll_backoff(),
|| self.closed.load(std::sync::atomic::Ordering::SeqCst),
|| async { self.paused.read().await.clone() },
|remaining| self.poll(remaining),
)
.await
}
#[must_use = "stream does nothing unless polled"]
pub fn stream(&self) -> ConsumerStream<'_> {
ConsumerStream::new(self)
}
pub async fn commit(&self) -> Result<()> {
let offsets_snapshot = {
let offsets = self.offsets.read().await;
if offsets.is_empty() {
debug!("No offsets to commit");
return Ok(());
}
self.committable_snapshot(&offsets)
};
self.metrics.commits.inc();
let assigned_set = if self.group_coordinator.is_some() {
let assignments = self.assignments.read().await;
Some(
assignments
.iter()
.flat_map(|(topic, parts)| parts.iter().map(move |&p| (topic.clone(), p)))
.collect::<HashSet<_>>(),
)
} else {
None
};
let commit_epochs = self.committed_leader_epochs(&offsets_snapshot).await;
let commit_offsets = Self::build_commit_offsets(
&offsets_snapshot,
&commit_epochs,
assigned_set.as_ref(),
self.group_coordinator.is_some(),
)?;
if commit_offsets.is_empty() {
debug!("No assigned partition offsets to commit");
return Ok(());
}
let committed_offsets = Self::build_committed_offsets(&commit_offsets);
if let Some(coordinator) = self.group_coordinator.clone() {
Self::commit_group_offsets_with_retry(
coordinator,
self.interceptor.clone(),
commit_offsets,
committed_offsets,
)
.await
} else {
for ((topic, partition), offset) in &committed_offsets {
debug!("Committed offset for {}-{}: {}", topic, partition, offset);
}
info!(
"Committed {} partition offsets (local only)",
committed_offsets.len()
);
Ok(())
}
}
pub async fn commit_sync(&self) -> Result<()> {
self.commit().await
}
pub fn commit_async(&self) -> OffsetCommitHandle {
let assigned_set = if self.group_coordinator.is_some() {
match self.assignments.try_read() {
Ok(guard) => Some(
guard
.iter()
.flat_map(|(topic, parts)| parts.iter().map(move |&p| (topic.clone(), p)))
.collect::<HashSet<_>>(),
),
Err(_) => {
return OffsetCommitHandle::ready(Err(KrafkaError::invalid_state(
"commit_async: assignments lock contention",
)));
}
}
} else {
None
};
let offsets_snapshot = match self.offsets.try_read() {
Ok(guard) => {
if guard.is_empty() {
return OffsetCommitHandle::ready(Ok(()));
}
self.metrics.commits.inc();
let committable = self.committable_snapshot(&guard);
let epochs = match self.partition_state.try_read() {
Ok(state) => Self::leader_epochs_from_state(&committable, &state),
Err(_) => HashMap::new(),
};
match Self::build_commit_offsets(
&committable,
&epochs,
assigned_set.as_ref(),
self.group_coordinator.is_some(),
) {
Ok(offsets) => offsets,
Err(error) => return OffsetCommitHandle::ready(Err(error)),
}
}
Err(_) => {
return OffsetCommitHandle::ready(Err(KrafkaError::invalid_state(
"commit_async: offset lock contention",
)));
}
};
if offsets_snapshot.is_empty() {
debug!("Async commit: no eligible partition offsets to commit");
return OffsetCommitHandle::ready(Ok(()));
}
let committed_offsets: HashMap<(String, PartitionId), Offset> = offsets_snapshot
.iter()
.map(|((topic, partition), position)| ((topic.clone(), *partition), position.offset))
.collect();
let Some(coordinator) = self.group_coordinator.clone() else {
debug!("Async commit: no group coordinator, offsets stored locally");
return OffsetCommitHandle::ready(Ok(()));
};
OffsetCommitHandle::Task(tokio::spawn(Self::commit_group_offsets_with_retry(
coordinator,
self.interceptor.clone(),
offsets_snapshot,
committed_offsets,
)))
}
fn committable_snapshot(
&self,
offsets: &HashMap<(String, PartitionId), Offset>,
) -> HashMap<(String, PartitionId), Offset> {
let buffer = self.recv_buffer.lock();
committable_positions(offsets, &buffer)
}
async fn committed_leader_epochs(
&self,
offsets: &HashMap<(String, PartitionId), Offset>,
) -> HashMap<(String, PartitionId), i32> {
let state = self.partition_state.read().await;
Self::leader_epochs_from_state(offsets, &state)
}
fn leader_epochs_from_state(
offsets: &HashMap<(String, PartitionId), Offset>,
state: &HashMap<(String, PartitionId), PartitionState>,
) -> HashMap<(String, PartitionId), i32> {
offsets
.keys()
.filter_map(|key| {
state
.get(key)
.and_then(|s| s.last_fetched_epoch)
.filter(|&epoch| epoch >= 0)
.map(|epoch| (key.clone(), epoch))
})
.collect()
}
fn build_commit_offsets(
offsets: &HashMap<(String, PartitionId), Offset>,
epochs: &HashMap<(String, PartitionId), i32>,
assigned_set: Option<&HashSet<(String, PartitionId)>>,
has_group: bool,
) -> Result<CommitRequestOffsets> {
if has_group && assigned_set.is_none() {
return Err(KrafkaError::invalid_state(
"commit_async: assignments snapshot unavailable",
));
}
Ok(offsets
.iter()
.filter(|((topic, partition), _)| {
!has_group
|| assigned_set
.is_some_and(|assigned| assigned.contains(&(topic.clone(), *partition)))
})
.map(|((topic, partition), offset)| {
let key = (topic.clone(), *partition);
let leader_epoch = epochs.get(&key).copied().unwrap_or(-1);
(
key,
CommitPosition {
offset: *offset,
leader_epoch,
metadata: None,
},
)
})
.collect())
}
fn build_committed_offsets(
commit_offsets: &CommitRequestOffsets,
) -> HashMap<(String, PartitionId), Offset> {
commit_offsets
.iter()
.map(|((topic, partition), position)| ((topic.clone(), *partition), position.offset))
.collect()
}
fn filter_commit_with_metadata_offsets(
offsets: HashMap<TopicPartition, OffsetAndMetadata>,
assigned_set: Option<&HashSet<(String, PartitionId)>>,
has_group: bool,
) -> Result<HashMap<TopicPartition, OffsetAndMetadata>> {
if has_group && assigned_set.is_none() {
return Err(KrafkaError::invalid_state(
"commit_with_metadata: assignments snapshot unavailable",
));
}
Ok(offsets
.into_iter()
.filter(|(tp, _)| {
!has_group
|| assigned_set.is_some_and(|assigned| {
assigned.contains(&(tp.topic.clone(), tp.partition))
})
})
.collect())
}
fn build_commit_offsets_with_metadata(
filtered_offsets: &HashMap<TopicPartition, OffsetAndMetadata>,
) -> CommitRequestOffsets {
filtered_offsets
.iter()
.map(|(tp, offset_meta)| {
(
(tp.topic.clone(), tp.partition),
CommitPosition {
offset: offset_meta.offset,
leader_epoch: offset_meta.leader_epoch.unwrap_or(-1),
metadata: offset_meta.metadata.clone(),
},
)
})
.collect()
}
async fn retry_commit_with<F, Fut>(mut commit_once: F) -> Result<()>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<()>>,
{
match commit_once().await {
Ok(()) => Ok(()),
Err(error) if error.is_retriable() => {
let mut last_error = error;
let backoffs = [Duration::from_millis(100), Duration::from_millis(250)];
for delay in &backoffs {
debug!(
"Commit failed with retriable error, retrying in {:?}: {last_error}",
delay
);
tokio::time::sleep(*delay).await;
match commit_once().await {
Ok(()) => return Ok(()),
Err(error) if error.is_retriable() => {
last_error = error;
}
Err(error) => return Err(error),
}
}
Err(last_error)
}
Err(error) => Err(error),
}
}
async fn commit_group_offsets_with_retry(
coordinator: Arc<GroupCoordinator>,
interceptor: Arc<dyn crate::interceptor::ConsumerInterceptor>,
commit_offsets: CommitRequestOffsets,
committed_offsets: HashMap<(String, PartitionId), Offset>,
) -> Result<()> {
let result = Self::retry_commit_with(|| coordinator.commit_offsets(&commit_offsets)).await;
match result {
Ok(()) => {
crate::interceptor::safe_on_commit(&*interceptor, &committed_offsets, None);
Ok(())
}
Err(error) => {
crate::interceptor::safe_on_commit(&*interceptor, &committed_offsets, Some(&error));
Err(error)
}
}
}
pub async fn commit_with_metadata(
&self,
offsets: HashMap<TopicPartition, OffsetAndMetadata>,
) -> Result<()> {
if offsets.is_empty() {
debug!("No offsets to commit");
return Ok(());
}
self.metrics.commits.inc();
let assigned_set = if self.group_coordinator.is_some() {
let assignments = self.assignments.read().await;
Some(
assignments
.iter()
.flat_map(|(topic, parts)| parts.iter().map(move |&p| (topic.clone(), p)))
.collect::<HashSet<_>>(),
)
} else {
None
};
let filtered_offsets = Self::filter_commit_with_metadata_offsets(
offsets,
assigned_set.as_ref(),
self.group_coordinator.is_some(),
)?;
if filtered_offsets.is_empty() {
debug!("No offsets to commit after filtering by assigned partitions");
return Ok(());
}
if let Some(coordinator) = self.group_coordinator.clone() {
let commit_offsets = Self::build_commit_offsets_with_metadata(&filtered_offsets);
let committed_offsets = Self::build_committed_offsets(&commit_offsets);
Self::commit_group_offsets_with_retry(
coordinator,
self.interceptor.clone(),
commit_offsets,
committed_offsets,
)
.await?;
let mut internal_offsets = self.offsets.write().await;
for (tp, offset_meta) in filtered_offsets {
internal_offsets.insert((tp.topic, tp.partition), offset_meta.offset);
}
} else {
for (tp, offset_meta) in &filtered_offsets {
let metadata_str = offset_meta.metadata.as_deref().unwrap_or("<none>");
debug!(
"Committed offset for {}-{}: {} (metadata: {})",
tp.topic, tp.partition, offset_meta.offset, metadata_str
);
}
let count = filtered_offsets.len();
let mut internal_offsets = self.offsets.write().await;
for (tp, offset_meta) in filtered_offsets {
internal_offsets.insert((tp.topic, tp.partition), offset_meta.offset);
}
info!(
"Committed {} partition offsets with metadata (local only)",
count
);
}
self.recompute_lag_metrics().await;
Ok(())
}
pub async fn position(&self, topic: &str, partition: PartitionId) -> Option<Offset> {
let key = (topic.to_string(), partition);
let offsets = self.offsets.read().await;
let fetch_position = offsets.get(&key).copied()?;
let buffer = self.recv_buffer.lock();
let undelivered = lowest_undelivered_offsets(&buffer);
Some(match undelivered.get(&key) {
Some(&first_undelivered) => fetch_position.min(first_undelivered),
None => fetch_position,
})
}
pub async fn fetch_position(&self, topic: &str, partition: PartitionId) -> Option<Offset> {
let offsets = self.offsets.read().await;
offsets.get(&(topic.to_string(), partition)).copied()
}
pub async fn assignment(&self) -> HashMap<String, Vec<PartitionId>> {
let assignments = self.assignments.read().await;
assignments.clone()
}
pub async fn subscription(&self) -> HashSet<String> {
let subscriptions = self.subscriptions.read().await;
subscriptions.clone()
}
pub async fn current_lag(&self, topic: &str, partition: PartitionId) -> Option<u64> {
let key = (topic.to_string(), partition);
let offsets = self.offsets.read().await;
let fetch_position = offsets.get(&key).copied()?;
let partition_state = self.partition_state.read().await;
let end = partition_state
.get(&key)
.and_then(|s| s.readable_end_offset(self.config.isolation_level))?;
let undelivered = lowest_undelivered_offsets(&self.recv_buffer.lock());
let position = match undelivered.get(&key) {
Some(&first_undelivered) => fetch_position.min(first_undelivered),
None => fetch_position,
};
Some((end - position).max(0) as u64)
}
pub async fn lag(&self) -> LagResult {
let offsets = self.offsets.read().await;
let partition_state = self.partition_state.read().await;
let now = Instant::now();
let threshold = self.config.lag_staleness_threshold;
let mut lag = HashMap::with_capacity(partition_state.len());
let mut stale_partitions = Vec::new();
let undelivered = lowest_undelivered_offsets(&self.recv_buffer.lock());
for (key, state) in partition_state.iter() {
if let (Some(end), Some(&fetch_position)) = (
state.readable_end_offset(self.config.isolation_level),
offsets.get(key),
) {
let position = match undelivered.get(key) {
Some(&first_undelivered) => fetch_position.min(first_undelivered),
None => fetch_position,
};
lag.insert(key.clone(), (end - position).max(0) as u64);
let is_stale = state
.watermark_updated_at
.is_none_or(|t| now.saturating_duration_since(t) > threshold);
if is_stale {
stale_partitions.push(key.clone());
}
}
}
LagResult {
lag,
stale_partitions,
}
}
pub async fn cached_beginning_offset(
&self,
topic: &str,
partition: PartitionId,
) -> Option<Offset> {
let key = (topic.to_string(), partition);
self.partition_state
.read()
.await
.get(&key)
.and_then(|s| s.log_start_offset)
}
pub async fn cached_end_offset(&self, topic: &str, partition: PartitionId) -> Option<Offset> {
let key = (topic.to_string(), partition);
self.partition_state
.read()
.await
.get(&key)
.and_then(|s| s.readable_end_offset(self.config.isolation_level))
}
pub async fn cached_high_watermark(
&self,
topic: &str,
partition: PartitionId,
) -> Option<Offset> {
let key = (topic.to_string(), partition);
self.partition_state
.read()
.await
.get(&key)
.and_then(|s| s.high_watermark)
}
pub async fn cached_last_stable_offset(
&self,
topic: &str,
partition: PartitionId,
) -> Option<Offset> {
let key = (topic.to_string(), partition);
self.partition_state
.read()
.await
.get(&key)
.and_then(|s| s.last_stable_offset)
}
pub async fn fetch_end_offset(&self, topic: &str, partition: PartitionId) -> Result<Offset> {
validate_topic_name(topic)?;
self.resolve_list_offset(topic, partition, -1).await
}
pub async fn is_caught_up(&self) -> bool {
let assignments = self.assignments.read().await;
if assignments.is_empty() {
return true; }
let offsets = self.offsets.read().await;
let partition_state = self.partition_state.read().await;
let undelivered = lowest_undelivered_offsets(&self.recv_buffer.lock());
for (topic, partitions) in assignments.iter() {
for &partition in partitions {
let key = (topic.clone(), partition);
let Some(end) = partition_state
.get(&key)
.and_then(|s| s.readable_end_offset(self.config.isolation_level))
else {
return false;
};
let fetch_position = offsets.get(&key).copied().unwrap_or(0);
let position = match undelivered.get(&key) {
Some(&first_undelivered) => fetch_position.min(first_undelivered),
None => fetch_position,
};
if position < end {
return false;
}
}
}
true
}
pub async fn unsubscribe(&self) -> Result<()> {
if self.config.enable_auto_commit
&& self.group_coordinator.is_some()
&& let Err(e) = self.commit().await
{
warn!("Commit during unsubscribe failed: {e}");
}
let revoked: Vec<TopicPartition> = {
let assignments = self.assignments.read().await;
assignments
.iter()
.flat_map(|(t, ps)| ps.iter().map(move |&p| TopicPartition::new(t, p)))
.collect()
};
if !revoked.is_empty() {
self.safe_on_partitions_revoked(&revoked).await;
}
let leave_group_result = if let Some(ref coordinator) = self.group_coordinator {
coordinator.leave_group().await
} else {
Ok(())
};
self.subscriptions.write().await.clear();
self.assignments.write().await.clear();
self.clear_partition_state().await;
self.metrics.assigned_partitions.set(0);
debug!("Unsubscribed from all topics");
leave_group_result
}
pub async fn pause(&self, topic: &str, partitions: &[PartitionId]) {
let mut paused = self.paused.write().await;
let topic_owned = topic.to_string();
for &partition in partitions {
paused.insert((topic_owned.clone(), partition));
}
self.metrics.paused_partitions.set(paused.len() as u64);
debug!("Paused partitions for {}: {:?}", topic, partitions);
}
pub async fn resume(&self, topic: &str, partitions: &[PartitionId]) {
let mut paused = self.paused.write().await;
let topic_key = topic.to_string();
for &partition in partitions {
paused.remove(&(topic_key.clone(), partition));
}
self.metrics.paused_partitions.set(paused.len() as u64);
debug!("Resumed partitions for {}: {:?}", topic, partitions);
}
pub async fn paused_partitions(&self) -> HashSet<(String, PartitionId)> {
self.paused.read().await.clone()
}
pub fn update_seed_brokers(&self, servers: Vec<String>) -> Result<()> {
self.metadata.update_seed_brokers(servers)
}
pub async fn refresh_tls(&self) -> Result<()> {
self.pool.refresh_tls().await
}
pub async fn rebootstrap(&self) {
self.metadata.rebootstrap().await;
}
fn select_close_result(
auto_commit_result: Result<()>,
leave_group_result: Result<()>,
) -> Result<()> {
match auto_commit_result {
Err(error) if Self::should_ignore_close_auto_commit_error(&error) => leave_group_result,
Ok(()) => leave_group_result,
Err(error) => Err(error),
}
}
fn should_ignore_close_auto_commit_error(error: &KrafkaError) -> bool {
matches!(
error,
KrafkaError::Broker {
code: crate::error::ErrorCode::UnknownMemberId
| crate::error::ErrorCode::IllegalGeneration
| crate::error::ErrorCode::RebalanceInProgress
| crate::error::ErrorCode::FencedMemberEpoch
| crate::error::ErrorCode::StaleMemberEpoch,
..
}
)
}
pub async fn close(&self) -> Result<()> {
if self.closed.swap(true, std::sync::atomic::Ordering::SeqCst) {
return Ok(());
}
let auto_commit_result = if self.config.enable_auto_commit {
self.commit().await
} else {
Ok(())
};
let revoked: Vec<TopicPartition> = {
let assignments = self.assignments.read().await;
assignments
.iter()
.flat_map(|(t, ps)| ps.iter().map(move |&p| TopicPartition::new(t, p)))
.collect()
};
if !revoked.is_empty() {
self.safe_on_partitions_revoked(&revoked).await;
}
let leave_group_result = if let Some(ref coordinator) = self.group_coordinator {
coordinator.leave_group().await
} else {
Ok(())
};
self.subscriptions.write().await.clear();
self.assignments.write().await.clear();
self.clear_partition_state().await;
self.metrics.assigned_partitions.set(0);
crate::interceptor::safe_consumer_close(&*self.interceptor);
if self.pool_owned {
self.pool.close_all().await;
info!("Consumer closed (connection pool torn down)");
} else {
info!("Consumer closed (shared connection pool left open)");
}
Self::select_close_result(auto_commit_result, leave_group_result)
}
#[inline]
#[must_use]
pub fn owns_pool(&self) -> bool {
self.pool_owned
}
#[inline]
pub fn is_closed(&self) -> bool {
self.closed.load(std::sync::atomic::Ordering::SeqCst)
}
pub async fn committed(
&self,
partitions: &[(&str, PartitionId)],
) -> Result<HashMap<(String, PartitionId), CommittedPosition>> {
if self.is_closed() {
return Err(KrafkaError::invalid_state("consumer is closed"));
}
let Some(ref coordinator) = self.group_coordinator else {
return Err(KrafkaError::invalid_state(
"committed() requires a group_id; an assign-only consumer has no \
coordinator to read committed offsets from",
));
};
if partitions.is_empty() {
return Ok(HashMap::new());
}
let mut by_topic: HashMap<String, Vec<PartitionId>> = HashMap::new();
for (topic, partition) in partitions {
validate_topic_name(topic)?;
by_topic
.entry((*topic).to_string())
.or_default()
.push(*partition);
}
coordinator.fetch_committed_offsets(&by_topic).await
}
#[inline]
pub fn wakeup(&self) {
self.wakeup_flag
.store(true, std::sync::atomic::Ordering::Release);
self.wakeup_notify.notify_waiters();
}
pub async fn group_metadata(&self) -> Option<ConsumerGroupMetadata> {
self.group_coordinator.as_ref()?.group_metadata().await
}
#[inline]
pub fn group_coordinator(&self) -> Option<&Arc<GroupCoordinator>> {
self.group_coordinator.as_ref()
}
#[inline]
pub fn metrics(&self) -> &Arc<ConsumerMetrics> {
&self.metrics
}
#[inline]
pub fn connection_metrics(&self) -> Arc<ConnectionMetrics> {
self.pool.metrics()
}
}
impl Drop for Consumer {
fn drop(&mut self) {
if !self.closed.load(std::sync::atomic::Ordering::SeqCst) && !std::thread::panicking() {
warn!(
"Consumer dropped without close(); group rebalance will be delayed \
until session.timeout.ms. Call `Consumer::close()` before drop."
);
}
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
use super::*;
use tokio::sync::RwLock;
#[tokio::test]
async fn test_consumer_builder_no_servers() {
let result = Consumer::builder().build().await;
assert!(result.is_err());
}
#[test]
fn test_paused_partitions_set() {
let mut paused: HashSet<(String, PartitionId)> = HashSet::new();
paused.insert(("topic1".to_string(), 0));
paused.insert(("topic1".to_string(), 1));
paused.insert(("topic2".to_string(), 0));
assert!(paused.contains(&("topic1".to_string(), 0)));
assert!(paused.contains(&("topic1".to_string(), 1)));
assert!(paused.contains(&("topic2".to_string(), 0)));
assert!(!paused.contains(&("topic2".to_string(), 1)));
paused.remove(&("topic1".to_string(), 0));
assert!(!paused.contains(&("topic1".to_string(), 0)));
}
#[test]
fn test_topic_partition() {
let tp = TopicPartition::new("my-topic", 3);
assert_eq!(tp.topic(), "my-topic");
assert_eq!(tp.partition(), 3);
let mut map = HashMap::new();
map.insert(TopicPartition::new("test", 0), 100i64);
map.insert(TopicPartition::new("test", 1), 200i64);
assert_eq!(map.get(&TopicPartition::new("test", 0)), Some(&100i64));
assert_eq!(map.get(&TopicPartition::new("test", 1)), Some(&200i64));
}
#[test]
fn test_offset_and_metadata() {
let offset = OffsetAndMetadata::new(100);
assert_eq!(offset.offset, 100);
assert!(offset.metadata.is_none());
let offset_with_meta = OffsetAndMetadata::with_metadata(200, "checkpoint-123");
assert_eq!(offset_with_meta.offset, 200);
assert_eq!(offset_with_meta.metadata.as_deref(), Some("checkpoint-123"));
let offset_with_epoch = OffsetAndMetadata::with_epoch(300, 5);
assert_eq!(offset_with_epoch.offset, 300);
assert_eq!(offset_with_epoch.leader_epoch, Some(5));
}
#[test]
fn test_partition_assignment_strategy_default() {
let config = ConsumerConfig::default();
assert_eq!(
config.partition_assignment_strategy(),
PartitionAssignmentStrategy::Range
);
}
#[test]
fn test_partition_assignment_strategy_protocol_name() {
assert_eq!(PartitionAssignmentStrategy::Range.protocol_name(), "range");
assert_eq!(
PartitionAssignmentStrategy::RoundRobin.protocol_name(),
"roundrobin"
);
assert_eq!(
PartitionAssignmentStrategy::CooperativeSticky.protocol_name(),
"cooperative-sticky"
);
}
#[test]
fn test_consumer_config_defaults() {
let config = ConsumerConfig::default();
assert!(config.fetch_max_bytes > 0);
assert!(config.fetch_min_bytes > 0);
assert!(config.max_partition_fetch_bytes > 0);
}
#[tokio::test]
async fn test_consumer_builder_rejects_bad_heartbeat() {
let result = Consumer::builder()
.bootstrap_servers("localhost:9092")
.group_id("test")
.session_timeout(Duration::from_secs(5))
.heartbeat_interval(Duration::from_secs(5))
.build()
.await;
match result {
Err(e) => assert!(e.to_string().contains("heartbeat_interval")),
Ok(_) => panic!("expected error for heartbeat_interval >= session_timeout"),
}
}
#[tokio::test]
async fn test_consumer_builder_rejects_heartbeat_greater_than_session() {
let result = Consumer::builder()
.bootstrap_servers("localhost:9092")
.group_id("test")
.session_timeout(Duration::from_secs(5))
.heartbeat_interval(Duration::from_secs(10))
.build()
.await;
assert!(result.is_err());
}
#[test]
fn test_max_poll_records_offset_recomputation() {
let records: Vec<ConsumerRecord> = (0..5)
.map(|i| ConsumerRecord {
topic: "topic1".to_string(),
partition: 0,
offset: 100 + i,
timestamp: 0,
timestamp_type: 0,
key: None,
value: Some(bytes::Bytes::from(format!("val-{i}"))),
headers: vec![],
leader_epoch: None,
delivery_count: None,
})
.collect();
let original_offset_updates: Vec<((String, PartitionId), Offset)> =
vec![(("topic1".to_string(), 0), 105)];
let max = 3usize;
let mut truncated = records;
truncated.truncate(max);
let mut delivered_offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
for r in &truncated {
let key = (r.topic.clone(), r.partition);
let entry = delivered_offsets.entry(key).or_insert(r.offset);
if r.offset > *entry {
*entry = r.offset;
}
}
let new_offset_updates: Vec<_> = delivered_offsets
.into_iter()
.map(|(key, offset)| (key, offset + 1))
.collect();
assert_eq!(new_offset_updates.len(), 1);
let (key, offset) = &new_offset_updates[0];
assert_eq!(key, &("topic1".to_string(), 0));
assert_eq!(*offset, 103);
assert_ne!(*offset, original_offset_updates[0].1);
}
#[test]
fn test_max_poll_records_multi_partition_offset() {
let mut records = Vec::new();
for i in 0..3 {
records.push(ConsumerRecord {
topic: "topic1".to_string(),
partition: 0,
offset: 50 + i,
timestamp: 0,
timestamp_type: 0,
key: None,
value: Some(bytes::Bytes::from("val")),
headers: vec![],
leader_epoch: None,
delivery_count: None,
});
}
for i in 0..3 {
records.push(ConsumerRecord {
topic: "topic1".to_string(),
partition: 1,
offset: 200 + i,
timestamp: 0,
timestamp_type: 0,
key: None,
value: Some(bytes::Bytes::from("val")),
headers: vec![],
leader_epoch: None,
delivery_count: None,
});
}
records.truncate(4);
let mut delivered_offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
for r in &records {
let key = (r.topic.clone(), r.partition);
let entry = delivered_offsets.entry(key).or_insert(r.offset);
if r.offset > *entry {
*entry = r.offset;
}
}
assert_eq!(
*delivered_offsets.get(&("topic1".to_string(), 0)).unwrap(),
52
);
assert_eq!(
*delivered_offsets.get(&("topic1".to_string(), 1)).unwrap(),
200
);
}
#[tokio::test]
async fn test_recv_buffer_returns_all_records() {
use std::collections::VecDeque;
let mut buffer = VecDeque::new();
buffer.push_back(ConsumerRecord {
topic: "t".into(),
partition: 0,
offset: 1,
timestamp: 0,
timestamp_type: 0,
key: None,
value: Some(bytes::Bytes::from("r1")),
headers: vec![],
leader_epoch: None,
delivery_count: None,
});
buffer.push_back(ConsumerRecord {
topic: "t".into(),
partition: 0,
offset: 2,
timestamp: 0,
timestamp_type: 0,
key: None,
value: Some(bytes::Bytes::from("r2")),
headers: vec![],
leader_epoch: None,
delivery_count: None,
});
assert_eq!(buffer.len(), 2);
let first = buffer.pop_front().unwrap();
assert_eq!(first.offset, 1);
let second = buffer.pop_front().unwrap();
assert_eq!(second.offset, 2);
assert!(buffer.is_empty());
}
#[test]
fn test_subscribe_replaces_subscriptions() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(async {
let subs: RwLock<HashSet<String>> = RwLock::new(HashSet::new());
{
let mut s = subs.write().await;
s.clear(); s.insert("topic1".to_string());
}
assert_eq!(subs.read().await.len(), 1);
assert!(subs.read().await.contains("topic1"));
{
let mut s = subs.write().await;
s.clear(); s.insert("topic2".to_string());
}
assert_eq!(subs.read().await.len(), 1);
assert!(subs.read().await.contains("topic2"));
assert!(!subs.read().await.contains("topic1"));
});
}
#[test]
fn test_unsubscribe_clears_all_state() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(async {
let offsets: RwLock<HashMap<(String, PartitionId), Offset>> =
RwLock::new(HashMap::new());
let paused: RwLock<HashSet<(String, PartitionId)>> = RwLock::new(HashSet::new());
let assignments: RwLock<HashMap<String, Vec<PartitionId>>> =
RwLock::new(HashMap::new());
let recv_buffer: RwLock<std::collections::VecDeque<ConsumerRecord>> =
RwLock::new(std::collections::VecDeque::new());
offsets.write().await.insert(("t".into(), 0), 100);
paused.write().await.insert(("t".into(), 0));
assignments.write().await.insert("t".into(), vec![0]);
recv_buffer.write().await.push_back(ConsumerRecord {
topic: "t".into(),
partition: 0,
offset: 0,
timestamp: 0,
timestamp_type: 0,
key: None,
value: None,
headers: vec![],
leader_epoch: None,
delivery_count: None,
});
offsets.write().await.clear();
paused.write().await.clear();
assignments.write().await.clear();
recv_buffer.write().await.clear();
assert!(offsets.read().await.is_empty());
assert!(paused.read().await.is_empty());
assert!(assignments.read().await.is_empty());
assert!(recv_buffer.read().await.is_empty());
});
}
#[test]
fn test_fetch_skips_untracked_partitions() {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(async {
let offsets: RwLock<HashMap<(String, PartitionId), Offset>> =
RwLock::new(HashMap::new());
offsets.write().await.insert(("t".into(), 0), 42);
let o = offsets.read().await;
assert_eq!(o.get(&("t".to_string(), 0)).copied(), Some(42));
assert_eq!(o.get(&("t".to_string(), 1)).copied(), None);
});
}
fn buffered(topic: &str, partition: PartitionId, offset: Offset) -> ConsumerRecord {
ConsumerRecord {
topic: topic.to_string(),
partition,
offset,
timestamp: 0,
timestamp_type: 0,
key: None,
value: None,
headers: Vec::new(),
leader_epoch: None,
delivery_count: None,
}
}
#[test]
fn purging_on_reposition_prevents_a_backwards_commit() {
let mut buffer: std::collections::VecDeque<ConsumerRecord> =
[buffered("orders", 0, 100), buffered("orders", 0, 101)]
.into_iter()
.collect();
let positions: HashMap<(String, PartitionId), Offset> =
[(("orders".to_string(), 0), 5_000)].into_iter().collect();
let clamped = committable_positions(&positions, &buffer);
assert_eq!(
clamped.get(&("orders".to_string(), 0)).copied(),
Some(100),
"this is the bug: the commit would move the group backwards"
);
let repositioned: HashSet<(String, PartitionId)> =
[("orders".to_string(), 0)].into_iter().collect();
assert_eq!(purge_buffered_records(&mut buffer, &repositioned), 2);
let clamped = committable_positions(&positions, &buffer);
assert_eq!(
clamped.get(&("orders".to_string(), 0)).copied(),
Some(5_000)
);
}
#[test]
fn purging_leaves_other_partitions_untouched() {
let mut buffer: std::collections::VecDeque<ConsumerRecord> = [
buffered("orders", 0, 10),
buffered("orders", 1, 20),
buffered("payments", 0, 30),
]
.into_iter()
.collect();
let repositioned: HashSet<(String, PartitionId)> =
[("orders".to_string(), 0)].into_iter().collect();
assert_eq!(purge_buffered_records(&mut buffer, &repositioned), 1);
assert_eq!(buffer.len(), 2);
assert!(
buffer
.iter()
.all(|r| !(r.topic == "orders" && r.partition == 0))
);
assert_eq!(purge_buffered_records(&mut buffer, &HashSet::new()), 0);
assert_eq!(buffer.len(), 2);
}
#[test]
fn undelivered_boundary_is_shared_by_position_lag_and_commit() {
let buffer: std::collections::VecDeque<ConsumerRecord> = [
buffered("orders", 0, 120),
buffered("orders", 0, 118),
buffered("orders", 1, 7),
]
.into_iter()
.collect();
let undelivered = lowest_undelivered_offsets(&buffer);
assert_eq!(undelivered.len(), 2, "one entry per distinct partition");
assert_eq!(
undelivered.get(&("orders".to_string(), 0)).copied(),
Some(118),
"the *lowest* buffered offset is the boundary, not the first seen"
);
assert_eq!(
undelivered.get(&("orders".to_string(), 1)).copied(),
Some(7)
);
let positions: HashMap<(String, PartitionId), Offset> = [
(("orders".to_string(), 0), 200),
(("orders".to_string(), 1), 10),
]
.into_iter()
.collect();
let committable = committable_positions(&positions, &buffer);
assert_eq!(
committable.get(&("orders".to_string(), 0)).copied(),
Some(118)
);
let partition_state: HashMap<(String, PartitionId), PartitionState> = [(
("orders".to_string(), 0),
PartitionState {
high_watermark: Some(200),
..PartitionState::default()
},
)]
.into_iter()
.collect();
let (with_buffer, _) = compute_aggregate_lag(
&positions,
&partition_state,
&undelivered,
IsolationLevel::ReadUncommitted,
);
assert_eq!(
with_buffer, 82,
"records parked in the buffer must count as lag"
);
let (drained, _) = compute_aggregate_lag(
&positions,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadUncommitted,
);
assert_eq!(drained, 0, "with nothing parked, the consumer is caught up");
}
#[test]
fn undelivered_boundary_is_empty_for_an_empty_buffer() {
let buffer: std::collections::VecDeque<ConsumerRecord> = std::collections::VecDeque::new();
assert!(lowest_undelivered_offsets(&buffer).is_empty());
}
#[test]
fn drain_withholds_paused_partitions_without_dropping_them() {
let mut buffer: std::collections::VecDeque<ConsumerRecord> = [
buffered("orders", 0, 1),
buffered("orders", 1, 2),
buffered("orders", 0, 3),
]
.into_iter()
.collect();
let paused: HashSet<(String, PartitionId)> =
[("orders".to_string(), 0)].into_iter().collect();
let mut batch = Vec::new();
drain_buffered_records(&mut buffer, &mut batch, 10, &paused);
assert_eq!(batch.len(), 1);
assert_eq!(batch[0].partition, 1);
assert_eq!(buffer.len(), 2);
assert!(buffer.iter().all(|r| r.partition == 0));
}
#[test]
fn drain_fast_path_preserves_order_and_respects_max_records() {
let mut buffer: std::collections::VecDeque<ConsumerRecord> =
(0..5).map(|i| buffered("orders", 0, i)).collect();
let mut batch = Vec::new();
drain_buffered_records(&mut buffer, &mut batch, 3, &HashSet::new());
assert_eq!(
batch.iter().map(|r| r.offset).collect::<Vec<_>>(),
vec![0, 1, 2]
);
assert_eq!(buffer.len(), 2);
}
#[test]
fn readable_end_offset_follows_the_isolation_level() {
let state = PartitionState {
high_watermark: Some(1_000),
last_stable_offset: Some(400),
..PartitionState::default()
};
assert_eq!(
state.readable_end_offset(IsolationLevel::ReadCommitted),
Some(400)
);
assert_eq!(
state.readable_end_offset(IsolationLevel::ReadUncommitted),
Some(1_000)
);
let offsets: HashMap<(String, PartitionId), Offset> =
[(("orders".to_string(), 0), 400)].into_iter().collect();
let partition_state: HashMap<(String, PartitionId), PartitionState> =
[(("orders".to_string(), 0), state)].into_iter().collect();
let (total, max) = compute_aggregate_lag(
&offsets,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadCommitted,
);
assert_eq!(
(total, max),
(0, 0),
"a drained read_committed consumer must not report phantom lag"
);
let (total, _) = compute_aggregate_lag(
&offsets,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadUncommitted,
);
assert_eq!(total, 600);
}
#[test]
fn readable_end_offset_falls_back_to_the_high_watermark() {
let state = PartitionState {
high_watermark: Some(1_000),
last_stable_offset: None,
..PartitionState::default()
};
assert_eq!(
state.readable_end_offset(IsolationLevel::ReadCommitted),
Some(1_000)
);
}
#[test]
fn record_budget_is_claimed_in_batches_and_never_underflows() {
use std::sync::atomic::{AtomicUsize, Ordering};
let budget = AtomicUsize::new(500);
assert_eq!(claim_record_budget(&budget, 200), 200);
assert_eq!(budget.load(Ordering::Relaxed), 300);
assert_eq!(claim_record_budget(&budget, 1_000), 300);
assert_eq!(budget.load(Ordering::Relaxed), 0);
assert_eq!(claim_record_budget(&budget, 1), 0);
assert_eq!(budget.load(Ordering::Relaxed), 0);
budget.fetch_add(50, Ordering::Relaxed);
assert_eq!(claim_record_budget(&budget, 80), 50);
}
#[test]
fn control_batch_type_is_read_from_the_marker_key() {
use crate::protocol::{Record, RecordBatch};
let marker = |control_type: i16| {
let mut key = bytes::BytesMut::new();
bytes::BufMut::put_i16(&mut key, 0); bytes::BufMut::put_i16(&mut key, control_type);
let mut batch = RecordBatch::new();
batch.attributes.is_control_batch = true;
batch.add_record(Record {
attributes: 0,
timestamp_delta: 0,
offset_delta: 0,
key: Some(key.freeze()),
value: Some(bytes::Bytes::new()),
headers: Vec::new(),
});
batch
};
assert!(control_batch_is_abort(&marker(0)), "type 0 is ABORT");
assert!(!control_batch_is_abort(&marker(1)), "type 1 is COMMIT");
let mut short = RecordBatch::new();
short.attributes.is_control_batch = true;
short.add_record(Record {
attributes: 0,
timestamp_delta: 0,
offset_delta: 0,
key: Some(bytes::Bytes::from_static(&[0, 0])),
value: None,
headers: Vec::new(),
});
assert!(!control_batch_is_abort(&short));
let mut keyless = RecordBatch::new();
keyless.attributes.is_control_batch = true;
assert!(!control_batch_is_abort(&keyless));
}
const WALK_EPOCH: i32 = 7;
fn walk_record(offset_delta: i32) -> crate::protocol::Record {
crate::protocol::Record {
attributes: 0,
timestamp_delta: 0,
offset_delta,
key: None,
value: Some(bytes::Bytes::from_static(b"v")),
headers: Vec::new(),
}
}
fn walk_batch(base_offset: i64, last_offset_delta: i32, deltas: &[i32]) -> RecordBatch {
let mut batch = RecordBatch::new();
batch.base_offset = base_offset;
batch.last_offset_delta = last_offset_delta;
batch.partition_leader_epoch = WALK_EPOCH;
for &delta in deltas {
batch.add_record(walk_record(delta));
}
batch
}
fn walk(
batches: &[RecordBatch],
fetch_offset: Offset,
budget: Option<&std::sync::atomic::AtomicUsize>,
) -> (Vec<ConsumerRecord>, PartitionDecodeOutcome) {
let mut buf = bytes::BytesMut::new();
for batch in batches {
buf.extend_from_slice(&batch.encode().expect("encode batch"));
}
let mut records = Vec::new();
let outcome = decode_partition_batches(
"walk-topic",
0,
buf.freeze(),
fetch_offset,
Vec::new(),
budget,
RecordBatch::MAX_DECOMPRESSED_SIZE,
&mut records,
);
(records, outcome)
}
#[test]
fn compaction_emptied_batch_is_skipped_not_a_stall() {
let (records, outcome) = walk(&[walk_batch(1000, 4, &[])], 1000, None);
assert!(records.is_empty(), "an emptied batch delivers nothing");
assert_eq!(
outcome.last_offset,
Some(1004),
"the position must advance through the emptied batch's span, \
or the next fetch re-reads it forever"
);
assert_eq!(outcome.last_epoch, WALK_EPOCH);
assert!(outcome.error.is_none());
assert!(!outcome.corrupt);
}
#[test]
fn compaction_emptied_batch_does_not_discard_the_rest_of_the_response() {
let batches = [walk_batch(1000, 4, &[]), walk_batch(1005, 1, &[0, 1])];
for budget in [None, Some(std::sync::atomic::AtomicUsize::new(512))] {
let (records, outcome) = walk(&batches, 1000, budget.as_ref());
let offsets: Vec<Offset> = records.iter().map(|r| r.offset).collect();
assert_eq!(offsets, vec![1005, 1006]);
assert_eq!(outcome.last_offset, Some(1006));
if let Some(budget) = budget {
assert_eq!(
budget.load(std::sync::atomic::Ordering::Relaxed),
510,
"the emptied batch must not consume budget slots"
);
}
}
}
#[test]
fn drained_compacted_batch_advances_past_its_removed_tail() {
let (records, outcome) = walk(&[walk_batch(100, 9, &[0, 3, 5])], 100, None);
let offsets: Vec<Offset> = records.iter().map(|r| r.offset).collect();
assert_eq!(offsets, vec![100, 103, 105]);
assert!(records.iter().all(|r| r.leader_epoch == Some(WALK_EPOCH)));
assert_eq!(
outcome.last_offset,
Some(109),
"a drained batch advances to its span end (Java client's nextFetchOffset)"
);
}
#[test]
fn position_inside_a_compacted_batch_escapes_it() {
let (records, outcome) = walk(&[walk_batch(100, 9, &[0, 3, 5])], 106, None);
assert!(
records.is_empty(),
"everything in the batch was already delivered"
);
assert_eq!(
outcome.last_offset,
Some(109),
"the position must escape the straddling batch"
);
assert_eq!(outcome.last_epoch, WALK_EPOCH);
}
#[test]
fn budget_cut_batch_re_fetches_its_remainder() {
let budget = std::sync::atomic::AtomicUsize::new(2);
let (records, outcome) = walk(&[walk_batch(100, 2, &[0, 1, 2])], 100, Some(&budget));
let offsets: Vec<Offset> = records.iter().map(|r| r.offset).collect();
assert_eq!(offsets, vec![100, 101]);
assert_eq!(
outcome.last_offset,
Some(101),
"a budget-cut batch must not be skipped over"
);
assert_eq!(budget.load(std::sync::atomic::Ordering::Relaxed), 0);
}
#[test]
fn batch_below_the_fetch_position_cannot_rewind_it() {
let (records, outcome) = walk(&[walk_batch(100, 4, &[])], 200, None);
assert!(records.is_empty());
assert_eq!(
outcome.last_offset, None,
"no position update at all beats a backwards one"
);
}
#[test]
fn truncated_tail_after_a_skipped_batch_is_benign() {
let mut buf = bytes::BytesMut::new();
buf.extend_from_slice(&walk_batch(1000, 4, &[]).encode().expect("encode"));
let tail = walk_batch(1005, 1, &[0, 1]).encode().expect("encode");
buf.extend_from_slice(&tail[..tail.len() - 5]);
let mut records = Vec::new();
let outcome = decode_partition_batches(
"walk-topic",
0,
buf.freeze(),
1000,
Vec::new(),
None,
RecordBatch::MAX_DECOMPRESSED_SIZE,
&mut records,
);
assert!(records.is_empty());
assert_eq!(outcome.last_offset, Some(1004));
assert!(outcome.error.is_none(), "a truncated tail is not a fault");
assert!(!outcome.corrupt);
}
#[test]
fn corrupt_first_batch_reports_a_fault() {
let mut bad = bytes::BytesMut::from(
walk_batch(1000, 1, &[0, 1])
.encode()
.expect("encode")
.as_ref(),
);
bad[30] ^= 0xFF;
let mut records = Vec::new();
let outcome = decode_partition_batches(
"walk-topic",
0,
bad.freeze(),
1000,
Vec::new(),
None,
RecordBatch::MAX_DECOMPRESSED_SIZE,
&mut records,
);
assert!(records.is_empty());
assert_eq!(outcome.last_offset, None);
assert!(
outcome.error.is_some(),
"an undecodable position is a reported fault"
);
assert!(outcome.corrupt);
}
#[test]
fn aborted_transaction_batches_are_filtered_but_advance_the_position() {
let mut aborted_data = walk_batch(100, 2, &[0, 1, 2]);
aborted_data.attributes.is_transactional = true;
aborted_data.producer_id = 9;
let mut marker_key = bytes::BytesMut::new();
bytes::BufMut::put_i16(&mut marker_key, 0); bytes::BufMut::put_i16(&mut marker_key, 0); let mut abort_marker = walk_batch(103, 0, &[]);
abort_marker.attributes.is_control_batch = true;
abort_marker.producer_id = 9;
abort_marker.add_record(crate::protocol::Record {
attributes: 0,
timestamp_delta: 0,
offset_delta: 0,
key: Some(marker_key.freeze()),
value: Some(bytes::Bytes::new()),
headers: Vec::new(),
});
let mut committed_data = walk_batch(104, 1, &[0, 1]);
committed_data.attributes.is_transactional = true;
committed_data.producer_id = 9;
let mut buf = bytes::BytesMut::new();
for batch in [&aborted_data, &abort_marker, &committed_data] {
buf.extend_from_slice(&batch.encode().expect("encode"));
}
let mut records = Vec::new();
let outcome = decode_partition_batches(
"walk-topic",
0,
buf.freeze(),
100,
vec![crate::protocol::AbortedTransaction {
producer_id: 9,
first_offset: 100,
}],
None,
RecordBatch::MAX_DECOMPRESSED_SIZE,
&mut records,
);
let offsets: Vec<Offset> = records.iter().map(|r| r.offset).collect();
assert_eq!(
offsets,
vec![104, 105],
"only the committed transaction is delivered"
);
assert_eq!(outcome.last_offset, Some(105));
}
#[test]
fn advance_through_batch_never_moves_backwards() {
let mut last_offset = Some(500);
let mut last_epoch = 3;
advance_through_batch(&mut last_offset, &mut last_epoch, 100, 400, 9);
assert_eq!(last_offset, Some(500));
assert_eq!(last_epoch, 3);
advance_through_batch(&mut last_offset, &mut last_epoch, 700, 600, 9);
assert_eq!(last_offset, Some(500));
advance_through_batch(&mut last_offset, &mut last_epoch, 100, 600, 9);
assert_eq!(last_offset, Some(600));
assert_eq!(last_epoch, 9);
}
#[test]
fn test_commit_filter_does_not_leak_stale_offsets() {
let offsets: HashMap<(String, PartitionId), Offset> = [
(("topic1".into(), 0), 100),
(("topic2".into(), 0), 200), ]
.into_iter()
.collect();
let assigned_set: HashSet<(String, PartitionId)> = HashSet::new();
let no_epochs = HashMap::new();
let filtered =
Consumer::build_commit_offsets(&offsets, &no_epochs, Some(&assigned_set), true)
.expect("empty assigned set is valid and must filter everything");
assert!(filtered.is_empty());
}
#[test]
fn test_commit_filter_requires_assignment_snapshot_for_group_commit() {
let offsets: HashMap<(String, PartitionId), Offset> =
[(("topic1".into(), 0), 100)].into_iter().collect();
let no_epochs = HashMap::new();
let error = Consumer::build_commit_offsets(&offsets, &no_epochs, None, true)
.expect_err("group commits require an assignment snapshot");
assert!(
error
.to_string()
.contains("assignments snapshot unavailable")
);
}
#[test]
fn test_commit_with_metadata_filter_does_not_leak_stale_offsets() {
let offsets: HashMap<TopicPartition, OffsetAndMetadata> = [
(
TopicPartition::new("topic1", 0),
OffsetAndMetadata::with_metadata(100, "keep"),
),
(
TopicPartition::new("topic2", 0),
OffsetAndMetadata::with_metadata(200, "stale"),
),
]
.into_iter()
.collect();
let assigned_set: HashSet<(String, PartitionId)> = HashSet::new();
let filtered =
Consumer::filter_commit_with_metadata_offsets(offsets, Some(&assigned_set), true)
.expect("empty assigned set is valid and must filter everything");
assert!(filtered.is_empty());
}
#[test]
fn test_commit_with_metadata_filter_keeps_all_offsets_without_group() {
let offsets: HashMap<TopicPartition, OffsetAndMetadata> = [
(
TopicPartition::new("topic1", 0),
OffsetAndMetadata::new(100),
),
(
TopicPartition::new("topic2", 1),
OffsetAndMetadata::new(200),
),
]
.into_iter()
.collect();
let filtered = Consumer::filter_commit_with_metadata_offsets(offsets, None, false)
.expect("standalone consumers should commit all provided offsets");
assert_eq!(filtered.len(), 2);
}
#[tokio::test]
async fn test_offset_commit_handle_ready_flattens_result() {
OffsetCommitHandle::ready(Ok(()))
.await
.expect("ready ok result");
let error = OffsetCommitHandle::ready(Err(KrafkaError::invalid_state("boom")))
.await
.expect_err("ready error must surface");
assert!(error.to_string().contains("boom"));
}
#[tokio::test]
async fn test_offset_commit_handle_flattens_task_result() {
let error = OffsetCommitHandle::Task(tokio::spawn(async {
Err(KrafkaError::invalid_state("task failed"))
}))
.await
.expect_err("task error must surface");
assert!(error.to_string().contains("task failed"));
}
#[tokio::test]
async fn test_retry_commit_with_succeeds_after_retriable_errors() {
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
Consumer::retry_commit_with({
let attempts = attempts.clone();
move || {
let attempts = attempts.clone();
async move {
let attempt = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
if attempt < 2 {
Err(KrafkaError::broker(
crate::error::ErrorCode::CoordinatorLoadInProgress,
"retry",
))
} else {
Ok(())
}
}
}
})
.await
.expect("retriable errors should eventually succeed");
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 3);
}
#[tokio::test]
async fn test_retry_commit_with_returns_last_retriable_error_after_exhaustion() {
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let error = Consumer::retry_commit_with({
let attempts = attempts.clone();
move || {
let attempts = attempts.clone();
async move {
attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Err(KrafkaError::broker(
crate::error::ErrorCode::CoordinatorLoadInProgress,
"retry",
))
}
}
})
.await
.expect_err("exhausted retriable errors must surface the final error");
assert!(matches!(
error,
KrafkaError::Broker {
code: crate::error::ErrorCode::CoordinatorLoadInProgress,
..
}
));
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 3);
}
#[tokio::test]
async fn test_retry_commit_with_stops_on_non_retriable_error() {
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let error = Consumer::retry_commit_with({
let attempts = attempts.clone();
move || {
let attempts = attempts.clone();
async move {
attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Err(KrafkaError::broker(
crate::error::ErrorCode::GroupAuthorizationFailed,
"stop",
))
}
}
})
.await
.expect_err("non-retriable errors must stop immediately");
assert!(matches!(
error,
KrafkaError::Broker {
code: crate::error::ErrorCode::GroupAuthorizationFailed,
..
}
));
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
}
#[test]
fn test_select_close_result_prefers_auto_commit_error() {
let error = Consumer::select_close_result(
Err(KrafkaError::invalid_state("commit failed")),
Err(KrafkaError::invalid_state("leave failed")),
)
.expect_err("auto-commit error must take precedence");
assert!(error.to_string().contains("commit failed"));
}
#[test]
fn test_select_close_result_ignores_rebalance_close_commit_error() {
let error = Consumer::select_close_result(
Err(KrafkaError::broker(
crate::error::ErrorCode::UnknownMemberId,
"rebalance needed",
)),
Err(KrafkaError::invalid_state("leave failed")),
)
.expect_err("leave-group error must surface when close-time commit error is benign");
assert!(error.to_string().contains("leave failed"));
}
#[test]
fn test_select_close_result_swallows_rebalance_close_commit_error_when_leave_succeeds() {
Consumer::select_close_result(
Err(KrafkaError::broker(
crate::error::ErrorCode::RebalanceInProgress,
"rebalance needed",
)),
Ok(()),
)
.expect("rebalance-related close-time commit errors should be ignored");
}
#[test]
fn test_select_close_result_returns_leave_group_error_when_commit_succeeds() {
let error =
Consumer::select_close_result(Ok(()), Err(KrafkaError::invalid_state("leave failed")))
.expect_err("leave-group error must surface when commit succeeded");
assert!(error.to_string().contains("leave failed"));
}
#[test]
fn test_max_poll_interval_used_for_rebalance() {
let config = ConsumerConfig::default();
assert_eq!(config.max_poll_interval, Duration::from_secs(300));
assert_eq!(
config.session_timeout,
Duration::from_secs(45),
"session_timeout matches Java/librdkafka since Kafka 3.0; the older 10s \
default caused spurious rebalances under GC pauses and is rejected by \
brokers with group.min.session.timeout.ms > 10000"
);
assert!(config.max_poll_interval > config.session_timeout);
}
#[test]
fn test_list_offsets_partition_grouping_by_leader() {
let leader_map: HashMap<(&str, PartitionId), crate::BrokerId> = [
(("topic1", 0), 1),
(("topic1", 1), 2),
(("topic2", 0), 1), (("topic2", 1), 3),
]
.into_iter()
.collect();
let mut partitions: HashMap<String, Vec<PartitionId>> = HashMap::new();
partitions.insert("topic1".to_string(), vec![0, 1]);
partitions.insert("topic2".to_string(), vec![0, 1]);
let mut by_leader: HashMap<crate::BrokerId, Vec<(String, PartitionId)>> = HashMap::new();
for (topic, parts) in &partitions {
for &p in parts {
if let Some(&leader) = leader_map.get(&(topic.as_str(), p)) {
by_leader
.entry(leader)
.or_default()
.push((topic.clone(), p));
}
}
}
assert_eq!(by_leader[&1].len(), 2);
assert!(by_leader[&1].contains(&("topic1".to_string(), 0)));
assert!(by_leader[&1].contains(&("topic2".to_string(), 0)));
assert_eq!(by_leader[&2].len(), 1);
assert_eq!(by_leader[&2][0], ("topic1".to_string(), 1));
assert_eq!(by_leader[&3].len(), 1);
assert_eq!(by_leader[&3][0], ("topic2".to_string(), 1));
}
#[test]
fn test_list_offsets_request_construction() {
let leader_partitions: Vec<(String, PartitionId)> = vec![
("topic1".to_string(), 0),
("topic1".to_string(), 2),
("topic2".to_string(), 1),
];
let timestamp = -1i64;
let mut topics_map: HashMap<String, Vec<ListOffsetsRequestPartition>> = HashMap::new();
for (topic, partition) in &leader_partitions {
topics_map
.entry(topic.clone())
.or_default()
.push(ListOffsetsRequestPartition {
partition_index: *partition,
current_leader_epoch: -1,
timestamp,
});
}
let topics: Vec<ListOffsetsRequestTopic> = topics_map
.into_iter()
.map(|(name, parts)| ListOffsetsRequestTopic {
name,
partitions: parts,
})
.collect();
let request = ListOffsetsRequest {
replica_id: -1,
isolation_level: 0,
topics,
timeout_ms: None,
};
assert_eq!(request.replica_id, -1);
assert_eq!(request.topics.len(), 2);
let t1 = request.topics.iter().find(|t| t.name == "topic1").unwrap();
assert_eq!(t1.partitions.len(), 2);
assert!(t1.partitions.iter().any(|p| p.partition_index == 0));
assert!(t1.partitions.iter().any(|p| p.partition_index == 2));
for p in &t1.partitions {
assert_eq!(p.timestamp, -1);
assert_eq!(p.current_leader_epoch, -1);
}
let t2 = request.topics.iter().find(|t| t.name == "topic2").unwrap();
assert_eq!(t2.partitions.len(), 1);
assert_eq!(t2.partitions[0].partition_index, 1);
}
#[test]
fn test_list_offsets_response_result_extraction() {
use crate::error::ErrorCode;
use crate::protocol::ListOffsetsResponsePartition;
use crate::protocol::ListOffsetsResponseTopic;
let response = ListOffsetsResponse {
topics: vec![
ListOffsetsResponseTopic {
name: "topic1".to_string(),
partitions: vec![
ListOffsetsResponsePartition {
partition_index: 0,
error_code: ErrorCode::None,
timestamp: -1,
offset: 42,
leader_epoch: -1,
},
ListOffsetsResponsePartition {
partition_index: 1,
error_code: ErrorCode::None,
timestamp: -1,
offset: 100,
leader_epoch: -1,
},
],
},
ListOffsetsResponseTopic {
name: "topic2".to_string(),
partitions: vec![ListOffsetsResponsePartition {
partition_index: 0,
error_code: ErrorCode::None,
timestamp: -1,
offset: 7,
leader_epoch: -1,
}],
},
],
};
let mut result: HashMap<(String, PartitionId), Result<Offset>> = HashMap::new();
apply_list_offsets_response(&response, &mut result);
assert_eq!(result.len(), 3);
assert_eq!(*result[&("topic1".to_string(), 0)].as_ref().unwrap(), 42);
assert_eq!(*result[&("topic1".to_string(), 1)].as_ref().unwrap(), 100);
assert_eq!(*result[&("topic2".to_string(), 0)].as_ref().unwrap(), 7);
}
#[test]
fn test_list_offsets_partial_failure_keeps_successes() {
use crate::error::ErrorCode;
use crate::protocol::ListOffsetsResponsePartition;
use crate::protocol::ListOffsetsResponseTopic;
let response = ListOffsetsResponse {
topics: vec![ListOffsetsResponseTopic {
name: "topic1".to_string(),
partitions: vec![
ListOffsetsResponsePartition {
partition_index: 0,
error_code: ErrorCode::None,
timestamp: -1,
offset: 42,
leader_epoch: -1,
},
ListOffsetsResponsePartition {
partition_index: 1,
error_code: ErrorCode::NotLeaderForPartition,
timestamp: -1,
offset: -1,
leader_epoch: -1,
},
ListOffsetsResponsePartition {
partition_index: 2,
error_code: ErrorCode::None,
timestamp: -1,
offset: 99,
leader_epoch: -1,
},
],
}],
};
let mut result: HashMap<(String, PartitionId), Result<Offset>> = HashMap::new();
apply_list_offsets_response(&response, &mut result);
assert_eq!(result.len(), 3);
assert_eq!(*result[&("topic1".to_string(), 0)].as_ref().unwrap(), 42);
assert_eq!(*result[&("topic1".to_string(), 2)].as_ref().unwrap(), 99);
assert!(result[&("topic1".to_string(), 1)].is_err());
let err_msg = result[&("topic1".to_string(), 1)]
.as_ref()
.unwrap_err()
.to_string();
assert!(
err_msg.contains("ListOffsets error"),
"unexpected: {err_msg}"
);
}
#[test]
fn test_list_offsets_all_failed_returns_error() {
use crate::error::ErrorCode;
use crate::protocol::ListOffsetsResponsePartition;
use crate::protocol::ListOffsetsResponseTopic;
let response = ListOffsetsResponse {
topics: vec![ListOffsetsResponseTopic {
name: "topic1".to_string(),
partitions: vec![ListOffsetsResponsePartition {
partition_index: 0,
error_code: ErrorCode::NotLeaderForPartition,
timestamp: -1,
offset: -1,
leader_epoch: -1,
}],
}],
};
let mut result: HashMap<(String, PartitionId), Result<Offset>> = HashMap::new();
apply_list_offsets_response(&response, &mut result);
assert_eq!(result.len(), 1);
assert!(result[&("topic1".to_string(), 0)].is_err());
let err_msg = result[&("topic1".to_string(), 0)]
.as_ref()
.unwrap_err()
.to_string();
assert!(
err_msg.contains("ListOffsets error"),
"unexpected: {err_msg}"
);
}
#[test]
fn test_list_offsets_request_encode_v1_v2() {
use bytes::BytesMut;
let request = ListOffsetsRequest {
replica_id: -1,
isolation_level: 1,
topics: vec![ListOffsetsRequestTopic {
name: "test-topic".to_string(),
partitions: vec![
ListOffsetsRequestPartition {
partition_index: 0,
current_leader_epoch: -1,
timestamp: -1, },
ListOffsetsRequestPartition {
partition_index: 1,
current_leader_epoch: -1,
timestamp: -2, },
],
}],
timeout_ms: None,
};
let mut buf = BytesMut::new();
request.encode_v1(&mut buf).unwrap();
let encoded_v1_len = buf.len();
assert!(encoded_v1_len > 0);
let mut buf_v2 = BytesMut::new();
request.encode_v2(&mut buf_v2).unwrap();
assert_eq!(buf_v2.len(), encoded_v1_len + 1);
}
fn cooperative_newly_assigned(
new: &HashMap<String, Vec<PartitionId>>,
old: &HashMap<String, Vec<PartitionId>>,
) -> Vec<TopicPartition> {
let old_sets: HashMap<&String, HashSet<PartitionId>> = old
.iter()
.map(|(t, ps)| (t, ps.iter().copied().collect()))
.collect();
let mut result = Vec::new();
for (topic, partitions) in new {
let old_set = old_sets.get(topic);
for &p in partitions {
let is_new = old_set.is_none_or(|os| !os.contains(&p));
if is_new {
result.push(TopicPartition::new(topic, p));
}
}
}
result
}
fn cooperative_revocations(
old: &HashMap<String, Vec<PartitionId>>,
new: &HashMap<String, Vec<PartitionId>>,
) -> Vec<TopicPartition> {
let new_sets: HashMap<&String, HashSet<PartitionId>> = new
.iter()
.map(|(t, ps)| (t, ps.iter().copied().collect()))
.collect();
let mut result = Vec::new();
for (topic, partitions) in old {
let new_set = new_sets.get(topic);
for &p in partitions {
let gone = new_set.is_none_or(|ns| !ns.contains(&p));
if gone {
result.push(TopicPartition::new(topic, p));
}
}
}
result
}
fn apply_revocations_to_assignments(
assignments: &mut HashMap<String, Vec<PartitionId>>,
revoked: &[(String, PartitionId)],
) {
let mut revoked_by_topic: HashMap<&str, HashSet<PartitionId>> = HashMap::new();
for (topic, partition) in revoked {
revoked_by_topic
.entry(topic.as_str())
.or_default()
.insert(*partition);
}
for (topic, revoked_parts) in &revoked_by_topic {
if let Some(parts) = assignments.get_mut(*topic) {
parts.retain(|p| !revoked_parts.contains(p));
if parts.is_empty() {
assignments.remove(*topic);
}
}
}
}
#[test]
fn test_cooperative_newly_assigned_fresh_join() {
let old: HashMap<String, Vec<PartitionId>> = HashMap::new();
let new: HashMap<String, Vec<PartitionId>> = [
("topic1".to_string(), vec![0, 1, 2]),
("topic2".to_string(), vec![0]),
]
.into_iter()
.collect();
let result = cooperative_newly_assigned(&new, &old);
assert_eq!(result.len(), 4);
assert!(result.contains(&TopicPartition::new("topic1", 0)));
assert!(result.contains(&TopicPartition::new("topic1", 1)));
assert!(result.contains(&TopicPartition::new("topic1", 2)));
assert!(result.contains(&TopicPartition::new("topic2", 0)));
}
#[test]
fn test_cooperative_newly_assigned_partial_overlap() {
let old: HashMap<String, Vec<PartitionId>> = [
("topic1".to_string(), vec![0, 1]),
("topic2".to_string(), vec![0]),
]
.into_iter()
.collect();
let new: HashMap<String, Vec<PartitionId>> = [
("topic1".to_string(), vec![1, 2]),
("topic3".to_string(), vec![0]),
]
.into_iter()
.collect();
let result = cooperative_newly_assigned(&new, &old);
assert_eq!(result.len(), 2);
assert!(result.contains(&TopicPartition::new("topic1", 2)));
assert!(result.contains(&TopicPartition::new("topic3", 0)));
assert!(!result.contains(&TopicPartition::new("topic1", 1))); }
#[test]
fn test_cooperative_newly_assigned_identical() {
let assignment: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0, 1])].into_iter().collect();
let result = cooperative_newly_assigned(&assignment, &assignment);
assert!(result.is_empty());
}
#[test]
fn test_cooperative_revocations_partial() {
let old: HashMap<String, Vec<PartitionId>> = [
("topic1".to_string(), vec![0, 1, 2]),
("topic2".to_string(), vec![0]),
]
.into_iter()
.collect();
let new: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![1])].into_iter().collect();
let result = cooperative_revocations(&old, &new);
assert_eq!(result.len(), 3);
assert!(result.contains(&TopicPartition::new("topic1", 0)));
assert!(result.contains(&TopicPartition::new("topic1", 2)));
assert!(result.contains(&TopicPartition::new("topic2", 0)));
}
#[test]
fn test_cooperative_revocations_full() {
let old: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0, 1])].into_iter().collect();
let new: HashMap<String, Vec<PartitionId>> = HashMap::new();
let result = cooperative_revocations(&old, &new);
assert_eq!(result.len(), 2);
}
#[test]
fn test_cooperative_revocations_none() {
let old: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0])].into_iter().collect();
let new: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0, 1])].into_iter().collect();
let result = cooperative_revocations(&old, &new);
assert!(result.is_empty());
}
#[test]
fn test_eager_cleanup_preserves_pause_for_retained_partitions() {
let old: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0, 1])].into_iter().collect();
let new: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![1, 2])].into_iter().collect();
let revoked = revoked_partitions_diff(&old, &new);
let revoked_tuples: Vec<(String, PartitionId)> = revoked
.into_iter()
.map(|tp| (tp.topic, tp.partition))
.collect();
let mut paused: HashSet<(String, PartitionId)> =
[("topic1".to_string(), 0), ("topic1".to_string(), 1)]
.into_iter()
.collect();
for key in &revoked_tuples {
paused.remove(key);
}
assert!(!paused.contains(&("topic1".to_string(), 0)));
assert!(paused.contains(&("topic1".to_string(), 1)));
}
#[test]
fn test_apply_revocations_removes_partitions() {
let mut assignments: HashMap<String, Vec<PartitionId>> = [
("topic1".to_string(), vec![0, 1, 2]),
("topic2".to_string(), vec![0, 1]),
]
.into_iter()
.collect();
let revoked = vec![
("topic1".to_string(), 0),
("topic1".to_string(), 2),
("topic2".to_string(), 1),
];
apply_revocations_to_assignments(&mut assignments, &revoked);
assert_eq!(assignments["topic1"], vec![1]);
assert_eq!(assignments["topic2"], vec![0]);
}
#[test]
fn test_apply_revocations_removes_empty_topics() {
let mut assignments: HashMap<String, Vec<PartitionId>> = [
("topic1".to_string(), vec![0]),
("topic2".to_string(), vec![0, 1]),
]
.into_iter()
.collect();
let revoked = vec![("topic1".to_string(), 0)];
apply_revocations_to_assignments(&mut assignments, &revoked);
assert!(!assignments.contains_key("topic1"));
assert_eq!(assignments.len(), 1);
assert_eq!(assignments["topic2"], vec![0, 1]);
}
#[test]
fn test_apply_revocations_nonexistent_partition() {
let mut assignments: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0, 1])].into_iter().collect();
let revoked = vec![
("topic1".to_string(), 5), ("topic3".to_string(), 0), ];
apply_revocations_to_assignments(&mut assignments, &revoked);
assert_eq!(assignments["topic1"], vec![0, 1]);
}
#[test]
fn test_cooperative_two_phase_rebalance_consistency() {
let phase0: HashMap<String, Vec<PartitionId>> = [
("topic1".to_string(), vec![0, 1, 2]),
("topic2".to_string(), vec![0]),
]
.into_iter()
.collect();
let phase1_target: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0, 1])].into_iter().collect();
let to_revoke = cooperative_revocations(&phase0, &phase1_target);
assert_eq!(to_revoke.len(), 2);
assert!(to_revoke.contains(&TopicPartition::new("topic1", 2)));
assert!(to_revoke.contains(&TopicPartition::new("topic2", 0)));
let mut current = phase0.clone();
let revoked_tuples: Vec<(String, PartitionId)> = to_revoke
.iter()
.map(|tp| (tp.topic.clone(), tp.partition))
.collect();
apply_revocations_to_assignments(&mut current, &revoked_tuples);
assert_eq!(current["topic1"], vec![0, 1]);
assert!(!current.contains_key("topic2"));
let phase2_final: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0, 1, 3])]
.into_iter()
.collect();
let newly_assigned = cooperative_newly_assigned(&phase2_final, ¤t);
assert_eq!(newly_assigned.len(), 1);
assert!(newly_assigned.contains(&TopicPartition::new("topic1", 3)));
let extra_revoke = cooperative_revocations(¤t, &phase2_final);
assert!(extra_revoke.is_empty());
}
#[tokio::test]
async fn test_cooperative_callback_ordering() {
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
struct OrderTracker {
revoke_seq: AtomicU64,
assign_seq: AtomicU64,
counter: AtomicU64,
}
impl ConsumerRebalanceListener for OrderTracker {
async fn on_partitions_assigned(&self, _: &[TopicPartition]) {
self.assign_seq.store(
self.counter.fetch_add(1, Ordering::SeqCst),
Ordering::SeqCst,
);
}
async fn on_partitions_revoked(&self, _: &[TopicPartition]) {
self.revoke_seq.store(
self.counter.fetch_add(1, Ordering::SeqCst),
Ordering::SeqCst,
);
}
}
let tracker = Arc::new(OrderTracker {
revoke_seq: AtomicU64::new(u64::MAX),
assign_seq: AtomicU64::new(u64::MAX),
counter: AtomicU64::new(0),
});
let revoked = vec![TopicPartition::new("topic1", 2)];
ConsumerRebalanceListener::on_partitions_revoked(&*tracker, &revoked).await;
let assigned = vec![
TopicPartition::new("topic1", 0),
TopicPartition::new("topic1", 1),
TopicPartition::new("topic1", 3),
];
ConsumerRebalanceListener::on_partitions_assigned(&*tracker, &assigned).await;
let revoke_order = tracker.revoke_seq.load(Ordering::SeqCst);
let assign_order = tracker.assign_seq.load(Ordering::SeqCst);
assert!(
revoke_order < assign_order,
"on_partitions_revoked (seq={revoke_order}) must fire before on_partitions_assigned (seq={assign_order})"
);
}
#[tokio::test]
async fn test_cooperative_on_assigned_fires_on_empty() {
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
struct EmptyTracker {
assigned_called: AtomicBool,
}
impl ConsumerRebalanceListener for EmptyTracker {
async fn on_partitions_assigned(&self, parts: &[TopicPartition]) {
assert!(parts.is_empty());
self.assigned_called.store(true, Ordering::SeqCst);
}
async fn on_partitions_revoked(&self, _: &[TopicPartition]) {}
}
let tracker = EmptyTracker {
assigned_called: AtomicBool::new(false),
};
ConsumerRebalanceListener::on_partitions_assigned(&tracker, &[]).await;
assert!(tracker.assigned_called.load(Ordering::SeqCst));
}
fn ps_with_hw(watermark: Offset) -> PartitionState {
PartitionState {
high_watermark: Some(watermark),
..Default::default()
}
}
#[test]
fn test_lag_computation_logic() {
let mut offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
let (total_lag, max_lag) = compute_aggregate_lag(
&offsets,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadUncommitted,
);
assert_eq!(total_lag, 0);
assert_eq!(max_lag, 0);
offsets.insert(("t".into(), 0), 50);
offsets.insert(("t".into(), 1), 100);
partition_state.insert(("t".into(), 0), ps_with_hw(80));
partition_state.insert(("t".into(), 1), ps_with_hw(120));
let (total_lag, max_lag) = compute_aggregate_lag(
&offsets,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadUncommitted,
);
assert_eq!(total_lag, 50); assert_eq!(max_lag, 30); }
#[test]
fn test_lag_negative_clamped_to_zero() {
let mut offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
offsets.insert(("t".into(), 0), 100);
partition_state.insert(("t".into(), 0), ps_with_hw(80));
let (total_lag, _) = compute_aggregate_lag(
&offsets,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadUncommitted,
);
assert_eq!(total_lag, 0);
}
#[test]
fn test_lag_partial_watermarks() {
let mut offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
offsets.insert(("t".into(), 0), 50);
offsets.insert(("t".into(), 1), 100);
partition_state.insert(("t".into(), 0), ps_with_hw(80));
let (total_lag, _) = compute_aggregate_lag(
&offsets,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadUncommitted,
);
assert_eq!(total_lag, 30); }
#[test]
fn test_lag_after_revocation() {
let mut offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
offsets.insert(("t".into(), 0), 50);
offsets.insert(("t".into(), 1), 100);
partition_state.insert(("t".into(), 0), ps_with_hw(100)); partition_state.insert(("t".into(), 1), ps_with_hw(200));
let revoked = vec![TopicPartition::new("t", 0)];
for tp in &revoked {
let key = (tp.topic.clone(), tp.partition);
offsets.remove(&key);
partition_state.remove(&key);
}
assert!(!partition_state.contains_key(&("t".into(), 0)));
assert!(partition_state.contains_key(&("t".into(), 1)));
let (total_lag, max_lag) = compute_aggregate_lag(
&offsets,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadUncommitted,
);
assert_eq!(total_lag, 100);
assert_eq!(max_lag, 100);
}
#[test]
fn test_lag_clear_resets_to_zero() {
let mut offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
offsets.insert(("t".into(), 0), 50);
partition_state.insert(("t".into(), 0), ps_with_hw(100));
offsets.clear();
partition_state.clear();
let (total_lag, _) = compute_aggregate_lag(
&offsets,
&partition_state,
&HashMap::new(),
IsolationLevel::ReadUncommitted,
);
assert_eq!(total_lag, 0);
}
#[test]
fn test_partition_state_revocation_is_atomic() {
let key = ("t".to_string(), 0_i32);
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
partition_state.insert(
key.clone(),
PartitionState {
last_stable_offset: None,
high_watermark: Some(100),
log_start_offset: Some(0),
preferred_replica: Some((3_i32, Instant::now() + Duration::from_secs(60))),
offset_retry_backoff: Some((Instant::now(), Duration::from_millis(100))),
watermark_updated_at: None,
last_fetched_epoch: Some(7),
position_validated: true,
},
);
let state = &partition_state[&key];
assert!(state.high_watermark.is_some());
assert!(state.log_start_offset.is_some());
assert!(state.preferred_replica.is_some());
assert!(state.offset_retry_backoff.is_some());
partition_state.remove(&key);
assert!(!partition_state.contains_key(&key));
}
fn ps_with_preferred(replica_id: crate::BrokerId, expiry: Instant) -> PartitionState {
PartitionState {
preferred_replica: Some((replica_id, expiry)),
..Default::default()
}
}
#[test]
fn test_routing_plan_uses_leader_when_no_preferred() {
let keys = vec![("t".into(), 0), ("t".into(), 1)];
let leaders = HashMap::from([(("t".into(), 0), 1), (("t".into(), 1), 2)]);
let plan = build_fetch_routing_plan(keys, &HashMap::new(), &leaders, Instant::now());
assert!(plan.expired_preferred.is_empty());
assert_eq!(plan.partitions_by_broker[&1], vec![("t".into(), 0)]);
assert_eq!(plan.partitions_by_broker[&2], vec![("t".into(), 1)]);
}
#[test]
fn test_routing_plan_routes_to_preferred_replica() {
let keys = vec![("t".into(), 0)];
let leaders = HashMap::from([(("t".into(), 0), 1)]);
let partition_state = HashMap::from([(
("t".into(), 0),
ps_with_preferred(3_i32, Instant::now() + Duration::from_secs(60)),
)]);
let plan = build_fetch_routing_plan(keys, &partition_state, &leaders, Instant::now());
assert!(plan.expired_preferred.is_empty());
assert_eq!(plan.partitions_by_broker.len(), 1);
assert_eq!(plan.partitions_by_broker[&3], vec![("t".into(), 0)]);
}
#[test]
fn test_routing_plan_falls_back_on_expired_preferred() {
let keys = vec![("t".into(), 0)];
let leaders = HashMap::from([(("t".into(), 0), 1)]);
let partition_state = HashMap::from([(
("t".into(), 0),
ps_with_preferred(3_i32, Instant::now() - Duration::from_secs(10)),
)]);
let plan = build_fetch_routing_plan(keys, &partition_state, &leaders, Instant::now());
assert_eq!(plan.partitions_by_broker[&1], vec![("t".into(), 0)]);
assert_eq!(plan.expired_preferred, vec![("t".into(), 0)]);
}
#[test]
fn test_routing_plan_skips_partitions_without_leader() {
let keys = vec![("t".into(), 0), ("t".into(), 1)];
let leaders = HashMap::from([(("t".into(), 0), 1)]);
let plan = build_fetch_routing_plan(keys, &HashMap::new(), &leaders, Instant::now());
let all: Vec<_> = plan.partitions_by_broker.values().flatten().collect();
assert_eq!(all.len(), 1);
assert_eq!(*all[0], ("t".into(), 0));
assert_eq!(plan.skipped, vec![("t".into(), 1)]);
}
#[test]
fn test_routing_plan_all_partitions_skipped() {
let keys = vec![("t".into(), 0), ("t".into(), 1)];
let plan = build_fetch_routing_plan(keys, &HashMap::new(), &HashMap::new(), Instant::now());
assert!(plan.partitions_by_broker.is_empty());
assert!(plan.expired_preferred.is_empty());
assert_eq!(plan.skipped.len(), 2);
}
#[test]
fn test_routing_plan_mixed_preferred_and_leader() {
let keys = vec![("t".into(), 0), ("t".into(), 1), ("t".into(), 2)];
let leaders = HashMap::from([
(("t".into(), 0), 1),
(("t".into(), 1), 1),
(("t".into(), 2), 2),
]);
let future = Instant::now() + Duration::from_secs(300);
let partition_state = HashMap::from([
(("t".into(), 0), ps_with_preferred(3_i32, future)),
(
("t".into(), 1),
ps_with_preferred(3_i32, Instant::now() - Duration::from_secs(1)),
),
]);
let plan = build_fetch_routing_plan(keys, &partition_state, &leaders, Instant::now());
assert!(plan.partitions_by_broker[&3].contains(&("t".into(), 0)));
assert!(plan.partitions_by_broker[&1].contains(&("t".into(), 1)));
assert!(plan.partitions_by_broker[&2].contains(&("t".into(), 2)));
assert_eq!(plan.expired_preferred, vec![("t".into(), 1)]);
}
#[test]
fn test_consumer_is_send_sync() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<Consumer>();
}
#[test]
fn test_consumer_stream_is_send() {
fn assert_send<T: Send>() {}
assert_send::<ConsumerStream<'_>>();
}
#[test]
fn test_offsets_for_times_grouping() {
let partitions: &[(&str, PartitionId)] =
&[("topic1", 0), ("topic1", 2), ("topic2", 1), ("topic1", 5)];
let grouped = group_topic_partitions(partitions);
assert_eq!(grouped.len(), 2);
assert_eq!(grouped["topic1"], vec![0, 2, 5]);
assert_eq!(grouped["topic2"], vec![1]);
}
fn make_record(topic: &str, partition: PartitionId, offset: Offset) -> ConsumerRecord {
ConsumerRecord {
topic: topic.to_string(),
partition,
offset,
timestamp: 0,
timestamp_type: 0,
key: None,
value: None,
headers: vec![],
leader_epoch: None,
delivery_count: None,
}
}
fn make_test_consumer() -> Consumer {
let config = ConsumerConfig::default();
let pool = Arc::new(ConnectionPool::new(ConnectionConfig::default()));
let metadata = Arc::new(
ClusterMetadata::new(
vec!["127.0.0.1:9092".to_string()],
pool.clone(),
config.metadata_max_age,
)
.with_topic_cache_ttl_disabled(),
);
Consumer {
config,
metadata,
pool,
pool_owned: true,
subscriptions: LeveledRwLock::new(HashSet::new()),
assignments: LeveledRwLock::new(HashMap::new()),
offsets: LeveledRwLock::new(HashMap::new()),
paused: LeveledRwLock::new(HashSet::new()),
closed: std::sync::atomic::AtomicBool::new(false),
wakeup_flag: std::sync::atomic::AtomicBool::new(false),
wakeup_notify: tokio::sync::Notify::new(),
group_coordinator: None,
metrics: Arc::new(ConsumerMetrics::default()),
rebalance_listener: Arc::new(NoOpRebalanceListener),
interceptor: Arc::new(crate::interceptor::NoOpConsumerInterceptor),
last_auto_commit: SyncMutex::new(Instant::now()),
recv_buffer: SyncMutex::new(std::collections::VecDeque::new()),
fetch_rotation: std::sync::atomic::AtomicUsize::new(0),
fetch_sessions: SyncMutex::new(FetchSessionCache::new()),
partition_state: LeveledRwLock::new(HashMap::new()),
key_deserializer: None,
value_deserializer: None,
}
}
#[tokio::test]
async fn test_batch_recv_public_api_returns_closed_when_consumer_closed() {
let consumer = make_test_consumer();
consumer
.closed
.store(true, std::sync::atomic::Ordering::SeqCst);
let outcome = consumer
.batch_recv(5, Duration::from_millis(10))
.await
.unwrap();
assert!(matches!(outcome, BatchRecvOutcome::Closed));
}
#[tokio::test]
async fn test_close_clears_local_state_and_is_idempotent() {
let consumer = make_test_consumer();
consumer.subscriptions.write().await.insert("orders".into());
consumer
.assignments
.write()
.await
.insert("orders".into(), vec![0]);
consumer
.offsets
.write()
.await
.insert(("orders".into(), 0), 42);
consumer.paused.write().await.insert(("orders".into(), 0));
consumer.partition_state.write().await.insert(
("orders".into(), 0),
PartitionState {
high_watermark: Some(100),
..PartitionState::default()
},
);
consumer
.recv_buffer
.lock()
.push_back(make_record("orders", 0, 42));
consumer.metrics.buffered_records.set(1);
consumer.metrics.paused_partitions.set(1);
consumer.metrics.lag.set(5);
consumer.metrics.lag_max.set(5);
consumer.close().await.expect("close succeeds");
assert!(consumer.is_closed());
assert!(consumer.subscriptions.read().await.is_empty());
assert!(consumer.assignments.read().await.is_empty());
assert!(consumer.offsets.read().await.is_empty());
assert!(consumer.paused.read().await.is_empty());
assert!(consumer.partition_state.read().await.is_empty());
assert!(consumer.recv_buffer.lock().is_empty());
assert_eq!(consumer.metrics.buffered_records.get(), 0);
assert_eq!(consumer.metrics.paused_partitions.get(), 0);
assert_eq!(consumer.metrics.lag.get(), 0);
assert_eq!(consumer.metrics.lag_max.get(), 0);
consumer
.close()
.await
.expect("second close remains a no-op");
}
#[tokio::test]
async fn test_unsubscribe_clears_local_state_and_is_idempotent() {
let consumer = make_test_consumer();
consumer.subscriptions.write().await.insert("orders".into());
consumer
.assignments
.write()
.await
.insert("orders".into(), vec![0]);
consumer
.offsets
.write()
.await
.insert(("orders".into(), 0), 42);
consumer.paused.write().await.insert(("orders".into(), 0));
consumer.partition_state.write().await.insert(
("orders".into(), 0),
PartitionState {
high_watermark: Some(100),
..PartitionState::default()
},
);
consumer
.recv_buffer
.lock()
.push_back(make_record("orders", 0, 42));
consumer.metrics.buffered_records.set(1);
consumer.metrics.paused_partitions.set(1);
consumer.metrics.lag.set(5);
consumer.metrics.lag_max.set(5);
consumer.metrics.assigned_partitions.set(1);
consumer.unsubscribe().await.expect("unsubscribe succeeds");
assert!(!consumer.is_closed());
assert!(consumer.subscriptions.read().await.is_empty());
assert!(consumer.assignments.read().await.is_empty());
assert!(consumer.offsets.read().await.is_empty());
assert!(consumer.paused.read().await.is_empty());
assert!(consumer.partition_state.read().await.is_empty());
assert!(consumer.recv_buffer.lock().is_empty());
assert_eq!(consumer.metrics.buffered_records.get(), 0);
assert_eq!(consumer.metrics.paused_partitions.get(), 0);
assert_eq!(consumer.metrics.lag.get(), 0);
assert_eq!(consumer.metrics.lag_max.get(), 0);
assert_eq!(consumer.metrics.assigned_partitions.get(), 0);
consumer
.unsubscribe()
.await
.expect("second unsubscribe remains a no-op");
}
#[tokio::test]
async fn test_batch_recv_public_api_uses_buffer_and_updates_metric() {
let consumer = make_test_consumer();
{
let mut buffer = consumer.recv_buffer.lock();
buffer.push_back(make_record("orders", 0, 1));
buffer.push_back(make_record("orders", 0, 2));
}
let outcome = consumer
.batch_recv(1, Duration::from_millis(10))
.await
.unwrap();
let BatchRecvOutcome::Records(records) = outcome else {
panic!("expected records outcome");
};
assert_eq!(records.len(), 1);
assert_eq!(records[0].offset, 1);
let metrics = consumer.metrics().snapshot();
assert_eq!(metrics.buffered_records, 1);
}
#[test]
fn requeued_records_are_ordered_ahead_of_the_parked_surplus() {
let consumer = make_test_consumer();
{
let mut buffer = consumer.recv_buffer.lock();
buffer.push_back(make_record("orders", 0, 3));
buffer.push_back(make_record("orders", 0, 4));
}
consumer.requeue_undelivered(vec![
make_record("orders", 0, 1),
make_record("orders", 0, 2),
]);
let buffer = consumer.recv_buffer.lock();
let offsets: Vec<Offset> = buffer.iter().map(|r| r.offset).collect();
assert_eq!(
offsets,
vec![1, 2, 3, 4],
"requeued records must precede the parked surplus"
);
}
#[tokio::test]
async fn a_deserializer_failure_puts_the_batch_back_and_names_the_record() {
struct AlwaysFails;
impl crate::serdes::Deserializer for AlwaysFails {
fn deserialize(
&self,
_payload: bytes::Bytes,
_topic: &str,
_is_key: bool,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = Result<bytes::Bytes>> + Send + '_>,
> {
Box::pin(async { Err(KrafkaError::serialization("bad magic byte")) })
}
}
let with_value = |offset: Offset| {
let mut record = make_record("orders", 2, offset);
record.value = Some(bytes::Bytes::from_static(b"payload"));
record
};
let mut consumer = make_test_consumer();
consumer.value_deserializer = Some(Arc::new(AlwaysFails));
let error = consumer
.finish_delivery(vec![with_value(40), with_value(41)])
.await
.expect_err("a failing deserializer must fail the poll");
match error {
KrafkaError::RecordDeserialization {
ref topic,
partition,
offset,
part,
..
} => {
assert_eq!(topic, "orders");
assert_eq!(partition, 2);
assert_eq!(
offset, 40,
"the *first* failing record is the one to seek past"
);
assert_eq!(part, "value");
}
other => panic!("expected RecordDeserialization, got {other:?}"),
}
let buffer = consumer.recv_buffer.lock();
let offsets: Vec<Offset> = buffer.iter().map(|r| r.offset).collect();
assert_eq!(
offsets,
vec![40, 41],
"no record may be lost to a deserialization failure"
);
}
#[tokio::test]
async fn test_batch_recv_with_returns_empty_request_for_zero_max_records() {
let buffer = SyncMutex::new(std::collections::VecDeque::new());
let outcome = batch_recv_with(
&buffer,
|_| {},
0,
Duration::from_millis(10),
Duration::from_millis(10),
|| false,
|| async { HashSet::new() },
|_| async { Ok(vec![]) },
)
.await
.unwrap();
assert!(matches!(outcome, BatchRecvOutcome::EmptyRequest));
}
#[tokio::test]
async fn test_batch_recv_with_returns_closed_when_no_records_and_closed() {
let buffer = SyncMutex::new(std::collections::VecDeque::new());
let outcome = batch_recv_with(
&buffer,
|_| {},
10,
Duration::from_millis(20),
Duration::from_millis(10),
|| true,
|| async { HashSet::new() },
|_| async { Ok(vec![]) },
)
.await
.unwrap();
assert!(matches!(outcome, BatchRecvOutcome::Closed));
}
#[tokio::test]
async fn test_batch_recv_with_rebuffers_partial_batch_on_poll_error() {
let mut q = std::collections::VecDeque::new();
q.push_back(make_record("t", 0, 10));
q.push_back(make_record("t", 0, 11));
let buffer = SyncMutex::new(q);
let result = batch_recv_with(
&buffer,
|_| {},
10,
Duration::from_millis(20),
Duration::from_millis(10),
|| false,
|| async { HashSet::new() },
|_| async {
Err(KrafkaError::network(std::io::Error::other(
"simulated poll failure",
)))
},
)
.await;
assert!(result.is_err());
let buffer = buffer.lock();
assert_eq!(buffer.len(), 2);
assert_eq!(buffer[0].offset, 10);
assert_eq!(buffer[1].offset, 11);
}
#[tokio::test]
async fn test_batch_recv_with_timeout_returns_timed_out_without_oversleeping() {
let buffer = SyncMutex::new(std::collections::VecDeque::new());
let start = tokio::time::Instant::now();
let outcome = batch_recv_with(
&buffer,
|_| {},
10,
Duration::from_millis(15),
Duration::from_millis(10),
|| false,
|| async { HashSet::new() },
|_| async { Ok(vec![]) },
)
.await
.unwrap();
assert!(matches!(outcome, BatchRecvOutcome::TimedOut));
assert!(start.elapsed() < Duration::from_millis(60));
}
#[tokio::test]
async fn test_batch_recv_with_requeues_overflow_in_order() {
let buffer = SyncMutex::new(std::collections::VecDeque::new());
let poll_records = SyncMutex::new(Some(vec![
make_record("t", 0, 1),
make_record("t", 0, 2),
make_record("t", 0, 3),
]));
let outcome = batch_recv_with(
&buffer,
|_| {},
2,
Duration::from_millis(50),
Duration::from_millis(10),
|| false,
|| async { HashSet::new() },
|_| async { Ok(poll_records.lock().take().unwrap_or_default()) },
)
.await
.unwrap();
let BatchRecvOutcome::Records(batch) = outcome else {
panic!("expected records outcome");
};
assert_eq!(batch.len(), 2);
assert_eq!(batch[0].offset, 1);
assert_eq!(batch[1].offset, 2);
let buffer = buffer.lock();
assert_eq!(buffer.len(), 1);
assert_eq!(buffer[0].offset, 3);
}
#[test]
fn test_assignment_offset_precedence_uses_initial_when_no_committed() {
let assigned: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0])].into_iter().collect();
let committed: HashMap<(String, PartitionId), Offset> = HashMap::new();
let initial_offsets: HashMap<(String, PartitionId), Offset> =
[(("topic1".to_string(), 0), 500)].into_iter().collect();
let mut stored: HashMap<(String, PartitionId), Offset> = HashMap::new();
let need_reset = apply_assignment_offset_precedence(
&assigned,
&committed,
&initial_offsets,
&mut stored,
);
assert!(need_reset.is_empty());
assert_eq!(stored.get(&("topic1".to_string(), 0)), Some(&500));
}
#[test]
fn test_seed_initial_offsets_for_assigned_filters_and_vacant_only() {
let assigned: HashMap<String, Vec<PartitionId>> = [
("topic1".to_string(), vec![0, 1]),
("topic2".to_string(), vec![0]),
]
.into_iter()
.collect();
let initial_offsets: HashMap<(String, PartitionId), Offset> = [
(("topic1".to_string(), 0), 100),
(("topic1".to_string(), 2), 200), (("topic3".to_string(), 0), 300), ]
.into_iter()
.collect();
let mut stored: HashMap<(String, PartitionId), Offset> =
[(("topic1".to_string(), 1), 999)].into_iter().collect();
let inserted = seed_initial_offsets_for_assigned(&assigned, &initial_offsets, &mut stored);
assert_eq!(inserted, 1);
assert_eq!(stored.get(&("topic1".to_string(), 0)), Some(&100));
assert_eq!(stored.get(&("topic1".to_string(), 1)), Some(&999));
assert!(!stored.contains_key(&("topic1".to_string(), 2)));
assert!(!stored.contains_key(&("topic3".to_string(), 0)));
}
#[test]
fn test_assignment_offset_precedence_committed_wins_initial() {
let assigned: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0])].into_iter().collect();
let committed: HashMap<(String, PartitionId), Offset> =
[(("topic1".to_string(), 0), 999)].into_iter().collect();
let initial_offsets: HashMap<(String, PartitionId), Offset> =
[(("topic1".to_string(), 0), 500)].into_iter().collect();
let mut stored: HashMap<(String, PartitionId), Offset> = HashMap::new();
let need_reset = apply_assignment_offset_precedence(
&assigned,
&committed,
&initial_offsets,
&mut stored,
);
assert!(need_reset.is_empty());
assert_eq!(stored.get(&("topic1".to_string(), 0)), Some(&999));
}
#[test]
fn test_assignment_offset_precedence_missing_offsets_require_reset() {
let assigned: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0])].into_iter().collect();
let committed: HashMap<(String, PartitionId), Offset> = HashMap::new();
let initial_offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
let mut stored: HashMap<(String, PartitionId), Offset> = HashMap::new();
let need_reset = apply_assignment_offset_precedence(
&assigned,
&committed,
&initial_offsets,
&mut stored,
);
assert_eq!(need_reset, vec![("topic1".to_string(), 0)]);
assert!(stored.is_empty());
}
#[test]
fn test_assignment_offset_precedence_preserves_existing_user_offset() {
let assigned: HashMap<String, Vec<PartitionId>> =
[("topic1".to_string(), vec![0])].into_iter().collect();
let committed: HashMap<(String, PartitionId), Offset> =
[(("topic1".to_string(), 0), 999)].into_iter().collect();
let initial_offsets: HashMap<(String, PartitionId), Offset> =
[(("topic1".to_string(), 0), 500)].into_iter().collect();
let mut stored: HashMap<(String, PartitionId), Offset> =
[(("topic1".to_string(), 0), 42)].into_iter().collect();
let need_reset = apply_assignment_offset_precedence(
&assigned,
&committed,
&initial_offsets,
&mut stored,
);
assert!(need_reset.is_empty());
assert_eq!(stored.get(&("topic1".to_string(), 0)), Some(&42));
}
#[test]
fn test_apply_seek_many_offsets_updates_multiple_partitions() {
let mut stored: HashMap<(String, PartitionId), Offset> =
[(("orders".to_string(), 0), 10)].into_iter().collect();
let updates: HashMap<(String, PartitionId), Offset> = [
(("orders".to_string(), 0), 20),
(("orders".to_string(), 1), 30),
]
.into_iter()
.collect();
let updated = apply_seek_many_offsets(&mut stored, &updates);
assert_eq!(updated, 2);
assert_eq!(stored.get(&("orders".to_string(), 0)), Some(&20));
assert_eq!(stored.get(&("orders".to_string(), 1)), Some(&30));
}
#[tokio::test]
async fn test_seek_many_public_api_recomputes_lag_and_increments_metric() {
let consumer = make_test_consumer();
{
let mut offsets = consumer.offsets.write().await;
offsets.insert(("orders".to_string(), 0), 100);
}
{
let mut state = consumer.partition_state.write().await;
state.insert(
("orders".to_string(), 0),
PartitionState {
high_watermark: Some(120),
..PartitionState::default()
},
);
}
let updates: HashMap<(String, PartitionId), Offset> =
[(("orders".to_string(), 0), 110)].into_iter().collect();
consumer.seek_many(&updates).await.unwrap();
let metrics = consumer.metrics().snapshot();
assert_eq!(metrics.seeks, 1);
assert_eq!(metrics.lag, 10);
assert_eq!(metrics.lag_max, 10);
}
#[test]
fn test_committable_position_matches_fetch_position_when_buffer_empty() {
let mut positions = HashMap::new();
positions.insert(("t".to_string(), 0), 500);
let buffer = std::collections::VecDeque::new();
let committable = committable_positions(&positions, &buffer);
assert_eq!(committable.get(&("t".to_string(), 0)), Some(&500));
}
#[test]
fn test_committable_position_clamped_to_lowest_buffered_offset() {
let mut positions = HashMap::new();
positions.insert(("t".to_string(), 0), 500);
let buffer: std::collections::VecDeque<ConsumerRecord> =
(3..500).map(|o| make_record("t", 0, o)).collect();
let committable = committable_positions(&positions, &buffer);
assert_eq!(
committable.get(&("t".to_string(), 0)),
Some(&3),
"must commit only up to the first undelivered record"
);
}
#[test]
fn test_committable_position_is_per_partition() {
let mut positions = HashMap::new();
positions.insert(("t".to_string(), 0), 100);
positions.insert(("t".to_string(), 1), 200);
let buffer: std::collections::VecDeque<ConsumerRecord> =
vec![make_record("t", 0, 40), make_record("t", 0, 41)].into();
let committable = committable_positions(&positions, &buffer);
assert_eq!(committable.get(&("t".to_string(), 0)), Some(&40));
assert_eq!(
committable.get(&("t".to_string(), 1)),
Some(&200),
"a partition with no buffered records is unaffected"
);
}
#[test]
fn test_committable_position_uses_minimum_not_first_buffered() {
let mut positions = HashMap::new();
positions.insert(("t".to_string(), 0), 100);
let buffer: std::collections::VecDeque<ConsumerRecord> = vec![
make_record("t", 0, 70),
make_record("t", 0, 55),
make_record("t", 0, 90),
]
.into();
let committable = committable_positions(&positions, &buffer);
assert_eq!(committable.get(&("t".to_string(), 0)), Some(&55));
}
#[test]
fn test_committable_position_never_exceeds_fetch_position() {
let mut positions = HashMap::new();
positions.insert(("t".to_string(), 0), 10);
let buffer: std::collections::VecDeque<ConsumerRecord> =
vec![make_record("t", 0, 999)].into();
let committable = committable_positions(&positions, &buffer);
assert_eq!(committable.get(&("t".to_string(), 0)), Some(&10));
}
#[test]
fn test_fetch_update_applied_when_position_unchanged() {
let mut offsets = HashMap::new();
offsets.insert(("t".to_string(), 0), 1000);
let discarded = apply_fetch_offset_updates(
&mut offsets,
vec![(
("t".to_string(), 0),
FetchOffsetUpdate {
epoch: -1,
requested: 1000,
next: 1500,
},
)],
);
assert!(discarded.is_empty());
assert_eq!(offsets.get(&("t".to_string(), 0)), Some(&1500));
}
#[test]
fn test_fetch_update_discarded_after_concurrent_seek() {
let mut offsets = HashMap::new();
offsets.insert(("t".to_string(), 0), 100);
let discarded = apply_fetch_offset_updates(
&mut offsets,
vec![(
("t".to_string(), 0),
FetchOffsetUpdate {
epoch: -1,
requested: 1000,
next: 1500,
},
)],
);
assert_eq!(discarded, vec![("t".to_string(), 0)]);
assert_eq!(
offsets.get(&("t".to_string(), 0)),
Some(&100),
"seek() must survive an in-flight fetch from the old position"
);
}
#[test]
fn test_fetch_update_discarded_for_revoked_partition() {
let mut offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
let discarded = apply_fetch_offset_updates(
&mut offsets,
vec![(
("t".to_string(), 0),
FetchOffsetUpdate {
epoch: -1,
requested: 5,
next: 10,
},
)],
);
assert_eq!(discarded, vec![("t".to_string(), 0)]);
assert!(offsets.is_empty());
}
#[test]
fn test_fetch_updates_are_evaluated_independently() {
let mut offsets = HashMap::new();
offsets.insert(("t".to_string(), 0), 100); offsets.insert(("t".to_string(), 1), 200);
let discarded = apply_fetch_offset_updates(
&mut offsets,
vec![
(
("t".to_string(), 0),
FetchOffsetUpdate {
epoch: -1,
requested: 1000,
next: 1500,
},
),
(
("t".to_string(), 1),
FetchOffsetUpdate {
epoch: -1,
requested: 200,
next: 250,
},
),
],
);
assert_eq!(discarded, vec![("t".to_string(), 0)]);
assert_eq!(offsets.get(&("t".to_string(), 0)), Some(&100));
assert_eq!(offsets.get(&("t".to_string(), 1)), Some(&250));
}
#[test]
fn a_seeked_position_commits_no_leader_epoch() {
let key = ("events".to_string(), 0);
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
partition_state
.entry(key.clone())
.or_default()
.last_fetched_epoch = Some(7);
let offsets: HashMap<(String, PartitionId), Offset> = [(key.clone(), 42)].into();
let epochs = Consumer::leader_epochs_from_state(&offsets, &partition_state);
assert_eq!(epochs.get(&key), Some(&7));
invalidate_position_epoch(&mut partition_state, &key);
let epochs = Consumer::leader_epochs_from_state(&offsets, &partition_state);
assert!(
epochs.get(&key).is_none(),
"a position moved by seek() must commit -1, not the epoch it held before"
);
let committed = Consumer::build_commit_offsets(&offsets, &epochs, None, false).unwrap();
assert_eq!(committed[&key].leader_epoch, -1);
assert_eq!(committed[&key].offset, 42);
}
#[test]
fn test_invalidate_position_epoch_clears_epoch_and_validation() {
let key = ("t".to_string(), 0);
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
let entry = partition_state.entry(key.clone()).or_default();
entry.last_fetched_epoch = Some(11);
entry.position_validated = true;
entry.high_watermark = Some(900);
invalidate_position_epoch(&mut partition_state, &key);
let state = &partition_state[&key];
assert_eq!(state.last_fetched_epoch, None);
assert!(!state.position_validated);
assert_eq!(state.high_watermark, Some(900));
}
#[test]
fn test_invalidate_position_epoch_creates_entry_when_absent() {
let key = ("t".to_string(), 3);
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
invalidate_position_epoch(&mut partition_state, &key);
assert!(!partition_state[&key].position_validated);
}
#[tokio::test]
async fn test_seek_marks_position_for_revalidation() {
let consumer = make_test_consumer();
let key = ("orders".to_string(), 0);
{
let mut partition_state = consumer.partition_state.write().await;
let entry = partition_state.entry(key.clone()).or_default();
entry.last_fetched_epoch = Some(4);
entry.position_validated = true;
}
consumer.seek("orders", 0, 500).await.unwrap();
let partition_state = consumer.partition_state.read().await;
assert_eq!(partition_state[&key].last_fetched_epoch, None);
assert!(!partition_state[&key].position_validated);
}
#[tokio::test]
async fn test_seek_many_marks_all_positions_for_revalidation() {
let consumer = make_test_consumer();
let keys = [("orders".to_string(), 0), ("orders".to_string(), 1)];
{
let mut partition_state = consumer.partition_state.write().await;
for key in &keys {
let entry = partition_state.entry(key.clone()).or_default();
entry.last_fetched_epoch = Some(4);
entry.position_validated = true;
}
}
let mut targets: HashMap<(String, PartitionId), Offset> = HashMap::new();
targets.insert(keys[0].clone(), 10);
targets.insert(keys[1].clone(), 20);
consumer.seek_many(&targets).await.unwrap();
let partition_state = consumer.partition_state.read().await;
for key in &keys {
assert_eq!(partition_state[key].last_fetched_epoch, None);
assert!(!partition_state[key].position_validated);
}
}
#[tokio::test]
async fn test_truncation_rewinds_position_and_drops_buffered_records() {
let consumer = make_test_consumer();
let key = ("orders".to_string(), 0);
consumer.offsets.write().await.insert(key.clone(), 1_000);
{
let mut partition_state = consumer.partition_state.write().await;
let entry = partition_state.entry(key.clone()).or_default();
entry.last_fetched_epoch = Some(7);
}
{
let mut buffer = consumer.recv_buffer.lock();
buffer.push_back(make_record("orders", 0, 940)); buffer.push_back(make_record("orders", 0, 950)); buffer.push_back(make_record("orders", 0, 980)); buffer.push_back(make_record("orders", 1, 999)); }
consumer
.truncate_to_diverging_offset(
"orders",
0,
crate::protocol::DivergingEpoch {
epoch: 6,
end_offset: 950,
},
)
.await;
assert_eq!(consumer.offsets.read().await[&key], 950);
let partition_state = consumer.partition_state.read().await;
assert_eq!(partition_state[&key].last_fetched_epoch, None);
assert!(partition_state[&key].position_validated);
drop(partition_state);
let buffered: Vec<(String, Offset)> = consumer
.recv_buffer
.lock()
.iter()
.map(|r| (format!("{}-{}", r.topic, r.partition), r.offset))
.collect();
assert_eq!(
buffered,
vec![("orders-0".to_string(), 940), ("orders-1".to_string(), 999)]
);
}
#[tokio::test]
async fn test_truncation_does_not_apply_auto_offset_reset() {
let consumer = make_test_consumer();
let key = ("orders".to_string(), 0);
consumer.offsets.write().await.insert(key.clone(), 5_000);
consumer
.truncate_to_diverging_offset(
"orders",
0,
crate::protocol::DivergingEpoch {
epoch: 2,
end_offset: 4_096,
},
)
.await;
assert_eq!(consumer.offsets.read().await[&key], 4_096);
}
#[test]
fn test_routing_plan_uses_the_metadata_leader() {
let now = Instant::now();
let key = ("orders".to_string(), 0);
let partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
let mut leaders = HashMap::new();
leaders.insert(key.clone(), 9);
let plan = build_fetch_routing_plan(vec![key.clone()], &partition_state, &leaders, now);
assert_eq!(plan.partitions_by_broker[&9], vec![key]);
assert!(plan.skipped.is_empty());
}
#[test]
fn test_routing_plan_skips_partitions_without_a_leader() {
let now = Instant::now();
let key = ("orders".to_string(), 0);
let partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
let plan =
build_fetch_routing_plan(vec![key.clone()], &partition_state, &HashMap::new(), now);
assert!(plan.partitions_by_broker.is_empty());
assert_eq!(plan.skipped, vec![key]);
}
#[test]
fn test_routing_plan_preferred_replica_wins_over_the_leader() {
let now = Instant::now();
let key = ("orders".to_string(), 0);
let mut partition_state: HashMap<(String, PartitionId), PartitionState> = HashMap::new();
partition_state
.entry(key.clone())
.or_default()
.preferred_replica = Some((5, now + Duration::from_secs(60)));
let mut leaders = HashMap::new();
leaders.insert(key.clone(), 9);
let plan = build_fetch_routing_plan(vec![key.clone()], &partition_state, &leaders, now);
assert_eq!(plan.partitions_by_broker[&5], vec![key]);
}
#[tokio::test]
async fn test_broker_address_uses_an_endpoint_from_a_leader_hint() {
let consumer = make_test_consumer();
assert_eq!(consumer.broker_address(42), None);
assert!(consumer.metadata.apply_leader_hint(
"orders",
0,
42,
5,
Some(BrokerInfo::new(
42,
"broker-42.internal".to_string(),
9092,
None
)),
));
assert_eq!(
consumer.broker_address(42),
Some("broker-42.internal:9092".to_string())
);
}
#[test]
fn test_fetch_offset_update_carries_epoch_only_when_applied() {
let mut offsets: HashMap<(String, PartitionId), Offset> = HashMap::new();
offsets.insert(("t".to_string(), 0), 100);
offsets.insert(("t".to_string(), 1), 200);
let updates = vec![
(
("t".to_string(), 0),
FetchOffsetUpdate {
requested: 100,
next: 150,
epoch: 5,
},
),
(
("t".to_string(), 1),
FetchOffsetUpdate {
requested: 180,
next: 250,
epoch: 6,
},
),
];
let discarded = apply_fetch_offset_updates(&mut offsets, updates);
assert_eq!(discarded, vec![("t".to_string(), 1)]);
assert_eq!(offsets[&("t".to_string(), 0)], 150);
assert_eq!(offsets[&("t".to_string(), 1)], 200);
}
}