如何在Kafka物化视图中访问自定义Header以实现DLQ重试逻辑？

我正在构建一个基于Header信息的自定义DLQ（死信队列）重试机制，重试操作由调度器触发，而这个调度器需要从Kafka Streams的物化视图中查询消息的Header信息。以下是我的实现代码和配置细节：

首先定义绑定接口和服务类，用来将DLQ主题的数据持久化为物化视图：

public interface DlqBinding {
    String DLQ_TOPIC = "dlq";

    @Input(DLQ_TOPIC)
    KTable<?, ?> dlqInput();
}

@Slf4j
@EnableBinding(DlqBinding.class)
public class DlqRetryService {
    @StreamListener
    public void readTable(@Input(DlqBinding.DLQ_TOPIC) KTable<String, String> table) {
        // 这里不需要额外逻辑，绑定后Kafka Streams会自动将DLQ主题的数据物化到指定存储中
    }
}

然后是Spring Boot配置，指定物化视图的名称以及序列化方式：

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            brokers: localhost:29092
          configuration:
            default:
              key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
              value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
      bindings:
        dlq:
          consumer:
            materializedAs: currentDL
            keySerde: org.apache.kafka.common.serialization.Serdes$StringSerde
            valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde

注意：配置中的materializedAs: currentDL是物化视图的存储名称，后续调度器会通过这个名称获取存储实例。

调度器通过InteractiveQueryService获取物化视图的只读存储，遍历其中的消息并执行重试逻辑：

public void processDL() {
    // 通过存储名称获取物化视图的只读键值存储
    ReadOnlyKeyValueStore<Object, Object> currentDL = interactiveQueryService.getQueryableStore("currentDL", QueryableStoreTypes.keyValueStore());
    KeyValueIterator<Object, Object> all = currentDL.all();
    
    while (all.hasNext()) {
        KeyValue<Object, Object> next = all.next();
        log.info("Found Entry in currentDL: {}", next);
        
        // 在这里添加基于Header的重试逻辑
        // 例如：解析消息中的Header信息，判断是否满足重试条件，调用重试方法等
    }
}

关键说明： #

• 确保interactiveQueryService已经正确注入到调度器类中，它是Spring Cloud Stream Kafka Streams提供的交互查询服务。
• 在实际的重试逻辑中，你可以根据消息的Header信息（比如重试次数、重试时间窗口等）来决定是否触发重试，以及如何重试（比如重新发送到原主题）。

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