MongoDB同步Elasticsearch时删除操作未生效问题排查

问题核心 #

按顺序执行MongoDB文档A的创建→更新→删除操作后，Elasticsearch中仍保留更新后的文档，但_version为3（说明三个事件都已执行）。结合你的配置和环境，核心问题出在Logstash输出阶段的批量处理逻辑。

你的假设验证：成立 #

Logstash的Elasticsearch输出插件使用Bulk API时，若使用多个独立的输出块（如你配置中分别处理delete和其他操作），同一个批次内的事件会被拆分到不同的批量队列中，导致发送到ES的Bulk请求顺序与Logstash处理事件的顺序不一致。

具体原因：

• 你配置了两个elasticsearch输出实例，创建、更新事件进入第一个输出的批量池，删除事件进入第二个输出的批量池
• 两个输出实例的批量刷新是独立执行的，可能出现删除操作的Bulk请求先于更新操作的请求到达ES并执行，最终更新操作覆盖了删除结果

验证方法 #

1. 开启Logstash debug日志：在logstash.yml中添加logger.logstash.outputs.elasticsearch: debug，查看输出阶段事件的处理顺序，以及生成的Bulk请求内容，确认两个输出的批量发送顺序。
2. 开启Elasticsearch审计日志：在ES的elasticsearch.yml中配置审计日志，记录所有写入请求的执行顺序，直接验证三个操作的实际执行时序。

解决方案 #

方案1：合并为单个Elasticsearch输出（推荐） #

将两个输出块合并为一个，通过条件判断动态设置action和pipeline，确保所有事件进入同一个批量队列，严格按处理顺序生成Bulk请求：

output {
    elasticsearch {
        hosts => "http://es01:9200"
        user => "elastic"
        password => "changeme"
        index => "index_name"
        document_id => "%{[mongoId]}"
        # 直接复用MongoDB的operationType作为ES的action（insert/update/delete与ES的action值完全匹配）
        action => "%{[message][operationType]}"
        # 删除操作不需要执行ingestion pipeline，通过条件判断设置为空
        pipeline => "%{[message][operationType] == 'delete' ? '' : 'index_pipeline'}"
    }
}

ES的Bulk API会严格按请求内的操作顺序执行，只要Logstash发送的Bulk请求中操作顺序是创建→更新→删除，就能保证最终结果与MongoDB一致。

方案2：强制单个输出的顺序（不推荐，影响性能） #

如果必须保留两个输出块，可给delete输出设置强制立即刷新，确保删除操作在更新之后执行：

# 处理delete的输出
elasticsearch {
    hosts => "http://es01:9200"
    user => "elastic"
    password => "changeme"
    index => "index_name"
    document_id => "%{[mongoId]}"
    action => "delete"
    flush_size => 1
    batch_delay => 0
}

此方法会降低批量处理效率，仅适合测试或低流量场景。

补充说明 #

你配置的pipeline.workers:1和pipeline.ordered:true仅能保证Logstash在filter阶段的事件顺序，但输出阶段若存在多个独立的输出实例，它们的批量处理逻辑会打破原有顺序，这是核心误区。

内容的提问来源于stack exchange，提问作者Aman Yadav