Allow ignore create topic if topic already exist in Kafka/Redpanda client

[wutianchen]

potential line of change: https://github.com/DataSQRL/sqrl/blob/main/sqrl-tools/sqrl-run/src/main/java/com/datasqrl/DatasqrlRun.java#L294 (a.k.a put some configurations to ignore topic creation if already exists)

error triggered on my test to restart datasqrl/cmd run with persisted redpanda data. Because when a new redpanda cluster starts with persistend data in /data/redpanda, the topic is already there along with the data stored in the topic. And the client can not handle that

error message

INFO  2025-02-12 02:38:29,626 [shard 0:main] cluster - topics_frontend.cc:346 - Create topics [{configuration: { topic: {kafka/hightemp_1}, partition_count: 1, replication_factor: 1, is_migrated: false, properties: { compression: {nullopt}, cleanup_policy_bitflags: {nullopt}, compaction_strategy: {nullopt}, retention_bytes: {{nullopt}}, retention_duration_ms: {{nullopt}}, segment_size: {nullopt}, timestamp_type: {nullopt}, recovery_enabled: {nullopt}, shadow_indexing: {disabled}, read_replica: {nullopt}, read_replica_bucket: {nullopt}, remote_topic_namespace_override: {nullopt}, remote_topic_properties: {nullopt}, batch_max_bytes: {nullopt}, retention_local_target_bytes: {{nullopt}}, retention_local_target_ms: {{nullopt}}, remote_delete: true, segment_ms: {{nullopt}}, record_key_schema_id_validation: {nullopt}, record_key_schema_id_validation_compat: {nullopt}, record_key_subject_name_strategy: {nullopt}, record_key_subject_name_strategy_compat: {nullopt}, record_value_schema_id_validation: {nullopt}, record_value_schema_id_validation_compat: {nullopt}, record_value_subject_name_strategy: {nullopt}, record_value_subject_name_strategy_compat: {nullopt}, initial_retention_local_target_bytes: {{nullopt}}, initial_retention_local_target_ms: {{nullopt}}, mpx_virtual_cluster_id: {nullopt}, write_caching: {nullopt}, flush_ms: {nullopt}, flush_bytes: {nullopt}, remote_label: {nullopt}, iceberg_mode: disabled, leaders_preference: {nullopt}, delete_retention_ms: {disabled}, iceberg_delete: {nullopt}}}, custom_assignments: {}},]
WARN  2025-02-12 02:38:29,626 [shard 0:main] kafka - create_topics.cc:185 - Failed to create topic(s) {{kafka/hightemp_1}} error_code observed: cluster::errc::topic_already_exists
02:38:29.629 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.common.utils.AppInfoParser - App info kafka.admin.client for adminclient-1 unregistered
02:38:29.636 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.common.metrics.Metrics - Metrics scheduler closed
02:38:29.636 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.common.metrics.Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter
02:38:29.636 [kafka-admin-client-thread | adminclient-1] INFO  org.apache.kafka.common.metrics.Metrics - Metrics reporters closed
Exception in thread "main" java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: The topic has already been created
        at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
        at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2005)
        at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
        at com.datasqrl.DatasqrlRun.initKafka(DatasqrlRun.java:298)
        at com.datasqrl.DatasqrlRun.run(DatasqrlRun.java:90)
        at com.datasqrl.DatasqrlRun.main(DatasqrlRun.java:67)
Caused by: org.apache.kafka.common.errors.TopicExistsException: The topic has already been created

Test Case (feel free to @wutianchen )

merge change into branch rewire-redpanda-volume (waiting for cicd workflow runs through)

then run more than two times

docker run -it -p 8888:8888 -p 8081:8081 -v $PWD:/build -v /data/redpanda:/data/redpanda datasqrl/cmd:sqrltest run sensors.sqrl sensors.graphqls

without the above create topic error appears

[mbroecheler]

@velo You already fixed this right?