You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/10/14 13:16:17 UTC
(iggy) 02/02: feat(gateways): answer Kafka ListOffsets by timestamp and MAX_TIMESTAMP
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch kafka_list_offsets_by_timestamp
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit d3a7ab2da9c01df412d4fbd03b024ce8925d1f55
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Wed Oct 14 15:15:47 2026 +0200
feat(gateways): answer Kafka ListOffsets by timestamp and MAX_TIMESTAMP
---
gateways/kafka/docs/MANUAL_TESTING.md | 4 +-
gateways/kafka/docs/kafka_api_keys_reference.md | 4 +-
gateways/kafka/src/protocol/api.rs | 4 +
gateways/kafka/src/protocol/handlers/fetch.rs | 32 ++-
.../kafka/src/protocol/handlers/list_offsets.rs | 257 +++++++++++++++++----
gateways/kafka/tests/kafka_client_e2e_tests.rs | 6 +-
6 files changed, 255 insertions(+), 52 deletions(-)
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index d9ee15c87..d3ef45776 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -411,8 +411,8 @@ docker run -d --name akhq-iggy --network host \
| ------ | ------ | --------------- |
| -1 | UNKNOWN_SERVER_ERROR | Produce with a bridge: Iggy error with no closer code, or bad bridge login |
| 0 | NONE | Fetch top-level error field only (`ec=0` there does not mean per-partition success - see A6) |
-| 3 | UNKNOWN_TOPIC_OR_PARTITION | Metadata stub, per topic. Produce with a bridge: missing topic or partition |
-| 6 | NOT_LEADER_OR_FOLLOWER | Produce/Fetch/ListOffsets stub (not stored). Produce with a bridge: Iggy unreachable, or the request budget ran out. Fetch with a bridge: Iggy unreachable, too slow, loading the partition, or an offset past the end in its first 30 s on the connection. ListOffsets with a bridge: partition loading, Iggy refused the read for now, or a lookup by time polled empty without proof that no record matches |
+| 3 | UNKNOWN_TOPIC_OR_PARTITION | Metadata stub, per topic. Produce or ListOffsets with a bridge: missing topic or partition |
+| 6 | NOT_LEADER_OR_FOLLOWER | Produce/Fetch/ListOffsets stub (not stored). Produce with a bridge: Iggy unreachable, or the request budget ran out. Fetch with a bridge: Iggy unreachable, too slow, loading the partition, or an offset past the end in its first 30 s on the connection. ListOffsets with a bridge: Iggy unreachable, partition loading, Iggy refused the read for now, or a lookup by time polled empty without proof that no record matches |
| 7 | REQUEST_TIMED_OUT | Produce with a bridge: deadline passed, or connection lost mid-send (may be stored). ListOffsets with a bridge: past the 100-topic cap, the request deadline passed, or a lookup by time ran out of its time |
| 10 | MESSAGE_TOO_LARGE | Produce with a bridge: record, send or partition too large, even alone |
| 17 | INVALID_TOPIC_EXCEPTION | Produce with a bridge: bad topic name |
diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md
index 5bad9e9a2..871053feb 100644
--- a/gateways/kafka/docs/kafka_api_keys_reference.md
+++ b/gateways/kafka/docs/kafka_api_keys_reference.md
@@ -270,8 +270,8 @@ Key new minimums:
| Category | Count | Notes |
| ---------- | :-----: | ------- |
-| 🔴 Bridge (data path) | 9 | Produce, Fetch, Metadata, OffsetCommit, OffsetFetch, SaslHandshake, ApiVersions, SaslAuthenticate, ShareFetch |
-| 🟠 Required Stub (client state machine) | 13 | ListOffsets, consumer group (8-14), CreateTopics, InitProducerId (22), ConsumerGroupHeartbeat (68), ShareGroupHeartbeat (77), ShareAcknowledge (80) |
+| 🔴 Bridge (data path) | 10 | Produce, Fetch, ListOffsets, Metadata, OffsetCommit, OffsetFetch, SaslHandshake, ApiVersions, SaslAuthenticate, ShareFetch |
+| 🟠 Required Stub (client state machine) | 10 | Consumer group (10-14), CreateTopics (19), InitProducerId (22), ConsumerGroupHeartbeat (68), ShareGroupHeartbeat (77), ShareAcknowledge (80) |
| Implemented | 2 | DescribeGroups (15), ListGroups (16) |
| 🟡 Optional Stub (admin/observability) | 39 | Can return `UNSUPPORTED_VERSION` or `NOT_CONTROLLER` safely |
| ❌ Unadvertised | 5 | AddPartitionsToTxn (24), AddOffsetsToTxn (25), EndTxn (26), TxnOffsetCommit (28), and DeleteRecords (21). Absent from ApiVersions, so a conforming client never sends one |
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index 13e7498ed..43de08738 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -336,6 +336,9 @@ pub struct GatewayState {
/// Per Kafka topic and partition, how long probes read it as loading. Fetch and `ListOffsets`
/// share it, since every consumer sees the same probe.
pub(crate) loading: Mutex<fetch::Spells>,
+ /// Per Kafka topic and partition, how long `ListOffsets` lookups by time poll empty without
+ /// proof that no record matches. Only the first of a spell warns.
+ pub(crate) unproven_empty: Mutex<fetch::Spells>,
/// Topic probes that every Fetch shares.
pub(crate) probe_board: ProbeBoard,
}
@@ -374,6 +377,7 @@ impl GatewayState {
produce_large: Arc::new(Semaphore::new(PRODUCE_LARGE)),
groups,
loading: Mutex::default(),
+ unproven_empty: Mutex::default(),
probe_board: ProbeBoard::default(),
}
}
diff --git a/gateways/kafka/src/protocol/handlers/fetch.rs b/gateways/kafka/src/protocol/handlers/fetch.rs
index bd90fe5e4..622abc44d 100644
--- a/gateways/kafka/src/protocol/handlers/fetch.rs
+++ b/gateways/kafka/src/protocol/handlers/fetch.rs
@@ -1273,7 +1273,7 @@ impl Spell {
}
}
-/// Per Kafka topic and partition, a spell of probes that read it as loading.
+/// Per Kafka topic and partition, a spell of sightings, such as probes that read it as loading.
pub(crate) type Spells = PartitionMap<Spell>;
impl Spells {
@@ -1316,6 +1316,21 @@ pub(crate) fn sight_loading(
sighting.within_grace()
}
+/// Counts a sighting of `partition` of `topic`, and returns whether it starts a new spell, so a
+/// caller can log the first sighting louder than the retries that follow.
+pub(crate) fn sight_first(
+ spells: &Mutex<Spells>,
+ topic: &str,
+ partition: u32,
+ now: Instant,
+) -> bool {
+ spells
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner)
+ .sight(topic, partition, now)
+ .is_new()
+}
+
/// Per Kafka topic, then partition, the offset that one connection's Fetches cannot place. A read
/// in range ends it.
pub(crate) type Unplaced = PartitionMap<UnplacedOffset>;
@@ -2346,6 +2361,21 @@ mod tests {
);
}
+ #[test]
+ fn given_repeated_sightings_when_counted_should_report_only_the_first_of_each_spell() {
+ let spells = Mutex::default();
+ let start = Instant::now();
+ let at = |seconds| start + Duration::from_secs(seconds);
+ assert!(sight_first(&spells, "a", 0, at(0)), "a new spell");
+ assert!(!sight_first(&spells, "a", 0, at(1)));
+ assert!(sight_first(&spells, "a", 1, at(1)), "per partition");
+ assert!(!sight_first(&spells, "a", 0, at(20)));
+ assert!(
+ sight_first(&spells, "a", 0, at(20) + UNPLACED_GRACE),
+ "a quiet grace ends the spell"
+ );
+ }
+
#[test]
fn given_an_offset_past_the_end_when_it_stays_so_should_answer_6_for_the_grace_then_1() {
let unplaced = Mutex::default();
diff --git a/gateways/kafka/src/protocol/handlers/list_offsets.rs b/gateways/kafka/src/protocol/handlers/list_offsets.rs
index 58e2027fb..9c5573c07 100644
--- a/gateways/kafka/src/protocol/handlers/list_offsets.rs
+++ b/gateways/kafka/src/protocol/handlers/list_offsets.rs
@@ -34,11 +34,11 @@ use crate::bridge::{BridgeError, IggyBridge};
use crate::error::Result;
use crate::protocol::api::{
API_KEY_LIST_OFFSETS, ApiVersionRange, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_REQUEST_TIMED_OUT,
- ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, REQUEST_DEADLINE,
- UNKNOWN_OFFSET,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome,
+ REQUEST_DEADLINE, UNKNOWN_OFFSET,
};
use crate::protocol::bounds_guard::validate_list_offsets_shape;
-use crate::protocol::handlers::fetch::{Spells, sight_loading};
+use crate::protocol::handlers::fetch::{Spells, sight_first, sight_loading};
use crate::protocol::handlers::{decode_guarded, decode_request, encode_message, respond_or_close};
pub const RANGE: ApiVersionRange = ApiVersionRange {
@@ -99,7 +99,7 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
// No `timeout_ms` below v10, so the ceiling is fixed. It bounds the sum of the topic calls.
let deadline = Instant::now() + REQUEST_DEADLINE;
let resolved = resolve_all_topics(bridge, &state.loading, &req.topics, deadline).await;
- let topics = locate_records(bridge, resolved, deadline).await;
+ let topics = locate_records(bridge, &state.unproven_empty, resolved, deadline).await;
let resp = ListOffsetsResponse::default().with_topics(topics);
respond_or_close(encode_message(&resp, api_version, 256), "ListOffsets")
}
@@ -271,7 +271,9 @@ enum PartitionAnswer {
/// A record lookup still to run, with what the probe saw of its partition.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct PendingLookup {
- index: i32,
+ /// The requested index, as the response echoes it.
+ partition_index: i32,
+ /// The same index, as Iggy takes it.
partition: u32,
lookup: RecordLookup,
/// Messages the probe counted. 0 also when the server has no stats for the partition.
@@ -369,7 +371,7 @@ fn resolve_one_partition(
millis => RecordLookup::AppendedAtOrAfter(millis),
};
return PartitionAnswer::Pending(PendingLookup {
- index,
+ partition_index: index,
partition: partition_index,
lookup,
retained: probe.messages_count,
@@ -402,6 +404,7 @@ fn not_found_response(partition: i32) -> ListOffsetsPartitionResponse {
/// looked up once.
async fn locate_records(
bridge: &Arc<IggyBridge>,
+ unproven_empty: &Mutex<Spells>,
topics: Vec<ResolvedTopic>,
deadline: Instant,
) -> Vec<ListOffsetsTopicResponse> {
@@ -423,7 +426,15 @@ async fn locate_records(
let answer = if let Some(answer) = answered.get(&key) {
answer.clone()
} else {
- let answer = locate_one(bridge, &mut slot, &topic.name, pending, deadline).await;
+ let answer = locate_one(
+ bridge,
+ unproven_empty,
+ &mut slot,
+ &topic.name,
+ pending,
+ deadline,
+ )
+ .await;
answered.insert(key, answer.clone());
answer
};
@@ -442,31 +453,32 @@ async fn locate_records(
/// 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 {
- index,
+ 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(index, ERROR_REQUEST_TIMED_OUT);
+ 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(index, ERROR_REQUEST_TIMED_OUT),
+ None => return error_response(partition_index, ERROR_REQUEST_TIMED_OUT),
}
}
_ => None,
};
- lookup_answer(topic, pending, found, last.as_ref())
+ 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
@@ -506,57 +518,97 @@ fn lookup_deadline(now: Instant, request_deadline: Instant) -> Instant {
/// 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. Anything else answers [`ERROR_NOT_LEADER_OR_FOLLOWER`] to retry.
+/// 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 {
- index: partition,
+ 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)
+ .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, error.to_list_offsets_error_code())
+ error_response(partition_index, error.to_list_offsets_error_code())
}
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)
- && u64::try_from(last.offset)
- .is_ok_and(|offset| offset.saturating_add(1) >= high_watermark)
+ last.appended_before(millis) && reaches(last, high_watermark)
}
- _ => false,
+ (RecordLookup::AppendedAtOrAfter(_), None) => false,
};
if proven {
- return not_found_response(partition);
+ return not_found_response(partition_index);
}
- tracing::warn!(
- topic,
- partition,
- retained,
- high_watermark,
- ?lookup,
- ?last,
- "ListOffsets lookup polled empty without proof that no record matches; \
- answering retriable"
- );
- error_response(partition, ERROR_NOT_LEADER_OR_FOLLOWER)
+ 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 {
+ match error.to_list_offsets_error_code() {
+ ERROR_UNKNOWN_SERVER_ERROR => ERROR_NOT_LEADER_OR_FOLLOWER,
+ code => code,
+ }
+}
+
fn offset_response(partition: i32, offset: i64) -> ListOffsetsPartitionResponse {
ListOffsetsPartitionResponse::default()
.with_partition_index(partition)
@@ -650,7 +702,7 @@ mod tests {
fn pending(lookup: RecordLookup, retained: u64, high_watermark: u64) -> PendingLookup {
PendingLookup {
- index: 3,
+ partition_index: 3,
partition: 3,
lookup,
retained,
@@ -715,7 +767,7 @@ mod tests {
panic!("timestamp {requested} answered without a lookup: {answer:?}");
};
let want = PendingLookup {
- index: 0,
+ partition_index: 0,
partition: 0,
lookup: record_lookup,
retained: 42,
@@ -755,9 +807,21 @@ mod tests {
fn an_uncounted_partition_whose_lookups_both_poll_empty_is_not_found() {
// Every message expired: the probe and both polls agree nothing is retained.
let time = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS);
- let resp = lookup_answer("orders", pending(time, 0, 42), Ok(None), Some(&Ok(None)));
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(time, 0, 42),
+ Ok(None),
+ Some(&Ok(None)),
+ );
assert_eq!(answer(&resp), (ERROR_NONE, -1, -1));
- let resp = lookup_answer("orders", pending(RecordLookup::Last, 0, 42), Ok(None), None);
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(RecordLookup::Last, 0, 42),
+ Ok(None),
+ None,
+ );
assert_eq!(answer(&resp), (ERROR_NONE, -1, -1));
}
@@ -765,7 +829,13 @@ mod tests {
fn an_uncounted_partition_answers_the_record_a_lookup_finds() {
let time = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS);
let found = Ok(Some(record(17, REQUESTED_MILLIS + 5)));
- let resp = lookup_answer("orders", pending(time, 0, 42), found, None);
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(time, 0, 42),
+ found,
+ None,
+ );
assert_eq!(answer(&resp), (ERROR_NONE, 17, REQUESTED_MILLIS + 5));
}
@@ -838,7 +908,13 @@ mod tests {
}
}
let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS);
- let resp = lookup_answer("orders", pending(lookup, 42, 42), Err(refused()), None);
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(lookup, 42, 42),
+ Err(refused()),
+ None,
+ );
assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER);
}
@@ -860,7 +936,13 @@ mod tests {
fn a_found_record_answers_its_offset_and_timestamp() {
let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS);
let found = Ok(Some(record(17, REQUESTED_MILLIS + 123)));
- let resp = lookup_answer("orders", pending(lookup, 42, 42), found, None);
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(lookup, 42, 42),
+ found,
+ None,
+ );
assert_eq!(answer(&resp), (ERROR_NONE, 17, REQUESTED_MILLIS + 123));
}
@@ -868,7 +950,13 @@ mod tests {
fn a_failed_lookup_answers_its_error() {
let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS);
let failed = Err(BridgeError::Timeout);
- let resp = lookup_answer("orders", pending(lookup, 42, 42), failed, None);
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(lookup, 42, 42),
+ failed,
+ None,
+ );
assert_eq!(
resp.error_code,
BridgeError::Timeout.to_list_offsets_error_code()
@@ -881,6 +969,7 @@ mod tests {
for high_watermark in [41, 42] {
let last = Some(Ok(Some(record(41, REQUESTED_MILLIS - 1))));
let resp = lookup_answer(
+ &Mutex::default(),
"orders",
pending(lookup, 42, high_watermark),
Ok(None),
@@ -895,7 +984,13 @@ mod tests {
fn an_empty_poll_whose_last_record_falls_short_of_the_high_watermark_answers_retriable() {
let lookup = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS);
let last = Some(Ok(Some(record(40, REQUESTED_MILLIS - 1))));
- let resp = lookup_answer("orders", pending(lookup, 42, 42), Ok(None), last.as_ref());
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(lookup, 42, 42),
+ Ok(None),
+ last.as_ref(),
+ );
assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER);
}
@@ -907,19 +1002,92 @@ mod tests {
Some(Ok(Some(record(41, REQUESTED_MILLIS)))),
Some(Ok(Some(record(41, REQUESTED_MILLIS + 1)))),
Some(Ok(None)),
- Some(Err(BridgeError::Timeout)),
None,
];
for last in unproven {
let shown = format!("{last:?}");
- let resp = lookup_answer("orders", pending(lookup, 42, 42), Ok(None), last.as_ref());
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(lookup, 42, 42),
+ Ok(None),
+ last.as_ref(),
+ );
assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER, "{shown}");
}
}
+ #[test]
+ fn an_uncounted_partition_whose_last_record_matches_the_time_answers_retriable() {
+ let time = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS);
+ let last = Some(Ok(Some(record(41, REQUESTED_MILLIS + 1))));
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(time, 0, 42),
+ Ok(None),
+ last.as_ref(),
+ );
+ assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER);
+ }
+
+ #[test]
+ fn a_failed_last_poll_answers_its_error_but_never_unknown_server_error() {
+ let time = RecordLookup::AppendedAtOrAfter(REQUESTED_MILLIS);
+ let failures = [
+ (BridgeError::Timeout, ERROR_REQUEST_TIMED_OUT),
+ (
+ BridgeError::InvalidConfig("bad".to_string()),
+ ERROR_NOT_LEADER_OR_FOLLOWER,
+ ),
+ ];
+ for (error, code) in failures {
+ let shown = format!("{error:?}");
+ let last = Some(Err(error));
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(time, 42, 42),
+ Ok(None),
+ last.as_ref(),
+ );
+ assert_eq!(resp.error_code, code, "{shown}");
+ }
+ }
+
+ /// The polled node can lag the probed one and serve a last record below its high watermark.
+ #[test]
+ fn a_last_record_short_of_the_high_watermark_answers_retriable() {
+ for retained in [0, 42] {
+ let short = Ok(Some(record(40, REQUESTED_MILLIS)));
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(RecordLookup::Last, retained, 42),
+ short,
+ None,
+ );
+ assert_eq!(resp.error_code, ERROR_NOT_LEADER_OR_FOLLOWER, "{retained}");
+ let reaching = Ok(Some(record(41, REQUESTED_MILLIS)));
+ let resp = lookup_answer(
+ &Mutex::default(),
+ "orders",
+ pending(RecordLookup::Last, retained, 42),
+ reaching,
+ None,
+ );
+ assert_eq!(
+ answer(&resp),
+ (ERROR_NONE, 41, REQUESTED_MILLIS),
+ "{retained}"
+ );
+ }
+ }
+
#[test]
fn an_empty_last_poll_while_messages_are_retained_answers_retriable() {
let resp = lookup_answer(
+ &Mutex::default(),
"orders",
pending(RecordLookup::Last, 42, 42),
Ok(None),
@@ -960,6 +1128,7 @@ mod tests {
let passed = Instant::now();
let resp = locate_one(
&bridge,
+ &Mutex::default(),
&mut slot,
"orders",
pending(lookup, 42, 42),
@@ -978,7 +1147,7 @@ mod tests {
)),
partitions: vec![PartitionAnswer::Pending(pending(time, 42, 42))],
}];
- let responses = locate_records(&bridge, topics, passed).await;
+ let responses = locate_records(&bridge, &Mutex::default(), topics, passed).await;
assert_eq!(
responses[0].partitions[0].error_code,
ERROR_REQUEST_TIMED_OUT
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs
index d34aebb4c..3fcc633fc 100644
--- a/gateways/kafka/tests/kafka_client_e2e_tests.rs
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -1411,11 +1411,11 @@ fn now_millis() -> i64 {
i64::try_from(elapsed.as_millis()).expect("now fits an i64")
}
-/// `millis` as the `YYYY-MM-DDTHH:mm:SS.sss` that `--to-datetime` reads, in UTC, which is the
-/// client container's zone.
+/// `millis` as the `YYYY-MM-DDTHH:mm:SS.sssZ` that `--to-datetime` reads. The zone is explicit,
+/// so the client container's own zone does not matter.
fn utc_datetime(millis: i64) -> String {
let micros = u64::try_from(millis).expect("a time after the epoch") * 1000;
- IggyTimestamp::from(micros).to_utc_string("%Y-%m-%dT%H:%M:%S%.3f")
+ IggyTimestamp::from(micros).to_utc_string("%Y-%m-%dT%H:%M:%S%.3fZ")
}
/// Runs `command` under the same GNU `timeout` bound as [`run_client`].