SpringBoot集成EmbeddedKafka测试时schema.registry.url缺失报错如何解决

根因：Spring Kafka 官方自动配置仅会识别spring.kafka.producer.*、spring.kafka.consumer.*下的标准配置，你在spring.kafka根节点、自定义工具类中添加的schema.registry.url不属于默认加载的生产者配置范围，因此不会被KafkaTemplate关联的ProducerFactory读取，导致Avro序列化器初始化时缺少必要配置。

可按以下优先级解决：

• 方法1：通过application.yml直接配置生产者属性（无需额外写配置类，优先级最高）

spring:
  kafka:
    bootstrap-servers: 你的kafka地址:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      properties:
        # 序列化器所需的自定义配置统一放在producer.properties下，会被自动加载
        schema.registry.url: http://你的schema注册中心地址:8081

• 方法2：自定义ProducerFactory显式注入配置（适合需要自定义KafkaTemplate逻辑的场景）

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaProducerConfig {
    @Value("${spring.kafka.schema.registry.url}")
    private String schemaRegistryUrl;
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        // 显式传入schema注册中心地址
        configs.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
        return new DefaultKafkaProducerFactory<>(configs);
    }

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

• 测试场景特殊排查
  • 检查测试类是否使用@TestPropertySource、@SpringBootTest(properties = {})等注解覆盖了生产环境的schema配置
  • 若测试类手动实例化KafkaTemplate而非通过Spring容器注入，需确保实例化时传入的ProducerFactory包含schema.registry.url配置
  • 确认confluent schema序列化器版本与Spring Kafka版本兼容，避免版本不匹配导致配置读取逻辑异常

内容的提问来源于stack exchange，提问作者Khilarian