Some documents missing in Elasticsearch

[ben.zeghers]

I'm using Logstash (hosted in AWS Fargate) to import json files from my S3 bucket to Elasticsearch (hosted by Elastic Cloud). It works perfectly file for most of the files but some (~1%) never show up when I query for them using Kibana. If I re-upload the same file to S3, which causes the document to be re-indexed, it then shows up as expected. I've enabled debug logging for Logstash and can see that it's processing the file (see below).

I'm not sure how to debug this problem any further or if it's an issue with Logstash or Elasticsearch. Any help would be much appreciated.

Logstash debug logs:

[2018-03-21T19:48:11,679][DEBUG][logstash.inputs.s3 ] S3 input processing {:bucket=>"my-bucket", :key=>"142439.json"}
[2018-03-21T19:48:11,680][DEBUG][logstash.inputs.s3 ] S3 input: Download remote file {:remote_key=>"142439.json", :local_filename=>"/tmp/logstash/142439.json"}
[2018-03-21T19:48:11,691][DEBUG][logstash.inputs.s3 ] Processing file {:filename=>"/tmp/logstash/142439.json"}
[2018-03-21T19:48:11,768][DEBUG][logstash.pipeline ] filter received {"event"=>{"sale_id"=>142439, snip..., "@version"=>"1", "@timestamp"=>2018-03-21T19:48:11.691Z}}
[2018-03-21T19:48:11,768][DEBUG][logstash.util.decorators ] filters/LogStash::Filters::Mutate: adding value to field {"field"=>"[@metadata][action]", "value"=>["index"]}
[2018-03-21T19:48:11,768][DEBUG][logstash.pipeline ] output received {"event"=>{"sale_id"=>142439, snip..., "@version"=>"1", "@timestamp"=>2018-03-21T19:48:11.691Z}}

My Logstash configuration file is:

input {
    s3 {
        bucket => "${BUCKET}"
        codec => "json"
    }
}

filter {
    if [remove] {
        mutate {
            add_field => {
                "[@metadata][action]" => "delete"
            }
        }
    } else {
        mutate {
            add_field => {
                "[@metadata][action]" => "index"
            }
        }
    }
}

output {
    elasticsearch {
        hosts => ["${ELASTICSEARCH_HOST}"]
        user => "${ELASTICSEARCH_USER}"
        password => "${ELASTICSEARCH_PASSWORD}"
        index => "used-retail-sales"
        document_id => "%{sale_id}"
        action => "%{[@metadata][action]}"
    }
}