KafkaSink

启动类

connector-kafka.jar

com.oceanbase.oms.connector.kafka.sink.KafkaSinkBooster

【元信息】servers

• 功能：kafka servers 信息

• 默认值：无，必填

• 值的选择范围：string

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【元信息】enableAsync

• 功能：是否采用异步写入

• 默认值：true

• 值的选择范围：boolean

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【分发】partition

• 功能：如果 partitionMode=ONE，表示固定写入的分区名称

• 默认值：0

• 值的选择范围：int

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【分发】partitionMode

• 功能：分发逻辑，确定哪条数据写入哪个分区

• 默认值：HASH

• 值的选择范围：HASH（主键 Hash 确定）/ ONE（固定一个）/ TABLE（按照 Table 的名称 Hash）

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

• 备注：请注意在 ONE 模式下，由于用户期望是顺序执行，所以 workerNum 需要设置为 1，否则会存在写并发问题

【写入线程】workerNum

• 功能：写入线程数

• 默认值：4

• 值的选择范围：int

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【Kafka 配置项】batchSize

• 功能：kafka 的 batch.size，缓冲区大写，单位为字节

• 默认值：16M

• 值的选择范围：int

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【Kafka 配置项】lingerMs

• 功能：Kafka 攒批时间

• 默认值：100ms

• 值的选择范围：int

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【Kafka 配置项】properties

• 功能：透传 Kafka 配置项

• 默认值：

    {
      "max.request.size": "1024 * 1024 * 1024", // 1 GB
      "buffer.memory": "128 * 1024 * 1024", // 128 MB
      "retries": "1000",
      "max.block.ms": "30 * 60 * 1000", // 30 分钟
      "request.timeout.ms": "30 * 60 * 1000", // 30 分钟
      "acks": "1",
      "max.in.flight.requests.per.connection": "1",
      "compression.type": "lz4",
      "receive.buffer.bytes": "16 * 1024 * 1024",
      "send.buffer.bytes": "16 * 1024 * 1024"
    }
  

• 值的选择范围：kafka producer config 配置项

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【鉴权】securityProtocol

• 功能：鉴权协议

• 默认值：空

• 值的选择范围：SASL_PLAINTEXT/SASL_SSL/PLAINTEXTSASL/SSLSASL

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【鉴权】sslTruststoreLocation

• 功能：SSL 鉴权文件地址

• 默认值：空

• 值的选择范围：path

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【鉴权】saslConfig

• 功能：SASL 配置项

• 默认值：无

• 值的选择范围：

    {
      "mechanism": "鉴权机制 GSSAPI/PLAIN/SCRAM-SHA-256/SCRAM-SHA-512",
      "username": "如果是 PLAIN/SCRAM-SHA-256/SCRAM-SHA-512，则需要账号密码",
      "password": "如果是 PLAIN/SCRAM-SHA-256/SCRAM-SHA-512，则需要账号密码",
      "kerberosPrincipal": "",
      "kerberosKeyTabFileLocation": "当鉴权是 GSSAPI 时，则需要填写，表示鉴权配置文件",
      "jaasConfFileLocation": "当鉴权是 GSSAPI 时，则需要填写，表示鉴权配置文件",
      "kerberosConfFileLocation": "当鉴权是 GSSAPI 时，表示 krb conf 文件地址"
    }
  

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

【序列化】serializerType

• 功能：序列化格式

• 默认值：DEFAULT

• 值的选择范围：DEFAULT/DEFAULT_WITH_SCHEMA/SHAREPLEX/CANAL/DATAWORKS_V2/DATAWORKS_V1

• 支持的版本：OMS 社区版 V3.3.1-CE 及之后版本

isTopicMandatory

• 功能：是否开启 Topic 存在性强制检查

• 默认值：true

• 值的选择范围：true 或 false

• 支持的版本：OMS 社区版 V4.1.1-CE 及之后版本

consumerGroup

• 功能：用于检查 Topic 是否存在 consumerGroup

• 默认值：test

• 值的选择范围：string

• 支持的版本：OMS 社区版 V4.1.1-CE 及之后版本

【时区转换】dateTimezone

• 功能：设置 dateTime 所在的时区

  例如，配置为 +08:00，表示 dateTime 是 +8 时区的时间

• 默认值：+00:00

• 值的选择范围：+18:00 ~ -18:00

• 支持的版本：OMS 社区版 V4.2.7-CE 及之后版本

【时区转换】dateConvertTimezone

• 功能：设置时间转换后的时区

  例如，配置为 +08:00，表示将时间转换为 +8 时区的时间。

• 默认值：+00:00

• 值的选择范围：+18:00 ~ -18:00

• 支持的版本：OMS 社区版 V4.2.7-CE 及之后版本

【时区转换】dateConvertFormat

• 功能：时间格式化配置

  例如，yyyy-MM-dd HH:mm:ss，配置后 datetime 和 timestamp 会转换为根据指定格式输出时间。

• 默认值：空

• 值的选择范围：string

• 支持的版本：OMS 社区版 V4.2.7-CE 及之后版本

时区转换相关参数的配置示例如下。

dateTimezone：'+08:00'

dateConvertTimezone: '+08:00'