Return remaining messages in batch when auto commit is disabled (#1375)

iggy-rs · Dec 4, 2024 · 733037c · 733037c
1 parent a135c10
commit 733037c
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 9 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/sdk/Cargo.toml b/sdk/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "iggy"
-version = "0.6.50"
+version = "0.6.51"
 description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
 edition = "2021"
 license = "MIT"

diff --git a/sdk/src/clients/consumer.rs b/sdk/src/clients/consumer.rs
@@ -505,7 +505,7 @@ impl IggyConsumer {
                 )
                 .await;
 
-            if let Ok(polled_messages) = polled_messages {
+            if let Ok(mut polled_messages) = polled_messages {
                 if polled_messages.messages.is_empty() {
                     return Ok(polled_messages);
                 }
@@ -522,12 +522,17 @@ impl IggyConsumer {
                     last_consumed_offset.insert(partition_id, AtomicU64::new(0));
                 }
 
-                if has_consumed_offset && consumed_offset >= polled_messages.messages[0].offset {
-                    return Ok(PolledMessages {
-                        messages: EMPTY_MESSAGES,
-                        current_offset: polled_messages.current_offset,
-                        partition_id,
-                    });
+                if has_consumed_offset {
+                    polled_messages
+                        .messages
+                        .retain(|message| message.offset > consumed_offset);
+                    if polled_messages.messages.is_empty() {
+                        return Ok(PolledMessages {
+                            messages: EMPTY_MESSAGES,
+                            current_offset: polled_messages.current_offset,
+                            partition_id,
+                        });
+                    }
                 }
 
                 let stored_offset;