如何为Apache Camel Kafka生产者配置AWS MSK Schema Registry校验

我已在AWS Glue的MSK Schema Registry中创建了Schema，但无法将其配置到应用的Kafka Producer中。

已创建的MSK Schema #

已在AWS Glue MSK Schema Registry中完成Schema创建

当前Kafka配置代码 #

class KafkaConfig {

    @Value("${brokerHost}")
    String brokerHost;

    @Value("${saslMechanism}")
    String saslMechanism;
    @Value("${securityProtocol}")
    String securityProtocol;
    @Value("${brokerConfig}")
    String brokerConfig;
    @Autowired
    CamelContext camelContext;

    @Bean
    public KafkaConfiguration kafkaConfiguration() {
        KafkaConfiguration kafkaConfiguration = new KafkaConfiguration();
        kafkaConfiguration.setBrokers(brokerHost);
        kafkaConfiguration.setSaslMechanism(saslMechanism);
        kafkaConfiguration.setSecurityProtocol(securityProtocol);
        kafkaConfiguration.setSaslJaasConfig(brokerConfig);
        return kafkaConfiguration;
    }

    @Bean(name = "appKafkaProducer")
    public KafkaComponent kafkaComponentProducer() throws Exception {
        KafkaComponent kafkaComponentProducer = new KafkaComponent();
        kafkaComponentProducer.setConfiguration(kafkaConfiguration());;
        camelContext.addComponent("appKafkaProducer", kafkaComponentProducer);
        return kafkaComponentProducer;
    }

    @Bean(name = "appKafkaConsumer")
    public KafkaComponent kafkaComponentConsumer() {
        KafkaComponent kafkaComponentConsumer = new KafkaComponent();
        kafkaComponentConsumer.setConfiguration(kafkaConfiguration());
        camelContext.addComponent("appKafkaConsumer", kafkaComponentConsumer);
        return kafkaComponentConsumer;
    }
}

生产者路由配置 #

<get uri="/refresh">
    <description>Refresh the Route</description>
    <route id="refreshRoute">
        <doTry>
        
            <to uri="appKafkaProducer:Checking" />
            <setBody>
                <constant>ok</constant>
            </setBody>
            <doCatch>
                <exception>java.lang.Exception</exception>
                <to uri="bean:commonsExchangeUtil?method=setErrorResponse(500, 'Internal Server Error', ${exception.message}, *)" />
            </doCatch>
        </doTry>
    </route>
</get>

查阅Camel官方文档未找到合适配置方案，请问需要添加哪些配置才能让生产者对接Schema Registry并实现消息校验？

要让Camel Kafka Producer对接AWS Glue MSK Schema Registry并实现消息校验，需完成以下配置：

1. 添加必要依赖 #

确保项目中引入AWS Glue Schema Registry的Kafka序列化器依赖：

<!-- AWS Glue Schema Registry Kafka SerDe -->
<dependency>
    <groupId>software.amazon.glue</groupId>
    <artifactId>schema-registry-serde</artifactId>
    <version>1.1.17</version> <!-- 使用最新稳定版本 -->
</dependency>
<!-- AWS SDK v2 Glue依赖 -->
<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>glue</artifactId>
    <version>2.20.0</version>
</dependency>

2. 配置Kafka生产者的序列化器与Schema Registry参数 #

修改KafkaConfig中的kafkaConfiguration()方法，添加Schema Registry相关配置：

@Bean
public KafkaConfiguration kafkaConfiguration() {
    KafkaConfiguration kafkaConfiguration = new KafkaConfiguration();
    kafkaConfiguration.setBrokers(brokerHost);
    kafkaConfiguration.setSaslMechanism(saslMechanism);
    kafkaConfiguration.setSecurityProtocol(securityProtocol);
    kafkaConfiguration.setSaslJaasConfig(brokerConfig);

    // 配置Schema Registry相关参数
    kafkaConfiguration.setProducerProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    // 根据Schema类型选择序列化器：JSON用GlueSchemaRegistryJsonSerializer，Avro用GlueSchemaRegistryAvroSerializer
    kafkaConfiguration.setProducerProperty("value.serializer", "software.amazon.glue.serializers.GlueSchemaRegistryJsonSerializer");
    // 指定AWS区域
    kafkaConfiguration.setProducerProperty("schema.registry.region", "${aws.region}");
    // 与Glue中创建的Schema名称保持一致
    kafkaConfiguration.setProducerProperty("schema.name", "你的Schema名称");
    // 开启自动注册Schema（若Schema未存在）
    kafkaConfiguration.setProducerProperty("auto.register.schemas", "true");
    // 强制开启消息结构校验
    kafkaConfiguration.setProducerProperty("validate.messages", "true");

    return kafkaConfiguration;
}

3. 配置AWS权限 #

确保应用运行的IAM角色具备以下权限：

• glue:GetSchema
• glue:RegisterSchemaVersion
• glue:GetSchemaVersion

本地测试时，需通过环境变量、~/.aws/credentials文件或IAM角色配置AWS凭证。

4. 验证消息结构 #

确保生产者发送的消息结构与Glue中定义的Schema完全匹配，否则会触发校验失败。

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