Logstash kafka output works but kafka input doesn't

[foresightyj]

I have been using logstash for the past two years and consider myself to be fairly familiar with configurations. Actually this configuration worked before and just stopped working a few days ago.

To make a minimal case, I downloaded the latest logstash 5.5.2, and start new with this config:

input { 
    kafka {
        topics => ["test_kafka_logstash_2"]
        bootstrap_servers => "192.168.0.112:9092"
        id => "test_kafka_input"
        auto_offset_reset => "latest"

    }
    
    stdin {
        id => "test_stdin"
    }

}
filter {
    mutate {
        add_field => { "hello" => "world" }
    }
}

output {
    stdout { codec => rubydebug }
    
    kafka {
        codec => plain {
            format => "%{message}"
        }
        topic_id => "test_logstash_kafka_output"
        bootstrap_servers => "192.168.0.112:9092"
        id => "test_kafka_output"
    }
}

Run it with this command:

Then enter two lines:

And the proof that the two lines were written to the output kafka topic successfully

Exactly two lines are written to the output topic.

Now for the kafka input, I have script that writes to the test_kafka_logstash_2 topic and this is total count of messages after the test:

Below is the complete log that is logged:

https://gist.github.com/foresightyj/5037361e2d81b1ad7746c3e24241b67f