logstash-babel.conf

input {
  http {
    id => "farfallino"
    port => 9090
    add_field => { "confused_language" => "farfallino" }
  }

  http {
     id => "hacker"
     port => 9091
     add_field => { "confused_language" => "hacker" }
   }
}

filter {
    if [http][method] != "POST"{
        drop {}
    }

    mutate {
        remove_field => ["@timestamp", "@version", "url", "host", "http", "event", "user_agent"]
    }
}

output {
  kafka {
    codec => plain { format => "%{message}" }
    topic_id => "babel-input"
    bootstrap_servers => "broker:9092"
    message_key => "%{confused_language}"
  }
}