Storing state of a Debezium connector

Table of Contents

• Overview

• Kafka

  • Offset Store
  • Internal schema history store

• File

  • Offset Store
  • Internal schema history store

• Memory

  • Offset Store
  • Internal schema history store

• JDBC

  • Offset Store
  • Offset Table defaults
  • Internal schema history store
  • History Table defaults

• Redis

  • Offset Store
  • Internal schema history store
  • Redis Configuration Examples

• Amazon S3

  • Internal schema history store

• Azure Blob Storage

  • Internal schema history store

• RocketMQ

  • Internal schema history store

• Chronicle Queue

  • JVM setup
  • Chronicle Queue provider
  • Hybrid Chronicle Queue provider

Overview

Debezium connectors require persistent storage to preserve their state between restarts. All connectors require a mechanism to provide persistent storage for the offsets. In addition, connectors such as Db2, MySQL, Oracle, and SQL Server, require additional storage for their so-called internal schema history, which records changes to table schema in the database.

Offset storage is provided automatically for deployments within the Kafka Connect runtime through one of the following mechanisms:

If you run a connector in Debezium Engine or Debezium Server, you must explicitly configure the offset store. For connectors that work with schema-based databases, you configure internal schema history storage by setting connector properties.

Kafka

Debezium can use Kafka to store its state, including source offsets and schema history. Connectors implement KafkaOffsetBackingStore to store offsets in a Kafka topic (for example, connect-offsets). These offsets ensure that after a connector restarts, it is able to resume reading from the correct position. Connectors store their schema history in a separate compacted topic, such as schema-changes.inventory.

Offset Store

Internal schema history store

File

It’s possible to persist the position (offsets) of the connector in a local file on disk. These offsets ensure that after a connector restarts, Debezium can resume reading from the last read position. File storage provides a simple, fast mechanism for storing offsets that is ideal for single-node applications or testing scenarios.

Offset Store

Internal schema history store

Memory

MemoryOffsetBackingStore is a volatile, in-memory store that Debezium Embedded uses to track source offsets. Storing offsets in memory preserves the offset state only during the application’s runtime. Offset records are lost if the connector shuts down or crashes. Memory storage is ideal for testing, or for short-lived tasks, but it is not suitable for production environments that require persistent offsets.

Offset Store

Internal schema history store

JDBC

The store uses an arbitrary relational database to store the offset data. You must provide the JDBC driver for the database. Debezium can store data in the same source database from which it captures events, or you can configure it to use a different database.

Debezium provides pre-configured DML and DDL statements. You can use these default statements, or you can override the defaults with your own statements to provide compatibility with database dialects or to customize them for specific use cases.

Offset Store

Deprecated configuration prior to 3.2

Offset Table defaults

Create Query

CREATE TABLE %s (
id VARCHAR(36)      NOT NULL,
offset_key          VARCHAR(1255),
offset_val          VARCHAR(1255),
record_insert_ts    TIMESTAMP NOT NULL,
record_insert_seq   INTEGER NOT NULL),
PRIMARY KEY (id)

Select query

SELECT id, offset_key, offset_val FROM %s ORDER BY record_insert_ts, record_insert_seq

Insert query

INSERT INTO %s(id, offset_key, offset_val, record_insert_ts, record_insert_seq)
    VALUES ( ?, ?, ?, ?, ? )

Delete query

DELETE FROM %s

Internal schema history store

Deprecated configuration prior to 3.2

History Table defaults

Create Query

CREATE TABLE %s (
    id VARCHAR(36) NOT NULL,
    history_data VARCHAR(65000),
    history_data_seq INTEGER,
    record_insert_ts TIMESTAMP NOT NULL,
    record_insert_seq INTEGER NOT NULL,
    PRIMARY KEY (id, history_data_seq)
)

Select Query

SELECT id, history_data FROM %s
    ORDER BY record_insert_ts, record_insert_seq, id, history_data_seq

Data exist Query

SELECT * FROM %s LIMIT 1

Insert Query

INSERT INTO %s(id, history_data, history_data_seq, record_insert_ts, record_insert_seq) VALUES ( ?, ?, ?, ?, ? )

Redis

Debezium can use a Jedis client to store data in a Redis cache.

Debezium can use either a single Redis instance or use Redis Cluster mode:

Single instance mode

Connects to a single Redis server instance.

Cluster mode

Connects to a Redis Cluster for high availability and horizontal scaling.

To enable Redis Cluster mode, set the redis.cluster.enabled property to true and provide comma-separated host:port addresses in the redis.address property.

Offset Store

Internal schema history store

Redis Configuration Examples

Single Instance Mode

