EmbeddedKafka搭配MockSchemaRegistryClient测试Spring Cloud Stream Kafka Streams Avro报错

问题核心原因 #

你遇到的问题由两个常见错误导致：

1. YAML配置缩进错误，spring.cloud.stream被缩进到了spring.cloud.function的子级，导致Spring Cloud Stream的流配置完全不生效，stream1根本没有被初始化，自然不会有数据写入topic2
2. MockSchemaRegistryClient实例不互通：mock实例是按对象隔离的，你在Spring上下文、测试生产者、自定义Deserializer中分别新建的实例互不共享注册的schema，序列化时生成的schema id在消费侧的mock实例中不存在，就会抛出找不到schema的异常

具体修复方案 #

1. 先修正application.yaml缩进

spring.application.name: processingapp
spring.cloud:
  function.definition: stream1 # 后续再扩展;stream2;stream3
  stream: # 此处要和function平级，不能缩进在function下
    bindings:
      stream1-in-0:
        destination: topic1
      stream1-out-0:
        destination: topic2
    kafka:
      binder:
        min-partition-count: 1
        replication-factor: 1
        auto-create-topics: true
        auto-add-partitions: true
      bindings:
        default:
          consumer:
            autoRebalanceEnabled: true
            resetOffsets: true
            startOffset: earliest
        stream1-in-0:
          consumer:
            keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
            valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
        stream1-out-0:
          producer:
            keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
            valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
      streams:
        binder:
          configuration:
            schema.registry.url: mock://localtest
            specific.avro.reader: true
            auto.register.schemas: true # 确保流处理时自动注册schema到mock registry
            value.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicNameStrategy # 统一subject命名规则

2. 统一使用共享的MockSchemaRegistryClient实例

不要在各个逻辑中新建mock实例，直接注入Spring上下文初始化的同一个实例即可，测试代码修改如下：

@RunWith(SpringRunner.class)
@SpringBootTest
public class Test {

    private static final String INPUT_TOPIC = "topic1";
    private static final String OUTPUT_TOPIC = "topic2";

    // 直接注入Spring上下文生成的共享MockSchemaRegistryClient实例
    @Autowired
    private SchemaRegistryClient schemaRegistryClient;

    @ClassRule
    public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 1, INPUT_TOPIC, OUTPUT_TOPIC);

    @BeforeClass
    public static void setup() {
        System.setProperty("spring.cloud.stream.kafka.binder.brokers", embeddedKafka.getEmbeddedKafka().getBrokersAsString());
    }

    @org.junit.Test
    public void testSendReceive() throws IOException {
        Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
        senderProps.put("key.serializer", LongSerializer.class);
        senderProps.put("value.serializer", SpecificAvroSerializer.class);
        // 生产者使用共享的schema registry实例
        senderProps.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_CLIENT_CONFIG, schemaRegistryClient);
        AvroFileParser fileParser = new AvroFileParser();
        DefaultKafkaProducerFactory<Long, Test1> pf = new DefaultKafkaProducerFactory<>(senderProps);
        KafkaTemplate<Long, Test1> template = new KafkaTemplate<>(pf, true);
        Test1 test1 = fileParser.parseTest1("src/test/resources/mocks/test1.json");

        template.send(INPUT_TOPIC, 123456L, test1);
        
        // 等待流处理完成，避免过早消费不到数据
        Awaitility.await().atMost(Duration.ofSeconds(15)).until(() -> {
            Map<String, Object> consumer2Props = KafkaTestUtils.consumerProps("testConsumer2", "false", embeddedKafka.getEmbeddedKafka());
            consumer2Props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            consumer2Props.put("key.deserializer", StringDeserializer.class);
            consumer2Props.put("value.deserializer", SpecificAvroDeserializer.class);
            consumer2Props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_CLIENT_CONFIG, schemaRegistryClient);
            consumer2Props.put(AbstractKafkaAvroSerDeConfig.SPECIFIC_AVRO_READER_CONFIG, true);

            DefaultKafkaConsumerFactory<String, Test2> consumer2Factory = new DefaultKafkaConsumerFactory<>(consumer2Props);
            Consumer<String, Test2> consumer2 = consumer2Factory.createConsumer();
            consumer2.subscribe(Collections.singleton(OUTPUT_TOPIC));
            ConsumerRecords<String, Test2> records2 = consumer2.poll(Duration.ofSeconds(2));
            return records2.count() > 0;
        });
    }
}

3. 删除自定义的TestAvroDeserializer

不需要额外自定义Deserializer，只要使用共享的mock实例，流处理阶段自动注册的schema在消费时就可以正常识别。

可选替代方案 #

如果不想处理Mock实例的适配问题，可以直接使用Testcontainers启动真实的Kafka和Schema Registry容器，测试逻辑完全和生产环境一致，不需要适配Mock的特殊规则，更适合复杂的Kafka Streams+Avro场景测试。

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