DeadLetterPublishingRecoverer-死信发布失败与InvalidTopicException名称主题在Topic分区结束与_ERR

Question

当我更改DeadLetterPublishingRecoverer destionationResolver时，我发现了一个错误。

当我使用：

private static final BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition>
        DESTINATION_RESOLVER = (cr, e) -> new TopicPartition(cr.topic() + ".ERR", cr.partition());

它工作完美。

但是，如果使用\u ERR而不是。错误，出现错误：

2020-08-05 12:53:10,277 [kafka-producer-network-thread | producer-kafka-tx-group1.ABC_TEST_XPTO.0] WARN  o.apache.kafka.clients.NetworkClient - [Producer clientId=producer-kafka-tx-group1.ABC_TEST_XPTO.0, transactionalId=kafka-tx-group1.ABC_TEST_XPTO.0] Error while fetching metadata with correlation id 7 : {ABC_TEST_XPTO_ERR=INVALID_TOPIC_EXCEPTION}

2020-08-05 12:53:10,278[kafka-生产者-网络-线程|生产者-kafka-tx-group1.ABC_TEST_XPTO.0]ERRORorg.apache.kafka.clients.元数据-[生产者clientId=生产者-kafka-tx-group1.ABC_TEST_XPTO.0， transactionalId=kafka-tx-group1.ABC_TEST_XPTO.0]元数据响应报告无效主题[ABC_TEST_XPTO_ERR]2020-08-05 12:53:10,309[org.springframework.kafka.KafkaListenerEndpoint Container#0-0-C-1]ERROR o. s. k. s. LoggingProducerListener-发送消息时抛出的异常key='null'和payload='XPTOEvent（Super=Event（id=CAPBA2548，目标=ABC_TEST_XPTO， he...'到主题ABC_TEST_XPTO_ERR和分区0：org.apache.kafka.common.errors.InvalidTopicException：无效主题：[ABC_TEST_XPTO_ERR]2020-08-05 12:53:10,320[org.springframework.kafka.KafkaListenerEndpoint Container#0-C-1]ERROR o. s. k. l.DeadLetterPublishingRecoverer-死信发布失败：生产者记录（主题=ABC_TEST_XPTO_ERR，分区=0，标题=记录标题（标题=...org.springframework.kafka.KafkaException：发送失败；嵌套异常org.apache.kafka.common.errors.无效主题异常：无效主题：[ABC_TEST_XPTO_ERR]在org.springframework.kafka.core.KafkaTemplate.do发送（KafkaTemplate.java:573）在org.springframework.kafka.core.KafkaTemplate.send（KafkaTemplate.java:388）在org.springframework.kafka.listener.DeadLetterPublishingRecoverer.publish（DeadLetterPublishingRecoverer.java:290）在org.springframework.kafka.listener.DeadLetterPublishingRecoverer.accept（DeadLetterPublishingRecoverer.java:226）在org.springframework.kafka.listener.DeadLetterPublishingR（DeadLetterPublishingRecoverer.java:54）在org.springframework.kafka.listener.FailedRecordTracker.skip（FailedRecordTracker.java:106）强文本

“我的主题”在名称中间使用了“\uu”，例如ABC\u TEST\u XPTO，因此如果可能的话，我希望将死信主题设置为“ERR”

我的环境

>

Spring Boot 2.3.2。发布

Spring-Kafka 2.5.3。释放，但2.5.4出现同样的问题。释放

Java11

私有静态最终BiFunction

@组件类ContainerFactoryConfigurer{

ContainerFactoryConfigurer(ConcurrentKafkaListenerContainerFactory<?, ?> factory,
                           ChainedKafkaTransactionManager<?, ?> tm,
                           KafkaTemplate<Object, Object> template) {

    factory.getContainerProperties().setTransactionManager(tm);
    DefaultAfterRollbackProcessor rollbackProcessor = new DefaultAfterRollbackProcessor((record, exception) -> {
    }, new FixedBackOff(0L, Long.valueOf(maxAttemps)), template, true);
    factory.setAfterRollbackProcessor(rollbackProcessor);
    SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
            new DeadLetterPublishingRecoverer(Collections.singletonMap(Object.class, template), DESTINATION_RESOLVER), new FixedBackOff(0L, Long.valueOf(maxAttemps)));
    errorHandler.setCommitRecovered(true);
    errorHandler.setAckAfterHandle(true);
    factory.setErrorHandler(errorHandler);
}

}

谢谢DPG

Answer 1

这对我来说很好...

@SpringBootApplication
public class So63270367Application {

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

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

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

    @KafkaListener(id = "so63270367", topics = "ABC_TEST_XPTO_ERR")
    public void listen(String in) {
        System.out.println(in);
    }

}

spring.kafka.consumer.auto-offset-reset=earliest

也许你的经纪人对主题名称有一些规定；也许看看经纪人的日志？

编辑

正如我在评论中所说；记录从哪里发布并不重要；这对我仍然有效。。。

@SpringBootApplication
public class So63270367Application {

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

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

    @Bean
    public NewTopic topicErr() {
        return TopicBuilder.name("ABC_TEST_XPTO_ERR").partitions(1).replicas(1).build();
    }

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

    @KafkaListener(id = "so63270367", topics = "ABC_TEST_XPTO")
    public void listen(String in) {
        System.out.println(in);
        throw new RuntimeException("test");
    }

    @KafkaListener(id = "so63270367err", topics = "ABC_TEST_XPTO_ERR")
    public void listenErr(String in) {
        System.out.println("From DLT:" + in);
    }

    @Bean
    public SeekToCurrentErrorHandler eh(KafkaOperations<String, String> template) {
        return new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(
                template,
                (cr, e) -> new TopicPartition(cr.topic() + "_ERR", cr.partition())),
                new FixedBackOff(0L, 0L));
    }

}

From DLT:foo