[Bug] inconsistency between paimon table buckets and kafka topic partitions causes data write errors

### Search before asking

- [X] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.

### Paimon version

paimon-1.0-snapshot

### Compute Engine

flink 1.18.0

### Minimal reproduce step

1.create an paimon catalog and use kafka as the logsystem.

```shell
CREATE CATALOG paimon_catalog WITH (
'type' = 'paimon',
'log.system.auto-register' = 'true',
'warehouse' = '/Users/paimon/path'
)
```

2.create an primary table and use kafka as the logsystem.

```shell
use paimon_catalog;

CREATE TABLE IF NOT EXISTS test_tb (
user_id BIGINT,
item_id BIGINT,
behavior STRING,
dt STRING,
hh STRING,
PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
)
with (
'bucket' ='10',
'kafka.topic' ='test_topic',
'log.consistency' ='eventual',
'log.system.partitions' ='1',
'log.system.replication' ='1',
'kafka.bootstrap.servers' ='127.0.0.1:9092',
'log.system' ='kafka'
));
```

3.create an datagen table

```shell
CREATE temporary TABLE datagen_table (
user_id BIGINT,
item_id BIGINT,
behavior STRING,
dt STRING,
hh STRING
) WITH (
'connector' = 'datagen',
'fields.user_id.min'='1',
'fields.user_id.max'='10000000',
'rows-per-second'='10'
);
```

4.use the datagen table's data insert into test_tb

```shell
insert into test_tb select * from datagen_table
```

### What doesn't meet your expectations?

```shell
org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to send data to Kafka: Partition 5 of topic test_topic with partition count 1 is not present in metadata after 60000 ms.
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.checkErroneous(FlinkKafkaProducer.java:1428)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:859)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.invoke(FlinkKafkaProducer.java:99)
at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction.invoke(TwoPhaseCommitSinkFunction.java:240)
at org.apache.paimon.flink.sink.RowDataStoreWriteOperator.processElement(RowDataStoreWriteOperator.java:136)
at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:237)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:146)
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:562)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:858)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:807)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.TimeoutException: Partition 5 of topic test_topic with partition count 1 is not present in metadata after 60000 ms.
```

The bucket number of the paimon table is 10, but the partition number of the Kakfa topic is 1, which causes the data to be sent to a non-existent partition when it is sent to Kafka.

### Anything else?

In this code, the partitions of the data produced by Kafka are obtained based on the bucket. Does this mean that the number of partitions of Kafka must be consistent with the number of buckets in the table? In our production scenario, the number of partitions of Kafka is always the default

### Are you willing to submit a PR?

- [X] I'm willing to submit a PR!

Reproduce the SQL setup with a 10-bucket Paimon table and a one-partition Kafka topic. Start at RowDataStoreWriteOperator.processElement and the FlinkKafkaProducer stack frames to trace how bucket 5 reaches Kafka; done means the expected handling of mismatched bucket and partition counts is established and covered by an appropriate regression test.

Written by the indexing model from the issue text.