Fluentd fails forwarding logs to Kafka broker with error getting request for apikey

Issue

• Forwarding logs to a Kafka broker fails in fluentd with error:

  2020-12-10 15:24:06 +0000 [warn]: Send exception occurred: Failed to send messages to flux-openshift-v4/1
  2020-12-10 15:24:06 +0000 [warn]: Exception Backtrace : /usr/local/share/gems/gems/fluent-plugin-kafka-0.13.1/lib/fluent/plugin/kafka_producer_ext.rb:240:in `deliver_messages_with_retries'
  /usr/local/share/gems/gems/fluent-plugin-kafka-0.13.1/lib/fluent/plugin/kafka_producer_ext.rb:126:in `deliver_messages'
  /usr/local/share/gems/gems/fluent-plugin-kafka-0.13.1/lib/fluent/plugin/out_kafka2.rb:270:in `write'
  /usr/local/share/gems/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:1125:in `try_flush'
  /usr/local/share/gems/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:1431:in `flush_thread_run'
  /usr/local/share/gems/gems/fluentd-1.7.4/lib/fluent/plugin/output.rb:461:in `block (2 levels) in start'
  /usr/local/share/gems/gems/fluentd-1.7.4/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create'
  2020-12-10 15:24:06 +0000 [warn]: failed to flush the buffer. retry_time=13 next_retry_seconds=2020-12-10 15:29:26 +0000 chunk="5b5907447d28804082b18b56af32c054" error_class=Kafka::DeliveryFailed error="Failed to send messages to flux-openshift-v4/1"
    2020-12-10 15:24:06 +0000 [warn]: suppressed same stacktrace
  

• Forwarding logs to a Kafka broker fails with the next logs in the Kafka broker:

  [2020-12-30 15:48:05,250] ERROR Closing socket for 172.31.29.210:9092-209.132.189.136:2153 because of error (kafka.network.Processor)
  org.apache.kafka.common.errors.InvalidRequestException: Error getting request for apiKey: 0 and apiVersion: 3
  Caused by: java.lang.IllegalArgumentException: Invalid version for API key 0: 3
          at org.apache.kafka.common.protocol.ProtoUtils.schemaFor(ProtoUtils.java:31)
          at org.apache.kafka.common.protocol.ProtoUtils.requestSchema(ProtoUtils.java:44)
          at org.apache.kafka.common.protocol.ProtoUtils.parseRequest(ProtoUtils.java:60)
          at org.apache.kafka.common.requests.ProduceRequest.parse(ProduceRequest.java:134)
          at org.apache.kafka.common.requests.AbstractRequest.getRequest(AbstractRequest.java:42)
          at kafka.network.RequestChannel$Request.liftedTree2$1(RequestChannel.scala:96)
          at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:91)
          at kafka.network.Processor$$anonfun$processCompletedReceives$1.apply(SocketServer.scala:492)
          at kafka.network.Processor$$anonfun$processCompletedReceives$1.apply(SocketServer.scala:487)
          at scala.collection.Iterator$class.foreach(Iterator.scala:727)
          at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
          at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
          at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
          at kafka.network.Processor.processCompletedReceives(SocketServer.scala:487)
          at kafka.network.Processor.run(SocketServer.scala:417)
          at java.lang.Thread.run(Thread.java:745)
  [2020-12-30 15:48:07,695] ERROR Closing socket for 172.31.29.210:9092-209.132.189.136:21620 because of error (kafka.network.Processor)