The default serializer for the ElasticSearch sink (ElasticSearchLogStashEventSerializer) duplicates fields that are mapped to default logstash fields.
For instance timestamp, source, host. Those appear both as logstash fields ("@timestamp", "@source_host" etc.), and both as fields under the @fields ("@fields.timestamp", "@fields.host").
When inserting a field from the headers as a logstash system field it should be removed from the dictionary so it wouldn't get written again under the "@fields" field.