This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 46bb40a9a fix(sdk): delete_offset(None) for a standalone consumer
should delete current_partition_id offset (#3882)
46bb40a9a is described below
commit 46bb40a9a55024470d827b02a216919fb5ece34c
Author: haubur <[email protected]>
AuthorDate: Fri Aug 14 21:47:52 2026 +0200
fix(sdk): delete_offset(None) for a standalone consumer should delete
current_partition_id offset (#3882)
Co-authored-by: Hubert Gruszecki <[email protected]>
---
core/integration/tests/sdk/consumer_offset.rs | 139 ++++++++++++++++++++++++++
core/integration/tests/sdk/mod.rs | 1 +
core/sdk/src/clients/consumer.rs | 7 +-
3 files changed, 146 insertions(+), 1 deletion(-)
diff --git a/core/integration/tests/sdk/consumer_offset.rs
b/core/integration/tests/sdk/consumer_offset.rs
new file mode 100644
index 000000000..3bf715327
--- /dev/null
+++ b/core/integration/tests/sdk/consumer_offset.rs
@@ -0,0 +1,139 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::str::FromStr;
+
+use futures::StreamExt;
+use iggy::prelude::*;
+use integration::iggy_harness;
+use tokio::time::{Duration, timeout};
+
+const STREAM_NAME: &str = "delete-offset-stream";
+const TOPIC_NAME: &str = "delete-offset-topic";
+const CONSUMER_NAME: &str = "delete-offset-consumer";
+const ASSIGNED_PARTITION_ID: u32 = 3;
+const POLL_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[iggy_harness]
+async fn standalone_consumer_deletes_partition_zero_on_none_in_delete_offset(
+ harness: &TestHarness,
+) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
+ let stream_id = Identifier::named(STREAM_NAME).unwrap();
+ let topic_id = Identifier::named(TOPIC_NAME).unwrap();
+
+ client.create_stream(STREAM_NAME).await.unwrap();
+ client
+ .create_topic(
+ &stream_id,
+ TOPIC_NAME,
+ 4,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+
+ // send messages to all partitions
+ for partition_id in [0, ASSIGNED_PARTITION_ID] {
+ let mut messages = vec![
+ IggyMessage::from_str("message_1").unwrap(),
+ IggyMessage::from_str("message_2").unwrap(),
+ IggyMessage::from_str("message_3").unwrap(),
+ ];
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(partition_id),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+ }
+
+ // get consumer assigned to last partition
+ let mut consumer = client
+ .consumer(
+ CONSUMER_NAME,
+ STREAM_NAME,
+ TOPIC_NAME,
+ ASSIGNED_PARTITION_ID,
+ )
+ .unwrap()
+ .auto_commit(AutoCommit::Disabled)
+ .batch_length(1)
+ .build();
+ consumer.init().await.unwrap();
+
+ timeout(POLL_TIMEOUT, consumer.next())
+ .await
+ .expect("Consumer should receive a message before timeout")
+ .expect("Consumer stream should remain open")
+ .expect("Consumer should poll from the assigned partition");
+ assert_eq!(consumer.partition_id(), ASSIGNED_PARTITION_ID);
+
+ // set offsets for assigned partition and partition 0 manually
+ consumer
+ .store_offset(2, Some(ASSIGNED_PARTITION_ID))
+ .await
+ .unwrap();
+ consumer.store_offset(1, Some(0)).await.unwrap();
+
+ // Check that the offset was stored on the server as intended.
+ let mut stored_offset =
+ fetch_stored_offset(&client, &stream_id, &topic_id,
ASSIGNED_PARTITION_ID).await;
+ assert_eq!(stored_offset, Some(2));
+
+ let mut stored_offset_p0 = fetch_stored_offset(&client, &stream_id,
&topic_id, 0).await;
+ assert_eq!(stored_offset_p0, Some(1));
+
+ // assigned partition was untouched
+ stored_offset =
+ fetch_stored_offset(&client, &stream_id, &topic_id,
ASSIGNED_PARTITION_ID).await;
+ assert_eq!(stored_offset, Some(2));
+
+ // delete with none and check both again
+ consumer.delete_offset(None).await.unwrap();
+ // this fails, since partition 0 got deleted
+ stored_offset_p0 = fetch_stored_offset(&client, &stream_id, &topic_id,
0).await;
+ assert_eq!(stored_offset_p0, Some(1));
+}
+
+/// Read the offset the server has stored for the consumer.
+async fn fetch_stored_offset(
+ client: &IggyClient,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partition_id: u32,
+) -> Option<u64> {
+ client
+ .get_consumer_offset(
+ &Consumer::new(Identifier::named(CONSUMER_NAME).unwrap()),
+ stream_id,
+ topic_id,
+ Some(partition_id),
+ )
+ .await
+ .unwrap()
+ .map(|offset| offset.stored_offset)
+}
diff --git a/core/integration/tests/sdk/mod.rs
b/core/integration/tests/sdk/mod.rs
index a59335337..cdf8dfb4e 100644
--- a/core/integration/tests/sdk/mod.rs
+++ b/core/integration/tests/sdk/mod.rs
@@ -17,6 +17,7 @@
mod consumer_group;
mod consumer_group_membership;
+mod consumer_offset;
mod hello_world;
mod http_refresh;
mod producer;
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index 5400d7c85..a935ca24d 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -280,7 +280,12 @@ impl IggyConsumer {
}
/// Deletes the consumer offset on the server either for the current
partition or the provided partition ID.
- pub async fn delete_offset(&self, partition_id: Option<u32>) -> Result<(),
IggyError> {
+ pub async fn delete_offset(&self, mut partition_id: Option<u32>) ->
Result<(), IggyError> {
+ // `None` is only resolved server-side for consumer groups. For a
standalone consumer
+ // explicitly assign the current partition_id.
+ if partition_id.is_none() && !self.is_consumer_group {
+ partition_id = Some(self.current_partition_id.load(ORDERING));
+ }
let client = self.client.read().await;
client
.delete_consumer_offset(
