SAP ODP Source Connector - AutoMQ Documentation

概述

SAP ODP Source Connector 从 SAP Operational Data Provisioning（ODP）数据源读取数据，并将 ODP 返回的数据记录写入 Kafka Topic。Connector 使用 SAP Java Connector（JCo）连接 SAP，通过 ODP RFC 接口打开订阅、读取字段元数据、按包抓取数据并在任务停止时关闭订阅。每条记录按 ODP 字段生成 Connect Struct；十进制字段映射为 FLOAT64，其他字段映射为字符串。 Connector 适合将 SAP ODP 暴露的数据集接入 Kafka，供下游实时处理、数据集成或分析使用。I_EXTRACTION_MODE 选择 ODP 打开请求的提取模式，Topic 指定所有记录写入的 Kafka Topic。

前置条件

SAP 系统已启用目标 ODP 数据源，并提供 Connector 使用的 ODP 上下文、订阅者类型、订阅者名称和订阅者进程。
Connect Worker 的插件目录中已安装 Connector ZIP 解压后的插件文件，并额外提供与 Worker 平台匹配的 SAP Java Connector 运行时；SAP JCo 需要从 SAP 官方渠道单独获取。
用于连接 SAP 的账号具有调用 RODPS_REPL_ODP_OPEN、RODPS_REPL_ODP_GET_DETAIL、RODPS_REPL_ODP_FETCH 和 RODPS_REPL_ODP_CLOSE 的权限。

授权许可

使用 Apache License 2.0。

快速开始

准备 Connect Cluster、Kafka 以及可访问的 SAP ODP 数据源，确认 SAP 连接信息、ODP 订阅参数和 Kafka Topic 已准备完成。Connector 的通用管理操作请参阅 AutoMQ 的管理 Connector。下面的配置是一个最小可用配置示例。将尖括号中的值替换为实际环境值，并将这些键值作为 Connector 配置提交。

connector.class=com.confluent.SAPSourceConnector
jco.client.ashost=<sap-application-host>
jco.client.sysnr=<sap-system-number>
jco.client.client=<sap-client>
jco.client.user=<sap-user>
jco.client.passwd=<sap-password>
jco.client.lang=<sap-language>
jco.destination.pool_capacity=<pool-capacity>
jco.destination.peak_limit=<peak-limit>
I_ODPNAME=<odp-name>
I_CONTEXT=<odp-context>
I_SUBSCRIBER_TYPE=<subscriber-type>
I_SUBSCRIBER_NAME=<subscriber-name>
I_SUBSCRIBER_PROCESS=<subscriber-process>
Topic=<kafka-topic>
I_MAXPACKAGESIZE=<package-size>
I_EXTRACTION_MODE=D

jco.client.passwd 只用于连接 SAP，不要写入日志、工单或版本库。I_EXTRACTION_MODE=D 使用 Connector 的默认提取模式；如果需要 SAP ODP 支持的其他模式，请先确认对应订阅配置和源端语义。

配置

SAP JCo 连接

`jco.client.ashost`

SAP 应用服务器的主机名或地址。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：填写 SAP JCo 可访问的应用服务器地址。必填。

`jco.client.sysnr`

SAP 系统编号。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：填写 SAP 系统接受的系统编号。必填。

`jco.client.client`

SAP Client 编号。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：填写目标 SAP Client。必填。

`jco.client.user`

连接 SAP 的用户名。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：账号必须具备目标 ODP RFC 调用权限。必填。

`jco.client.passwd`

连接 SAP 的用户密码。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：必填。将其视为敏感信息，使用 Connect 的安全配置或外部密钥管理能力保护，不要记录到日志。

`jco.client.lang`

SAP 登录语言代码。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：填写 SAP 系统接受的语言代码。必填。

`jco.destination.pool_capacity`

SAP JCo 目标连接池容量。

