​​​​​​​Note – Apache Kafka is supported as a Target since version 6, by Repstance Advanced Edition only (see Supported Database Requirements).

Repstance publishes the changes of a Source Database to Apache Kafka. The Capture Process reads the changes of the Source Database as it does for any other Target. The Apply Process, in place of changing a Target Database, sends every changed row as a message to a Kafka topic, a topic for every table. Applications and tools read the topics: they update a search index or a cache, feed a stream processor, load a data lake or a data warehouse, or react to the change of a row. The messages are JSON with the fields before, after, source, op and ts_ms (see Kafka Message Format).

This example publishes the schema dbo of a SQL Server database to a Kafka cluster. The steps are the ones of Basic Replication Configuration. Any database that Repstance supports as a Source (PostgreSQL and MySQL since version 6) can be used in place of SQL Server: only the commands for the Source Database change.

What you need

• The same as in Basic Replication Configuration: Repstance Advanced Edition is running and repcli can reach it. The Source Database is testDB on db-src.example.com and the login repuser has the privileges that are described in Database Privileges.
• A Kafka cluster whose brokers the Repstance server can reach, here kafka1.example.com:9092. The Apply Process needs the permission to write to the topics and, when it creates them, to create topics. A cluster with TLS or with a login is described in Secured Clusters below.

Check the brokers

Kafka has no database in which Repstance keeps its objects, so the Target needs no preparation: prepare database and remove database do nothing. validate database asks the brokers for their metadata, so it tells whether a broker answers with the settings and the login that are given (see Kafka as Target):

Prepare the Source Database and create the Capture Process

The Source Database and the Capture Process are made as for any other Target (see Basic Replication Configuration). loadinclude=dbo.% adds an Initial Load: the rows that the tables have now are published first, as snapshot messages, and the changes follow them (see Initial Load):

Create the Apply Process

The Apply Process reads the trail files of the Capture Process and publishes the changes. dbtype=kafka is the type of the Target, and server lists the bootstrap brokers: the Apply Process finds the other brokers of the cluster by itself. kafka.topicprefix=prod is the first part of the names of the topics, so the table dbo.emp goes to the topic prod.dbo.emp. kafka.createtopics=1 lets the process create the topics that do not exist, with kafka.partitions=3 partitions and kafka.replicationfactor=3 replicas (a cluster refuses more replicas than it has brokers). Without it the topics must exist before the process runs (see Kafka Apply Parameters):

Check the process (see Validate Process). A new process starts with the first transaction that is available in the trail:

When the Capture Process has already written the definition of the tables to the trail, the command also looks at the topics. With kafka.createtopics=0 it lists the topics that do not exist. It warns about a table without a primary key or unique index, whose messages have no key, and about tables that share a topic. It fails when tables with different columns share a topic, when a table would go to the topic of the DDL events, when two topic names differ only in . and _ (Kafka refuses to create both), and when a table without a key goes to a compacted topic.

Start the processes

Start the Capture Process first and then the Apply Process (see Run any Capture or Apply Processes):

To start a process when it is created, and every time the Repstance server starts, add autostart=1 to the prepare command of the process.

Check the replication

The command status shows the Apply Process as it shows the Apply Process of any Target (see Status of any Capture or Apply Processes). Its first line has the brokers in place of the server of a database. Kafka has no database, so DB is empty:

Read the topics

Any Kafka consumer reads the topics. For a first look use the console consumer of the Kafka distribution. It prints the key of every message, a tab and the value:

This is the message of an insert of the row 4 of the table dbo.emp (the fields are described in Kafka Message Format):

Topics, Keys and Order

• Topics – A topic holds the messages of one table: {prefix}.{schema}.{table}, with the names after the mapping rules. A rule can send several tables to one topic. The changes of the structure of the tables are events in one more topic, prod.ddl (see Changes of the Tables below).
• Keys – The key of a message is the primary key, or the unique index, of the row: {"id":4}. All the messages of a row have the same key.
• Order – The messages with the same key go to the same partition, so a consumer gets the changes of a row in the order in which they were committed in the Source Database. A consumer reads the partitions and the topics independently of each other, so the order between rows in different partitions, and between tables and topics, is not guaranteed. To put such messages in order use source.lsn, the LSN of the transaction, and source.seq, the number of the change in its transaction. An LSN is 0x and 20 hexadecimal digits, so the LSNs of one Source Database compare as text.
• Tables without a key – The messages of a table without a primary key or unique index have no key. When the process creates the topic of such a table, the topic gets one partition, so that the order of the messages is kept. A mapping rule with the attribute PK can name the columns that are the key (see Objects Mapping and Possible Transforms).

Delivery and Big Transactions

• At least once – The Apply Process publishes the messages of the committed transactions in the order of the trail, waits until the brokers have acknowledged them (all the in-sync replicas) and only then saves its position. After a failure or a restart it can publish the last messages again. A repeated message is identical to the first one, with the same source.lsn and source.seq, so a consumer drops it, or applies it again by the key with the same result.
• Rolled back transactions – A transaction that was rolled back in the Source Database is not published.
• Big transactions – A transaction bigger than maxtxbytes (120 MiB by default) is read and published in parts. All its messages are in one Kafka transaction, which is committed with the last part. A consumer with isolation.level=read_committed sees all the messages of the transaction at once, or none of them, also when the Apply Process fails or is killed in the middle of it, and after the restart the transaction is published once. A consumer with the default read_uncommitted sees the messages as they are published, and also the messages of an attempt that was aborted. A Kafka transaction may stay open for kafka.transactiontimeout seconds (600 by default); the brokers limit it with transaction.max.timeout.ms.

