Get error when config “value_serializer” and “key_serializer” in output part

[rookie-fpc]

I am testing transfer data between mutile kafka cluster my configuration is below:

`input { kafka { bootstrap_servers => "10.62.169.206:9092,10.62.220.44:9092,10.62.220.150:9092" topics => ["prod-skywalking-meters"] group_id => "prod-skywalking-meters" value_deserializer_class => "org.apache.kafka.common.serialization.BytesDeserializer" codec => plain } }

output { kafka { codec => plain { format => "%{message}" } bootstrap_servers => ["10.62.169.141:9092,10.62.109.72:9092,10.62.174.33:9092"] topic_id => "pre-skywalking-meters" value_serializer => "org.apache.kafka.common.serialization.BytesSerializer" key_serializer => "org.apache.kafka.common.serialization.BytesSerializer" } }`

when I start logstash, I got error:

[ERROR] 2024-01-15 17:55:08.893 [[main]-pipeline-manager] javapipeline - Pipeline error {:pipeline_id=>"main", :exception=>#<NameError: uninitialized constant LogStash::Outputs::Kafka::ConfigurationError>, :backtrace=>["org/jruby/RubyModule.java:3766:inconst_missing'", "/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-integration-kafka-10.9.0-java/lib/logstash/outputs/kafka.rb:211:in register'", "org/logstash/config/ir/compiler/OutputStrategyExt.java:131:inregister'", "org/logstash/config/ir/compiler/AbstractOutputDelegatorExt.java:68:in register'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:233:inblock in register_plugins'", "org/jruby/RubyArray.java:1821:in each'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:232:inregister_plugins'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:598:in maybe_setup_out_plugins'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:245:instart_workers'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:190:in run'", "/usr/share/logstash/logstash-core/lib/logstash/java_pipeline.rb:142:inblock in start'"], "pipeline.sources"=>["/etc/logstash/conf.d/skywalking-meters2.conf"], :thread=>"#"} [INFO ] 2024-01-15 17:55:08.895 [[main]-pipeline-manager] javapipeline - Pipeline terminated {"pipeline.id"=>"main"} [ERROR] 2024-01-15 17:55:08.918 [Converge PipelineAction::Create

] agent - Failed to execute action {:id=>:main, :action_type=>LogStash::ConvergeResult::FailedAction, :message=>"Could not execute action: PipelineAction::Create

, action_result: false", :backtrace=>nil} [INFO ] 2024-01-15 17:55:08.998 [LogStash::Runner] runner - Logstash shut down`

what should I do? can you give some advice? The data is serialized. I want to transmit the data intact thank you