Skip to main content

​
Overview

The SingleStore Sink Connector consumes records from Kafka topics and writes record values to SingleStore tables in batches through LOAD DATA LOCAL INFILE. It sits between Kafka event streams and SingleStore databases used for real-time analytics or transaction processing, making it suitable for continuously ingesting business events, application logs, and state updates. For a single-topic input, the connector uses the topic name as the destination table name by default, or it can map that topic to a fixed table. With record field routing enabled, it can also group records from the same batch into multiple destination tables based on field values. Fields in Struct records with a Connect schema usually map to columns with the same names, while scalar records with a Connect schema are written to the data column. Schemaless Maps can be written to existing tables with compatible structures. The connector can create a missing table from a record with a Connect schema, but it does not perform schema evolution for existing tables.

​
Prerequisites

  • The destination database must already exist. The database account must be able to query table metadata and execute LOAD DATA LOCAL INFILE. It also needs permission to create tables when automatic table creation is used, and permission to create, read, and write the metadata table when the default metadata feature is retained.
  • The SingleStore server and JDBC connection must allow LOAD DATA LOCAL INFILE and support the selected GZip, LZ4, or uncompressed transfer mode.
  • Destination tables must be created in advance for schema-less records. The column names, types, nullability, and primary or unique keys of existing tables must accept the rows generated by the Connector. Subsequent Schema changes must be maintained on the database side.

​
License

Licensed under the Apache License 2.0.

​
Quick Start

Prepare a Connect Cluster, Kafka topic, SingleStore database, and a database account with the required permissions, and verify network connectivity and access permissions. For preparation and management instructions, see Manage Connectors. The following minimal practical configuration reads string records through a SingleStore Helios workspace endpoint and writes them to a table with the same name as the topic.
Replace the placeholders with actual resources. Specify the endpoint in the host:port format accepted by the JDBC driver. Create the destination table in advance with the same name as the topic and a data column that can accept strings. Supply the password through a secure credential management method, and do not commit actual credentials to version control.

​
Configuration

​
Connector Identity and Input Subscription

​
connector.class

Selects the SingleStore Sink Connector implementation class.
  • Type: string
  • Default: None
  • Importance: High
  • Required: Yes
  • Valid values / Notes: Use com.singlestore.kafka.SingleStoreSinkConnector.

​
tasks.max

Sets the maximum number of Sink Tasks that Kafka Connect can create.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid values / Notes: Must be at least 1. Actual effective parallelism is limited by the number of input topic partitions and partition assignment. Multiple Tasks do not provide write ordering across Tasks or serialization for the same key.

​
topics

Specifies the list of Kafka topics consumed by the Connector.
  • Type: list
  • Default: Empty list
  • Importance: High
  • Required: Mutually exclusive with topics.regex
  • Valid values / Notes: Use comma-separated topic names. Exactly one of this setting and a non-empty topics.regex must be configured. Validation fails if both are set or both are empty.

​
topics.regex

Selects the Kafka topics consumed by the Connector through a Java regular expression.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Required: Mutually exclusive with topics
  • Valid values / Notes: Must be a valid, non-empty Java regular expression and must be mutually exclusive with topics. When a new matching topic is added, also verify its destination table routing and table structure.

​
Record Conversion

​
key.converter

Overrides the Kafka key converter that this connector would otherwise inherit from the worker.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: null means that the Worker configuration is inherited. An explicit value must be a Converter class with a public no-argument constructor. The Connector does not use the Kafka Key to generate destination columns or row identifiers.

​
value.converter

Overrides the Kafka value converter that this connector would otherwise inherit from the worker.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid values / Notes: null means that the Worker configuration is inherited. An explicit value must be a Converter class with a public no-argument constructor. The Schema and value produced by the Converter determine automatic table creation, field mapping, and row serialization behavior.

​
SingleStore Connection and Authentication

​
connection.ddlEndpoint

Sets the DDL endpoint used for table discovery, table creation, and queries in self-managed deployments.
  • Type: string
  • Default: null
  • Importance: High
  • Required: Mutually exclusive with connection.clientEndpoint
  • Valid values / Notes: Mutually exclusive with connection.clientEndpoint. When a non-empty connection.dmlEndpoints is not configured, this endpoint is also used for DML writes. The JDBC driver validates the address format and connectivity.

​
connection.clientEndpoint

Sets a SingleStore Helios workspace endpoint used for both DDL and DML.
  • Type: string
  • Default: null
  • Importance: High
  • Required: Mutually exclusive with connection.ddlEndpoint
  • Valid values / Notes: Mutually exclusive with connection.ddlEndpoint and a non-empty connection.dmlEndpoints. When selected, this endpoint is used for all database operations.

