Unable to write after restarting kafka cluster

When I restarted the kafka cluster, the following problems occurred:
```
2020-01-17 07:29:35 +0000 [warn]: #0 failed to flush the buffer. retry_time=10 next_retry_seconds=2020-01-17 07:30:33 +0000 chunk="59c39b45987d2889fdf53f5e2d87f58c" error_class=Kafka::UnknownTopicOrPartition error="Kafka::UnknownTopicOrPartition"
2020-01-17 07:29:35 +0000 [warn]: #0 suppressed same stacktrace
2020-01-17 07:29:45 +0000 [warn]: #0 Send exception occurred: Kafka::UnknownTopicOrPartition
2020-01-17 07:29:45 +0000 [warn]: #0 Exception Backtrace : /var/lib/gems/2.5.0/gems/ruby-kafka-0.6.8/lib/kafka/protocol.rb:130:in `handle_error'
/var/lib/gems/2.5.0/gems/ruby-kafka-0.6.8/lib/kafka/protocol/metadata_response.rb:144:in `partitions_for'
/var/lib/gems/2.5.0/gems/ruby-kafka-0.6.8/lib/kafka/cluster.rb:155:in `partitions_for'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:190:in `assign_partitions!'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:153:in `block in deliver_messages_with_retries'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:148:in `loop'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:148:in `deliver_messages_with_retries'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:102:in `deliver_messages'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/out_kafka2.rb:220:in `write'
/var/lib/gems/2.5.0/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:1125:in `try_flush'
/var/lib/gems/2.5.0/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:1431:in `flush_thread_run'
/var/lib/gems/2.5.0/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:461:in `block (2 levels) in start'
/var/lib/gems/2.5.0/gems/fluentd-1.7.4/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create'
2020-01-17 07:29:45 +0000 [warn]: #0 failed to flush the buffer. retry_time=10 next_retry_seconds=2020-01-17 07:30:46 +0000 chunk="59c39a5ff585997b0db7f41cc30228dd" error_class=Kafka::UnknownTopicOrPartition error="Kafka::UnknownTopicOrPartition"
2020-01-17 07:29:45 +0000 [warn]: #0 suppressed same stacktrace
2020-01-17 07:29:45 +0000 [warn]: #0 Send exception occurred: Kafka::UnknownTopicOrPartition
2020-01-17 07:29:45 +0000 [warn]: #0 Exception Backtrace : /var/lib/gems/2.5.0/gems/ruby-kafka-0.6.8/lib/kafka/protocol.rb:130:in `handle_error'
/var/lib/gems/2.5.0/gems/ruby-kafka-0.6.8/lib/kafka/protocol/metadata_response.rb:144:in `partitions_for'
/var/lib/gems/2.5.0/gems/ruby-kafka-0.6.8/lib/kafka/cluster.rb:155:in `partitions_for'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:190:in `assign_partitions!'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:153:in `block in deliver_messages_with_retries'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:148:in `loop'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:148:in `deliver_messages_with_retries'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/kafka_producer_ext.rb:102:in `deliver_messages'
/var/lib/gems/2.5.0/gems/fluent-plugin-kafka-0.7.9/lib/fluent/plugin/out_kafka2.rb:220:in `write'
/var/lib/gems/2.5.0/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:1125:in `try_flush'
/var/lib/gems/2.5.0/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:1431:in `flush_thread_run'
/var/lib/gems/2.5.0/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:461:in `block (2 levels) in start'
/var/lib/gems/2.5.0/gems/fluentd-1.7.4/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create'
2020-01-17 07:29:45 +0000 [warn]: #0 failed to flush the buffer. retry_time=11 next_retry_seconds=2020-01-17 07:30:50 +0000 chunk="59c39a6a7a6e508dc8d8df3078918470" error_class=Kafka::UnknownTopicOrPartition error="Kafka::UnknownTopicOrPartition"

```