MessageConversionException尝试序列化Avro规范记录列表并使用@Splitter拆分它们时

Question

我有一个带有RabbitMQ绑定器的Spring Cloud Stream（Elmhurst Releas）Splitter。我正在尝试更新它，以便它将使用Avro模式作为有效负载。我得到了一个转换异常，这使得它看起来像是Avro转换器没有被调用，JSON转换器接收消息并在其上跳闸。

原因：组织。springframework。消息传递。转换器。MessageConversionException：无法写入JSON：不是映射：{“type”：“record”，“name”：“SkinnyMessage”，“namespace”：“com.example.avro”，“doc”：“用于传递消息对象引用的轻消息”，“fields”：[{“name”：“id”，“type”：“string”}，{“name”：“guid”，“type”：“string”}]}（通过引用链：com.example.avro.skinymessage[“schema”]-

我已经确认，我可以使用生成的Avro类（maven-Avro插件）创建对象并将其序列化到磁盘，所以这一部分似乎是正确的。我已经将该项目从Groovy转换为Java，但仍然会遇到相同的错误，所以我认为这也被排除了。

这是Avro模式：

{
  "namespace": "com.example.avro",
  "type": "record",
  "name": "SkinnyMessage",
  "doc": "Light message for passing references to Message objects",
  "fields": [
    {
      "name": "id",
      "type": "string"
    },
    {
      "name": "guid",
      "type": "string"
    }
  ]
}

以及课程的相关部分：

@EnableBinding(Processor.class)
@EnableSchemaRegistryClient
class PagingQueryProcessorApplication {
    @Timed(value = 'paging.query')
    @Splitter(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
    List<SkinnyMessage> queryExecutor(def trigger){
        log.debug 'Building query'
        def query = queryConfiguration.buildQuery().toUriString()

        log.info "Executing query: ${query}"
        def response = service.getRecordings(query)

        log.info "Returning response collection: ${response.body.content.size()}"
        // We build a slim notification on each of the query responses
        def skinnyMessages = response.body.content.collect{
            new SkinnyMessage(it.getLink('self').getHref(), it.content.guid)
        }
        skinnyMessages
    }
...
}

编辑：当我逐步使用调试器时，我可以看到AvroSchemaRegistryClientMessageConverter失败了canConvertTo（有效负载，标头）调用，因为标头中的mimeType是application/json，而不是application/*avro，所以它继续尝试转换器链的其余部分。

Answer 1

如果我构建了一条消息并在其上设置了一个avro内容类型头，这似乎是可行的，但这似乎是一种黑客行为。

List<Message<SkinnyMessage>> skinnyMessages = response.body.content.collect{
    MessageBuilder.withPayload(
            new SkinnyMessage(it.getLink('self').getHref(), it.content.recordingGuid))
            .setHeader('contentType', 'application/*+avro')
            .build()
}

创建在RabbitMQ UI中正确显示的消息：

内容类型：应用程序/vnd。skinnymessage。v1 avro相关ID:f8be74d6-f780-efcc-295d-338a8b7f2ea0内容\u类型：应用程序/八位字节流有效负载96字节编码：字符串

thttps://example.com/message/2597061H9a688e40-3e30-4b17-80e9-cf4f897e8a91

如果我正确理解了文档，那么从应用程序中的设置来看，这应该是透明的。yml：（在模式注册表示例中）：

spring:cloud:stream:bindings:output:contentType:application/*avro