我们使用logstash（2.3.3）收听Kafka的多个主题，使用Kafka的新插件（3.0.2）。然后根据主题名称（作为元数据添加）将每个主题的数据重新定向到s3 bucket中的特定文件夹。然而，在当前的配置中，只有第一个s3输出的数据似乎落在它的s3 bucket/文件夹中。
有人能告诉我这里出了什么问题吗？我很确定有更好的方式来编写这个配置，可以满足我们的要求！

input
{
 kafka
 {
  bootstrap_servers => "10.0.0.5:9093,10.0.1.5:9093"
  topics => "topic"
  codec => "json"
  ssl => true
  ssl_keystore_location => "/opt/logstash/ssl/server.keystore.jks"
  ssl_keystore_password => "<snipped>"
  ssl_truststore_location => "/opt/logstash/ssl/server.truststore.jks"
  ssl_truststore_password => "<snipped>"
  add_field => { "[@metadata][topic]" => "topic" }
 }
 kafka
 {
  bootstrap_servers => "10.0.0.5:9093,10.0.1.5:9093"
  topics => "topic-test"
  codec => "json"
  ssl => true
  ssl_keystore_location => "/opt/logstash/ssl/server.keystore.jks"
  ssl_keystore_password => "<snipped>"
  ssl_truststore_location => "/opt/logstash/ssl/server.truststore.jks"
  ssl_truststore_password => "<snipped>"
  add_field => { "[@metadata][topic]" => "topic-test" }
 }
 kafka
 {
  bootstrap_servers => "10.0.0.5:9093,10.0.1.5:9093"
  topics => "daily_batch"  
  ssl => true
  ssl_keystore_location => "/opt/logstash/ssl/server.keystore.jks"
  ssl_keystore_password => "<snipped>"
  ssl_truststore_location => "/opt/logstash/ssl/server.truststore.jks"
  ssl_truststore_password => "<snipped>"
  add_field => { "[@metadata][topic]" => "daily_batch" }
 }
}

output
{
 if [@metadata][topic] == "topic"
 {
  s3
    {
     region => "us-east-1"
     bucket => "our-s3-storage/topic"
     size_file => 20971520
     temporary_directory => "/logstash"
     use_ssl => "true"
     codec => json_lines     
    }
 }
 if [@metadata][topic] == "topic-test"
 {
  s3
    {
     region => "us-east-1"
     bucket => "our-s3-storage/topic-test"
     size_file => 2097152
     temporary_directory => "/logstash"
     use_ssl => "true"
     codec => json_lines     
    }
 }
 if [@metadata][topic] == "daily_batch"
 {
  s3
    {
     region => "us-east-1"
     bucket => "our-s3-storage/daily_batch"
     size_file => 41943
     temporary_directory => "/logstash"
     use_ssl => "true"
    }
 }
}