在Spring Cloud Stream Kafka中正确管理DLQ [英] Correctly manage DLQ in Spring Cloud Stream Kafka

问题描述

我想使用kafka在Spring Cloud Stream中管理DLQ.

I want manage a DLQ in Spring Cloud Stream using kafka.

application.yaml

server:
    port: 8091
eureka:
    client:
        serviceUrl:
            defaultZone: http://IP:8761/eureka
spring:
    application:
        name: employee-consumer
    cloud:
        stream:
            kafka:
                binder:
                    brokers: IP:9092
                bindings:
                    greetings-in:
                        destination: greetings
                        contentType: application/json
                    greetings-out:
                        destination: greetings
                        contentType: application/json
            bindings:
                greetings-out:
                    consumer:
                        enableDlq: true
                        dlqName: dead-out
    kafka:
      consumer:
        group-id: A

如您所见，在我的配置中，我启用了dlq并将名称设置为dlq主题.

As you can see in my configuration I enable dlq and set a name to the dlq topic.

为了测试DLQ行为，我在某些消息上抛出了异常

To test DLQ behaviour I throw an exception on certain messages

我的听众组件

@StreamListener("greetings-out")
    public void handleGreetingsInput(@Payload Greetings greetings) throws Exception {
        logger.info("Greetings input -> {}", greetings);
        if (greetings.getMessage().equals("ciao")) {
            throw new Exception("eer");
        }
    }

这样，等于"ciao"的消息将引发异常，并且在日志中，我看到它被处理了三遍

In this way, the message that is equal to "ciao" throws an exception and in logs I see that it gets processed three times

2018-07-09 13:19:57.256  INFO 1 --- [container-0-C-1] com.mitro.service.GreetingsListener      : Greetings input -> com.mitro.model.Greetings@3da9d701[timestamp=0,message=ciao]
2018-07-09 13:19:58.259  INFO 1 --- [container-0-C-1] com.mitro.service.GreetingsListener      : Greetings input -> com.mitro.model.Greetings@5bd62aaf[timestamp=0,message=ciao]
2018-07-09 13:20:00.262  INFO 1 --- [container-0-C-1] com.mitro.service.GreetingsListener      : Greetings input -> com.mitro.model.Greetings@c26f92b[timestamp=0,message=ciao]
2018-07-09 13:20:00.266 ERROR 1 --- [container-0-C-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessagingException: Exception thrown while invoking com.mitro.service.GreetingsListener#handleGreetingsInput[1 args]; nested exception is java.lang.Exception: eer, failedMessage=GenericMessage [payload=byte[32], headers={kafka_offset=3, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@510302cb, deliveryAttempt=3, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, contentType=application/json, kafka_receivedTopic=greetings-out, kafka_receivedTimestamp=1531142397248}]

这对我来说很好，但是我不明白为什么会创建一个名为停顿"的主题(请看下面的图片).

This is fine for me, but I don't understand why a topic called dead-out is created as result (take a look at the image below).

我在做什么错了?

编辑1 :(仍然不会为DLQ创建主题)

server:
    port: 8091
eureka:
    client:
        serviceUrl:
            defaultZone: http://IP:8761/eureka
spring:
    application:
        name: employee-consumer
    cloud:
        stream:
            kafka:
                streams:
                    binder:
                        serdeError: sendToDlq
                binder:
                    brokers: IP:9092
                    auto-create-topics: true
                bindings:
                    greetings-out:
                        destination: greetings-out
                        contentType: application/json
                        consumer:
                          enableDql: true
                          dlqName: dead-out
                          autoCommitOnError: true
                          autoCommitOffset: true
            bindings:
                greetings-out:
                    destination: greetings-out
                    contentType: application/json
                    consumer:
                        enableDlq: true
                        dlqName: dead-out
                        autoCommitOnError: true
                        autoCommitOffset: true
    kafka:
      consumer:
        group-id: A

推荐答案

看起来您的属性被颠倒了；通用属性-destination，contentType-必须在 spring.cloud.stream.bindings 下.特定于kafka的属性(enableDlq，dlqName)必须位于 spring.clound.stream.kafka.bindings 下.

Looks like your properties are reversed; the common properties - destination, contentType - must be under spring.cloud.stream.bindings. The kafka-specific properties (enableDlq, dlqName) must be under spring.clound.stream.kafka.bindings.

您将它们颠倒了.

编辑

(修改后的)配置存在两个问题.

There are two problems with your (modified) config.

1. typo enableDql 而不是 enableDlq
2. 没有群组-您不能与匿名使用者一起使用DLQ:

原因:java.lang.IllegalArgumentException:DLQ支持不适用于匿名订阅

Caused by: java.lang.IllegalArgumentException: DLQ support is not available for anonymous subscriptions

这很好:

spring:
  application:
    name: employee-consumer
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          auto-create-topics: true
        bindings:
          input:
            consumer:
              enableDlq: true
              dlqName: dead-out
              autoCommitOnError: true
              autoCommitOffset: true
      bindings:
        input:
          group: so51247113
          destination: greetings-out
          contentType: application/json

和

@SpringBootApplication
@EnableBinding(Sink.class)
public class So51247113Application {

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

    @StreamListener(Sink.INPUT)
    public void in(String in) {
        System.out.println(in);
        throw new RuntimeException("fail");
    }

    @KafkaListener(id = "foo", topics = "dead-out")
    public void dlq(Message<?> in) {
        System.out.println("DLQ:" + in);
    }

}

这篇关于在Spring Cloud Stream Kafka中正确管理DLQ的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！