​
connection.database

Sets the SingleStore database used for data and metadata writes.
  • Type: string
  • Default: None
  • Importance: High
  • Required: Yes
  • Valid values / Notes: The database name is added to the JDBC URL. The configuration definition does not reject an empty string, but the database must exist and be accessible to the account or the Task cannot start.

​
connection.user

Sets the SingleStore user used by the JDBC connection.
  • Type: string
  • Default: root
  • Importance: High
  • Valid values / Notes: Use a least-privilege account that satisfies the table discovery, table creation, and write requirements. The credentials are validated when the Task establishes the JDBC connection during startup.

​
connection.password

Sets the password for the SingleStore user.
  • Type: password
  • Default: null
  • Importance: High
  • Valid values / Notes: Supply the actual password through a secure configuration injection method. Do not expose credentials in logs, documentation, or version control.

​
connection.dmlEndpoints

Sets the list of SingleStore Aggregator endpoints used for DML writes.
  • Type: list
  • Default: null
  • Importance: Medium
  • Valid values / Notes: Use comma-separated endpoints. This setting can only be combined with connection.ddlEndpoint and conflicts with connection.clientEndpoint. When omitted or set to an empty list, DML uses the selected DDL or Client endpoint.

​
params.<value>

Passes dynamic connection parameters to the SingleStore JDBC driver.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: Specify an actual parameter as params.<driver-property>=<value>. The connector removes the params. prefix before passing the parameter to the driver. Parameter names and values follow the JDBC driver bundled with the connector. Some parameters might contain keys or certificate passwords. Do not set params.allowLocalInfile to a value that disables local file streaming.

​
Table Structure and Field Mapping

​
tableKey.<index_type>[.<name>]

Adds a key definition to tables automatically created by the Connector.
  • Type: list
  • Default: null
  • Importance: Low
  • Valid values / Notes: Index types are case-insensitive and can be PRIMARY, COLUMNSTORE, UNIQUE, SHARD, or KEY, optionally followed by a key name. The value is a comma-separated list of column names. This setting applies only when missing tables are created, does not modify existing tables, and uses the same set of key configurations for all automatically created tables.

​
fields.whitelist

Retains only the listed top-level record fields for subsequent routing and writing.
  • Type: list
  • Default: null
  • Importance: Medium
  • Valid values / Notes: Use comma-separated, case-sensitive field names. This setting applies to Structs and schema-less Maps. null or an empty list means that fields are not restricted. When used with the blacklist, the whitelist is applied first.

​
fields.blacklist

Excludes the listed top-level fields from records.
  • Type: list
  • Default: null
  • Importance: Medium
  • Valid values / Notes: Use comma-separated, case-sensitive field names. This setting applies to Structs and schema-less Maps. null or an empty list means that no fields are excluded. If the same field appears in both the whitelist and blacklist, the blacklist ultimately removes it.

​
singlestore.columnToField.<tableName>.<columnName>

Maps a destination table column to a field path in the Kafka record.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: The table name and column name in the property name must each occupy exactly one dot-separated segment and cannot contain additional dots. The property value can use a dot-separated nested field path. The mapping applies only when the parsed destination table name matches exactly. For an existing table, a missing path produces SQL NULL. If the connector must create the table from a schema but cannot resolve the mapped field schema, table creation fails.

​
Destination Table Routing

​
singlestore.tableName.<topicName>

Maps a Kafka topic to a fixed SingleStore table.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: Replace <topicName> with the actual topic name and set the value to the destination table name. When no mapping matches, the topic name is used as the table name. This setting cannot be used with singlestore.recordToTable.mappingField.

​
singlestore.recordToTable.mappingField

Specifies the field path used to select the destination table for each record.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: You can use a dot-separated Struct field or Map key path and must configure the corresponding singlestore.recordToTable.mapping.<value>. This setting cannot be used with any topic-to-table mapping. Records are skipped when the field path does not exist, the field value is null, or no mapping matches the value.

​
singlestore.recordToTable.mapping.<value>

Maps a routing field value to a SingleStore table.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: Replace <value> with the actual routing value and set the property value to the destination table name. At least one concrete mapping requires singlestore.recordToTable.mappingField to be set as well. Only records that match a mapping are written.

​
Writes, Deduplication, and Compression

​
singlestore.filter

Adds a WHERE filter expression to the generated LOAD DATA statement.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid values / Notes: The expression is appended directly to the SQL without Connector parsing or parameterization. Use only trusted expressions that apply to all destination tables, and do not place external input directly in this setting.

​
singlestore.upsert