# Offset storage configuration
offset.storage=io.debezium.storage.redis.offset.RedisOffsetBackingStore
offset.storage.redis.address=localhost:6379
offset.storage.redis.password=password
offset.storage.redis.cluster.enabled=false

# Schema history storage configuration
schema.history.internal=io.debezium.storage.redis.history.RedisSchemaHistory
schema.history.internal.storage.redis.address=localhost:6379
schema.history.internal.storage.redis.password=password
schema.history.internal.storage.redis.cluster.enabled=false

Cluster Mode

# Offset storage configuration
offset.storage=io.debezium.storage.redis.offset.RedisOffsetBackingStore
offset.storage.redis.address=redis-node-1:7001,redis-node-2:7002,redis-node-3:7003
offset.storage.redis.password=password
offset.storage.redis.cluster.enabled=true

# Schema history storage configuration
schema.history.internal=io.debezium.storage.redis.history.RedisSchemaHistory
schema.history.internal.storage.redis.address=redis-node-1:7001,redis-node-2:7002,redis-node-3:7003
schema.history.internal.storage.redis.password=password
schema.history.internal.storage.redis.cluster.enabled=true

Amazon S3

Debezium can use the Amazon S3 object storage service. Typically, you would use S3 storage when you deploy Debezium with Amazon Managed Streaming for Apache Kafka (Amazon MSK).

Internal schema history store

Property Default Description

schema.history.internal

No default

Must be set to io.debezium.storage.s3.history.S3SchemaHistory.

schema.history.internal.s3.access.key.id

No default

(Optional) The identifier of the static access key that Debezium uses to authenticate with S3.

schema.history.internal.s3.secret.access.key

No default

(Optional) The Amazon Web Services (AWS) secret key that Debezium uses to authenticate to S3.

schema.history.internal.s3.region.name

No default

(Optional) Specifies the name of the region that hosts the S3 bucket.

schema.history.internal.s3.bucket.name

No default

Specifies the name of the S3 bucket that stores the schema history.

schema.history.internal.s3.object.name

No default

Specifies the object name in the bucket that stores the schema history.

schema.history.internal.s3.endpoint

No default

(Optional) Specifies a custom URL that Debezium uses to access the S3 service.
Provide the URL in the following format: http://<server>:<port>;

Azure Blob Storage

Debezium can use the Azure Blob storage service to store data. Typically, you would use Azure Blob storage when you deploy Debezium in the Apache Kafka in Azure HDInsight service.

Internal schema history store

RocketMQ

Debezium can use the RocketMqSchemaHistory class to store and retrieve database schema changes in Apache RocketMQ.

Internal schema history store

Chronicle Queue

Debezium connectors buffer change events in an internal queue before delivering them to the runtime framework. By default, Debezium uses an in-memory bounded queue. You can replace the default queue with a Chronicle Queue backed implementation by setting the queue.provider.type connector property. Chronicle Queue writes events to memory-mapped files on disk, reducing heap pressure during high-throughput periods.

Two Chronicle Queue implementations are available:

chronicle

Spills all change events to disk. This removes events from the JVM heap entirely, which is useful when you expect sustained high throughput and want to minimize garbage collection overhead.

hybrid_chronicle

Uses an in-memory queue as the primary buffer and spills overflow to disk when the buffer reaches its capacity. In low-traffic scenarios, no serialization or disk I/O occurs.

JVM setup

When using the debezium/server or debezium/connect container images, all the required JVM setups will be configured for you by setting the environment variable ENABLE_CHRONICLE_QUEUE to true.

When using other deployment alternatives, Chronicle Queue’s access to several internal JVM APIs must be explicitly provided by your configuration. This requires specifying the following JVM startup arguments, which grants Chronicle Queue reflective access to core Java internals required for its off-heap memory management and low-latency operations.

--add-opens=java.base/sun.nio.ch=ALL-UNNAMED
--add-opens=java.base/java.lang=ALL-UNNAMED
--add-opens=java.base/java.lang.reflect=ALL-UNNAMED
--add-opens=java.base/java.io=ALL-UNNAMED
--add-opens=java.base/java.util=ALL-UNNAMED

Chronicle Queue provider

To use the chronicle provider, set queue.provider.type to chronicle.

Hybrid Chronicle Queue provider

To use the hybrid_chronicle provider, set queue.provider.type to hybrid_chronicle.

The in-memory buffer holds the most recent events up to the capacity defined by max.queue.size. When the buffer reaches capacity, the oldest event is evicted to disk before a new event is added. When polling, Debezium drains evicted events from Chronicle Queue first, then from the in-memory buffer, preserving strict FIFO ordering.