使用Mock Kafka Schema Registry测试时出现Schema获取失败问题求助

问题描述 #

我在Spring配置文件中做了如下设置：

spring.kafka.properties.schema.registry.url=mock://localhost

但运行测试时抛出以下错误：

Caused by: org.apache.kafka.common.errors.SerializationException: 获取ID为1的未知Avro schema失败
    at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.toKafkaException(AbstractKafkaSchemaSerDe.java:259)
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.schemaFromRegistry(AbstractKafkaAvroDeserializer.java:341)
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:113)
    at io.confluent.kafka.serializers.KafkaAvroDeserializer.deserialize(KafkaAvroDeserializer.java:55)
    at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:60)
    at org.springframework.kafka.support.serializer.ErrorHandlingDeserializer.deserialize(ErrorHandlingDeserializer.java:201)
    ... 16 common frames omitted
Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: 主题不存在；错误码：40401
    at io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient.getSchemaBySubjectAndIdFromRegistry(MockSchemaRegistryClient.java:177)
    at io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient.getSchemaBySubjectAndId(MockSchemaRegistryClient.java:262)
    at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.schemaFromRegistry(AbstractKafkaAvroDeserializer.java:330)
    ... 20 common frames omitted

解决方案 #

• 提前向Mock Schema Registry注册目标Schema
  MockSchemaRegistryClient不会自动加载任何Schema，必须手动注册测试用到的ID为1的Schema。可以在测试配置类中创建并初始化客户端：

  import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient;
  import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
  import org.apache.avro.Schema;
  import org.springframework.boot.test.context.TestConfiguration;
  import org.springframework.context.annotation.Bean;
  
  @TestConfiguration
  public class TestKafkaConfig {
  
      @Bean
      public SchemaRegistryClient schemaRegistryClient() {
          MockSchemaRegistryClient client = new MockSchemaRegistryClient();
          // 替换为你实际的Avro Schema内容
          String targetSchemaStr = "{\"type\":\"record\",\"name\":\"YourTestRecord\",\"fields\":[{\"name\":\"id\",\"type\":\"int\"}]}";
          Schema schema = new Schema.Parser().parse(targetSchemaStr);
          try {
              // 注册指定ID的Schema，subject名称要和业务使用的一致
              client.register("your-topic-value", schema, 1);
          } catch (Exception e) {
              throw new RuntimeException("注册Mock Schema失败", e);
          }
          return client;
      }
  }
  

• 校验Subject名称匹配
  错误提示“主题不存在”，要确认注册的subject名称和生产者/消费者使用的完全一致。Avro默认的subject命名规则是topic-name-value（值序列化）或topic-name-key（键序列化），如果自定义了命名策略，需同步调整注册的subject名称。

• 确保消息使用已注册的Schema ID
  如果是生产者发送测试消息，要保证序列化时使用的是已注册的ID（1）；如果是消费外部消息，需提前将消息携带的Schema ID对应的Schema注册到Mock客户端中。

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