filebeat结合logstash推送到不同的源存在过滤问题

Question

版本

logstash-7.6.2-1.noarch
filebeat-7.6.2-1.x86_64

问题描述

基于output的if判断过滤失败，passageway值为default时数据也会随机推送到kafka和ES对应的数据源中

经过测试 当数据量较大且多个filebeat同时推送会出现该情况，logstash是否存在批量处理过滤的异常？

logstash代码如下：

input {
    beats {
        host => ""
        port => 5044
    }
}
output {
    if [fields][passageway] == "kafka" {
        kafka {
            bootstrap_servers => ""
            codec => plain {
                format => "%{[messageData]}"
            }
            topic_id => "%{[fields][kafka_topic]}"
        }
    } else if [fields][passageway] == "es" {
        elasticsearch {
            hosts => ""
            action => "index"
            index => "%{[fields][type]}"
        }
    }
}