kafka_input.conf

input {
  kafka{
    codec => json
    bootstrap_servers => "localhost:9092"
    topics => ["elastic-test"]
  }
}

filter {
    mutate {
      add_field => {
        "id" => "%{[data][id]}"
      }
      add_field => {
        "firstName" => "%{[data][firstName]}"
      }
      add_field => {
        "lastName" => "%{[data][lastName]}"
      }
      add_field => {
        "city" => "%{[data][city]}"
      }
      add_field => {
        "country" => "%{[data][country]}"
      }
      add_field => {
        "email" => "%{[data][email]}"
      }
      add_field => {
        "phoneNumber" => "%{[data][phoneNumber]}"
      }
      add_field => {
        "createdAt" => "%{[data][createdAt]}"
      }
      remove_field => ["data", "@version", "@timestamp", "message", "event", "globalId"]
    }  
}

output {
  stdout {  
        codec => json_lines  
    } 
  elasticsearch {
      hosts => ["localhost:9200"]
      index => "kafka_test"
      document_id => "%{id}"
  }
}