feat(kafka source&sink): expose rdkafka fetch params

[fuyufjh]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Expose these configurations for the convenience of debugging in the future.

I'll set better parameters in a separate PR, after sufficient performance test, etc.

related #10888 #11229

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR contains user-facing changes.

Click here for Documentation

Types of user-facing changes

Please keep the types that apply to your changes, and remove the others.

• Connector (sources & sinks)

Release note

allow user specify kafka params in rw SQL (description in details available here)

valid for both source and sink

config name in RW	config name in Kafka	type
properties.message.max.bytes	message.max.bytes	int
properties.receive.message.max.bytes	receive.message.max.bytes	int

only valid in source

config name in RW	config name in Kafka	type
properties.queued.min.messages	queued.min.messages	int
properties.queued.max.messages.kbytes	queued.max.messages.kbytes	int
properties.fetch.wait.max.ms	fetch.wait.max.ms	int
properties.fetch.max.bytes	fetch.max.bytes	int

only valid for sink

config name in RW	config name in Kafka	type
properties.queue.buffering.max.messages	queue.buffering.max.messages	int
properties.queue.buffering.max.kbytes	queue.buffering.max.kbytes	int
properties.queue.buffering.max.ms	queue.buffering.max.ms	float
properties.enable.idempotence	enable.idempotence	bool (only accept 'true' and 'false')
properties.message.send.max.retries	message.send.max.retries	int
properties.retry.backoff.ms	retry.backoff.ms	int
properties.batch.num.messages	batch.num.messages	int
properties.batch.size	batch.size	int

[codecov]

Codecov Report

Merging #11203 (90bf75c) into main (19263e7) will decrease coverage by 0.02%.
Report is 68 commits behind head on main.
The diff coverage is 56.86%.

@@            Coverage Diff             @@
##             main   #11203      +/-   ##
==========================================
- Coverage   69.74%   69.73%   -0.02%     
==========================================
  Files        1313     1313              
  Lines      224257   224410     +153     
==========================================
+ Hits       156414   156496      +82     
- Misses      67843    67914      +71     

Flag	Coverage Δ
rust	69.73% <56.86%> (-0.02%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files Changed	Coverage Δ
...rc/connector/src/source/kafka/enumerator/client.rs	0.00% <0.00%> (ø)
src/connector/src/source/kafka/source/reader.rs	0.00% <0.00%> (ø)
src/connector/src/common.rs	2.19% <9.09%> (+0.94%)	⬆️
src/connector/src/sink/kafka.rs	37.00% <59.49%> (+3.89%)	⬆️
src/connector/src/source/kafka/mod.rs	69.35% <67.24%> (-30.65%)	⬇️

... and 5 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[fuyufjh]

Depends on #11232 because fetch.queue.backoff.ms was added in rdkafka 2.2.0

Update: I have removed fetch.queue.backoff.ms from the test cases, so that we can merge this first.

[fuyufjh]

@tabVersion LGTM. You may give an approval by yourself 🙌

[shanicky]

LGTM