logstash input使用codec从kafka解析json数组输出到es问题

logstash  版本6.5.4
kafka 版本 2.11-2.3.0
json数组格式是这样的[{"a":1,"b":2},{"a":1,"b":2},{"a":1,"b":2},{"a":1,"b":2}]
数据已经正确存到kafka，在kafkatool里能看到json数组，
并且在output用stdout打印也打印出所有的json，但是导入es后发现json数组里只有随机 的条数据进入了es，其他数据都丢失了，想半天不懂哪里出了问题，一下是kafka logstash配置

input {

    kafka { 

		auto_offset_reset => "earliest"

		bootstrap_servers => "localhost:9092"

		client_id => "items" 

		consumer_threads => 1

		group_id => "items"  

		topics=>["items"] 

		codec => "json" 

		add_field => { "[@metadata][type]" => "items" }



    }  

}

 

output {  

		elasticsearch {

			hosts => "localhost:9200"

			index => "%{[@metadata][type]}"  

			document_type => "doc"  

			document_id => "%{a}%{b}"

		} 

		stdout { }

}