Add support to more parameters to Kafka Sink for creating topics

[mbombonato]

Is your feature request related to a problem? Please describe.

No

Describe the solution you'd like

As we have the allow.auto.create.topics parameter, it would be nice to add some extra important parameters, so our topics are created correctly.

I'm missing the following basic parameters:

• Number of Partitions (int)
• Replicator Factor (int)
• Cleanup Policy (delete, compact and delete and compact)
• Retention Time (int as milliseconds)

Describe alternatives you've considered

CREATE SINK kafka_sink AS SELECT
   "user_id", "user_name"
FROM my_view
WITH (
    connector='kafka',
    properties.bootstrap.server='localhost:9092',
    topic='topic',
    allow.auto.create.topics='true',
    create.topic.partitions=10,
    create.topic.replication.factor=3,
    create.topic.cleanup.policy='compact',
    create.topic.retention.time=86400000
)
FORMAT UPSERT ENCODE JSON;

Additional context

Without those parameters, we have to create the topics manually beforehand and we can't use the parameter allow.auto.create.topics

[tabVersion]

Hi @mbombonato, for now, RisingWave connect to kafka broker via rust-rdkafka, so we can only support options listed in https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md.

[mbombonato]

Hello @tabVersion
Thanks for you reply. I notice we don't have those configurations there. I guess we need to open a feature request there and link with our issue right? It's weird they don't support those parameters as they are very basic. All popular GUI set those parameters when you create a topic manually.

[mbombonato]

Also, I've found this FAQ about topic configuration:
https://github.com/confluentinc/librdkafka/wiki/FAQ#topic-configuration
Are those configs "generic" or only the configs they have documented?

[tabVersion]

Hello @tabVersion Thanks for you reply. I notice we don't have those configurations there. I guess we need to open a feature request there and link with our issue right? It's weird they don't support those parameters as they are very basic. All popular GUI set those parameters when you create a topic manually.

Kafka Admin Api can do some of the options you mentioned above, but I think it goes too far as a streaming database.
I understand that the options are frequently used but RisingWave does not serve as a Kafka tool but an individual component.

Also, I've found this FAQ about topic configuration:
Wiki: FAQ (topic configuration) (confluentinc/librdkafka)
Are those configs "generic" or only the configs they have documented?

As far as I know, the confluentinc/librdkafka@master/CONFIGURATION.md lists all accepted configs. We don't plan to make any modifications to the SDK.

[mbombonato]

I understand. No problem, we'll figure out how to automate the topic creation before setting up our sinks. We're transitioning from ksqlDB, where it automatically created our compact topics with the partitions we set on the sink stream.