类型：int
默认值：无
重要级别：高
有效值 / 注意事项：填写 SAP JCo 接受的整数。必填，并根据 SAP 端连接容量规划设置。

`jco.destination.peak_limit`

SAP JCo 目标连接池峰值连接数限制。

类型：int
默认值：无
重要级别：高
有效值 / 注意事项：填写 SAP JCo 接受的整数。必填，并确保不超过 SAP 端允许的连接能力。

ODP 订阅

`I_ODPNAME`

要读取的 ODP 名称。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：必须是 SAP 系统中可用的 ODP 名称。必填。

`I_CONTEXT`

ODP 上下文。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：必须与目标 ODP 在 SAP 中使用的上下文匹配。必填。

`I_SUBSCRIBER_TYPE`

ODP 订阅者类型。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：填写 SAP ODP RFC 接口接受的订阅者类型。必填。

`I_SUBSCRIBER_NAME`

ODP 订阅者名称。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：填写 SAP ODP 订阅配置中使用的名称。必填。

`I_SUBSCRIBER_PROCESS`

ODP 订阅者进程。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：填写目标 SAP ODP 接口接受的订阅者进程。必填。

`I_EXTRACTION_MODE`

选择打开 ODP 订阅时使用的提取模式。

类型：string
默认值：D
重要级别：高
有效值 / 注意事项：配置定义提供 D、F 和 R 作为推荐值。项目资料将 D 描述为增量提取、F 描述为全量提取；使用 R 前应确认目标 SAP ODP 的具体语义。

Kafka 输出

`Topic`

写入 ODP 记录的 Kafka Topic 名称。

类型：string
默认值：无
重要级别：高
有效值 / 注意事项：填写已创建且 Connect Worker 有写入权限的 Kafka Topic。必填。

批量抓取

`I_MAXPACKAGESIZE`

每次向 SAP ODP 请求抓取的数据包大小。

类型：int
默认值：无
重要级别：高
有效值 / 注意事项：填写 SAP ODP RFC 接口接受的正整数。值越大，单次请求携带的数据越多，但也会增加单次请求的内存和处理压力。必填。

最佳实践

首次接入时建立 SAP ODP 数据基线

适用业务场景：首次将 SAP ODP 数据源接入 Kafka，需要先读取源端提供的全量数据，再按 ODP 配置继续获取后续数据。使用快速开始中的完整连接和订阅配置，将 I_EXTRACTION_MODE 设置为 F，并保留与目标 ODP 匹配的订阅者参数。

I_EXTRACTION_MODE=F

关键说明：全量模式由 SAP ODP 的打开请求决定，实际可用范围取决于源端 ODP 是否支持该模式。首次运行前确认目标 Topic 的保留策略和下游消费起点，避免全量基线与下游初始化计划不一致。

已有数据基线时只接入后续变化

适用业务场景：下游已经通过其他方式完成历史数据基线，只需要从 SAP ODP 接收后续变化。使用快速开始中的完整配置，将 I_EXTRACTION_MODE 设置为 D，并使用与既有基线相同的 ODP 名称、上下文和订阅者信息。

I_EXTRACTION_MODE=D

关键说明：D 是 ConfigDef 的默认值，但仍需确认 SAP ODP 源端已为目标数据集启用增量语义。该 Connector 不在 SourceRecord 中提供源分区和源 Offset；下游需要自行设计重复处理和重启后的数据核对策略。

按数据包大小平衡吞吐与请求压力

适用业务场景：Connector 已能稳定读取 SAP ODP，但需要根据数据量和 SAP 系统负载调整单次抓取规模。保留其他配置不变，仅调整 I_MAXPACKAGESIZE：

I_MAXPACKAGESIZE=<package-size>

关键说明：较大的数据包可以减少请求次数，但会增加单次 SAP RFC 调用的处理和内存压力；较小的数据包更容易控制单次请求规模，但可能增加调用频率。应结合 SAP 端负载、Worker 内存和 Topic 消费延迟逐步调整。

监控