Initial Load

The rows that the Capture Process loads (loadinclude, see Initial Load) are published to the topic of their table as snapshot messages, with "op":"r" and "snapshot":"true". source.seq is the number of the row in the load. The changes that follow the load are published after its rows. The Apply Process saves the progress of a load every ldtxsize rows (500000 by default). A load that was interrupted goes on after the last save, so the rows after it can be published twice.

The load option of a table (the part after the colon in loadinclude) cannot clean a topic, because a topic has no delete and no truncate. It is published as an event of DDL before the first row, so that a consumer knows that the table starts again: D (delete) and T (truncate) give a truncate event, C (create) a create event and R (recreate) a drop event and then a create event. A (preserve) and M (merge) give no event.

To publish the rows of a table again, for example for a new consumer, put the table into loadinclude of the Capture Process again. The Capture Process loads it when it runs (see Alter Process).

Changes of the Tables

A change of the structure of a table in the Source Database is published as an event to the topic of the DDL events, prod.ddl (or the topic of kafka.ddltopic). It is a new table, a new, dropped, renamed or modified column, a new or dropped primary key or unique index, a rename, a drop or a truncate of the table, for the tables that ddlinclude of the Capture Process selects. A new index that does not change the key columns is no event. The event has the columns of the table before and after the change and the statements of the Source Database. The Apply Process does not run them anywhere: Kafka has nothing to alter.

The topic of the DDL events has one partition, so its events are in the order of the Source Database. The messages that were published before a change do not have a new column, and a consumer that needs to know the columns of its messages reads this topic. The events and the messages are in different topics: compare their LSNs to see which one came first.

Choose What Is Published

The rules of the Apply Process decide which tables and columns are published and how the topics are named (see Objects Mapping and Possible Transforms and Objects Filtering). Leave out the tables that start with audit_, leave out the column salary of the table emp, make the column id of the table log the key, send the table orders to the schema sales (its topic is then prod.sales.orders) and publish only the inserts and the updates of emp:

The name of the topic is made of the names after the rules. The fields source.schema and source.table of a message keep the names of the Source Database. A rule that adds a column, and a column that is filled by a SQL function, are not part of the messages. A filter skips whole operations of a table and has no condition. The skipped operation is no message, and a delete or a change of the key has no tombstone either. The changes are used after the process has been stopped and run again (see Alter Process).

Compacted Topics

A compacted topic keeps the last message of every key, so it holds the current state of a table. Set kafka.topicconfig for the topics that the process creates; it does not change a topic that exists, and it is never used for the topic of the DDL events:

A delete is followed by a tombstone, the key and no value, which lets Kafka forget the row (kafka.tombstones=1, the default). Every table of a compacted topic needs a key, because Kafka refuses a message without a key there. validate process reports a table without a key that goes to a compacted topic as an error, and a mapping rule with the attribute PK can name the key.

Net Changes

With batchmode=2 the changes of a row within a batch of transactions are published as one message: a row that was changed three times is one update, an insert and a delete of the same row give nothing. It is for a consumer that needs the state of the rows and not their history. The order of the transactions is not kept, and the messages of a batch that is published again can differ from the first ones, so the consumer applies them by the key:

Secured Clusters

Give the Apply Process the security settings of the cluster. This command connects with TLS, checks the certificates of the brokers against the file of the certificate authority and logs in with SCRAM:

The logins plain, scram-sha-256 and scram-sha-512 are supported, and a client certificate (kafka.certfile and kafka.keyfile) for a cluster that asks for one. The files must be readable by the Repstance server. They are read when the process starts. validate database and validate process use the same settings and tell if the cluster refuses them (see Kafka as Target and Kafka Apply Parameters).

Web UI

In the Web UI click the Kafka logo to open the settings of the Target of an Apply Process (see Setup First Replication). The form “Database Settings” has the brokers, the port, TLS and the login, and it has no “Prepare for Replication” button, because there is nothing to prepare. The tab “Process Parameters” has “Topic Prefix”, “Create Topics”, “Partitions”, “Replication Factor” and “Tombstones”, and the other Kafka parameters are added as Advanced parameters. The tab “Data Filter” takes the operations of a table, the field for a SQL condition is disabled with the text “Not available for Kafka”, and the tab “Transformation Rules” takes the rules for tables and columns. “Validate” is validate process.

Limits

• Kafka is a Target only. It cannot be compared with the Source Database, because there is no database to read (see Requirements and Limits of Data Comparison).
• The messages are JSON. Avro, Protobuf and a schema registry are not supported.
• The logins are PLAIN and SCRAM. Kerberos, OAUTHBEARER and the IAM login of Amazon MSK are not supported.
• A key and a value together are at most kafka.maxmessagebytes bytes. The longest strings of a bigger row are cut, and the message lists the columns that were cut (see Kafka Message Format). The statements of a DDL event are left out when they are longer than 64 KB.
• The Kafka transactions of an Apply Process use the transactional id repstance-apply-{id}. When two Repstance servers publish to one cluster, their Apply Processes need different ids.

Stop and Remove

stop stops the Apply Process, and it goes on from its saved position at the next run. remove removes the process with its saved position and the progress of its load. The topics and their messages stay in Kafka, because they belong to the consumers:

Apache, Apache Kafka, Kafka and the Kafka logo are either registered trademarks or trademarks of the Apache Software Foundation in the United States and/or other countries. No endorsement by The Apache Software Foundation is implied by the use of these marks.