Enables REPLACE behavior for LOAD DATA.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid values / Notes: true or false. Existing rows are replaced only when a PRIMARY or UNIQUE key conflict occurs in the destination table. This is a full-row replacement, not a partial field merge, and it does not provide a transactional guarantee between Kafka offsets and the database.

​
singlestore.metadata.allow

Enables the metadata table and duplicate-batch suppression based on the first record identifier in a batch.
  • Type: boolean
  • Default: true
  • Importance: Medium
  • Valid values / Notes: When enabled, the Connector creates or uses the metadata table, checks the batch identifier before writing, and places the metadata write and the table writes for that batch in the same database transaction. This mechanism only covers batch replays with the same first Kafka coordinate and must not be treated as end-to-end exactly-once delivery.

​
singlestore.metadata.table

Sets the name of the metadata table used by the Connector.
  • Type: string
  • Default: kafka_connect_transaction_metadata
  • Importance: Low
  • Valid values / Notes: Applies when singlestore.metadata.allow=true. Use a table name that can be created, queried, and written in the current database. The configuration does not validate whether the identifier is legal.

​
singlestore.loadDataCompression

Selects the compression method for the LOAD DATA LOCAL INFILE data stream.
  • Type: string
  • Default: GZip
  • Importance: Low
  • Valid values / Notes: Case-insensitive. Valid values are GZip, LZ4, and Skip. Skip disables compression. The driver and server must support the corresponding file extension and transfer method.

​
Retries and Metric Tags

​
max.retries

Sets the maximum number of Connector-level retries after an SQL exception.
  • Type: int
  • Default: 10
  • Importance: Medium
  • Valid values / Notes: Must be greater than or equal to 0. 0 means that the first SQL exception causes failure. The retry budget resets after a successful write and is independent of Kafka Connect’s general error-handling configuration.

​
retry.backoff.ms

Sets the requested wait time before retrying an SQL write.
  • Type: int
  • Default: 3000
  • Importance: Medium
  • Valid values / Notes: Must be greater than or equal to 0, in milliseconds. This setting is used only while max.retries still has remaining attempts. There is no exponential backoff or jitter.

​
custom.metric.tags

Adds custom tags to the JMX ObjectName for each Task.
  • Type: list
  • Default: null
  • Importance: Low
  • Valid values / Notes: Use comma-separated key=value entries. Each entry must contain exactly one =, and a later entry takes precedence for a duplicate Key. Values are sanitized for JMX, but Keys are added to the ObjectName as written, so avoid invalid JMX characters.

​
Best Practices

​
Map a Business Topic to a Stable Destination Table Name

Applicable business scenario: During initial onboarding, the Kafka topic name contains an environment, version, or organizational prefix, but the database needs a stable, concise business table name. The destination table should remain unchanged if the topic naming convention changes later. Configuration example:
Replace the connection placeholders with actual resources. Create orders_events in advance or grant the account permission to create the table automatically from the input schema. Key considerations: An explicit mapping decouples Kafka resource naming from database table naming, which simplifies permission, Schema, and downstream query maintenance. Topic mappings and record field routing are mutually exclusive. To split records by content, use singlestore.recordToTable.* instead and configure a destination table for every valid routing value.

​
Replace Duplicate State Rows Using a Destination Table Business Key

Applicable business scenario: The destination table stores the current state of customers, orders, or devices. Updates for the same business entity already have the required ordering within Kafka partitions, but replayed or duplicate events might arrive again, and the existing full row should be replaced based on a database PRIMARY or UNIQUE key. Configuration example:
Replace the connection placeholders with actual resources. Create the customers table in advance with a PRIMARY or UNIQUE key that uniquely identifies the business entity. Input records must use the schema-enabled JSON representation accepted by JsonConverter, contain fields compatible with the destination columns, and include the fields corresponding to the database key. Key considerations: singlestore.upsert=true uses SingleStore LOAD DATA REPLACE, which replaces the full row on conflict instead of updating only changed fields. Replacement follows processing order; it does not select the latest version by event time, and writes for the same business key across partitions are not coordinated across tasks. This setting can reduce the impact of duplicate writes, but Kafka offset commits and database transactions are not atomic. The metadata table also suppresses only some replays based on the first record in a batch, so these settings do not provide exactly-once delivery.

​
Increase Write Parallelism Gradually by Kafka Partition

Applicable business scenario: The Connector is running steadily, the input topic has multiple partitions, and consumer lag continues to grow. Additional Tasks are needed so multiple partitions can write to SingleStore concurrently. Configuration example:
Replace the placeholders and verify that the input topic has multiple assignable partitions. Increase tasks.max in small increments while monitoring consumer lag, Task write latency, SQL retries, database load, and Worker JVM resources. Key considerations: tasks.max is an upper limit. Increasing it beyond the number of assignable partitions does not produce more effective work. There is no global ordering or same-key coordination across Tasks. If the same business key can be written concurrently from multiple partitions, first define the partition key, database key, and REPLACE semantics clearly.

