Spring-Kafka 使用 DeadLetterPublishingRecoverer 将自定义记录而不是失败记录发送到 DLT答案

【问题标题】：Spring-Kafka Sending custom record instead of failed record using DeadLetterPublishingRecoverer to a DLTSpring-Kafka 使用 DeadLetterPublishingRecoverer 将自定义记录而不是失败记录发送到 DLT
【发布时间】：2021-02-11 21:17:10
【问题描述】：

我正在使用 DeadLetterPublishingRecoverer 将失败的记录自动发送到 DLT。我正在尝试向 DLT 发送自定义记录而不是失败记录。是否有可能做到这一点。请帮我配置一下。我的 DeadLetterPublishingRecoverer 配置如下。

@Bean
DeadLetterPublishingRecoverer deadLetterPublishingRecoverer(KafkaTemplate<String, byte[]> byteArrayTemplate) {
    return new DeadLetterPublishingRecoverer([
            (byte[].class)                           : byteArrayTemplate],)

}

【问题讨论】：

标签： spring-boot spring-kafka

【解决方案1】：

创建DeadLetterPublishingRecoverer 的子类并覆盖createProducerRecord() 方法。

/**
 * Subclasses can override this method to customize the producer record to send to the
 * DLQ. The default implementation simply copies the key and value from the consumer
 * record and adds the headers. The timestamp is not set (the original timestamp is in
 * one of the headers). IMPORTANT: if the partition in the {@link TopicPartition} is
 * less than 0, it must be set to null in the {@link ProducerRecord}.
 * @param record the failed record
 * @param topicPartition the {@link TopicPartition} returned by the destination
 * resolver.
 * @param headers the headers - original record headers plus DLT headers.
 * @param data the value to use instead of the consumer record value.
 * @param isKey true if key deserialization failed.
 * @return the producer record to send.
 * @see KafkaHeaders
 */
protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?, ?> record,

        TopicPartition topicPartition, Headers headers, @Nullable byte[] data, boolean isKey) {

在即将发布的 2.7 版本中，这将更改为

/**
 * Subclasses can override this method to customize the producer record to send to the
 * DLQ. The default implementation simply copies the key and value from the consumer
 * record and adds the headers. The timestamp is not set (the original timestamp is in
 * one of the headers). IMPORTANT: if the partition in the {@link TopicPartition} is
 * less than 0, it must be set to null in the {@link ProducerRecord}.
 * @param record the failed record
 * @param topicPartition the {@link TopicPartition} returned by the destination
 * resolver.
 * @param headers the headers - original record headers plus DLT headers.
 * @param key the key to use instead of the consumer record key.
 * @param value the value to use instead of the consumer record value.
 * @return the producer record to send.
 * @see KafkaHeaders
 */
protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?, ?> record,
        TopicPartition topicPartition, Headers headers, @Nullable byte[] key, @Nullable byte[] value) {

【讨论】：

我尝试覆盖该方法，但由于它是一个内部方法，所以无法调用它
我不熟悉 Kotlin internal 语义；如果 Kotlin 不允许您覆盖该方法，我觉得这很奇怪；您可能必须为这一类使用 Java。
stackoverflow.com/questions/49284094 那里的答案暗示您必须使用 Java。似乎是 Kotlin 的严重限制。