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 <[email protected]> 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`].
