Spring Kafka Producer没有发送到Kafka 1.0.0(Magic v1不支持记录头)

Question

我正在使用这个docker-compose设置在本地设置Kafka:https://github.com/wurstmeister/kafka-docker/

docker-compose up 工作正常,通过shell创建主题工作正常.

现在我尝试通过连接到Kafka spring-kafka:2.1.0.RELEASE

启动Spring应用程序时,它会打印正确版本的Kafka:

o.a.kafka.common.utils.AppInfoParser     : Kafka version : 1.0.0
o.a.kafka.common.utils.AppInfoParser     : Kafka commitId : aaa7af6d4a11b29d

我尝试发送这样的消息

kafkaTemplate.send("test-topic", UUID.randomUUID().toString(), "test");

在客户端发送失败

UnknownServerException: The server experienced an unexpected error when processing the request

在服务器控制台中,我收到消息Magic v1不支持记录头

Error when handling request {replica_id=-1,max_wait_time=100,min_bytes=1,max_bytes=2147483647,topics=[{topic=test-topic,partitions=[{partition=0,fetch_offset=39,max_bytes=1048576}]}]} (kafka.server.KafkaApis)
java.lang.IllegalArgumentException: Magic v1 does not support record headers

谷歌搜索建议版本冲突,但版本似乎适合(org.apache.kafka:kafka-clients:1.0.0在类路径中).

有线索吗？谢谢!

编辑:我缩小了问题的根源.发送纯字符串有效,但通过JsonSerializer发送Json会导致给定的问题.这是我的生产者配置的内容:

@Value("\${kafka.bootstrap-servers}")
lateinit var bootstrapServers: String

@Bean
fun producerConfigs(): Map<String, Any> =
        HashMap<String, Any>().apply {
            // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
            put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
            put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java)
            put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer::class.java)
        }

@Bean
fun producerFactory(): ProducerFactory<String, MyClass> =
        DefaultKafkaProducerFactory(producerConfigs())

@Bean
fun kafkaTemplate(): KafkaTemplate<String, MyClass> =
        KafkaTemplate(producerFactory())

Answer 1

我有类似的问题.如果我们使用JsonSerializer或JsonSerde为值,Kafka默认添加标题.为了防止出现此问题,我们需要禁用添加信息标头.

如果您使用默认的json序列化,那么使用以下(这里的关键点是ADD_TYPE_INFO_HEADERS):

Map<String, Object> props = new HashMap<>(defaultSettings);
props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
ProducerFactory<String, Object> producerFactory = new DefaultKafkaProducerFactory<>(props);

但是如果你需要一个JsonSerializer特定的自定义ObjectMapper(比如with PropertyNamingStrategy.SNAKE_CASE),你应该禁止显式添加信息标题JsonSerializer,因为spring kafka忽略了DefaultKafkaProducerFactory属性 ADD_TYPE_INFO_HEADERS (对我而言,这是一个糟糕的spring kafka设计)

JsonSerializer<Object> valueSerializer = new JsonSerializer<>(customObjectMapper);
valueSerializer.setAddTypeInfo(false);
ProducerFactory<String, Object> producerFactory = new DefaultKafkaProducerFactory<>(props, Serdes.String().serializer(), valueSerializer);

或者如果我们使用JsonSerde,那么:

Map<String, Object> jsonSerdeProperties = new HashMap<>();
jsonSerdeProperties.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
JsonSerde<T> jsonSerde = new JsonSerde<>(serdeClass);
jsonSerde.configure(jsonSerdeProperties, false);

Answer 2

解决了。问题既不是代理、某些 docker 缓存，也不是 Spring 应用程序。

问题是我并行使用的控制台消费者进行调试。这是一位“老”消费者kafka-console-consumer.sh --topic=topic --zookeeper=...

它实际上在启动时打印一条警告：Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].

--bootstrap-server应该使用带有选项的“新”消费者（特别是在将 Kafka 1.0 与 JsonSerializer 一起使用时）。注意：这里使用旧的消费者确实会影响生产者。