Elasticsearch filter plugin is not working(logstash-filter-elasticsearch)

[dangal]

Hi,
here is my logstash conf and ss1.csv file has only passengerId field and the same field had in index titanic_kfk_shrink....so i would like that particular field in both the inputs (i mean input file and elasticsearch filter) and query the fields mentioned in fields option of elasticsearch filter

input {
 file {
   path => "/home/sgpl/logstash-5.4.0/config/ss1.csv"
   type => "end"
   sincedb_path => "/tmp/titanic"
    start_position => "beginning"
     }
     }
filter
{
  csv {
columns => ["PassengerId"]
separator => ","
 }

mutate {
 convert => { "PassengerId" => "integer" }
}
#if [type] == "end" {
elasticsearch {
            hosts => ["http://192.168.1.59:9233"]
           index => "titanic_kfk_shrink"
           #query => 'Age.keyword:%{Age.keyword}'
#query => "Pclass.keyword:1"
             query =>  'PassengerId.keyword:"%{[PassengerId]}"'
             fields => ["Age.keyword","Gender.keyword"]

          # sort => "Age.keyword:desc"            
#         }
#date {
#            match => ["[started]", "ISO8601"]
#            target => "[started]"
#         }
}
}
#}
output {
    stdout { codec => rubydebug }
}

it is throwing error like
[2017-11-14T20:55:14,898][WARN ][logstash.filters.elasticsearch] Failed to query elasticsearch for previous event {:index=>"titanic_kfk_shrink", :query=>"PassengerId.keyword:"659"", :event=>2017-11-14T15:25:13.855Z elasticsearch_masternode 659, :error=>#<TypeError: can't convert nil into String>}

please suggest what is wrong in my config file ASAP
thanks in advance.