feat: rework pubsub source to support parallel read and at-least-once #16733
Conversation
Signed-off-by: xxchan <xxchan22f@gmail.com>
- (message) change the offset column from ts to ack_id - (split) don't update start_offset - (reader) don't ack on read - refactor WaitCheckpointWorker to support update_task_on_chunk Signed-off-by: xxchan <xxchan22f@gmail.com>
Signed-off-by: xxchan <xxchan22f@gmail.com>
Signed-off-by: xxchan <xxchan22f@gmail.com>
Signed-off-by: xxchan <xxchan22f@gmail.com>
Signed-off-by: xxchan <xxchan22f@gmail.com>
There was a problem hiding this comment.
the pr generally LGTM
please update in the pr's note that pubsub does not rely on the SplitImpl's offset to start from a position. It prefers letting the broker side deal with offset, which is different from other connectors.
btw, idk if reusing offset in the SplitImpl is a good practice, it may be confusing if users want to query from the state table.
Sure. BTW, I've mentioned that it will be stateless from the RisingWave side.
I don't understand your point. What do you mean by reusing Anyway, I don't think any average user care about query from state table. |
| /// pubsub reader. | ||
| pub(crate) start_offset: Option<String>, | ||
|
|
||
| /// `stop_offset` is a numeric timestamp. | ||
| /// When not `None`, the `PubsubReader` stops reading messages when the `offset` property of | ||
| /// the `SourceMessage` is greater than or equal to the `stop_offset`. | ||
| pub(crate) stop_offset: Option<String>, | ||
| #[serde(rename = "stop_offset")] | ||
| #[serde(skip_serializing)] | ||
| pub(crate) __deprecated_stop_offset: Option<String>, |
There was a problem hiding this comment.
I guess start_offset can also be deprecated.
stop_offset is actually never used.
Signed-off-by: xxchan <xxchan22f@gmail.com>
Signed-off-by: xxchan <xxchan22f@gmail.com>
Signed-off-by: xxchan <xxchan22f@gmail.com>
Signed-off-by: xxchan <xxchan22f@gmail.com>
…limit: 524288 bytes. Signed-off-by: xxchan <xxchan22f@gmail.com>
|
simple manual load testing 
|
Signed-off-by: xxchan xxchan22f@gmail.comI hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
close #16572
Behavior changes
tstoack_idseek)WaitCheckpointWorkerto supportupdate_task_on_chunk: store allack_idsbefore next checkpointpubsub.parallelismWITH option. (Previously calledsplit_count, but I likeparallelismbetter, because "split" is not a user-facing concept feat: pre-release pubsub #14531)SourceEnumeratorenumContext.currentParallelism(). We don't have such a context in source manager yet.retain_acked_messages.Note: although there are huge changes, I think all are none breaking.
Explanations
Previously, we ack immediately and rely on
retain_acked_messages+seekto timestamp to replay messages. This is awkward, and not like designed usage.Now we ack on checkpoint and don't
seekat all. It's stateless on RisingWave's side. This corresponds more the queueing semantics (delete after ack).WaitCheckpointWorkerTesting
checkpoint_frequencyto very large (andset visibility_mode TO 'all'). Create Table and check load successfully. Kill and restart RisingWave, check the data can still be loaded without missing data (probably duplicated).Checklist
./risedev check(or alias,./risedev c)Documentation
Release note
Add
pubsub.parallelismWITHoption for PubSub source, which specifies the number of parallel consumers to run for the subscription. If not specified, the parallelism will be one.