​
Monitoring

​
What to Monitor

Monitor Kafka Connect Worker health, Connector and Task status, input throughput, consumer lag, processing latency, offset commits, errors, SQL retries, and Worker JVM CPU, memory, and garbage collection signals. Also compare the actual number of rows written to SingleStore and database errors. Monitor DLQ activity only when the deployment enables the corresponding Kafka Connect error handling. Internal routing, serialization, and write errors from the Connector do not necessarily enter the DLQ.

​
Import the Grafana Dashboard

Verify that Kafka Connect metrics are available in a Grafana data source and that the collected labels meet the dashboard filtering requirements. Download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

​
Limitations

  • SingleStore data writes and Kafka offset commits are not part of the same transaction. A failure after the database commit but before the offset commit can replay records. The default metadata table only suppresses some duplicates based on the first record identifier in a batch and cannot provide end-to-end exactly-once delivery.
  • Schema-less records cannot be used to create missing tables automatically. Existing tables do not receive automatic Schema evolution. Fields that are added, removed, renamed, or changed to a different type must first be handled on the database side.
  • Kafka record keys and headers are not written to destination columns and do not automatically become table primary keys or update identifiers. Kafka tombstones do not delete destination rows.
  • singlestore.upsert=true depends on a PRIMARY or UNIQUE key in the destination table and performs a full-row REPLACE, not a partial field merge.
  • Record field routing writes only records that match a singlestore.recordToTable.mapping.<value>. Records are skipped when the field path does not exist, the field value is null, or no mapping is configured, and there is no dedicated skipped-record metric or DLQ report.
  • The Connector applies max.retries and retry.backoff.ms only to SQL exceptions. Internal Task errors involving routing, Schema access, serialization, metric registration, or local file streaming do not use this retry budget and are not written to the Kafka Connect DLQ by the Connector.
  • The Connector does not provide an independent record-count or byte-count batch setting for an individual write. The effective batch is determined by the Kafka Connect Consumer Poll and routing results, and multiple destination tables are written serially within one Task.

​
FAQ

​
How do I resolve an endpoint configuration conflict or a SingleStore connection failure during Task startup?

Verify that only one of connection.ddlEndpoint and connection.clientEndpoint is set. Remove connection.dmlEndpoints when using connection.clientEndpoint. Then check that connection.database exists, the account credentials are correct, and the account can access and query the destination database from every Worker running a Task. When a self-managed deployment requires multiple write endpoints, retain connection.ddlEndpoint and configure comma-separated connection.dmlEndpoints.

​
Why was the destination table not created automatically, or why does the write report mismatched columns?

Automatic table creation requires record values with a Connect Schema and a database account with table discovery and table creation permissions. Schema-less Maps and other schema-less values require a pre-created table. The Connector does not execute ALTER TABLE for an existing table. Check that the fields produced by the Value Converter, fields.whitelist, fields.blacklist, and singlestore.columnToField.* match the destination column names, types, and nullability, and complete the database Schema change first.

​
Why are some messages not written after record field routing is enabled?

Check whether the dot-separated field path in singlestore.recordToTable.mappingField can be resolved in the Struct or Map, and verify that every actual field value has a corresponding singlestore.recordToTable.mapping.<value>. A record is skipped if the path does not exist, the field value is null, or the value has no mapping. Also verify that a conflicting singlestore.tableName.<topicName> is not configured at the same time.

​
Why do SQL errors repeat before the Task eventually fails?

The connector retries the entire write batch for every SQL exception, up to max.retries times, and waits for retry.backoff.ms before each retry. After those retries are exhausted, another SQL exception for the batch causes the task to fail. Check the worker logs and SingleStore errors to distinguish transient problems such as connection interruptions from permanent problems involving permissions, SQL expressions, table structure, or duplicate keys. Correct the root cause before resuming the task instead of only increasing the retry count to mask an unrecoverable error.

​
Why do duplicate rows appear after failure recovery?

After a database write succeeds, Kafka Connect must still commit the Kafka offset separately. A failure between those operations causes the same batch of records to be delivered again. Keeping singlestore.metadata.allow=true can suppress batch replays with the same first Kafka coordinate. Tables that store the latest state can also use a stable PRIMARY or UNIQUE key with singlestore.upsert=true, but both mechanisms have applicability boundaries. Audit duplicate data with a business-unique identifier and design the pipeline to allow for replay.