From 90f9d1ee00b06445bb7d4fa4949b536c342fff63 Mon Sep 17 00:00:00 2001 From: Andrea Ross <168456375+solace-aross@users.noreply.github.com> Date: Sat, 3 Oct 2026 11:42:37 -0400 Subject: [PATCH 1/5] fix: ListOffsets by timestamp returns -1, not 0, when no record matches Kafka's real broker answers a timestamp-based ListOffsets lookup with offset=-1/timestamp=-1 (UNKNOWN_OFFSET/UNKNOWN_TIMESTAMP) when no record matches, via KafkaApis.scala's buildErrorResponse. Nisshi was answering offset=0/timestamp=None instead, which the Java client's offsetsForTimes() treats as a valid hit (it only filters out entries where offset == UNKNOWN_OFFSET), so callers would then crash constructing OffsetAndTimestamp from a negative timestamp. Earliest/Latest on an empty partition still correctly return offset 0; the bug was specific to Timestamp lookups with no match. Fixed across all six storage backends (pg, lite, limbo, dynostore, slatedb, null) and corrected the service-layer default from 0 to -1. Signed-off-by: Andrea Ross <168456375+solace-aross@users.noreply.github.com> Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01D2qdgPhMyLGZN5dLR8CVHs Signed-off-by: Andrea Ross <168456375+solace-aross@users.noreply.github.com> --- nisshi-broker/tests/it/list_offsets.rs | 92 ++++++++++- .../tests/it/storage_list_offsets.rs | 143 ++++++++++++++++++ nisshi-storage-dynostore/src/dynostore.rs | 12 +- nisshi-storage-null/src/lib.rs | 14 +- nisshi-storage-slatedb/src/storage.rs | 7 +- nisshi-storage-sql/src/limbo.rs | 11 +- nisshi-storage-sql/src/lite.rs | 11 +- nisshi-storage-sql/src/pg.rs | 11 +- nisshi-storage/src/service/list_offsets.rs | 9 +- 9 files changed, 293 insertions(+), 17 deletions(-) diff --git a/nisshi-broker/tests/it/list_offsets.rs b/nisshi-broker/tests/it/list_offsets.rs index 2c3915e45..7be337467 100644 --- a/nisshi-broker/tests/it/list_offsets.rs +++ b/nisshi-broker/tests/it/list_offsets.rs @@ -442,9 +442,12 @@ pub async fn multiple_record(broker: Broker) -> Result<()> { .is_some_and(|timestamp| timestamp > 0) ); + // Other partitions are empty: no record matches this timestamp, so this is + // a not-found result, not an empty-partition result. Offset is -1, matching + // Kafka's buildErrorResponse default. for partition in partitions[1..].iter() { assert_eq!(i16::from(ErrorCode::None), partition.error_code); - assert_eq!(Some(0), partition.offset); + assert_eq!(Some(-1), partition.offset); assert_eq!(Some(-1), partition.timestamp); } @@ -492,9 +495,12 @@ pub async fn multiple_record(broker: Broker) -> Result<()> { .is_some_and(|timestamp| timestamp > 0) ); + // Other partitions are empty: no record matches this timestamp, so this is + // a not-found result, not an empty-partition result. Offset is -1, matching + // Kafka's buildErrorResponse default. for partition in partitions[1..].iter() { assert_eq!(i16::from(ErrorCode::None), partition.error_code); - assert_eq!(Some(0), partition.offset); + assert_eq!(Some(-1), partition.offset); assert_eq!(Some(-1), partition.timestamp); } @@ -542,9 +548,12 @@ pub async fn multiple_record(broker: Broker) -> Result<()> { .is_some_and(|timestamp| timestamp > 0) ); + // Other partitions are empty: no record matches this timestamp, so this is + // a not-found result, not an empty-partition result. Offset is -1, matching + // Kafka's buildErrorResponse default. for partition in partitions[1..].iter() { assert_eq!(i16::from(ErrorCode::None), partition.error_code); - assert_eq!(Some(0), partition.offset); + assert_eq!(Some(-1), partition.offset); assert_eq!(Some(-1), partition.timestamp); } @@ -638,8 +647,11 @@ where assert!(!items.is_empty()); + // The topic has no records at all, so no record matches this timestamp: + // it's a not-found result at the raw storage layer (the -1 default is + // applied by the broker service, not the storage backend itself). for (_toptition, response) in items { - assert_eq!(Some(0), response.offset); + assert_eq!(None, response.offset); assert_eq!(None, response.timestamp); } @@ -733,8 +745,11 @@ where .list_offsets(IsolationLevel::ReadUncommitted, &offsets[..]) .await?; + // No record's timestamp is >= `after`: not-found at the raw storage layer + // (the -1 default is applied by the broker service, not the storage + // backend itself). assert_eq!(1, responses.len()); - assert_eq!(Some(0), responses[0].1.offset); + assert_eq!(None, responses[0].1.offset); Ok(()) } @@ -931,6 +946,73 @@ mod lite { } } +#[cfg(feature = "turso")] +mod turso { + use super::*; + use nisshi_storage::ArcDynStorage; + + async fn storage_container( + cluster: impl Into + Clone, + node: i32, + ) -> Result { + common::storage_container( + StorageType::Turso, + cluster, + node, + Url::parse("tcp://127.0.0.1/")?, + None, + ) + .await + } + + #[ignore] + #[tokio::test] + async fn multiple_record() -> Result<()> { + let _guard = init_tracing()?; + + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + + let sc = storage_container(cluster_id, broker_id).await?; + register_broker(cluster_id, broker_id, &sc).await?; + + let broker = broker(sc)?; + super::multiple_record(broker).await + } + + #[ignore] + #[tokio::test] + async fn new_topic() -> Result<()> { + let _guard = init_tracing()?; + + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + + super::new_topic( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } + + #[ignore] + #[tokio::test] + async fn single_record() -> Result<()> { + let _guard = init_tracing()?; + + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + + super::single_record( + cluster_id, + broker_id, + storage_container(cluster_id, broker_id).await?, + ) + .await + } +} + #[cfg(feature = "slatedb")] mod slatedb { use super::*; diff --git a/nisshi-broker/tests/it/storage_list_offsets.rs b/nisshi-broker/tests/it/storage_list_offsets.rs index 1d44caaec..c535cc40f 100644 --- a/nisshi-broker/tests/it/storage_list_offsets.rs +++ b/nisshi-broker/tests/it/storage_list_offsets.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::time::SystemTime; + use crate::common::{ alphanumeric_string, init_tracing, lite_storage, memory_storage, postgres_storage, slate_storage, @@ -107,6 +109,91 @@ async fn simple(storage: impl Storage + Clone, broker_id: i32) -> Result<()> { Ok(()) } +/// A Timestamp lookup that matches no record (here: an empty topic) is a +/// not-found result, not an empty-partition result. At the wire level, Kafka's +/// `buildErrorResponse` (KafkaApis.scala) sends offset=-1/timestamp=-1 with +/// error_code=NONE for this case, and `ListOffsetsService` must apply that +/// default rather than the Earliest/Latest empty-partition default of 0. +async fn timestamp_no_match(storage: impl Storage + Clone, broker_id: i32) -> Result<()> { + let extensions = Extensions::default(); + + let create_topic = CreateTopicsService { + storage: storage.clone(), + }; + + let topic = &alphanumeric_string(15)[..]; + + let num_partitions = 1; + let replication_factor = 0; + + { + let response = create_topic + .serve(RequestInput { + request: CreateTopicsRequest::default() + .validate_only(Some(false)) + .topics(Some( + [CreatableTopic::default() + .name(topic.into()) + .num_partitions(num_partitions) + .replication_factor(replication_factor) + .assignments(Some([].into())) + .configs(Some([].into()))] + .into(), + )), + extensions: extensions.clone(), + }) + .await?; + + let topics = response.topics.as_deref().unwrap_or_default(); + assert_eq!(1, topics.len()); + assert_eq!(ErrorCode::None, ErrorCode::try_from(topics[0].error_code)?); + } + + let service = ListOffsetsService { + storage: storage.clone(), + }; + + let response = service + .serve(RequestInput { + request: ListOffsetsRequest::default() + .isolation_level(Some(IsolationLevel::ReadUncommitted.into())) + .replica_id(broker_id) + .topics(Some( + [ListOffsetsTopic::default() + .name(topic.into()) + .partitions(Some( + [ListOffsetsPartition::default() + .current_leader_epoch(Some(-1)) + .max_num_offsets(Some(3)) + .partition_index(0) + .timestamp(ListOffset::Timestamp(SystemTime::now()).try_into()?)] + .into(), + ))] + .into(), + )), + extensions: extensions.clone(), + }) + .await?; + + let topics = response.topics.as_deref().unwrap_or_default(); + assert_eq!(1, topics.len()); + assert_eq!(topic, topics[0].name); + + let partitions = topics[0].partitions.as_deref().unwrap_or_default(); + assert_eq!(1, partitions.len()); + assert_eq!(0, partitions[0].partition_index); + assert!(partitions[0].old_style_offsets.is_none()); + assert_eq!( + ErrorCode::None, + ErrorCode::try_from(partitions[0].error_code)? + ); + assert_eq!(Some(-1), partitions[0].timestamp); + assert_eq!(Some(-1), partitions[0].offset); + assert_eq!(Some(0), partitions[0].leader_epoch); + + Ok(()) +} + #[cfg(feature = "dynostore")] mod in_memory { use super::*; @@ -131,6 +218,20 @@ mod in_memory { Ok(()) } + + #[tokio::test] + async fn timestamp_no_match() -> Result<()> { + let _guard = init_tracing()?; + + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + + let storage = storage_container(cluster_id, broker_id).await?; + + super::timestamp_no_match(storage, broker_id).await?; + + Ok(()) + } } #[cfg(feature = "libsql")] @@ -157,6 +258,20 @@ mod lite { Ok(()) } + + #[tokio::test] + async fn timestamp_no_match() -> Result<()> { + let _guard = init_tracing()?; + + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + + let storage = storage_container(cluster_id, broker_id).await?; + + super::timestamp_no_match(storage, broker_id).await?; + + Ok(()) + } } #[cfg(feature = "slatedb")] @@ -183,6 +298,20 @@ mod slatedb { Ok(()) } + + #[tokio::test] + async fn timestamp_no_match() -> Result<()> { + let _guard = init_tracing()?; + + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + + let storage = storage_container(cluster_id, broker_id).await?; + + super::timestamp_no_match(storage, broker_id).await?; + + Ok(()) + } } #[cfg(feature = "postgres")] @@ -209,4 +338,18 @@ mod pg { Ok(()) } + + #[tokio::test] + async fn timestamp_no_match() -> Result<()> { + let _guard = init_tracing()?; + + let cluster_id = Uuid::now_v7(); + let broker_id = rng().random_range(0..i32::MAX); + + let storage = storage_container(cluster_id, broker_id).await?; + + super::timestamp_no_match(storage, broker_id).await?; + + Ok(()) + } } diff --git a/nisshi-storage-dynostore/src/dynostore.rs b/nisshi-storage-dynostore/src/dynostore.rs index 1dfd45e01..93941880d 100644 --- a/nisshi-storage-dynostore/src/dynostore.rs +++ b/nisshi-storage-dynostore/src/dynostore.rs @@ -1339,11 +1339,21 @@ impl Storage for DynoStore { }, )) } else { + // Earliest/Latest on an empty partition is offset 0 (nothing + // written yet). A Timestamp lookup that matches no record is + // "not found", not "empty": leave offset as None so the caller + // applies Kafka's not-found default (-1), matching + // KafkaApis.scala's buildErrorResponse. + let offset = match offset_request { + ListOffset::Earliest | ListOffset::Latest => Some(0), + ListOffset::Timestamp(_) => None, + }; + responses.push(( topition.to_owned(), ListOffsetResponse { error_code: ErrorCode::None, - offset: Some(0), + offset, ..Default::default() }, )) diff --git a/nisshi-storage-null/src/lib.rs b/nisshi-storage-null/src/lib.rs index 2377b4c15..23ddf86bd 100644 --- a/nisshi-storage-null/src/lib.rs +++ b/nisshi-storage-null/src/lib.rs @@ -221,13 +221,23 @@ impl Storage for Engine { ) -> Result> { Ok(offsets .iter() - .map(|(topition, _)| { + .map(|(topition, list_offset)| { + // Earliest/Latest on an empty (here: nonexistent) partition is + // offset 0. A Timestamp lookup never matches a record in this + // backend, so it's "not found": leave offset as None so the + // caller applies Kafka's not-found default (-1), matching + // KafkaApis.scala's buildErrorResponse. + let offset = match list_offset { + ListOffset::Earliest | ListOffset::Latest => Some(0), + ListOffset::Timestamp(_) => None, + }; + ( topition.to_owned(), ListOffsetResponse { error_code: ErrorCode::None, timestamp: None, - offset: Some(0), + offset, }, ) }) diff --git a/nisshi-storage-slatedb/src/storage.rs b/nisshi-storage-slatedb/src/storage.rs index e06967b27..7e7bc85fb 100644 --- a/nisshi-storage-slatedb/src/storage.rs +++ b/nisshi-storage-slatedb/src/storage.rs @@ -1387,10 +1387,13 @@ impl Storage for Engine { offset: Some(offset), timestamp: to_system_time(ts).ok(), }, - // Match PostgreSQL behavior: return offset 0 when no match found + // No record's timestamp is >= target: "not found", not + // "empty". Leave offset/timestamp as None so the caller + // applies Kafka's not-found default (-1), matching + // KafkaApis.scala's buildErrorResponse. None => ListOffsetResponse { error_code: ErrorCode::None, - offset: Some(0), + offset: None, timestamp: None, }, } diff --git a/nisshi-storage-sql/src/limbo.rs b/nisshi-storage-sql/src/limbo.rs index ed123db50..83105e756 100644 --- a/nisshi-storage-sql/src/limbo.rs +++ b/nisshi-storage-sql/src/limbo.rs @@ -2035,8 +2035,15 @@ impl Storage for Engine { .inspect(|result| debug!(?result))? .map_or_else( || { - let timestamp = None; - let offset = Some(0); + // Earliest/Latest on an empty partition is offset 0 (nothing + // written yet). A Timestamp lookup that matches no record is + // "not found", not "empty": leave offset/timestamp as None so + // the caller applies Kafka's not-found default (-1), matching + // KafkaApis.scala's buildErrorResponse. + let (offset, timestamp) = match offset_type { + ListOffsetRequest::Earliest | ListOffsetRequest::Latest => (Some(0), None), + ListOffsetRequest::Timestamp(_) => (None, None), + }; debug!( cluster = self.cluster, ?topition, diff --git a/nisshi-storage-sql/src/lite.rs b/nisshi-storage-sql/src/lite.rs index ca5ee2054..99de39111 100644 --- a/nisshi-storage-sql/src/lite.rs +++ b/nisshi-storage-sql/src/lite.rs @@ -3333,8 +3333,15 @@ impl Storage for Delegate { .inspect(|result| debug!(?result))? .map_or_else( || { - let timestamp = None; - let offset = Some(0); + // Earliest/Latest on an empty partition is offset 0 (nothing + // written yet). A Timestamp lookup that matches no record is + // "not found", not "empty": leave offset/timestamp as None so + // the caller applies Kafka's not-found default (-1), matching + // KafkaApis.scala's buildErrorResponse. + let (offset, timestamp) = match offset_type { + ListOffset::Earliest | ListOffset::Latest => (Some(0), None), + ListOffset::Timestamp(_) => (None, None), + }; debug!( cluster = self.cluster, ?topition, diff --git a/nisshi-storage-sql/src/pg.rs b/nisshi-storage-sql/src/pg.rs index ebecac730..9c2d63a60 100644 --- a/nisshi-storage-sql/src/pg.rs +++ b/nisshi-storage-sql/src/pg.rs @@ -2737,8 +2737,15 @@ impl Storage for Postgres { .inspect(|result| debug!(?result))? .map_or_else( || { - let timestamp = None; - let offset = Some(0); + // Earliest/Latest on an empty partition is offset 0 (nothing + // written yet). A Timestamp lookup that matches no record is + // "not found", not "empty": leave offset/timestamp as None so + // the caller applies Kafka's not-found default (-1), matching + // KafkaApis.scala's buildErrorResponse. + let (offset, timestamp) = match offset_type { + ListOffset::Earliest | ListOffset::Latest => (Some(0), None), + ListOffset::Timestamp(_) => (None, None), + }; debug!( cluster = self.cluster, ?topition, diff --git a/nisshi-storage/src/service/list_offsets.rs b/nisshi-storage/src/service/list_offsets.rs index f39403ed2..e82249b90 100644 --- a/nisshi-storage/src/service/list_offsets.rs +++ b/nisshi-storage/src/service/list_offsets.rs @@ -165,7 +165,14 @@ where .unwrap_or(Some(-1)) .or(Some(-1)), ) - .offset(offset.offset().or(Some(0))) + // Kafka's buildErrorResponse (KafkaApis.scala) + // sends offset=-1 on every error path, not just + // when a Timestamp lookup finds no match. This + // default fires today: SlateDb's storage.rs + // leaves offset: None on its "topic not found" and + // "partition out of range" paths, both returning + // UnknownTopicOrPartition. + .offset(offset.offset().or(Some(-1))) .leader_epoch(Some(0)), ) } else { From 7f6ce57db7d1b6b196633ffe609535a189dfbf74 Mon Sep 17 00:00:00 2001 From: Andrea Ross <168456375+solace-aross@users.noreply.github.com> Date: Sat, 3 Oct 2026 12:00:58 -0400 Subject: [PATCH 2/5] fix: avoid same-millisecond race in list_offsets single_record test `after` was captured with SystemTime::now() immediately after produce returned. The record's stored timestamp is captured just before produce as integer milliseconds (to_timestamp(&SystemTime::now())), so when both captures land in the same millisecond, lite's `r.timestamp >= $4` lookup legitimately matches at offset 0 instead of returning no match. The new L752 assertion (Some(0) -> None) made this previously-invisible race observable, failing ~5-10% of the time under concurrent test load (0/30 isolated). Push `after` a full second past SystemTime::now() so it's deterministically later than any record's millisecond timestamp, regardless of scheduling delay between capturing the record's timestamp and this one. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01D2qdgPhMyLGZN5dLR8CVHs Signed-off-by: Andrea Ross <168456375+solace-aross@users.noreply.github.com> --- nisshi-broker/tests/it/list_offsets.rs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/nisshi-broker/tests/it/list_offsets.rs b/nisshi-broker/tests/it/list_offsets.rs index 7be337467..35cd14307 100644 --- a/nisshi-broker/tests/it/list_offsets.rs +++ b/nisshi-broker/tests/it/list_offsets.rs @@ -707,7 +707,14 @@ where .inspect(|offset| debug!(?offset))? ); - let after = SystemTime::now(); + // Strictly later than any produced record's millisecond timestamp: the + // record's stored timestamp is captured just before `produce` as + // `to_timestamp(&SystemTime::now())` (integer milliseconds). Taking + // `after` immediately after `produce` returns can land in the same + // millisecond, which makes lite's `r.timestamp >= $4` lookup match at + // offset 0 instead of returning no match, flaking this assertion under + // load (observed ~5-10% under concurrent test load, 0/30 isolated). + let after = SystemTime::now() + Duration::from_secs(1); debug!(after = to_timestamp(&after)?); let offsets = [(topition.clone(), ListOffset::Latest)]; From 00d7261e213c59bf9af503805424b713018e6d59 Mon Sep 17 00:00:00 2001 From: Andrea Ross <168456375+solace-aross@users.noreply.github.com> Date: Mon, 5 Oct 2026 16:47:43 -0400 Subject: [PATCH 3/5] docs: drop a remote comment pointing at another backend's internals The offset=-1 fallback comment in list_offsets.rs named a specific backend file and the current paths that leave offset: None on it. That's a stale-the-day-it's-wrong remote comment (.claude/rules/comments.md): nothing in the named file would notify an editor that this comment exists, and the enumeration goes wrong the moment another backend's behavior changes. State the durable requirement instead. Co-Authored-By: Claude Sonnet 5 Signed-off-by: Andrea Ross <168456375+solace-aross@users.noreply.github.com> --- nisshi-storage/src/service/list_offsets.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/nisshi-storage/src/service/list_offsets.rs b/nisshi-storage/src/service/list_offsets.rs index e82249b90..d53595b0b 100644 --- a/nisshi-storage/src/service/list_offsets.rs +++ b/nisshi-storage/src/service/list_offsets.rs @@ -167,11 +167,9 @@ where ) // Kafka's buildErrorResponse (KafkaApis.scala) // sends offset=-1 on every error path, not just - // when a Timestamp lookup finds no match. This - // default fires today: SlateDb's storage.rs - // leaves offset: None on its "topic not found" and - // "partition out of range" paths, both returning - // UnknownTopicOrPartition. + // when a Timestamp lookup finds no match, so a + // backend that leaves offset: None on any + // not-found path gets the same default here. .offset(offset.offset().or(Some(-1))) .leader_epoch(Some(0)), ) From 418cf120f9864121bba60250d07f2bfc53b528ea Mon Sep 17 00:00:00 2001 From: Andrea Ross <168456375+solace-aross@users.noreply.github.com> Date: Tue, 6 Oct 2026 03:09:37 -0400 Subject: [PATCH 4/5] docs(storage): state the ListOffsetResponse no-offset rule once Document on ListOffsetResponse::offset when an answer has no offset and that ListOffsetsService sends it as -1, and shrink each backend's copy of the rule to one line that says what the code does. Pin the Kafka citation to the 3.9.1 tag, note the leader-epoch difference (Kafka sends -1, nisshi 0), trim the test comment, and add a CHANGELOG entry. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011P97dHLhTRdJ35fMJFpYPg Signed-off-by: Andrea Ross <168456375+solace-aross@users.noreply.github.com> --- CHANGELOG.md | 1 + nisshi-broker/tests/it/list_offsets.rs | 10 +++------- nisshi-broker/tests/it/storage_list_offsets.rs | 2 ++ nisshi-storage-dynostore/src/dynostore.rs | 7 ++----- nisshi-storage-null/src/lib.rs | 7 ++----- nisshi-storage-slatedb/src/storage.rs | 6 ++---- nisshi-storage-sql/src/limbo.rs | 7 ++----- nisshi-storage-sql/src/lite.rs | 7 ++----- nisshi-storage-sql/src/pg.rs | 7 ++----- nisshi-storage/src/lib.rs | 7 ++++++- nisshi-storage/src/service/list_offsets.rs | 10 +++++----- 11 files changed, 29 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a3819c0c..b938036a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,3 +35,4 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 instead of panicking the decoder. - SlateDB compaction skips a stored batch it cannot inflate, with a warning, instead of abandoning the whole maintenance pass. +- ListOffsets by timestamp now answers offset -1 and timestamp -1 with error NONE when no record has a timestamp at or after the target, as Apache Kafka does. The broker answered offset 0 before. That made the Java consumer's `offsetsForTimes()` throw `IllegalArgumentException: Invalid negative timestamp`, and it sent a client that seeks to the returned offset back to the start of the partition. A partition answered with an error code now also carries offset -1 instead of 0. diff --git a/nisshi-broker/tests/it/list_offsets.rs b/nisshi-broker/tests/it/list_offsets.rs index 35cd14307..52a941145 100644 --- a/nisshi-broker/tests/it/list_offsets.rs +++ b/nisshi-broker/tests/it/list_offsets.rs @@ -707,13 +707,9 @@ where .inspect(|offset| debug!(?offset))? ); - // Strictly later than any produced record's millisecond timestamp: the - // record's stored timestamp is captured just before `produce` as - // `to_timestamp(&SystemTime::now())` (integer milliseconds). Taking - // `after` immediately after `produce` returns can land in the same - // millisecond, which makes lite's `r.timestamp >= $4` lookup match at - // offset 0 instead of returning no match, flaking this assertion under - // load (observed ~5-10% under concurrent test load, 0/30 isolated). + // A Timestamp lookup matches a record at or after the target, at + // millisecond precision. `after` is one second past now, so it is + // later than the record's timestamp. let after = SystemTime::now() + Duration::from_secs(1); debug!(after = to_timestamp(&after)?); diff --git a/nisshi-broker/tests/it/storage_list_offsets.rs b/nisshi-broker/tests/it/storage_list_offsets.rs index c535cc40f..0c9992840 100644 --- a/nisshi-broker/tests/it/storage_list_offsets.rs +++ b/nisshi-broker/tests/it/storage_list_offsets.rs @@ -114,6 +114,8 @@ async fn simple(storage: impl Storage + Clone, broker_id: i32) -> Result<()> { /// `buildErrorResponse` (KafkaApis.scala) sends offset=-1/timestamp=-1 with /// error_code=NONE for this case, and `ListOffsetsService` must apply that /// default rather than the Earliest/Latest empty-partition default of 0. +/// Kafka sends leader epoch -1 here; nisshi sends 0, a known difference that +/// clients ignore when the offset is -1. async fn timestamp_no_match(storage: impl Storage + Clone, broker_id: i32) -> Result<()> { let extensions = Extensions::default(); diff --git a/nisshi-storage-dynostore/src/dynostore.rs b/nisshi-storage-dynostore/src/dynostore.rs index 93941880d..e8fc158c6 100644 --- a/nisshi-storage-dynostore/src/dynostore.rs +++ b/nisshi-storage-dynostore/src/dynostore.rs @@ -1339,11 +1339,8 @@ impl Storage for DynoStore { }, )) } else { - // Earliest/Latest on an empty partition is offset 0 (nothing - // written yet). A Timestamp lookup that matches no record is - // "not found", not "empty": leave offset as None so the caller - // applies Kafka's not-found default (-1), matching - // KafkaApis.scala's buildErrorResponse. + // No matching object: Earliest/Latest answer 0, a Timestamp + // lookup answers no offset (see `ListOffsetResponse::offset`). let offset = match offset_request { ListOffset::Earliest | ListOffset::Latest => Some(0), ListOffset::Timestamp(_) => None, diff --git a/nisshi-storage-null/src/lib.rs b/nisshi-storage-null/src/lib.rs index 23ddf86bd..bb19d7728 100644 --- a/nisshi-storage-null/src/lib.rs +++ b/nisshi-storage-null/src/lib.rs @@ -222,11 +222,8 @@ impl Storage for Engine { Ok(offsets .iter() .map(|(topition, list_offset)| { - // Earliest/Latest on an empty (here: nonexistent) partition is - // offset 0. A Timestamp lookup never matches a record in this - // backend, so it's "not found": leave offset as None so the - // caller applies Kafka's not-found default (-1), matching - // KafkaApis.scala's buildErrorResponse. + // No records are stored: Earliest/Latest answer 0, a Timestamp + // lookup answers no offset (see `ListOffsetResponse::offset`). let offset = match list_offset { ListOffset::Earliest | ListOffset::Latest => Some(0), ListOffset::Timestamp(_) => None, diff --git a/nisshi-storage-slatedb/src/storage.rs b/nisshi-storage-slatedb/src/storage.rs index 7e7bc85fb..f086f754e 100644 --- a/nisshi-storage-slatedb/src/storage.rs +++ b/nisshi-storage-slatedb/src/storage.rs @@ -1387,10 +1387,8 @@ impl Storage for Engine { offset: Some(offset), timestamp: to_system_time(ts).ok(), }, - // No record's timestamp is >= target: "not found", not - // "empty". Leave offset/timestamp as None so the caller - // applies Kafka's not-found default (-1), matching - // KafkaApis.scala's buildErrorResponse. + // No record at or after the target: no offset (see + // `ListOffsetResponse::offset`). None => ListOffsetResponse { error_code: ErrorCode::None, offset: None, diff --git a/nisshi-storage-sql/src/limbo.rs b/nisshi-storage-sql/src/limbo.rs index 83105e756..f8a0cceca 100644 --- a/nisshi-storage-sql/src/limbo.rs +++ b/nisshi-storage-sql/src/limbo.rs @@ -2035,11 +2035,8 @@ impl Storage for Engine { .inspect(|result| debug!(?result))? .map_or_else( || { - // Earliest/Latest on an empty partition is offset 0 (nothing - // written yet). A Timestamp lookup that matches no record is - // "not found", not "empty": leave offset/timestamp as None so - // the caller applies Kafka's not-found default (-1), matching - // KafkaApis.scala's buildErrorResponse. + // No record row: Earliest/Latest answer 0, a Timestamp + // lookup answers no offset (see `ListOffsetResponse::offset`). let (offset, timestamp) = match offset_type { ListOffsetRequest::Earliest | ListOffsetRequest::Latest => (Some(0), None), ListOffsetRequest::Timestamp(_) => (None, None), diff --git a/nisshi-storage-sql/src/lite.rs b/nisshi-storage-sql/src/lite.rs index 99de39111..c54a02862 100644 --- a/nisshi-storage-sql/src/lite.rs +++ b/nisshi-storage-sql/src/lite.rs @@ -3333,11 +3333,8 @@ impl Storage for Delegate { .inspect(|result| debug!(?result))? .map_or_else( || { - // Earliest/Latest on an empty partition is offset 0 (nothing - // written yet). A Timestamp lookup that matches no record is - // "not found", not "empty": leave offset/timestamp as None so - // the caller applies Kafka's not-found default (-1), matching - // KafkaApis.scala's buildErrorResponse. + // No record row: Earliest/Latest answer 0, a Timestamp + // lookup answers no offset (see `ListOffsetResponse::offset`). let (offset, timestamp) = match offset_type { ListOffset::Earliest | ListOffset::Latest => (Some(0), None), ListOffset::Timestamp(_) => (None, None), diff --git a/nisshi-storage-sql/src/pg.rs b/nisshi-storage-sql/src/pg.rs index 9c2d63a60..d20656482 100644 --- a/nisshi-storage-sql/src/pg.rs +++ b/nisshi-storage-sql/src/pg.rs @@ -2737,11 +2737,8 @@ impl Storage for Postgres { .inspect(|result| debug!(?result))? .map_or_else( || { - // Earliest/Latest on an empty partition is offset 0 (nothing - // written yet). A Timestamp lookup that matches no record is - // "not found", not "empty": leave offset/timestamp as None so - // the caller applies Kafka's not-found default (-1), matching - // KafkaApis.scala's buildErrorResponse. + // No record row: Earliest/Latest answer 0, a Timestamp + // lookup answers no offset (see `ListOffsetResponse::offset`). let (offset, timestamp) = match offset_type { ListOffset::Earliest | ListOffset::Latest => (Some(0), None), ListOffset::Timestamp(_) => (None, None), diff --git a/nisshi-storage/src/lib.rs b/nisshi-storage/src/lib.rs index eea46dc07..5ead58eae 100644 --- a/nisshi-storage/src/lib.rs +++ b/nisshi-storage/src/lib.rs @@ -614,6 +614,10 @@ pub type ListOffsetRequest = ListOffset; pub struct ListOffsetResponse { pub error_code: ErrorCode, pub timestamp: Option, + /// `None` when a [`ListOffset::Timestamp`] lookup matches no record, or + /// when `error_code` is not [`ErrorCode::None`]; `ListOffsetsService` + /// sends `None` as -1, as Kafka does. [`ListOffset::Earliest`] and + /// [`ListOffset::Latest`] with [`ErrorCode::None`] are always `Some`. pub offset: Option, } @@ -1409,7 +1413,8 @@ pub trait Storage: Debug + Send + Sync + 'static { /// Query the offset stage for a topic partition. async fn offset_stage(&self, topition: &Topition) -> Result; - /// Query the offsets for one or more topic partitions. + /// Query the offsets for one or more topic partitions. See + /// [`ListOffsetResponse::offset`] for when an answer has no offset. async fn list_offsets( &self, isolation_level: IsolationLevel, diff --git a/nisshi-storage/src/service/list_offsets.rs b/nisshi-storage/src/service/list_offsets.rs index d53595b0b..5806f69c0 100644 --- a/nisshi-storage/src/service/list_offsets.rs +++ b/nisshi-storage/src/service/list_offsets.rs @@ -165,12 +165,12 @@ where .unwrap_or(Some(-1)) .or(Some(-1)), ) - // Kafka's buildErrorResponse (KafkaApis.scala) - // sends offset=-1 on every error path, not just - // when a Timestamp lookup finds no match, so a - // backend that leaves offset: None on any - // not-found path gets the same default here. + // `None` is sent as -1, as Kafka does: + // https://github.com/apache/kafka/blob/3.9.1/core/src/main/scala/kafka/server/KafkaApis.scala#L1171-L1177 .offset(offset.offset().or(Some(-1))) + // Kafka sends -1 (the schema default) when + // the offset is unknown; nisshi always sends + // 0. Clients ignore the epoch for offset -1. .leader_epoch(Some(0)), ) } else { From aba42867f25ab2cc57d087864eaea55061c865b2 Mon Sep 17 00:00:00 2001 From: Andrea Ross <168456375+solace-aross@users.noreply.github.com> Date: Tue, 6 Oct 2026 18:50:54 -0400 Subject: [PATCH 5/5] test: pin ListOffsets no-match on a non-empty partition and the null backend multiple_record now looks up one second past its last record and asserts every partition, including the one holding records, answers offset -1 and timestamp -1 with no error. single_record also asserts the no-match answer carries no timestamp, and the null backend gets a unit test for its Earliest/Latest/Timestamp answers. Signed-off-by: Andrea Ross <168456375+solace-aross@users.noreply.github.com> Co-Authored-By: Claude Sonnet 5.5 --- Cargo.lock | 1 + nisshi-broker/tests/it/list_offsets.rs | 45 ++++++++++++++++++++++++++ nisshi-storage-null/Cargo.toml | 3 ++ nisshi-storage-null/src/lib.rs | 42 ++++++++++++++++++++++++ 4 files changed, 91 insertions(+) diff --git a/Cargo.lock b/Cargo.lock index 62a875bc2..60f707a40 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6184,6 +6184,7 @@ dependencies = [ "nisshi-sans-io", "nisshi-storage", "regex", + "tokio", "tracing", "url", "uuid", diff --git a/nisshi-broker/tests/it/list_offsets.rs b/nisshi-broker/tests/it/list_offsets.rs index 52a941145..d51cdc5c3 100644 --- a/nisshi-broker/tests/it/list_offsets.rs +++ b/nisshi-broker/tests/it/list_offsets.rs @@ -557,6 +557,50 @@ pub async fn multiple_record(broker: Broker) -> Result<()> { assert_eq!(Some(-1), partition.timestamp); } + debug!(phase = "after last record"); + let timestamp = ListOffset::Timestamp(third + Duration::from_secs(1)) + .try_into() + .inspect(|after| debug!(?after))?; + + let response = broker + .serve(RequestInput { + request: ListOffsetsRequest::default() + .isolation_level(isolation) + .replica_id(replica_id) + .topics(Some( + [ListOffsetsTopic::default() + .name(topic_name.into()) + .partitions(Some( + (0..num_partitions) + .map(|partition_index| { + ListOffsetsPartition::default() + .partition_index(partition_index) + .max_num_offsets(max_num_offsets) + .timestamp(timestamp) + .current_leader_epoch(Some(current_leader_epoch)) + }) + .collect::>(), + ))] + .into(), + )), + extensions: extensions.clone(), + }) + .await?; + + let topics = response.topics.as_deref().unwrap_or_default(); + assert_eq!(1, topics.len()); + assert_eq!(topic_name, topics[0].name); + let partitions = topics[0].partitions.as_deref().unwrap_or_default(); + assert_eq!(num_partitions as usize, partitions.len()); + + // Partition 0 holds records, but none at or after the target: Kafka + // answers offset -1 and timestamp -1 with no error. + for partition in partitions { + assert_eq!(i16::from(ErrorCode::None), partition.error_code); + assert_eq!(Some(-1), partition.offset); + assert_eq!(Some(-1), partition.timestamp); + } + Ok(()) } @@ -753,6 +797,7 @@ where // backend itself). assert_eq!(1, responses.len()); assert_eq!(None, responses[0].1.offset); + assert_eq!(None, responses[0].1.timestamp); Ok(()) } diff --git a/nisshi-storage-null/Cargo.toml b/nisshi-storage-null/Cargo.toml index 20741aff9..06f5e8c99 100644 --- a/nisshi-storage-null/Cargo.toml +++ b/nisshi-storage-null/Cargo.toml @@ -18,5 +18,8 @@ tracing.workspace = true url.workspace = true uuid.workspace = true +[dev-dependencies] +tokio.workspace = true + [lints] workspace = true diff --git a/nisshi-storage-null/src/lib.rs b/nisshi-storage-null/src/lib.rs index bb19d7728..7172c6d89 100644 --- a/nisshi-storage-null/src/lib.rs +++ b/nisshi-storage-null/src/lib.rs @@ -577,3 +577,45 @@ impl Storage for Engine { }) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn list_offsets_timestamp_has_no_offset() -> Result<()> { + let engine = Engine::new( + "cluster".into(), + 111, + Url::parse("tcp://127.0.0.1:9092/").expect("valid url"), + ); + let topition = Topition::new("t", 0); + + let responses = engine + .list_offsets( + IsolationLevel::ReadUncommitted, + &[ + (topition.clone(), ListOffset::Earliest), + (topition.clone(), ListOffset::Latest), + (topition.clone(), ListOffset::Timestamp(SystemTime::now())), + ], + ) + .await?; + + let offsets = responses + .iter() + .map(|(_, response)| (response.error_code, response.offset)) + .collect::>(); + + assert_eq!( + vec![ + (ErrorCode::None, Some(0)), + (ErrorCode::None, Some(0)), + (ErrorCode::None, None), + ], + offsets + ); + + Ok(()) + } +}