logstash的input配置成kafka和file，有什么区别

我现在的问题是elk环境大搭建，数据本意是从kafka来输出（input），然后推送到es中。现在遇见的问题是input分别配置成kafka和file，但是只有配置成file的时候才能正常在es中接到数据，kafka不行。
我怕的inout配置分别是：
input {
kafka {
zk_connect => "10.10.16.252:2181,10.10.16.253:2181,10.10.16.249:2181"
group_id => "test-consumer-group"
topic_id => "logStash2"
codec => "line"
reset_beginning => false # boolean (optional)， default: false
consumer_threads => 5 # number (optional)， default: 1
decorate_events => true # boolean (optional)， default: false
}
}


filter{
grok{
patterns_dir => "./patterns"
match => { "message" => ["%{IP:SRC_ADDRESS},%{ANYCHAR:DEST_I_POSITION}"]}
}
}

output {
elasticsearch{

host => "localhost"
}
}

input {
file{
path => "/home/hadoop1/bms/logstash-1.5.4/mylog/http.log"
}
}


filter{
grok{
patterns_dir => "./patterns"
match => { "message" => ["%{IP:SRC_ADDRESS},%{ANYCHAR:DEST_I_POSITION}"]}
}
}

output {
elasticsearch{

host => "localhost"
}
}
求问这是为什么。},
  "hits" : {
    "total" : 1,
    "max_score" : 1.0,
    "hits" : [ {
      "_index" : "logstash-2015.11.06",
      "_type" : "logs",
      "_id" : "AVDcH7qy1viDqLY4Ne-6",
      "_score" : 1.0,
      "_source":{"message":"6.6.6.6,word","tags":["_jsonparsefailure","_grokparsefailure"],"@version":"1","@timestamp":"2015-11-06T09:27:58.057Z","kafka":{"msg_size":12,"topic":"logStash2","consumer_group":"test-consumer-group","partition":1,"key":null}}
    } ]
  }
其中es那面报错是这样的：