如何配置Spring/Kafka自动发布Avro Schema到Schema Registry，无需向对应主题发送记录

要实现不发送Kafka消息就自动注册Avro Schema，核心是直接调用Schema Registry的客户端API完成注册，而非依赖生产者发送消息时的自动触发逻辑。以下是具体配置和实现步骤：

1. 添加必要依赖 #

确保项目中包含Spring Kafka、Avro以及Schema Registry客户端相关依赖。以Maven为例：

<dependencies>
    <!-- Spring Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>2.9.0</version> <!-- 适配你的Spring Boot版本 -->
    </dependency>
    <!-- Avro -->
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.11.0</version>
    </dependency>
    <!-- Schema Registry Client -->
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-schema-registry-client</artifactId>
        <version>7.3.0</version> <!-- 适配你的Confluent平台版本 -->
    </dependency>
</dependencies>

2. 配置Schema Registry连接 #

在application.properties或application.yml中配置Schema Registry的地址：

# Schema Registry地址
spring.kafka.properties.schema.registry.url=http://localhost:8081

3. 手动注册Schema #

创建一个启动时执行的Bean，通过Schema Registry客户端直接注册你的Avro Schema。假设你已经有一个Avro生成的类User：

import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
import org.apache.avro.Schema;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;

import java.io.IOException;

@Configuration
public class SchemaAutoRegisterConfig {

    private static final String SCHEMA_REGISTRY_URL = "http://localhost:8081";
    // 定义要注册的Subject，格式通常为{topic}-value或{topic}-key
    private static final String SUBJECT_NAME = "user-topic-value";

    @Bean
    public SchemaRegistryClient schemaRegistryClient() {
        return new CachedSchemaRegistryClient(SCHEMA_REGISTRY_URL, 100);
    }

    @EventListener(ApplicationReadyEvent.class)
    public void registerSchemaOnStartup(SchemaRegistryClient client) throws IOException, RestClientException {
        // 从Avro类中获取Schema
        Schema schema = User.getClassSchema();
        // 注册Schema到指定Subject
        client.register(SUBJECT_NAME, schema);
    }
}

关键说明 #

• Subject命名规则：默认情况下，Kafka生产者会使用{topic}-value或{topic}-key作为Subject名称，如果你需要自定义Subject，直接修改SUBJECT_NAME即可。
• Schema版本管理：Schema Registry会自动处理版本递增，重复注册相同Schema不会创建新的版本。
• 启动触发时机：通过ApplicationReadyEvent确保在Spring容器完全启动后执行注册操作，避免依赖未初始化的Bean。

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