我们有一个系统，可以将数据从MongoDB同步到Elasticsearch。以下是关键组件：

1. MongoDB源连接器（Kafka connect connector）：该组件从MongoDB操作日志中读取事件并生成关于Kafka主题的消息。
2. Logstash：Logstash从Kafka接收这些消息并将其发送给Elasticsearch。我们已经为Logstash配置了特定的设置，包括pipeline.workers：1和pipeline.ordered：true，以确保事件按照接收的顺序处理。
3. ElasticSearch：1个节点集群，索引只有一个主分片。

使用的ELK协议栈版本：8.7.1
**问题：**我们有一个用例，在mongoDB中以顺序方式执行以下操作：

1.创建文档A
1.更新文件A
1.删除文件A
虽然我们可以看到这些操作在Logstash中按顺序处理，但我们遇到了一个问题：文档A的删除没有反映在Elasticsearch中。Elasticsearch中文档上的_version是3，表明所有3个事件都被执行。这表明删除操作可能在更新操作之前处理，导致文档A在MongoDB中被删除后仍保留在Elasticsearch中。
logstash管道如下：

input {
    kafka {
        id => "my_plugin_id"
        group_id => "logstash"
        bootstrap_servers => "broker:29092"
        topics => ["topic"]
        auto_offset_reset => "earliest"
        consumer_threads => 1
    }
}

filter {
    json {
        source => "message"
        target => "message"
        add_field => { "mongoId" => "%{[message][documentKey][_id][$oid]}" }
    }
}

output {
    if [message][operationType] == "delete" {
        elasticsearch {
        hosts => "http://es01:9200"
        user => "elastic"
        password => "changeme"
        index => "index_name"
        document_id => "%{[mongoId]}"
        action => "delete"
        }
    }
    else {
        elasticsearch {
        hosts => "http://es01:9200"
        index => "index_name"
        document_id => "%{[mongoId]}"
        user => "elastic"
        password => "changeme"
        pipeline => "index_pipeline"
        }
    }
}

字符串

备注：正如上面的配置中所提到的，对于 delete 以外的其他操作，我们使用一个摄取管道来重构要索引的文档数据。document_id被设置为mongold。

一个假设，尽管有待验证：据我所知，Elasticsearch输出插件（用于logstash管道）使用批量API将数据发送到ElasticSearch。事件是否可能在单个批处理中处理的（子请求）不遵守严格的顺序，即删除可能正在运行在更新事件之前（或完成），因此，ElasticSearch中最终可见的文档对应于更新操作。刷新索引设置为默认值，即1次/秒。此外，我增加了pipeline.batch.delay（logstash pipeline config）增加到5000 ms（默认为50 ms），以确保所有事件都在同一批中。