ConsumeKafka

输出策略

该 Processor 提供多种输出策略（通过 Processor 属性 'Output Strategy' 配置），用于把 Kafka 记录转换为 FlowFile。

• 'Use Content as Value'（默认）输出的 FlowFile 记录只包含 Kafka 记录的值。
• 'Use Wrapper' 输出的 FlowFile 记录包含 Kafka 记录的键、值和 header，以及 Kafka 记录中的额外元数据。
• 'Inject Metadata' 输出的 FlowFile 记录包含 Kafka 记录的值，其中添加了一个子记录字段，用于保存元数据、 header 和键信息。

选择 "Use Wrapper" 时使用的记录 schema 如下（Avro 格式）：

{
  "type": "record",
  "name": "lynxusRecord",
  "namespace": "cn.org.acal.lynxus",
  "fields": [{
      "name": "key",
      "type": [{
          < Schema 由 Key Record Reader 决定，或根据 "Key Format" 属性为 "string" 或 "bytes"（详见下文） >
        }, "null"]
    },
    {
      "name": "value",
      "type": [
        {
          < Schema 由 Record Reader 决定 >
        },
        "null"
      ]
    },
    {
      "name": "headers",
      "type": [
        { "type": "map", "values": "string", "default": {}},
        "null"]
    },
    {
      "name": "metadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "topic", "type": ["string", "null"] },
            { "name": "partition", "type": ["int", "null"] },
            { "name": "offset", "type": ["int", "null"] },
            { "name": "timestamp", "type": ["long", "null"] }
          ]
        },
        "null"
      ]
    }
  ]
}

选择 "Inject Metadata" 时使用的记录 schema 如下（Avro 格式）：

{
  "type": "record",
  "name": "lynxusRecord",
  "namespace": "cn.org.acal.lynxus",
  "fields": [
      < 由 Record Reader 为 Kafka 消息确定的字段 >,
    {
      "name": "kafkaMetadata",
      "type": [
        {
          "type": "record",
          "name": "metadataType",
          "fields": [
            { "name": "topic", "type": ["string", "null"] },
            { "name": "partition", "type": ["int", "null"] },
            { "name": "offset", "type": ["int", "null"] },
            { "name": "timestamp", "type": ["long", "null"] },
            {
              "name": "headers",
              "type": [ { "type": "map", "values": "string", "default": {}}, "null"]
            },
            {
              "name": "key",
              "type": [{
                < Schema 由 Key Record Reader 决定，或根据 "Key Format" 属性为 "string" 或 "bytes"（详见下文） >
              }, "null"]
            }
          ]
        },
        "null"
      ]
    }
  ]
}

如果 Output Strategy 属性设为 'Use Wrapper' 或 'Inject Metadata'，会激活一个额外的 Processor 配置属性（'Key Format'）。该属性用于指定 Kafka 记录的键应如何写入 FlowFile。'Key Format' 的可选值如下：

• 'Byte Array' 以字节数组形式提供 Kafka 记录的键，与 Kafka 记录中接收到的完全一致。
• 'String' 使用 UTF-8 字符编码把 Kafka 记录键的字节转换为字符串。（无法把键字节解析为 UTF-8 时，记录会被路由到 'parse.failure' Relationship。）
• 'Record' 使用关联的 'Key Record Reader' Controller Service，把 Kafka 记录键的字节转换为反序列化的 Lynxus 记录。如果 Key Format 属性设为 'Record'，会出现名为 'Key Record Reader' 的额外 Processor 配置属性。该属性用于指定把 Kafka 记录的键解析为 Record 时使用的 Record Reader。

选择 "Use Wrapper" 策略时，JsonRecordSetWriter 输出的 FlowFile 内容示例：

[
  {
    "key": {
      "name": "Acme",
      "number": "AC1234"
    },
    "value": {
      "address": "1234 First Street",
      "zip": "12345",
      "account": {
        "name": "Acme",
        "number": "AC1234"
      }
    },
    "headers": {
      "attributeA": "valueA",
      "attributeB": "valueB"
    },
    "metadata": {
      "topic": "accounts",
      "partition": 0,
      "offset": 0,
      "timestamp": 0
    }
  }
]

选择 "Inject Metadata" 策略时，JsonRecordSetWriter 输出的 FlowFile 内容示例：

[
  {
    "address": "1234 First Street",
    "zip": "12345",
    "account": {
      "name": "Acme",
      "number": "AC1234"
    },
    "kafkaMetadata": {
      "topic": "accounts",
      "partition": 0,
      "offset": 0,
      "timestamp": 0,
      "headers": {
        "attributeA": "valueA",
        "attributeB": "valueB"
      },
      "key": {
        "name": "Acme",
        "number": "AC1234"
      }
    }
  }
]