用途
Configuring Topics 用于配置 Celigo Event Streams 中的主题（topic）。主题是一个具名的、持久化的消息通道：发布方追加消息，每个监听方都能收到全部消息，各自从自己的位置、按自己的节奏读取；消息在主题的保留窗口结束前一直可读，无论是否被读取过。
主题将产生事件的流程与消费事件的流程解耦：可以新增消费者而不改动生产者，暂停某个消费者而其他消费者照常运行，并让它从停止处继续追赶。
主题是账户资源，位于一个环境内、独立于任何集成，类似查找缓存。流程通过两个步骤使用主题，两者都不需要连接：
- 发布方（Publisher）：adaptorType 为 "TopicImport" 的导入步骤，收到的每条记录成为一条消息。
- 监听方（Listener）：adaptorType 为 "TopicExport"、type 为 "stream" 的导出步骤，是消息到达时运行流程的页面生成器。
主要能力与字段
主题涉及以下关注点：主题本身（名称、保留期、许可门槛、在所用集成上的注册）、发布、监听、消息（读取保留消息、清除消息）、位置（每个监听方在主题上的偏移量）。
主题字段规则：
- name：最多 200 字符；名称不唯一，步骤绑定的是 _id，因此重命名不会破坏绑定。
- description：最多 5,120 字符，仅用于展示。
- retentionMs：消息可读时长，默认 604800000（7 天），最少 3600000（1 小时），最多 7776000000（90 天）及许可的 topics.retentionCeilingDays；缩短保留期会丢弃早于新窗口的消息。
- externalId：自定义键，可通过列表过滤找回主题而不必保存 id。
- draft：仅在创建时设置，48 小时后自动删除，除非更新将其设为 false。
- __integrationIds：仅在创建时使用，把新主题注册到这些集成。
仅 name 和 description 可单独修改；其他任何变更都属完整更新，完整更新在请求体缺少字段时会清空 description 和 externalId 并将 retentionMs 重置为 7 天，因此应始终发送完整主题对象。
许可与配额
Event Streams 需要许可。账户许可包含 topics.enabled、topics.topicsAllowed（账户可持有的主题数）和 topics.retentionCeilingDays。列出主题在所有账户上可用；创建或更新主题以及所有消息操作需要相应授权，否则写入主题返回 422 topics_subscription_required，消息调用返回 403。配额在创建主题或从回收站恢复时检查（topics_limit_exceeded），接近配额时没有警告，回收站中的主题不计入。
使用方式
构建事件流的步骤：
- 规划发布方与监听方：明确一条消息代表什么（如订单、订单行）及其形状（任意 JSON 对象，序列化后不超过 1 MiB）；确定谁发布（流程中的发布方步骤、API、Tool，或通过限定到该主题的 API 令牌从 Celigo 外部直接发布）；确定谁监听（每个消费者一个监听流程，每个监听方都收到全部消息）；确定保留期（至少覆盖监听方可能暂停或失败的最长时间）；考虑重复处理，为每个监听流程设计幂等写入。
- 检查已有主题：列出账户主题，按 externalId 过滤，或查看某集成已注册的主题；名称不唯一，复用前应按 _id 和 description 确认，并查看其依赖。
- 创建主题并注册：设置名称、描述、保留期，可选 externalId；通过 __integrationIds 在创建时注册，或之后通过集成的主题注册完成，或保存绑定了该主题的流程时自动注册。
- 构建发布方：adaptorType 为 "TopicImport" 且设置 topic._topicId 的导入步骤，无连接、无 create/update/upsert 逻辑；无映射时消息即步骤收到的记录，有映射时会重塑；oneToMany: true 配合 pathToMany 时数组每个元素作为独立消息发布；topic.batchSize 默认 1000，控制每次发布请求的记录数。流程或 API 的测试运行不会真正发布，而 invoke 会真正发布到所有订阅流程。
- 构建监听方及其流程：adaptorType 为 "TopicExport"、type 为 "stream"、设置 topic._topicId 的导出步骤，无连接、无 cdc 对象、无 isLookup；将其设为流程的页面生成器，并以 disabled: true、无调度创建流程。topic.startPosition 默认 latest（仅投递订阅后发布的消息），earliest 会重放主题仍保留的全部消息；topic.maxWaitTimeMs 为收集消息交给流程的批处理窗口，最小 1000，默认 30000。
- 启用流程以完成订阅：监听方在流程从禁用变为启用时订阅，约半分钟内出现位置；创建时即启用的流程从未订阅，需先禁用再启用；latest 模式下订阅形成前发布的消息会被跳过，因此应先启用再发布。
- 为每个监听流程的写入实现幂等：消息可能多次到达目的地，应按业务键写入；earliest 监听方订阅时会重放全部保留消息；同一环境内克隆流程中的监听方也会处理每条消息；错误重试会重发记录；位置过期或被清除的监听方会从仍保留的最早消息恢复。
消息与位置
发布：请求体为 {"messages": [ {...}, ... ]}，每条消息为 JSON 对象，序列化后最多 1 MiB；裸数组返回 400 invalid_messages，空数组不发布任何内容。200 仅表示请求体有效，需读取每个 responses[].statusCode。单条超大消息返回 422 topic_publish_event_too_large，其余消息仍发布。首次向新主题发布会创建其流，耗时略长。
读取：消息按从旧到新返回，每页最多 100 条（默认 50），格式为 {timestamp, payload}；沿 nextPageURL 翻页，可用 time_gt 和 time_lte（epoch 毫秒）按发布时间过滤；没有搜索功能；读取不会移动任何监听方的位置。
清除：删除所有保留消息，保留主题、保留期、注册和订阅，不可撤销。清除后偏移量继续计数，已启用的监听方照常接收后续消息。
位置：对某主题上的某个监听方，位置为 {hasCursor, offset}，offset 是监听方下一条将读取的消息。hasCursor 为 false（offset -1）表示从未订阅；禁用的流程保留监听方位置，重新启用后接收期间发布的一切消息；当监听方位置上的消息已不再保留时，即使 hasCursor 为 true，下次运行也会从仍保留的最早消息恢复。
故障诊断
监听方从不运行时，依次检查：导出步骤是否具备 TopicExport、type: "stream" 及正确的 topic._topicId；流程是否已启用（创建即启用的流程需禁用再启用）；监听方位置（hasCursor 为 false 表示从未订阅）；主题消息（保留数减 offset 为未读数量）；批处理窗口（默认约在发布后 30 秒开始运行）；再按常规流程排查运行记录和错误。latest 模式下订阅形成前发布的消息会被跳过，超出保留期或被清除的消息会丢失。
删除、恢复与副本
当有导出或导入步骤绑定该主题（即使不在任何流程中），或包含此类步骤的流程存在时，删除会被拒绝，返回 422 dependencies_not_deleted 并列出各阻塞项；删除主题无需先注销，删除会从所有注册的集成中移除该主题。已删除主题在回收站保留 30 天，期间列表不可见、读取返回 404；恢复会带回保留消息但不恢复注册，需重新注册；从回收站清除会永久删除主题及其消息。
适用场景与限制
适用：一个事件多个独立消费者、在流程间传递记录、可暂停并追赶的消费者。不适用：在流程中途用已发布数据丰富记录（监听方不能作为查找步骤）。事件流受许可和主题数配额限制；保留期上下限受许可约束；消息最大 1 MiB；读取无搜索功能；清除不可撤销；发布与清除应事先与用户确认。