Problem in sending data from HTTP plugin to kafka plugin in logstash

[DZ1 (Jenni)]

I am trying to send JSON data to HTTP plugin and that output to Kafka using Kafka plugin. In this HTTP input plugin shows output in the first attempt and sends only a small amount of data to Kafka. When I hit it next time there is no data in logs but that small fraction of data comes at Kafka topic.

input {
 http {
    host => "0.0.0.0"
    port => "5000"
    response_headers => {
      "Access-Control-Allow-Origin" => "*"
      "Content-Type" => "text/plain"
      "Access-Control-Allow-Headers" => "Origin, X-Requested-With, Content-Type,
       Accept"
    }
  }
}

output {
      stdout {
        codec => json
       }
      kafka {
        #codec => json
        bootstrap_servers => "kafka:9092"
        topic_id => "NotificationTopic"
     }
}

I am getting this in Kafka Topic:

{
      "@version" => "1",
    "@timestamp" => 2019-08-28T12:23:52.368Z,
       "message" => "2019-08-28T12:23:51.647Z 192.168.144.1 %{message}"
}

The HTTP input plugin should be accepting data at every hit and should be sent the same to the Kafka output plugin.

[Badger]

If the http input knows from the content-type that the input body is json then it will automagically parse the JSON, so you will not have a [message] field. The default codec on the kafka output is plain, and what you are seeing there is the default format (since you do not have a [message] field the sprintf reference to it does not get substituted).

You need to either force the codec on the input so that it does not parse the JSON, or add a codec to the output. json_lines might work.

[DZ1 (Jenni)]

@Badger Yes, you are right. I tried it with codec and it worked fine. Thanks for your suggestion.
Here are my config that worked:

input {
 http {
    host => "0.0.0.0"
    port => "5000"
    codec => json
  }
}

filter {
  mutate {
      remove_field => ["host", "@version", "@timestamp", "headers"]
  }
}

output {
      stdout { }
      kafka {
        codec => json_lines
        bootstrap_servers => "metrics-kafka:9092"
        topic_id => "NotificationTopic"
     }
}