Spring Kafka producer更新到spring-boot 2.0后发送消息时失败

Question

@Configuration
public class KafkaProducerConfig {

@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;

@Bean
public ProducerFactory<String, KafkaMessage> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
}

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    return configProps;
}

@Bean
public KafkaTemplate<String, KafkaMessage> kafkaTemplate() {
    return new KafkaTemplate<>(producerFactory());
}
}

[2018-10-08 19:20:53,562] INFO Kafka version : 1.0.0 (org.apache.kafka.common.utils.AppInfoParser)
[2018-10-08 19:20:53,562] INFO Kafka commitId : aaa7af6d4a11b29d (org.apache.kafka.common.utils.AppInfoParser)

应用程序错误：

java.util.concurrent.ExecutionException: org.springframework.kafka.core.KafkaProducerException: Failed to send; nested exception is org.apache.kafka.common.errors.UnknownServerException: The server experienced an unexpected error when processing the request
Caused by: org.springframework.kafka.core.KafkaProducerException: Failed to send; nested exception is org.apache.kafka.common.errors.UnknownServerException: The server experienced an unexpected error when processing the request

Kafka日志错误：

java.lang.IllegalArgumentException: Magic v1 does not support record headers

Answer 1

Magic v1不支持当kafka broker版本和客户机版本不匹配时，会抛出记录头。

spring-kafka 2.1.10不支持kafka 2.0.0版本。您需要使用spring-kafka 2.2.x来与kafka 2.0.0一起工作。

对于Kafka客户端的兼容性，请查看Spring的官方文档。