krishvishal commented on code in PR #4545:
URL: https://github.com/apache/iggy/pull/4545#discussion_r4311062562
##########
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),
Review Comment:
A partition with no stats entry also reads high watermark 0
(`core/server/src/responses.rs:969`), so this -1 can send a reset by time to
LATEST.
Drop this arm, because `lookup_answer` proves -1 from its polls, and update
`BRIDGE_MAPPING.md:337`, `:352-354` and the test at line 843.
##########
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, retriable_code(&error))
+ }
+ Ok(None) => {
+ let proven = match (lookup, last) {
+ (_, Some(Err(error))) => {
+ tracing::warn!(topic, partition, %error, "ListOffsets last
record lookup failed");
+ return error_response(partition_index,
retriable_code(error));
+ }
+ (RecordLookup::Last, _) | (_, Some(Ok(None))) => retained == 0,
+ (RecordLookup::AppendedAtOrAfter(millis),
Some(Ok(Some(last)))) => {
+ last.appended_before(millis) && reaches(last,
high_watermark)
+ }
+ (RecordLookup::AppendedAtOrAfter(_), None) => false,
+ };
+ if proven {
+ return not_found_response(partition_index);
+ }
+ if sight_first(unproven_empty, topic, partition, Instant::now()) {
+ tracing::warn!(
+ topic,
+ partition,
+ retained,
+ high_watermark,
+ ?lookup,
+ ?last,
+ "ListOffsets lookup polled empty without proof that no
record matches; \
+ answering retriable"
+ );
+ } else {
+ tracing::debug!(
+ topic,
+ partition,
+ retained,
+ high_watermark,
+ ?lookup,
+ ?last,
+ "ListOffsets lookup still polls empty without proof that
no record matches"
+ );
+ }
+ error_response(partition_index, ERROR_NOT_LEADER_OR_FOLLOWER)
+ }
+ }
+}
+
+/// Whether `record` is the last below `high_watermark` or beyond it.
+fn reaches(record: &LocatedRecord, high_watermark: u64) -> bool {
+ u64::try_from(record.offset).is_ok_and(|offset| offset.saturating_add(1)
>= high_watermark)
+}
+
+/// `error`'s code, but 6 where it would be -1, which the `AdminClient` fails
the partition on.
+const fn retriable_code(error: &BridgeError) -> i16 {
Review Comment:
This folds a rejected bridge login (-1) into retriable 6, so the
`AdminClient` retries until its timeout. The probe path, Fetch and Produce keep
-1.
If `error.is_bridge_login_rejected()` is true, keep -1 and fold only the
other -1 codes.
--
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]
