PublishKafka

文档中心 › 组件索引 › PublishKafka

PublishKafka Processor

类型全限定名：cn.org.acal.lynxus.kafka.processors.PublishKafka

所属 Bundle：cn.org.acal.lynxus:lynxus-kafka-nar:2.11.8

描述

使用 Kafka Producer API 将 FlowFile 的内容作为消息或作为单独记录发送到 Apache Kafka。

标签

ApacheKafkaRecordcsvjsonavrologsPutSendMessagePubSub

属性

下表中，加粗的属性为必填项；其余为可选项。表格同时标注默认值、是否支持表达式语言，以及是否为敏感属性（值会被加密）。

名称默认值可选值描述
Kafka Connection Service控制器服务 API：
KafkaConnectionService
cn.org.acal.lynxus-lynxus-kafka-service-api-nar-2.11.8		提供与 Kafka Broker 的连接以发布 Kafka 记录
Topic Name处理器发布 Kafka 记录的 Kafka 主题的名称
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Failure StrategyRoute to Failure
  • Route to Failure — 当无法将记录发布到 Kafka 时，FlowFile 将路由到 failure 关系。
  • Rollback — 当无法将记录发布到 Kafka 时，FlowFile 将被放回队列以便重试。 对于 FlowFile 顺序很重要的数据流，可以使用此策略，同时确保每个处理器仅使用单个并发任务（Concurrent Task）。
指定处理器无法将数据发布到 Kafka 时如何处理 FlowFile
传递保证all
  • Guarantee Replicated Delivery — 除非消息根据主题配置被复制到相应数量的 Kafka 节点，否则 FlowFile 将路由到 failure
  • Guarantee Single Node Delivery — 如果消息被单个 Kafka 节点接收，无论是否被复制，FlowFile 都将路由到 success。这比 <Guarantee Replicated Delivery> 更快，但如果某个 Kafka 节点崩溃，可能导致数据丢失
  • Best Effort — 在成功将内容发送到 Kafka 节点后，FlowFile 将路由到 success，完全不等待该节点的任何确认。这提供了最佳性能，但可能导致数据丢失。
指定保证消息发送到Kafka的要求。 对应于Kafka Client acks属性。
压缩类型none
  • none
  • gzip
  • snappy
  • lz4
  • zstd
指定发送到 Kafka 的记录的压缩策略。 对应于Kafka Client的compression.type属性。
最大请求大小1 MB请求的最大大小（以字节为单位）。 对应于Kafka客户端max.request.size属性。
Transactions Enabledtrue
  • true
  • false
指定在与 Kafka 通信时是否提供事务保证。如果向 Kafka 发送数据时出现问题且此属性设置为 false，则已发送到 Kafka 的消息将继续传递给消费者。如果设置为 true，则 Kafka 事务将回滚，使这些消息不可用于消费者。将此设置为 true 要求 [Delivery Guarantee] 属性设置为 [Guarantee Replicated Delivery.]。
Transactional ID Prefix指定 KafkaProducer 配置 transactional.id 将是生成的 UUID，并将以配置的字符串作为前缀。
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
分区器类org.apache.kafka.clients.producer.internals.DefaultPartitioner
  • RoundRobinPartitioner — 消息将以轮询（round-robin）方式分配分区，将第一条消息发送到分区 1，下一条发送到分区 2，依此类推，必要时回绕。
  • DefaultPartitioner — 默认分区策略将选择严格统一粘性分区器（Strictly Uniform Sticky Partitioner）（有关更多详细信息以及可通过动态属性使用的可用配置选项，请参见 KIP-794）。
  • Expression Language Partitioner — 将 <Partition> 属性解释为将针对每个 FlowFile 求值的表达式语言（Expression Language）。该表达式将针对 FlowFile 求值一次，因此给定 FlowFile 中的所有记录都将进入同一分区。
