KafkaSink

Starter class

connector-kafka.jar

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

[Metadata] servers

• Description: specifies information about Kafka servers.

• Default value: None. This parameter is required.

• Value type: String.

• Supported version: OMS V3.3.1 and later.

[Metadata] enableAsync

• Description: specifies whether to enable asynchronous writing.

• Default value: true.

• Value type: Boolean.

• Supported version: OMS V3.3.1 and later.

[Distribution] partition

• Description: specifies the name of the partition into which data is written when partitionMode is set to ONE.

• Default value: 0.

• Value type: int.

• Supported version: OMS V3.3.1 and later.

[Distribution] partitionMode

• Description: specifies the distribution logic that determines the partition into which a data record is to be written.

• Default value: HASH.

• Valid values: HASH (hash value of the primary key), ONE (fixed to one partition), and TABLE (hash value of the table name)

• Supported version: OMS V3.3.1 and later.

• Note: In the ONE mode, workerNum must be set to 1 for serial execution. Otherwise, data may be concurrently written.

[Writing threads] workerNum

• Description: specifies the number of data writing threads.

• Default value: 4.

• Value type: int.

• Supported version: OMS V3.3.1 and later.

[Kafka parameters] batchSize

• Description: specifies the batch size of the Kafka instance, which is the same as the buffer size, in bytes.

• Default value: 16M.

• Value type: int.

• Supported version: OMS V3.3.1 and later.

[Kafka parameters] lingerMs

• Description: specifies the time to wait before sending a batch in Kafka.

• Default value: 100 ms.

• Value type: int.

• Supported version: OMS V3.3.1 and later.

[Kafka parameters] properties

• Description: specifies the transparent transmission parameters of Kafka.

• Default values:

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

• Valid values: Producer config parameters of Kafka.

• Supported version: OMS V3.3.1 and later.

[Authentication] securityProtocol

• Description: specifies the authentication protocol.

• Default value: Null.

• Valid values: SASL_PLAINTEXT, SASL_SSL, PLAINTEXTSASL, and SSLSASL.

• Supported version: OMS V3.3.1 and later.

[Authentication] sslTruststoreLocation

• Description: specifies the address of the SSL authentication files.

• Default value: Null.

• Valid values: Path.

• Supported version: OMS V3.3.1 and later.

[Authentication] saslConfig

• Description: specifies the SASL parameters.

• Default value: None.

• Value range:

    {
      "mechanism": "The authentication mechanism. Valid values: GSSAPI, PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512",
      "username": "The username and password are required for the PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512 mechanisms.",
      "password": "The username and password are required for the PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512mechanisms.",
      "kerberosPrincipal": "",
      "kerberosKeyTabFileLocation": "An authentication configuration file, which is required for the GSSAPI mechanism.",
      "jaasConfFileLocation": "An authentication configuration file, which is required for the GSSAPI mechanism.",
      "kerberosConfFileLocation": "The address of the Kerberos configuration file, which is required for the GSSAPI mechanism."
    }
  

• Supported version: OMS V3.3.1 and later.

[Serialization] serializerType

• Description: specifies the serialization format.

• Default value: DEFAULT.

• Valid values: DEFAULT, DEFAULT_WITH_SCHEMA, SHAREPLEX, CANAL, DATAWORKS_V2, and DATAWORKS_V1.

• Supported version: OMS V3.3.1 and later.

isTopicMandatory

• Description: specifies the whether to enable force existence check for topics.

• Default value: true.

• Valid values: true and false.

• Supported version: OMS V4.0.1 and later.

consumerGroup

• Description: specifies to check whether a topic has a consumerGroup.

• Default value: test.

• Value type: String.

• Supported version: OMS V4.0.1 and later.