如何使用SSL Bundle配置Kafka Schema Registry客户端？配置后无法连接

核心问题说明 #

Schema Registry客户端本质是HTTP客户端，并非Kafka协议客户端，所以你之前通过spring.kafka.properties.schema.registry.ssl.bundle配置的方式不会生效——这个参数仅作用于Kafka生产者/消费者的SSL配置，无法传递给Schema Registry的HTTP请求逻辑。

方案1：自定义Schema Registry Client Bean（推荐） #

如果使用Confluent的io.confluent:kafka-schema-registry-client依赖，可以手动创建客户端实例时，注入Spring管理的SSL Bundle构建SSL上下文：

1. 确保项目已正确配置SSL Bundle（示例application.yml）：

spring:
  ssl:
    bundles:
      theBundle:
        key-store: classpath:client-keystore.jks
        key-store-password: 你的密钥库密码
        trust-store: classpath:client-truststore.jks
        trust-store-password: 你的信任库密码

2. 添加自定义配置类，创建带SSL配置的Schema Registry Client：

import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import org.springframework.boot.ssl.SslBundle;
import org.springframework.boot.ssl.SslBundles;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.net.ssl.SSLContext;

@Configuration
public class SchemaRegistryConfig {

    @Bean
    public SchemaRegistryClient schemaRegistryClient(SslBundles sslBundles) throws Exception {
        String schemaRegistryUrl = "https://你的SchemaRegistry地址:8081";
        // 获取指定的SSL Bundle
        SslBundle sslBundle = sslBundles.getBundle("theBundle");
        // 构建SSL上下文
        SSLContext sslContext = sslBundle.createSslContext();
        
        // 初始化带SSL配置的客户端
        return new CachedSchemaRegistryClient(
            schemaRegistryUrl,
            100, // 缓存大小
            null,
            null,
            sslContext
        );
    }
}

3. 关联Kafka序列化器（以Avro为例）：
   如果用Spring Kafka的KafkaTemplate集成Schema Registry，需要让序列化器使用上述自定义客户端：

import io.confluent.kafka.serializers.KafkaAvroSerializer;
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 java.util.Map;

@Configuration
public class KafkaProducerConfig {

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate(
        DefaultKafkaProducerFactory<String, Object> producerFactory,
        SchemaRegistryClient schemaRegistryClient
    ) {
        KafkaAvroSerializer avroSerializer = new KafkaAvroSerializer(schemaRegistryClient);
        producerFactory.setValueSerializer(avroSerializer);
        return new KafkaTemplate<>(producerFactory);
    }
}

方案2：通过系统属性自动传递SSL配置 #

若不想自定义Bean，可以让Spring Boot将SSL Bundle的配置导出为系统属性，Schema Registry的HTTP客户端会自动读取：

在application.yml中添加导出配置：

spring:
  ssl:
    bundles:
      theBundle:
        key-store: classpath:client-keystore.jks
        key-store-password: 你的密钥库密码
        trust-store: classpath:client-truststore.jks
        trust-store-password: 你的信任库密码
        export:
          enabled: true
          alias: theBundle

开启后，Spring Boot会自动导出以下系统属性，Schema Registry客户端的HTTP请求会默认使用这些配置：

• javax.net.ssl.keyStore
• javax.net.ssl.keyStorePassword
• javax.net.ssl.trustStore
• javax.net.ssl.trustStorePassword

验证配置有效性 #

开启DEBUG日志查看SSL握手细节，排查连接问题：

logging:
  level:
    io.confluent.kafka.schemaregistry.client.rest: DEBUG
    org.apache.http: DEBUG

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