监控内容

监控 Kafka Connect Worker、Connector 和 Task 的状态，关注记录吞吐、处理延迟、Source Offset 提交、错误、重试、任务重启以及 Worker JVM 的堆内存和线程使用情况；同时观察 SAP 端 RFC 调用错误和连接池压力。由于该实现向 SourceRecord 提供空的源分区和源 Offset，不要仅依据 Offset 指标判断 SAP ODP 订阅是否已可靠推进。

导入 Grafana 大盘

从下载地址获取 Dashboard，使用已采集 Kafka Connect Worker、Connector 和 Task 指标的数据源，并在 Grafana 中导入 JSON 后选择对应数据源和标签。

限制条件

Connector 实现只生成一个 Task 配置；提高 tasks.max 不会将一个 SAP ODP 数据源拆分到多个 Task。
Connector 创建的 SourceRecord 不包含源分区和源 Offset，不能将其描述为基于 Kafka Connect Source Offset 的恢复机制。
运行 Connector 需要 SAP Java Connector 运行时和 SAP ODP RFC 接口；SAP JCo 不随公开 Connector ZIP 一起提供。

常见问题

为什么 Connector 启动后无法连接 SAP？

检查 jco.client.ashost、jco.client.sysnr、jco.client.client、jco.client.user、jco.client.passwd 和 jco.client.lang 是否与 SAP 登录参数一致，再确认 Connect Worker 能加载 SAP JCo 原生库和 Java 依赖。随后检查账号是否有目标 ODP RFC 函数的调用权限，以及 SAP 应用服务器是否允许来自 Worker 的连接。

为什么配置校验提示缺少参数？

除 I_EXTRACTION_MODE 外，Connector 的专属配置均未定义默认值。确认快速开始示例中的 SAP JCo、ODP 订阅、Topic 和 I_MAXPACKAGESIZE 配置都已填写，并检查属性名的大小写和下划线是否准确。

为什么 Kafka Topic 中没有记录？

先确认 Topic 是目标 Kafka Topic，Connector 账号有写入权限；再检查 I_ODPNAME、I_CONTEXT、I_SUBSCRIBER_TYPE、I_SUBSCRIBER_NAME 和 I_SUBSCRIBER_PROCESS 是否对应 SAP 中已存在且可订阅的 ODP。若使用 D，还要确认源端已有可供增量读取的数据；若使用 F，确认 ODP 支持全量提取并检查 SAP RFC 调用结果。

为什么设置较大的数据包后延迟或错误增加？

I_MAXPACKAGESIZE 会传给 RODPS_REPL_ODP_FETCH，较大的值会增加单次 SAP RFC 响应的大小和处理压力。结合 SAP 端连接池、Worker JVM 内存、Task 延迟和错误日志逐步降低该值，直到请求规模与系统负载平衡。

​概述

​前置条件

​授权许可

​快速开始

​配置

​SAP JCo 连接

​jco.client.ashost

​jco.client.sysnr

​jco.client.client

​jco.client.user

​jco.client.passwd

​jco.client.lang

​jco.destination.pool_capacity

​jco.destination.peak_limit

​ODP 订阅

​I_ODPNAME

​I_CONTEXT

​I_SUBSCRIBER_TYPE

​I_SUBSCRIBER_NAME

​I_SUBSCRIBER_PROCESS

​I_EXTRACTION_MODE

​Kafka 输出

​Topic

​批量抓取

​I_MAXPACKAGESIZE

​最佳实践

​首次接入时建立 SAP ODP 数据基线

​已有数据基线时只接入后续变化

​按数据包大小平衡吞吐与请求压力

​监控

​监控内容

​导入 Grafana 大盘

​限制条件

​常见问题

​为什么 Connector 启动后无法连接 SAP？

​为什么配置校验提示缺少参数？

​为什么 Kafka Topic 中没有记录？

​为什么设置较大的数据包后延迟或错误增加？

概述