Support compressed fluent input

[thomas-boehm-tractive]

Support input of compressed (gzip) fluent input

According to fluentbit and fluentd, the forward plugin, which we use to write date to logstash, supports compression by setting compress: gzip.
However, it appears that (in my example) the tcp plugin, with the fluent codec, doesn't support this yet.

I think it would make sense, especially for bigger documents, to allow the transfer of compressed data.

This would be a very simple configuration to verify the issue:

Logstash pipeline

input {
  tcp {
    port => "40005"
    codec => fluent {
      nanosecond_precision => true
    }
  }
}

output {
  stdout { 
    codec => json
  }
}

fluentbit-config:

[INPUT]
    name              tail
    path              /tmp/input/*.log
[OUTPUT]
    Name stdout
    Match *

[OUTPUT]
    Name forward
    Match *
    Host 172.17.0.3
    Port 40005
    compress gzip

This is the resulting error message, important part (Error) PackedForward with compression is not supported:

{"level":"ERROR","loggerName":"logstash.codecs.fluent","timeMillis":1716882646539,"thread":"nioEventLoopGroup-2-1","logEvent":{"message":"Fluent parse error, original data now in message field","error":{"stackTrace":[{"class":"RUBY","method":"decode_event","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-codec-fluent-3.4.2-java/lib/logstash/codecs/fluent.rb","line":116},{"class":"RUBY","method":"decode","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-codec-fluent-3.4.2-java/lib/logstash/codecs/fluent.rb","line":61},{"class":"org.msgpack.jruby.Unpacker","method":"each","file":"org/msgpack/jruby/Unpacker.java","line":222},{"class":"org.msgpack.jruby.Unpacker","method":"feed_each","file":"org/msgpack/jruby/Unpacker.java","line":209},{"class":"RUBY","method":"decode","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-codec-fluent-3.4.2-java/lib/logstash/codecs/fluent.rb","line":60},{"class":"RUBY","method":"decode_buffer","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-tcp-6.4.1-java/lib/logstash/inputs/tcp.rb","line":219},{"class":"RUBY","method":"decode","file":"/usr/share/logstash/vendor/bundle/jruby/3.1.0/gems/logstash-input-tcp-6.4.1-java/lib/logstash/inputs/tcp/decoder_impl.rb","line":23}],"exception":{"metaClass":{"metaClass":{"metaClass":{"error":"org.jruby.exceptions.Exception: (Error) PackedForward with compression is not supported","data":.....}]}}}}}}}

Also, I think, the error message is not super helpful, as I don't know what PackedForward is, why it is used, or if I can use some other type with compression?