指定使用哪个类来计算消息的分区 ID。 对应于Kafka Client的partitioner.class属性。
分区指定记录的 Kafka 分区目标。
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Message Demarcator指定用作 UTF-8 解释的字符串，用于在单个 FlowFile 中标记多条消息的分界。如果未指定，FlowFile 的整个内容将作为单条消息使用。如果指定，FlowFile 的内容将按此分隔符拆分，每部分作为单独的 Kafka 消息发送。要输入特殊字符（如'换行符'），请使用 CTRL+Enter 或 Shift+Enter，具体取决于操作系统。
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Record Reader控制器服务 API：
RecordReaderFactory
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		用于传入FlowFile的Record Reader
Record Writer控制器服务 API：
RecordSetWriterFactory
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		用于在发送到 Kafka 之前序列化数据的Record Writer
Publish StrategyUSE_VALUE
  • Use Content as Record Value — 仅将 FlowFile 内容写入 Kafka Record 值。
  • Use Wrapper — 将 Kafka Record 的键、值、头和元数据写入 Kafka Record 值。（更多信息参见处理器用法。）
用于将传入的 FlowFile 记录发布到 Kafka 的格式。
Message Key Field输入记录中应用作 Kafka 消息的键的字段名称。
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
FlowFile Attribute Header Pattern与所有 FlowFile 属性名称匹配的正则表达式。属性名称匹配此模式的任何属性都将作为 Header 添加到 Kafka 消息中。如果未指定，则不会将任何 FlowFile 属性添加为标头。
Header EncodingUTF-8对于作为 Kafka 记录标头添加的任何属性，此属性指示用于序列化标头的字符编码。
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Kafka Key用于消息的键。如果未指定，则使用 FlowFile 属性 'kafka.key' 作为消息键（如果存在）。请注意，同时设置 Kafka 键和分隔符可能导致多条 Kafka 消息具有相同的键。通常这不是问题，因为 Kafka 不强制或假设消息和键的唯一性。但是，同时设置分隔符和 Kafka 键会带来 Kafka 上数据丢失的风险。在 Kafka 进行主题压缩时，消息将基于此键进行去重。
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Kafka Key Attribute Encodingutf-8
  • UTF-8 Encoded — 键被解释为 UTF-8 编码的字符串。
  • Hex Encoded — 键被解释为任意二进制数据，并编码为小写十六进制字符串
  • Do Not Add Key as Attribute — 键不会作为属性添加
发出的 FlowFile 具有名为 'kafka.key' 的属性。此属性决定该属性的值应如何编码。
Record Key Writer控制器服务 API：
RecordSetWriterFactory
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		用于传出FlowFile的记录密钥编写器
Record Metadata StrategyFROM_PROPERTIES
  • Metadata From Record — Kafka Record 的主题和分区将分别通过查看 Record 的 /metadata/topic 和 /metadata/partition 字段来确定。如果这些字段无效或不存在，则将考虑处理器的 Topic Name 和 Partition/Partitioner class 属性。
  • Use Configured Values — Kafka Record 的主题将使用 'Topic Name' 处理器属性来确定。分区将使用 'Partition' 和 'Partitioner class' 属性来确定。
指定 Record 的元数据（主题和分区）应来自 Record 的元数据字段，还是来自配置的主题名称和分区/分区器类属性。

关系 (Relationships)

名称描述自动终止
failure任何无法发送到 Kafka 的 FlowFile 将被路由到此关系否
success所有内容都发送到 Kafka 的 FlowFiles。否

读取的 FlowFile 属性

名称描述
kafka.tombstone如果此属性设置为 'true'，且处理器未配置分隔符，且 FlowFile 的内容为 null，则将向 Kafka 发送零字节的墓碑消息。

写入的 FlowFile 属性

名称描述
msg.count发送到此 FlowFile 的 Kafka 消息数。此属性仅添加到路由到的 FlowFile

状态管理

该组件不存储状态。

受限 (Restricted)

该组件不受限。

输入要求

该组件要求有传入连接。

另请参阅

© 2026 ACAL Organization. Licensed under Apache License 2.0.