fix(kafka): overwrite the EntryId with Offset while consuming the Kaf…

…kaRecords
GreptimeTeam · Jan 11, 2024 · 7826d80 · 7826d80
1 parent 8ec1e42
commit 7826d80
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/src/log-store/src/kafka/log_store.rs b/src/log-store/src/kafka/log_store.rs
@@ -205,7 +205,11 @@ impl LogStore for KafkaLogStore {
                 }
 
                 // Tries to construct an entry from records consumed so far.
-                if let Some(entry) = maybe_emit_entry(record, &mut entry_records)? {
+                if let Some(mut entry) = maybe_emit_entry(record, &mut entry_records)? {
+                    // We don't rely on the EntryId generated by mito2. 
+                    // Instead, we use the offset return from Kafka as EntryId. 
+                    // Therefore, we MUST overwrite the EntryId with RecordOffset.
+                    entry.id = offset as u64;
                     yield Ok(vec![entry]);
                 }