[logstash] kafka - logstash - elasticseach

input {
  kafka {
    bootstrap_servers => "kafkahost:9092"
    topics => ["search_query_log"]
    group_id => "logstash"
    type => "search_query"
    consumer_threads => 1
  }
  kafka {
    bootstrap_servers => "kafkahost:9092"
    topics => ["search_query_log2"]
    group_id => "logstash"
    type => "search_query2"
    consumer_threads => 1
  }
}

filter {
  json {
    source => "message"
  }
  mutate {
    add_field => {
       "ls_timestamp" => "%{@timestamp}"
       "kst_timestamp" => ""
    }
    remove_field => ["message"]
  }
  ruby {
    code => "event.set('kst_timestamp', event.get('@timestamp').time.localtime('+09:00').strftime('%Y-%m-%d %H:%M:%S'))"
  }
  grok {
    match => {
      "kst_timestamp" => "%{YEAR:yyyy}-%{MONTHNUM:mm}-%{MONTHDAY:dd}%{GREEDYDATA}"
    }
    add_field => {
      "[@metadata][yymmdd]" => "%{yyyy}.%{mm}.%{dd}"
    }
    remove_field => [ "yyyy", "mm", "dd", "ls_timestamp" ]
  }
}

output {
 if [type] == "search_query" {
    elasticsearch {
    hosts => ["localhost:9200"]
    ilm_rollover_alias => "search-query-log"
    ilm_pattern => "{now/d{yyyy.MM.dd|+09:00}}-000001"
    ilm_policy => "search-querylog-policy"
    user => "ela"
    password => "elas"
   }
 }
 if [type] == "search_query2" {
    elasticsearch {
    hosts => ["localhost:9200"]
    ilm_rollover_alias => "search-query-log2"
    ilm_pattern => "{now/d{yyyy.MM.dd|+09:00}}-000001"
    ilm_policy => "search-querylog-policy"
    user => "elasti"
    password => "elasti"
   }
 }
  #stdout { codec => rubydebug }
}