test codec against class name string to prevent class equivalence bug…

… with a Delegator

Fixes elastic#11401

logstash-core/lib/logstash/inputs/base.rb

@@ -131,11 +131,12 @@ def fix_streaming_codecs
     require "logstash/codecs/line"
     require "logstash/codecs/json"
     require "logstash/codecs/json_lines"
-    case @codec
-      when LogStash::Codecs::Plain
+
+    case @codec.class.name
+      when "LogStash::Codecs::Plain"
         @logger.info("Automatically switching from #{@codec.class.config_name} to line codec", :plugin => self.class.config_name)
         @codec = LogStash::Codecs::Line.new("charset" => @codec.charset)
-      when LogStash::Codecs::JSON
+      when "LogStash::Codecs::JSON"
         @logger.info("Automatically switching from #{@codec.class.config_name} to json_lines codec", :plugin => self.class.config_name)
         @codec = LogStash::Codecs::JSONLines.new("charset" => @codec.charset)
     end

logstash-core/spec/logstash/inputs/base_spec.rb

@@ -113,4 +113,32 @@ def register; end
     tcp.instance_eval { fix_streaming_codecs }
     expect(tcp.codec.charset).to eq("CP1252")
   end
+
+  it "should switch plain codec to line" do
+    require "logstash/inputs/tcp"
+    require "logstash/codecs/plain"
+    require "logstash/codecs/line"
+
+    # it is important to use "codec" => "plain" here and not the LogStash::Codecs::Plain instance so that
+    # the config parsing wrap the codec into the delagator which was causing the codec identification bug
+    # per https://github.com/elastic/logstash/issues/11140
+    tcp = LogStash::Inputs::Tcp.new("codec" => "plain", "port" => 0)
+    tcp.register
+
+    expect(tcp.codec.class.name).to eq("LogStash::Codecs::Line")
+  end
+
+  it "should switch json codec to json_lines" do
+    require "logstash/inputs/tcp"
+    require "logstash/codecs/plain"
+    require "logstash/codecs/line"
+
+    # it is important to use "codec" => "json" here and not the LogStash::Codecs::Plain instance so that
+    # the config parsing wrap the codec into the delagator which was causing the codec identification bug
+    # per https://github.com/elastic/logstash/issues/11140
+    tcp = LogStash::Inputs::Tcp.new("codec" => "json", "port" => 0)
+    tcp.register
+
+    expect(tcp.codec.class.name).to eq("LogStash::Codecs::JSONLines")
+  end
 end