Do not block barrier when rate limiting with smaller than chunk_size

[kwannoel]

• Add test for rate limit on source. feat(connector): partition source reader batch_size by rate_limit #13800
• Add test for rate limit on snapshot read. feat(stream): apply rate_limit as chunk_size to snapshot read #13837
• Add test for rate limit on DML. feat(stream): add rate limit for dml #13902
• Support partition in non-transaction source by rate_limit feat(connector): partition source reader batch_size by rate_limit #13800
• Support partition in snapshot read by rate_limit. feat(stream): apply rate_limit as chunk_size to snapshot read #13837
• Support partition in transaction source by rate_limit
• Support rate limit on DML. feat(stream): add rate limit for dml #13902

[BugenZhao]

What's the plan for not blocking the barrier? I guess for those "source-like" executors such as Source or DML, it's natural. For other executors, we have to do "barrier stealing/rearrangement" and the consistency could be broken.

[kwannoel]

What's the plan for not blocking the barrier? I guess for those "source-like" executors such as Source or DML, it's natural. For other executors, we have to do "barrier stealing/rearrangement" and the consistency could be broken.

Yeah actually it's just for Source / DML.

The plan is to just throttle stream reader. I have a WIP PR where I will add more details soon.

[kwannoel]

Source is pending #14215 (comment)