[SUPPORT] TransactionParticipant should be created for each assigned partition, but has not been created for the topic/partition

**Describe the problem you faced**

I am trying to run the demo with kafka-connect and hudi-sink but when I create the sink connector I get the error TransactionParticipant should be created for each assigned partition. If I recreate the sink after the topic has received data it gives me a different error. I have to delete the sink connector delete the topic and then recreate it to get rid of the connectException error. But I am still stuck with TransactionParticipant error.

**To Reproduce**

Steps to reproduce the behavior:

Follow the instructions given [here](https://github.com/apache/hudi/blob/master/hudi-kafka-connect/README.md)

**Expected behavior**

A clear and concise description of what you expected to happen.

**Environment Description**

* Hudi version : 1.0.1

* Spark version : 3.5.5

* Hive version : 2.3.4

* Hadoop version : 2.10.2

* Storage (HDFS/S3/GCS..) : HDFS

* kafka-connect-hdfs: 10.1.0

* Running on Docker? (yes/no) : No

**Additional context**
this is my `connect-distributed.properties`

```bash
sudo cat /etc/kafka/connect-distributed.properties
# Written by Ansible
bootstrap.servers=hudi:9092
config.providers.secret.param.kafkastore.security.protocol=PLAINTEXT
config.storage.replication.factor=1
config.storage.topic=connect-cluster-configs
confluent.license.topic=_confluent-command
confluent.monitoring.interceptor.topic=_confluent-monitoring
connector.client.config.override.policy=All
consumer.bootstrap.servers=hudi:9092
consumer.confluent.monitoring.interceptor.bootstrap.servers=hudi:9092
consumer.confluent.monitoring.interceptor.security.protocol=PLAINTEXT
consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
consumer.security.protocol=PLAINTEXT
group.id=hudi-connect-cluster
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
key.converter.schema.registry.url=http://hudi:8081
listeners=http://0.0.0.0:8083
offset.flush.interval.ms=10000
offset.storage.replication.factor=1
offset.storage.topic=connect-cluster-offsets
plugin.path=/usr/share/java/connect_plugins,/usr/share/confluent-hub-components
producer.bootstrap.servers=hudi:9092
producer.confluent.monitoring.interceptor.bootstrap.servers=hudi:9092
producer.confluent.monitoring.interceptor.security.protocol=PLAINTEXT
producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
producer.security.protocol=PLAINTEXT
rest.advertised.host.name=hudi
rest.advertised.listener=http
rest.advertised.port=8083
rest.port=8083
security.protocol=PLAINTEXT
status.storage.replication.factor=1
status.storage.topic=connect-cluster-status
value.converter.schema.registry.url=http://hudi:8081
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
```

**Stacktrace**

```
.connect.runtime.WorkerSinkTask:617)
Mar 25 05:54:55 hudi connect-distributed[3710459]: org.apache.kafka.connect.errors.RetriableException: TransactionParticipant should be created for each assigned partition, but has not been created for the topic/partition: hudi-test-topic:0
Mar 25 05:54:55 hudi connect-distributed[3710459]: at org.apache.hudi.connect.HoodieSinkTask.put(HoodieSinkTask.java:111)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:593)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:340)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:238)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:207)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:229)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:284)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
Mar 25 05:54:55 hudi connect-distributed[3710459]: at java.base/java.lang.Thread.run(Thread.java:840)
```

Start with the kafka-connect README reproduction steps and inspect org.apache.hudi.connect.HoodieSinkTask.java at line 111, where the stack trace reports the failure. Compare connector assignment and transaction setup for hudi-test-topic:0, then verify that the demo sink starts and processes assigned partitions without the reported exception.

Written by the indexing model from the issue text.