---
title: "KafkaSink | OceanBase 文档中心"
description: KafkaSink 启动类 connector-kafka.jar com.oceanbase.oms.connector.kafka.sink.KafkaSinkBooster 【元信息】servers 功能：kafka servers 信息 默认值：无，必填 值的选择范围：string 支持的版本：OMS 社区版…
---
切换语言

- 中文站 - 简体中文
- International - English
- 日本站 - 日本語

文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*L03BS6f-o40AAAAAAAAAAAAADiGDAQ/original) 迁移服务 OMSV 4.2.8 社区版

# KafkaSink

更新时间：2026-04-14 15:40:47

[编辑](https://github.com/oceanbase/oms-doc/edit/V4.2.8/zh-CN/900.o-m-manual/400.components/500.component-parameters/400.sink-plugin/300.kafka-sink.md)  

## 启动类

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 配置项
 - 默认值：

  ```JSON
    {
      "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 配置项
 - 默认值：无
 - 值的选择范围：

  ```JSON
    {
      "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'

| 时间类型 | 数据库存储时间 | 同步后（不设置 dateConvertFormat） | 同步后（dateConvertFormat 为 yyyy-MM-dd HH:mm:ss） |
| --- | --- | --- | --- |
| datetime | 2024-12-01 08:00:00 | 1733011200000 | 2024-12-01 08:00:00 |
| timestamp | 2024-12-01 08:00:00 | 2024-12-01T08:00:00+08:00 | 2024-12-01 08:00:00 |

 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线