github-actions[bot] commented on code in PR #4545: URL: https://github.com/apache/iggy/pull/4545#discussion_r4272108687
########## gateways/kafka/docs/BRIDGE_MAPPING.md: ########## @@ -300,11 +300,66 @@ Produce takes the base offset from the send confirmation example for a request it classifies as a duplicate, in which case the response carries `-1` rather than a guessed offset; Kafka clients surface that as an unknown offset. -ListOffsets LATEST is the high watermark from `IggyBridge::high_watermarks`. EARLIEST has no +ListOffsets LATEST is the high watermark of the topic probe `IggyBridge::partition_probes` reads +on the bridge's probe client. EARLIEST has no server-side field today (`Partition` carries no log start offset), so it answers `0`. The `(messages_count, current_offset) == (0, 0)` ambiguity documented on `high_watermark` in `src/bridge/iggy_bridge/offsets.rs` applies to LATEST. +ListOffsets by timestamp ([#4516](https://github.com/apache/iggy/issues/4516)) polls one record +by time and answers its offset and its timestamp as Fetch reports it: the producer timestamp when +one was set, otherwise the append time. + +| Request | Answer | +| --- | --- | +| A timestamp | The first record Iggy appended at or after it | +| A timestamp past the last record, or any time on a partition that retains no message | No error, offset and timestamp `-1`, as Kafka answers | +| A lookup that polls empty without proof that no record matches (below) | `NOT_LEADER_OR_FOLLOWER` (6), which clients retry | +| `MAX_TIMESTAMP` (`-3`, v7) | The last record | +| A negative time other than the sentinels | The first record, as on a Kafka broker | Review Comment: nit: `resolve_one_partition` matches `MAX_TIMESTAMP` with no version check, so a v1-v6 request for `-3` gets the last record, while this table marks `-3` as v7's sentinel and the row for other negative times promises the first record. Say that `-3` answers the last record at every version, or gate the sentinel on the request version. ########## gateways/kafka/src/protocol/handlers/list_offsets.rs: ########## @@ -271,63 +308,304 @@ async fn resolve_all_topics( .iter() .map(|requested| resolve_one_partition(requested, lookup)) .collect(); - ListOffsetsTopicResponse::default() - .with_name(topic.name.clone()) - .with_partitions(partitions) + ResolvedTopic { + name: topic.name.clone(), + partitions, + } }) .collect() } -/// `lookup` is the whole topic's resolution outcome: an errored [`TopicLookup::Watermarks`] is a +/// `lookup` is the whole topic's resolution outcome: an errored [`TopicLookup::Probed`] is a /// call-level failure (e.g. the mapped stream doesn't exist) applying to every partition alike; /// the inner per-partition `Result` inside its `Ok` is [`BridgeError::PartitionOutOfRange`] for one /// bad index among otherwise resolvable ones. fn resolve_one_partition( requested: &ListOffsetsPartition, lookup: &TopicLookup, -) -> ListOffsetsPartitionResponse { - let Ok(partition_index) = u32::try_from(requested.partition_index) else { - return error_response(requested.partition_index, ERROR_UNKNOWN_TOPIC_OR_PARTITION); +) -> PartitionAnswer { + let index = requested.partition_index; + let Ok(partition_index) = u32::try_from(index) else { + return error_answer(index, ERROR_UNKNOWN_TOPIC_OR_PARTITION); }; let results = match lookup { TopicLookup::NotAttempted => { - return error_response(requested.partition_index, ERROR_REQUEST_TIMED_OUT); + return error_answer(index, ERROR_REQUEST_TIMED_OUT); } - TopicLookup::Watermarks(Err(call_err)) => { - return error_response(requested.partition_index, call_err.to_kafka_error_code()); + TopicLookup::Probed(Err(call_err)) => { + return error_answer(index, call_err.to_list_offsets_error_code()); } - TopicLookup::Watermarks(Ok(results)) => results, + TopicLookup::Probed(Ok(results)) => results, }; // `results` preserves the order of the sorted, deduped partition list `resolve_all_topics` - // passed to `high_watermarks` (`IggyBridge::high_watermarks` maps over its input in place), + // passed to `partition_probes` (`IggyBridge::partition_probes` maps over its input in place), // so a binary search is correct here, not just faster than the linear scan this replaced. - let Ok(found) = results.binary_search_by_key(&partition_index, |(index, _)| *index) else { - return error_response(requested.partition_index, ERROR_UNKNOWN_TOPIC_OR_PARTITION); + let Ok(found) = results.binary_search_by_key(&partition_index, |(id, _)| *id) else { + return error_answer(index, ERROR_UNKNOWN_TOPIC_OR_PARTITION); }; - let (_, watermark) = &results[found]; + let (_, probe) = &results[found]; - let watermark = match watermark { - Err(err) => return error_response(requested.partition_index, err.to_kafka_error_code()), - Ok(watermark) => *watermark, + let probe = match probe { + Err(err) => return error_answer(index, err.to_list_offsets_error_code()), + Ok(probe) => *probe, }; + let watermark = i64::try_from(probe.high_watermark).unwrap_or(i64::MAX); - match requested.timestamp { - LATEST_TIMESTAMP => offset_response(requested.partition_index, watermark), + let answer = match requested.timestamp { + LATEST_TIMESTAMP => offset_response(index, watermark), // Real only for a partition retention has never trimmed: Iggy tracks no rolling // low-watermark distinct from partition creation, so a `0` here for an older, // already-trimmed partition names a log-start offset that no longer exists - a real // consumer with `auto.offset.reset=earliest` would seek into a hole. Fetch there reads // from the oldest kept message, so the consumer skips the gap. Not fixable client-side; // needs the bridge to expose a real start offset. - EARLIEST_TIMESTAMP => offset_response(requested.partition_index, 0), - // Non-retriable, unlike ERROR_UNKNOWN_SERVER_ERROR: a Java client resolves this - // immediately instead of retrying the request until its own default.api.timeout.ms. - _ => error_response( - requested.partition_index, - ERROR_UNSUPPORTED_FOR_MESSAGE_FORMAT, - ), + EARLIEST_TIMESTAMP => offset_response(index, 0), + // Nothing was ever appended, so no poll is spent on it. A count of 0 alone is not enough: + // the server reports it without stats too. + _ if probe.high_watermark == 0 => not_found_response(index), + timestamp => { + let lookup = match timestamp { + MAX_TIMESTAMP => RecordLookup::Last, + millis => RecordLookup::AppendedAtOrAfter(millis), + }; + return PartitionAnswer::Pending(PendingLookup { + partition_index: index, + partition: partition_index, + lookup, + retained: probe.messages_count, + high_watermark: probe.high_watermark, + }); + } + }; + PartitionAnswer::Answered(answer) +} + +fn error_answer(partition: i32, error_code: i16) -> PartitionAnswer { + PartitionAnswer::Answered(error_response(partition, error_code)) +} + +/// Kafka's answer when no record matches a timestamp: no error, and -1 for both fields. +fn not_found_response(partition: i32) -> ListOffsetsPartitionResponse { + ListOffsetsPartitionResponse::default() + .with_partition_index(partition) + .with_timestamp(UNKNOWN_OFFSET) + .with_offset(UNKNOWN_OFFSET) +} + +/// Answers every partition [`resolve_all_topics`] left pending, one record lookup at a time +/// on one Fetch read slot, which each lookup reuses. +/// +/// Looked up one by one rather than at once: a consumer resetting a group by time asks for every +/// partition in one request, and running those in parallel would take every read slot from the +/// consumers that are fetching. A lookup that cannot finish within [`LOOKUP_BUDGET`] or by +/// `deadline` answers [`ERROR_REQUEST_TIMED_OUT`], which the client retries. Repeated entries are +/// looked up once. +async fn locate_records( + bridge: &Arc<IggyBridge>, + unproven_empty: &Mutex<Spells>, + topics: Vec<ResolvedTopic>, + deadline: Instant, +) -> Vec<ListOffsetsTopicResponse> { + let mut slot = None; + let mut answered: HashMap<(TopicName, PendingLookup), ListOffsetsPartitionResponse> = + HashMap::new(); + let mut responses = Vec::with_capacity(topics.len()); + for topic in topics { + let mut partitions = Vec::with_capacity(topic.partitions.len()); + for partition in topic.partitions { + let pending = match partition { + PartitionAnswer::Answered(answer) => { + partitions.push(answer); + continue; + } + PartitionAnswer::Pending(pending) => pending, + }; + let key = (topic.name.clone(), pending); + let answer = if let Some(answer) = answered.get(&key) { + answer.clone() + } else { + let answer = locate_one( + bridge, + unproven_empty, + &mut slot, + &topic.name, + pending, + deadline, + ) + .await; + answered.insert(key, answer.clone()); + answer + }; + partitions.push(answer); + } + responses.push( + ListOffsetsTopicResponse::default() + .with_name(topic.name) + .with_partitions(partitions), + ); + } + drop(slot); + responses +} + +/// Answers one `pending` lookup, reusing `slot` or taking a new one into it. +async fn locate_one( + bridge: &Arc<IggyBridge>, + unproven_empty: &Mutex<Spells>, + slot: &mut Option<FetchSlot>, + topic: &str, + pending: PendingLookup, + request_deadline: Instant, +) -> ListOffsetsPartitionResponse { + let PendingLookup { + partition_index, + partition, + lookup, + .. + } = pending; + let deadline = lookup_deadline(Instant::now(), request_deadline); + let Some(found) = poll_record(bridge, slot, topic, partition, lookup, deadline).await else { + return error_response(partition_index, ERROR_REQUEST_TIMED_OUT); + }; + let last = match (&found, lookup) { + (Ok(None), RecordLookup::AppendedAtOrAfter(_)) => { + match poll_record(bridge, slot, topic, partition, RecordLookup::Last, deadline).await { + Some(last) => Some(last), + None => return error_response(partition_index, ERROR_REQUEST_TIMED_OUT), + } + } + _ => None, + }; + lookup_answer(unproven_empty, topic, pending, found, last.as_ref()) +} + +/// One [`IggyBridge::locate_record`] on `slot`, or on a new slot for the first lookup or after a +/// timed-out poll lost it. +/// `None` if `deadline` has passed, no slot frees, or no answer comes by `deadline`. +async fn poll_record( + bridge: &Arc<IggyBridge>, + slot: &mut Option<FetchSlot>, + topic: &str, + partition: u32, + lookup: RecordLookup, + deadline: Instant, +) -> Option<core::result::Result<Option<LocatedRecord>, BridgeError>> { + // A poll started now would time out at once and drop the slot's client. + if Instant::now() >= deadline { + return None; + } + if slot.is_none() { + // A poll that timed out frees its old slot as it stops, so for a moment the request can + // hold two. + *slot = bridge.fetch_slot(deadline).await; + } + let (located, back) = bridge + .locate_record(slot.take()?, topic, partition, lookup, deadline) + .await?; + *slot = Some(back); + Some(located) +} + +fn lookup_deadline(now: Instant, request_deadline: Instant) -> Instant { + request_deadline.min(now + LOOKUP_BUDGET) +} + +/// The answer to one `pending` lookup, from what it `found` and, after an empty +/// `AppendedAtOrAfter`, the `last` record. +/// +/// Iggy also polls empty on a read fault or before a partition seeds its commit point, and a node +/// that lags the probed one can serve a shorter tail. So "no record" needs the last record appended +/// before the time and reaching the probed high watermark, or, where the probe counted no message, +/// a last poll that is empty too. A last record short of the high watermark is no answer to +/// `Last` either. Anything else answers [`ERROR_NOT_LEADER_OR_FOLLOWER`] to retry. +fn lookup_answer( + unproven_empty: &Mutex<Spells>, + topic: &str, + pending: PendingLookup, + found: core::result::Result<Option<LocatedRecord>, BridgeError>, + last: Option<&core::result::Result<Option<LocatedRecord>, BridgeError>>, +) -> ListOffsetsPartitionResponse { + let PendingLookup { + partition_index, + partition, + lookup, + retained, + high_watermark, + } = pending; + match found { + Ok(Some(record)) if lookup == RecordLookup::Last && !reaches(&record, high_watermark) => { + tracing::debug!( + topic, + partition, + offset = record.offset, + high_watermark, + "ListOffsets last record falls short of the probed high watermark; answering \ + retriable" + ); + error_response(partition_index, ERROR_NOT_LEADER_OR_FOLLOWER) + } + Ok(Some(record)) => ListOffsetsPartitionResponse::default() + .with_partition_index(partition_index) + .with_timestamp(record.timestamp) + .with_offset(record.offset), + Err(error) => { + tracing::warn!(topic, partition, %error, "ListOffsets record lookup by time failed"); + error_response(partition_index, error.to_list_offsets_error_code()) Review Comment: nit: `lookup_answer` answers a failed first poll through `to_list_offsets_error_code` and a failed last poll through `retriable_code`, so one `BridgeError` becomes 6 on one path and -1 on the other. Answer both arms through `retriable_code`, which keeps -1 away from an `AdminClient` that fails the partition on it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
