Kafka topic Schema version mismatch error

Problem description:

There is error messge when run spark app using spark streaming Kafka version 0.10.0.1(https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html) to send message to version 0.9 or 0.8 of Kafka broker.

2019-04-08 01:08:04,211 [ERROR] [kafka-producer-network-thread | producer-1]  Uncaught error in kafka producer I/O thread: 

org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'brokers': Error reading field 'host': Error reading string of length 25394, only 543 bytes available

                   at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:73)

                   at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:380)

                   at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:449)

                   at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:269)

                   at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:229)

                   at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:134)

                   at java.lang.Thread.run(Thread.java:745)

Root cause:

The Kafka broker version is still using version less than 0.10. Since there is big change after 0.10 that timestamp added in Schema. https://kafka.apache.org/documentation/

"

Soution:

Kafka broker should be upgrade and compatible to low version Schema format.

"1.5 Upgrading From Previous Versions

Upgrading from 0.8.x or 0.9.x to 0.10.0.0

0.10.0.0 has potential breaking changes (please review before upgrading) and possible performance impact following the upgrade. By following the recommended rolling upgrade plan below, you guarantee no downtime and no performance impact during and following the upgrade. 
Note: Because new protocols are introduced, it is important to upgrade your Kafka clusters before upgrading your clients.

Notes to clients with version 0.9.0.0: Due to a bug introduced in 0.9.0.0, clients that depend on ZooKeeper (old Scala high-level Consumer and MirrorMaker if used with the old consumer) will not work with 0.10.0.x brokers. Therefore, 0.9.0.0 clients should be upgraded to 0.9.0.1 beforebrokers are upgraded to 0.10.0.x. This step is not necessary for 0.8.X or 0.9.0.1 clients.

For a rolling upgrade:

1. Update server.properties file on all brokers and add the following properties:Upgrade the brokers. This can be done a broker at a time by simply bringing it down, updating the code, and restarting it.
   • inter.broker.protocol.version=CURRENT_KAFKA_VERSION (e.g. 0.8.2 or 0.9.0.0).
   • log.message.format.version=CURRENT_KAFKA_VERSION (See potential performance impact following the upgrade for the details on what this configuration does.)
3. Once the entire cluster is upgraded, bump the protocol version by editing inter.broker.protocol.version and setting it to 0.10.0.0. NOTE: You shouldn't touch log.message.format.version yet - this parameter should only change once all consumers have been upgraded to 0.10.0.0
4. Restart the brokers one by one for the new protocol version to take effect.
5. Once all consumers have been upgraded to 0.10.0, change log.message.format.version to 0.10.0 on each broker and restart them one by one.

Note: If you are willing to accept downtime, you can simply take all the brokers down, update the code and start all of them. They will start with the new protocol by default.

Note: Bumping the protocol version and restarting can be done any time after the brokers were upgraded. It does not have to be immediately after."