如何基于条件反序列化Kafka消息及处理Spring环境下的未知事件？

一、过滤指定事件类型并自动提交非目标事件偏移 #

要实现仅处理PersonCreatedEvent和PersonMaritalStatusChangedEvent，可以选择以下两种方式：

方式1：监听器内直接判断类型（适用于已兼容所有事件POJO的场景） #

在Kafka监听器中接收PersonDomainEvent后，直接判断实例的实际类型，仅处理目标事件，其他事件手动提交偏移：

@KafkaListener(topics = "person.domain-events", groupId = "person-event-consumer-group")
public void handlePersonDomainEvent(PersonDomainEvent event, Acknowledgment ack) {
    if (event instanceof PersonCreatedEvent createdEvent) {
        // 处理PersonCreatedEvent业务逻辑
        processPersonCreated(createdEvent);
    } else if (event instanceof PersonMaritalStatusChangedEvent maritalEvent) {
        // 处理PersonMaritalStatusChangedEvent业务逻辑
        processMaritalStatusChange(maritalEvent);
    } else {
        // 其他事件类型直接提交偏移，不做处理
        ack.acknowledge();
    }
}

方式2：前置过滤（更高效，可结合需求2实现） #

配合下文的拦截器实现前置过滤，直接将非目标事件排除在监听器处理流程外，自动提交偏移。

二、拦截未识别Schema，避免反序列化失败 #

当生产者新增消费者未更新POJO的事件时，必须在反序列化之前拦截消息，避免抛出反序列化异常。核心思路是利用Confluent Schema Registry的schema.id消息头，查询事件类型是否在消费者支持列表中，不支持则直接过滤。

步骤1：配置Schema Registry客户端 #

注入Schema Registry客户端用于查询schema信息：

@Bean
public SchemaRegistryClient schemaRegistryClient(@Value("${spring.kafka.properties.schema.registry.url}") String registryUrl) {
    return new CachedSchemaRegistryClient(registryUrl, 100);
}

步骤2：自定义消费者拦截器 #

实现ConsumerInterceptor，在消息进入反序列化流程前完成过滤：

@Component
@Slf4j
public class UnsupportedEventInterceptor implements ConsumerInterceptor<String, byte[]> {

    private final SchemaRegistryClient schemaRegistryClient;
    // 消费者支持的事件类型全限定名
    private static final Set<String> SUPPORTED_EVENT_TYPES = Set.of(
        "com.yourpackage.PersonCreatedEvent",
        "com.yourpackage.PersonMaritalStatusChangedEvent"
    );

    public UnsupportedEventInterceptor(SchemaRegistryClient schemaRegistryClient) {
        this.schemaRegistryClient = schemaRegistryClient;
    }

    @Override
    public ConsumerRecords<String, byte[]> onConsume(ConsumerRecords<String, byte[]> records) {
        Map<TopicPartition, List<ConsumerRecord<String, byte[]>>> filteredMap = new HashMap<>();

        for (ConsumerRecord<String, byte[]> record : records) {
            Integer schemaId = extractSchemaId(record);
            if (schemaId == null) {
                log.warn("No schema.id found in record, skipping");
                continue;
            }

            try {
                Schema unionSchema = schemaRegistryClient.getById(schemaId);
                String actualEventType = getActualEventType(record.value(), unionSchema);
                
                if (SUPPORTED_EVENT_TYPES.contains(actualEventType)) {
                    filteredMap.computeIfAbsent(record.topicPartition(), k -> new ArrayList<>()).add(record);
                }
                // 不支持的事件直接跳过，后续自动提交偏移
            } catch (IOException e) {
                log.error("Failed to fetch schema with id: {}", schemaId, e);
            }
        }

        return new ConsumerRecords<>(filteredMap);
    }

    private Integer extractSchemaId(ConsumerRecord<String, byte[]> record) {
        Header schemaIdHeader = record.headers().lastHeader("schema.id");
        if (schemaIdHeader == null) {
            return null;
        }
        return ByteBuffer.wrap(schemaIdHeader.value()).getInt();
    }

    private String getActualEventType(byte[] payload, Schema unionSchema) {
        // Confluent Avro格式：1字节魔术位 + 4字节schemaId + 4字节schema索引（Union类型专用）
        if (payload == null || payload.length < 9 || unionSchema.getType() != Schema.Type.UNION) {
            return unionSchema.getFullName();
        }
        // 解析Union类型中的实际事件schema索引
        int schemaIndex = ByteBuffer.wrap(payload, 5, 4).getInt();
        Schema actualSchema = unionSchema.getTypes().get(schemaIndex);
        return actualSchema.getFullName();
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {}

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

步骤3：将拦截器配置到消费者工厂 #

在消费者工厂中添加拦截器配置，确保拦截器生效：

@Bean
public ConsumerFactory<String, PersonDomainEvent> consumerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    // 基础Kafka配置
    configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-bootstrap-servers");
    configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "person-event-consumer-group");
    // 拦截器配置
    configProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, 
        Collections.singletonList(UnsupportedEventInterceptor.class.getName()));
    // Avro反序列化配置
    configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
    configProps.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "your-schema-registry-url");
    configProps.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);

    return new DefaultKafkaConsumerFactory<>(configProps);
}

最终效果 #

1. 所有非PersonCreatedEvent和PersonMaritalStatusChangedEvent的消息都会被拦截器过滤，不会进入监听器，自动提交偏移量。
2. 当生产者新增消费者未定义POJO的事件类型时，拦截器会识别到该类型不在支持列表中，直接跳过，避免触发反序列化失败异常。

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