fix: add kafka compress type and producer ack params

[rfyiamcool]

summary

Allows users to select parameters. 😁

The default waitforall is slow because it needs to wait synchronously for the brokers of the isr cluster to synchronize data.

Kafka's compression can appropriately reduce storage in kafka and pod/node bandwidth.

Of course, the effectiveness of parameters still depends on the scenario.

[sweep-ai]

Apply Sweep Rules to your PR?

• Apply: Leftover TODOs in the code should be handled.
• Apply: All new business logic should have corresponding unit tests in the tests/ directory.
• Apply: Any clearly inefficient or repeated code should be optimized or refactored.