Kafka ***** cannot be cast to class java.lang.String

Kafka在默认情况下， 将会使用 org.apache.kafka.common.serialization.StringSerializer 对 key 和 value 进行序列化， 在这种情况下， 只能序列化字符串对象。如果发送的消息为一般的自定义对象， 将会出现 **** 无法转变为 java.lang.String .

解决措施：
 将默认的 key 和 value 在生产者和消费者中都将他们明确地指定为别的序列化对象。比如， 如果使用Spring的Kafka集成， org.springframework.kafka.support.serializer.JsonSerializer将是一个很好的解决方案， 只需要在项目内添加如下依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.6.5</version>
</dependency>

如果使用的是 application.yml 配置文件的 Spring Boot 项目， 将属性Kafka的生产者和消费者设置为以下属性即可：

spring:
	kafka：
		producer:
      		key-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      		value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      	consumer:
	    	key-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
	    	value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

即可解决这个问题。

如果使用自定义的添加配置属性的 Kafka 客户端对象， 在 Properties 对象中添加对应的配置即可。
生产者：

properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonSerializer");
properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonSerializer");

消费者：

properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonDeserializer");
properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonDeserializer");