带有Avro消息的Spring Cloud Stream JSON死信队列

Question

我正在使用带有Avro和汇流模式注册表的Spring云流。我正在为所有服务使用一个单独的DLQ主题，因此具有不同模式的消息可能会落在这个主题中。我已禁用动态架构注册，以确保不传递错误消息(schemaspring.cloud.stream.schema.avro.DynamicSchemaGenerationEnabled=false)。

然而，问题是由于dlq上缺少模式，我可能会在进入这个主题时丢失一条消息。因此，我希望能够以JSON格式向dlq生成消息，并在管道的其余部分使用Avro。如果有人能帮助我如何做到这一点，或者能为我指出这件事的一个例子，我将不胜感激。

Answer 1

如果您使用的是Stream2.1或更高版本，请禁用绑定器中的DLQ处理，并使用ListenerContainerCustomizerbean向侦听器容器添加自定义的ErrorHandler；您可以将SeekToCurrEnterrorHandler与自定义恢复器一起使用-您可以将DeadLetterPublishingRecoverer作为起点-重写此方法...

/**
 * 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, RecordHeaders headers, @Nullable byte[] data, boolean isKey) {