Add alternative ingestion API with direct offset return #42
Conversation
Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>
elenagaljak-db
requested review from
dbatomic,
gotocoding-DB and
teodordelibasic-db
|
|
||
| ### API Changes | ||
|
|
||
| - Added `ingest_record_v2()` method to `ZerobusStream` for immediate offset return without Future wrapping |
There was a problem hiding this comment.
So maybe something that we can directly pair up with wait_for_offset.
Eg ingest_record_async_offset or ingest_record_offset_ack.
ingest_record_offset_ack indicates that you need to pair this with wait_for_offset so I would vote for that.
Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>
| now, now | ||
| ); | ||
| let offset_id = stream.ingest_record_v2(json_record_2).await.unwrap(); | ||
| println!("Record ingested with offset Id: {}", offset_id); |
There was a problem hiding this comment.
Why do we call it "record ingested" if only after ack we can say that the record was ingested?
Should be "Record sent with offset id: "
| // Example 2: Using v2 API, ingest_record_v2 returns offset immediately. | ||
| let json_record_2 = format!( |
There was a problem hiding this comment.
This part is not in line with Example 1 – there, we define the json record first and then have the comment.
| if let Some(offset) = offset { | ||
| if offset >= offset_to_wait { | ||
| info!(stream_id = %stream_id, "Stream is caught up to the given offset. Flushing complete."); | ||
| info!(stream_id = %stream_id, "Stream is caught up to the given offset. {} complete.", operation_name); |
There was a problem hiding this comment.
"Waiting for acknowledgement completed"?
Signed-off-by: elenagaljak-db <elena.galjak@databricks.com>
| /// # Ok(()) | ||
| /// # } | ||
| /// ``` | ||
| pub async fn ingest_record_v2( |
There was a problem hiding this comment.
Let's then just rename this to what Atomic proposed, mark previously existing methods as deprecated and change all the examples/documentation to primarily show this new API.
What changes are proposed in this pull request?
Adds alternative ingestion API methods (
ingest_record_v2()andingest_records_v2()) that return logical offset IDs directly instead of wrapped in Futures. Includes newwait_for_offset()method to explicitly wait for acknowledgment of specific offsets. Refactorsflush()andwait_for_offset()to share common waiting logic.How is this tested?
Tests Added: