Spring Cloud Stream Avro场景下DLQ消息JSON化配置问询

我刚好处理过类似的场景，给你一套可行的实现方案，分配置和代码两部分来拆解：

核心思路 #

我们需要给主业务绑定和DLQ绑定分别配置不同的消息处理策略：

• 主绑定：继续使用Avro序列化+Confluent Schema Registry，保持dynamicSchemaGenerationEnabled=false来拦截非法消息
• DLQ绑定：单独配置JSON序列化/反序列化，不需要依赖Schema Registry，确保消息内容完整留存

第一步：差异化绑定配置（application.yml） #

在配置文件中，给主绑定和DLQ绑定分别指定不同的消息转换器和序列化方式：

spring:
  cloud:
    stream:
      schema:
        avro:
          dynamicSchemaGenerationEnabled: false  # 全局禁用动态Schema生成
          schema-registry-client:
            endpoint: http://your-schema-registry:8081  # 主流程用的Schema Registry地址
      bindings:
        # 主业务生产者/消费者绑定（示例：business-output/business-input）
        business-output:
          destination: business-topic
          content-type: application/vnd.apache.avro.v1+json  # 也可使用application/avro-binary，按需选择
          producer:
            use-native-encoding: true
            schema-key-strategy: io.confluent.kafka.serializers.subject.TopicNameStrategy
        business-input:
          destination: business-topic
          content-type: application/vnd.apache.avro.v1+json
          consumer:
            use-native-decoding: true
        # DLQ生产者绑定配置
        dlq-output:
          destination: shared-dlq-topic
          content-type: application/json  # 明确指定JSON格式
          producer:
            use-native-encoding: false  # 禁用原生编码，使用Spring的消息转换器
        # 若需要消费DLQ，添加消费者绑定配置
        dlq-input:
          destination: shared-dlq-topic
          content-type: application/json
          consumer:
            use-native-decoding: false
      kafka:
        binder:
          brokers: your-kafka-broker:9092
          configuration:
            key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
            value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
            schema.registry.url: http://your-schema-registry:8081
        bindings:
          business-input:
            consumer:
              enable-dlq: true
              dlq-name: shared-dlq-topic  # 指定共享DLQ主题

第二步：自定义消息转换器配置 #

创建一个配置类，为DLQ绑定单独注册Jackson JSON转换器，确保主绑定仍然优先使用Avro转换器：

import org.springframework.cloud.stream.binder.MessageConverterConfigurer;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;

@Configuration
public class StreamMessageConverterConfig {

    @Bean
    public MessageConverterConfigurer dlqMessageConverterConfigurer() {
        return new MessageConverterConfigurer() {
            @Override
            public void configureCompositeMessageConverterFactory(CompositeMessageConverterFactory factory) {
                // 为DLQ绑定添加JSON转换器
                MappingJackson2MessageConverter jsonConverter = new MappingJackson2MessageConverter();
                jsonConverter.setSerializedPayloadClass(String.class);
                jsonConverter.setStrictContentTypeMatch(false);
                factory.addConverter(jsonConverter);
            }
        };
    }

    // 确保Avro转换器优先级高于JSON（若默认配置已生效，此步骤可省略）
    @Bean
    public MessageConverter avroMessageConverter() {
        return new org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter();
    }
}

第三步：DLQ消息序列化细节处理 #

当主流程的Avro消息处理失败被路由到DLQ时，Spring Cloud Stream会自动将原始消息包装成ErrorMessage对象。为了让这个对象能正确序列化成JSON，你可以选择：

1. 依赖Jackson默认序列化：ErrorMessage是Spring内置类，Jackson能直接序列化它，包含原始Payload、错误栈等信息
2. 自定义错误格式：实现ErrorHandler来定制DLQ消息内容

以下是自定义错误处理器的示例：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.messaging.Message;

@Configuration
public class DlqErrorHandlerConfig {

    @Bean
    public ErrorHandler dlqErrorHandler() {
        return (exception, message) -> {
            // 自定义错误消息内容，提取原始Payload和错误信息
            String errorContent = String.format(
                "Processing failed: %s, Original Payload: %s",
                exception.getMessage(),
                new String((byte[]) message.getPayload())
            );
            // 如需手动发送到DLQ，可在此处调用DLQ绑定的发送方法
            // 若使用默认DLQ机制，Spring会自动完成包装与发送
        };
    }
}

验证要点 #

1. 发送不符合Schema的Avro消息到主业务主题，会被拦截并路由到DLQ
2. 查看DLQ主题的消息，应为JSON格式，包含完整的错误信息与原始Payload
3. 主业务流程的消息仍保持Avro格式，依赖Schema Registry完成校验

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