Kafka 软件包 中的消费消息 操作 使您能够使用消费者组轮询 Apache Kafka 主题以获取一批消息，并可配置批大小、轮询超时、偏移重置策略和提交策略，将所有已消费的消息作为用于在您的自动化工作流中处理的表变量返回。
先决条件
- 使用连接 操作 建立的活动 Kafka 会话。 请参阅 Kafka 软件包 中的连接 操作。
- 目标 Kafka 主题存在于 broker 上，并且已认证用户对该主题具有消费者权限。
- 消费者组 ID 可用。
- 如果您计划使用手动提交策略，请确保在自动化工作流中的此 操作 之后添加一个提交偏移量 操作，以确认处理。 请参阅 Kafka 软件包 中的提交偏移量 操作。
设置
使用消费消息 操作 通过消费者组轮询 Apache Kafka 主题以获取一批消息。 操作 返回一个表变量，其中每一行代表一条消息。 您可以配置每次轮询的最大消息数、轮询超时时间（毫秒）、偏移重置策略（最早或最新），以及偏移是否自动提交或手动提交。 当您需要精确一次处理语义时，请使用手动提交策略；随后，自动化必须在成功处理每条消息后调用提交偏移量 操作。
-
会话名称： 选择以下一个选项，以指定要使用的 Kafka 会话：
- 会话名称： 选择由连接 操作 返回的活动 Kafka 会话，或接受默认会话名称 Default。
- 变量： 选择一个包含活动 Kafka 会话配置的会话变量。
- 主题： 输入要消费的 Kafka 主题的确切名称。 例如，transactions-data。
- 消费者组 ID： 输入用于跟踪已消费偏移量的消费者组标识符。 所有共享相同组 ID 的自动化共享分区分配和偏移量状态。 例如，erp-sync-group。
- 最大消息数： 输入一个正整数（1 或更大值），表示每次轮询请求要检索的最大消息数。 默认值为 10。 如果主题中可用的记录较少，返回的消息也可能更少。
- 轮询超时 (ms)： 输入一个非负整数，表示消费消息 操作 在返回消息之前等待的最大毫秒数。 默认值为 5000（5 秒）。 输入 0 以立即返回任何可用消息。
-
偏移量重置： 选择当该组和主题不存在已提交偏移量时，消费者如何定位自身。 默认为最早。 选择以下选项之一：
- 最早： 从主题中最旧的可用消息开始读取。 当您希望从头处理所有历史消息时，请使用此选项。
- 最新： 从消费者订阅后产生的最新消息开始读取。 仅在您只想处理后续的新消息时使用此选项。
-
提交策略： 选择消费者在轮询后如何提交偏移量。 默认为自动提交。 选择以下选项之一：
- 自动提交： 消费者会在固定间隔自动提交偏移量。 将此选项用于至少一次处理，在这种情况下，失败时偶尔重新处理消息是可以接受的。
- 手动： 仅当自动化显式调用提交偏移量 操作 时，偏移量才会被提交。 使用此选项可实现精确一次处理语义，在推进消费者位置之前必须确认处理成功。
- （可选）CSV 文件路径： 输入绝对文件路径，将已消费的消息保存为 CSV 文件。 如果留空，则消息仅在表格输出变量中返回，不会写入文件。
- 将输出分配到变量： 指定消费消息 操作 存储结果的变量。 消费消息 操作 返回一个包含来自 Kafka 主题的最多指定数量消息的表变量。 表中的每一行代表一条消息。 使用循环 操作 迭代这些行，并在后续自动化步骤中处理每条消息。