1

kafka-smallrye为了满足我的服务的可靠性，我需要使用and将所有无法反序列化的传入消息推送到死信主题中quarkus。

有关该主题的所有消息都应采用 avro 格式（但我无法确定），并在模式注册表上定义了模式。

我以这种方式设置了我的消费者的配置：

mp:
  messaging:
    incoming:
      test-in:
        connector: smallrye-kafka
        group:
          id: test-in-consumer-group
        topic: events-topic
        failure-strategy: dead-letter-queue
        schema:
          registry:
            url: http://localhost:8081
        value:
          deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
          subject:
            name:
              strategy: io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
        key:
          deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
          subject:
            name:
              strategy: io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
        specific:
          avro:
            reader: true

我的消费者代码：

@ApplicationScoped
public class Consumer {

    @Incoming("test-in")
    public CompletionStage<Void> store(KafkaRecord<Key,SpecificRecord> data ){
            String schemaFullName = data.getPayload().getSchema().getFullName();
            System.out.println(schemaFullName);

            // other consumer code
            return data.ack();
    }
}

当消费者无法反序列化消息时，消费进程被阻塞，而是将消息移动到死信并继续。我想反序列化错误不会产生，nack因此无法将消息移至死信。

有没有办法将不可反序列化的消息移动到死信主题？

4

0 回答 0