无法重试 Kafka 主主题（没有 RETRY 或 DLT 退出）答案

【问题标题】：Unable to retry on Kafka Main topic ( no RETRY or DLT exits)无法重试 Kafka 主主题（没有 RETRY 或 DLT 退出）
【发布时间】：2022-11-29 05:26:13
【问题描述】：

任何人都可以让我知道我们是否可以在主 Kafka 主题本身上重试，当在使用来自同一主主题的消息后发生异常时。我正在使用 spring-kaka，但看起来我们无法通过 spring-kafka 实现这一点。

【问题讨论】：

你的问题不清楚；默认错误处理程序将重试。显示您的代码和配置，并更清楚地解释您遇到的问题。
我只有一个话题。如果在服务级别使用事件后发生异常，我想将消息发回同一主题。我想在配置的时间内这样做。希望我的问题现在很清楚了。

标签： spring-boot apache-kafka spring-kafka

【解决方案1】：

这会做你想做的......

@SpringBootApplication
public class So74602308Application {

    public static void main(String[] args) {
        SpringApplication.run(So74602308Application.class, args);
    }

    @KafkaListener(id = "so74602308", topics = "so74602308")
    void listen(String in,
            @Header(KafkaHeaders.OFFSET) long offset,
            @Header(name = "my.retries", required = false) byte[] retries) {

        System.out.println(in + " @" + offset);
        int count = 0;
        if (retries != null) {
            ByteBuffer bb = ByteBuffer.wrap(retries);
            count = bb.getInt();
        }
        System.out.println("Count:" + count);
        if (count < 4) {
            throw new RuntimeException("retry this one to same queue, no delay");
        }
        else {
            System.out.println("Retries exhausted for record at offset " + offset);
        }
    }

    @Bean
    NewTopic topic() {
        return TopicBuilder.name("so74602308").partitions(1).replicas(1).build();
    }

    @Bean
    ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> {
            template.send("so74602308", "test");
        };
    }

    @Bean
    CommonErrorHandler errorHandler(KafkaTemplate<String, String> template) {
        return new DefaultErrorHandler(new DeadLetterPublishingRecoverer(template, (rec, ex) -> new TopicPartition("so74602308", -1)) {

            @Override
            protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?, ?> record,
                    TopicPartition topicPartition, Headers headers, @Nullable byte[] key, @Nullable byte[] value) {

                ProducerRecord<Object, Object> out = super.createProducerRecord(record, topicPartition, headers, key,
                        value);
                org.apache.kafka.common.header.Header header = out.headers().lastHeader("my.retries");
                if (header == null) {
                    header = new RecordHeader("my.retries", new byte[] { 0, 0, 0, 0 });
                    out.headers().add(header);
                }
                ByteBuffer bb = ByteBuffer.wrap(header.value());
                int retries = bb.getInt();
                bb.position(0);
                bb.putInt(++retries);
                return out;
            }

        }, new FixedBackOff(0L, 0L));
    }

}

test @21
Count:0
test @22
Count:1
test @23
Count:2
test @24
Count:3
test @25
Count:4
Retries exhausted for record at offset 25

【讨论】：