Plugin kafka_cdc_plugin

← Firebird streaming / CDC

The kafka_cdc_plugin plugin is designed to register data change events in Firebird database tables and save them in Kafka. This process is also known as Change Data Capture, henceforth CDC.

Each event is a document in JSON format, which is as close as possible to the debezium event format. In the next version it is planned to add the ability to serialize events in AVRO format.

Requirements

The plugin requires the librdkafka and liblz4 libraries. On Windows, they are included in the distribution. On Linux, you must install them manually.

Installation on Debian:

sudo apt install librdkafka1

sudo apt install liblz4-1

fb_streaming + kafka_cdc_plugin files and folders on Windows

• fb_streaming.exe - fb_streaming service;
• fbclient.dll - Firebird client library (version 4.0 and higher);
• icudt63.dll, icuin63.dll, icuuc63.dll, icudt63l.dat - ICU library files. On some Windows versions, the system ICU library is already installed, in which case these files are optional;
• fb_streaming.conf - fb_streaming service configuration file;
• fb_streaming.log - fb_streaming service log (created manually);
• zlib1.dll, zstd.dll, lz4.dll - data and traffic compression libraries required by the kafka_cdc_plugin plugin;
• libssl-3-x64.dll, libcrypto-3-x64.dll - OpenSSL encryption libraries for Windows-x64, libssl-3.dll, libcrypto-3.dll, required by the kafka_cdc_plugin plugin;
• libssl-3.dll, libcrypto-3.dll - OpenSSL encryption libraries for Windows-x86, required by the kafka_cdc_plugin plugin;
• doc - Documentation folder;
• stream_plugins - Folder for plugin library files;
• stream_plugins/kafka_cdc_plugin.dll - Kafka_cdc_plugin plugin library.

[!NOTE] Note: SASL is only supported in the 64-bit build.

How it works

The fb_streaming service (daemon) checks the contents of the directory specified for the task in the fb_streaming.conf configuration file, and if this directory contains unprocessed replication log files, then it analyzes them and generates appropriate events that are processed by the specified plugin. The last replication segment number processed is written to a control file named <database guid>. The location of this file can be specified in the lockDir configuration parameter. If this parameter is not specified, then by default the control file <database guid> will be created in the directory specified as the archive log directory for the task.

[!IMPORTANT] Important: Replication log files must be in an archived state.

The control file is necessary to restore operation after a sudden shutdown of the service (for example, the lights were turned off or the server was rebooted). It contains the following information:

• control information (database GUID, size, etc.);
• number of the last processed segment and position in it;
• the number of the segment with the oldest active transaction (a transaction can start in one segment and end in another);
• a list of all active transactions in the form of pairs {tnxNumber, segmentNumber}, where segmentNumber is the number of the segment in which the transaction began.

If an emergency situation occurs and the fb_streaming service has been stopped, then the next time it starts, it reads the control file and rereads all replication segments, starting with the number of the segment with the oldest active transaction, and ending with the number of the last processed segment. In the process of reading these segments, fb_streaming repeats all events from the list of active transactions, after which it goes into normal operation.

A replication segment file is deleted if its number is less than the segment number with the oldest active transaction.

The kafka_cdc_plugin generates only data change events for the given tables for each of the INSERT, UPDATE and DELETE operations. Each active transaction has its own buffer in which these events accumulate, and they are sent to Kafka only when the transaction completes, after which the buffer is cleared.

Events can enter Kafka in two formats:

• JSON without schema (parameters key_converter_schemas_enable and value_converter_schemas_enable equal to false);
• JSON with a schema (parameters key_converter_schemas_enable and value_converter_schemas_enable equal to true).

Future versions also plan to add support for event serialization in AVRO format, as well as support for schema registries.

Topic names

By default, kafka_cdc_plugin writes change events for all INSERT, UPDATE and DELETE operations, in one Apache Kafka theme. kafka_cdc_plugin uses the scheme described in the kafka_topic_cdc_event parameter equal to $[topicPrefix].cdc for the names of data change event topics.

Where $[topicPrefix] is a substitution that is replaced by a prefix for topics. This prefix is set by the kafka_topic_prefix parameter, by default it is fb_streaming. Thus, by default, all data change events are write in the topic fb_streaming.cdc (after applying substitution).

There are other substitutions. The substitution $[tableName] indicates the name of the table for which the event is generated. Thus, by specifying in the configuration

kafka_topic_cdc_event = $[topicPrefix].cdc.$[tableName]

each table has its own topic of data change events.

Examples of topics with data change events:

fb_streaming.cdc.products
fb_streaming.cdc.products_on_hand
fb_streaming.cdc.customers
fb_streaming.cdc.orders

The topic name for the metadata change event is specified in the kafka_topic_transaction parameter. Only the substitution $[topicPrefix] is available in this parameter.

Transaction metadata

The kafka_cdc_plugin can generate events that represent transaction boundaries and that enrich data change event messages. By default, generation of transaction boundary events is disabled. It can be enabled by setting the transaction_metadata_enable parameter to true.

The kafka_cdc_plugin generates transaction boundary events for the BEGIN and END delimiters in every transaction. Transaction boundary events contain the following fields:

• status – BEGIN or END;
• id – Unique transaction identifier;
• ts_ms – The time of a transaction boundary event (BEGIN or END event). Since there is no such time in the replication logs, the time when the event is processed by the kafka_cdc_plugin plugin is processed;
• event_count (for END events) – Total number of events emitted by the transaction;
• data_collections (for END events) – An array of pairs of data_collection and event_countelements that indicates the number of events that the kafka_cdc_plugin emits for changes that originate from a data collection.

Example:

{
    "status": "BEGIN",
    "id": 901,
    "ts_ms": 1713099401529,
    "event_count": null,
    "data_collections": null
}

{
    "status": "END",
    "id": 901,
    "ts_ms": 1713099404605,
    "event_count": 2,
    "data_collections": [
        {
            "data_collection": "fbstreaming.cdc.CUSTOMERS",
            "event_count": 2
        }
    ]
}

Unless overridden by the kafka_topic_transaction parameter, kafka_cdc_plugin sends transaction events to the $[topicPrefix].transaction topic.

Change data event enrichment

When transaction metadata is enabled (transaction_metadata_enable = true), the Envelope of the data message is populated with a new transaction field. This field provides information about each event as a collection of fields:

• id – Unique transaction identifier;
• total_order – The absolute position of the event among all events generated by the transaction;
• data_collection_order – The per-data collection position of the event among all events that were emitted by the transaction.

Example:

{
    "before": null,
    "after": {
        "ID": 20,
        "FIRST_NAME": "Anne",
        "LAST_NAME": "Kretchmar",
        "EMAIL": "[email protected]"
    },
    "source": {
       ...
    },
    "op": "c",
    "ts_ms": 1713099401533,
    "transaction": {
        "id": 901,
        "total_order": 1,
        "data_collection_order": 1
 }
}

Data Change Events

The kafka_cdc_plugin generates a data change event for each INSERT, UPDATE and DELETE operation at the record level. Each event contains a key and a value. The key and value structure depends on the modified table.

The fb_streaming service and the kafka_cdc_plugin plugin are designed for continuous streams of event messages. However, the structure of these events may change over time, which may be difficult for consumers to process. To solve this problem, each event contains a schema for its content or, if you are using a schema registry, a schema identifier that the consumer can use to retrieve the schema from the registry. This makes each event self-contained.

The following JSON skeleton shows the four main parts of a change event. The schema field is only in the change event when you configure the kafka_cdc_plugin plugin to create it. Likewise, the event key and event payload are only found in the change event if you have configured the kafka_cdc_plugin plugin to generate them. If you are using the JSON format and have configured the kafka_cdc_plugin to generate all four main parts of change events, the change events have the following structure:

{
  "schema": { // <1>
    ...
   },
  "payload": { // <2>
    ...
  }
}
{
  "schema": { // <3>
    ...
  },
  "payload": { // <4>
    ...
  }
}

Overview of change event basic content:

Change event keys

For a given table, the change event key has a structure that contains a field for each column of the table’s primary key at the time the event was created.

Consider the CUSTOMERS table, and an example of a change event key for that table.

CREATE TABLE CUSTOMERS (
  ID BIGINT GENERATED BY DEFAULT AS IDENTITY,
  FIRST_NAME VARCHAR(255) NOT NULL,
  LAST_NAME VARCHAR(255) NOT NULL,
  EMAIL VARCHAR(255) NOT NULL,
  CONSTRAINT PK_CUSTOMERS PRIMARY KEY(ID)
);

{
    "schema": { // <1>
        "type": "struct",
        "name": "fbstreaming.CUSTOMERS.Key", // <2>
        "optional": "false", // <3>
        "fields": [ // <4>
            {
                "type": "int64",
                "optional": false,
                "field": "ID"
            }
        ]
    },
    "payload": { // <5>
        "ID": 1
    }
}

Description of change event key:

Change event values

The value in a change event is a bit more complicated than the key. Like the key, the value has a schema section and a payload section. The schema section contains the schema that describes the Envelope structure of the payload section, including its nested fields. Change events for operations that create, update or delete data all have a value payload with an envelope structure.

Consider the same sample table that was used to show an example of a change event key:

CREATE TABLE CUSTOMERS (
  ID BIGINT GENERATED BY DEFAULT AS IDENTITY,
  FIRST_NAME VARCHAR(255) NOT NULL,
  LAST_NAME VARCHAR(255) NOT NULL,
  EMAIL VARCHAR(255) NOT NULL,
  CONSTRAINT PK_CUSTOMERS PRIMARY KEY(ID)
);

The value portion of a change event for a change to this table is described for:

• create events
• update events
• delete events

To demonstrate these events, let’s run the following set of SQL queries:

insert into CUSTOMERS (FIRST_NAME, LAST_NAME, EMAIL)
values ('Anne', 'Kretchmar', '[email protected]');

commit;

update CUSTOMERS
set FIRST_NAME = 'Anne Marie';

commit;

delete from CUSTOMERS;

commit;

Create events

The following example shows part of the change event value that fb_streaming generates for an operation that creates data in the CUSTOMERS table:

{
    "schema": { // <1>
        "type": "struct",
        "fields": [
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "int64",
                        "optional": false,
                        "field": "ID"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "FIRST_NAME"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "LAST_NAME"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "EMAIL"
                    }
                ],
                "optional": true,
                "name": "fbstreaming.CUSTOMERS.Value", // <2>
                "field": "before"
            },
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "int64",
                        "optional": false,
                        "field": "ID"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "FIRST_NAME"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "LAST_NAME"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "EMAIL"
                    }
                ],
                "optional": true,
                "name": "fbstreaming.CUSTOMERS.Value",
                "field": "after"
            },
            {
                "type": "struct",
                "fields": [
                    {
                        "type": "string",
                        "optional": false,
                        "field": "dbguid"
                    },
                    {
                        "type": "int64",
                        "optional": false,
                        "field": "sequence"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "filename"
                    },
                    {
                        "type": "string",
                        "optional": false,
                        "field": "table"
                    },
                    {
                        "type": "int64",
                        "optional": false,
                        "field": "tnx"
                    },
                    {
                        "type": "int64",
                        "optional": true,
                        "field": "ts_ms"
                    }
                ],
                "optional": false,
                "name": "fbstreaming.Source", // <3>
                "field": "source"
            },
            {
                "type": "int64",
                "optional": false,
                "field": "op"
            }
        ],
        "optional": false,
        "name": "fbstreaming.CUSTOMERS.Envelope" // <4>
    },
    "payload": { // <5>
       "before": null, // <6>
       "after": { // <7>
           "ID": 1,
           "FIRST_NAME": "Anne",
           "LAST_NAME": "Kretchmar",
           "EMAIL": "[email protected]"
        },
        "source": { // <8>
           "dbguid": "{9D66A972-A8B9-42E0-8542-82D1DA5F1692}",
           "sequence": 1,
           "filename": "TEST.FDB.journal-000000001",
           "table": "CUSTOMERS",
           "tnx": 200,
           "ts_ms": 1711288254908
        },
        "op": "c", // <9>
        "ts_ms": 1711288255056 // <10>
    }
}

Descriptions of create event value fields:

Update events

The change event value for the update operation in the example table CUSTOMERS has the same schema as the create event for that table. Likewise, the event value payload has the same structure. However, the event value payload contains different values in the update event. Here is an example of the change event value in the event that fb_streaming generates for an update on the CUSTOMERS table:

{
    "schema": { ... },
    "payload": {
        "before": { // <1>
            "ID": 1,
            "FIRST_NAME": "Anne",
            "LAST_NAME": "Kretchmar",
            "EMAIL": "[email protected]"
        },
        "after": { // <2>
            "ID": 1,
            "FIRST_NAME": "Anne Marie",
            "LAST_NAME": "Kretchmar",
            "EMAIL": "[email protected]"
        },
        "source": { // <3>
            "dbguid": "{9D66A972-A8B9-42E0-8542-82D1DA5F1692}",
            "sequence": 2,
            "filename": "TEST.FDB.journal-000000002",
            "table": "CUSTOMERS",
            "tnx": 219,
            "ts_ms": 1711288254908
        },
        "op": "u", // <4>
        "ts_ms": 1711288256121 // <5>
    }
}

Descriptions of update event value fields:

Delete events

The value in a delete change event has the same schema portion as create and update events for the same table. The payload portion in a delete event for the sample CUSTOMERS table looks like this:

{
    "schema": { ... },
    "payload": { // <1>
        "before": {
            "ID": 1,
            "FIRST_NAME": "Anne Marie",
            "LAST_NAME": "Kretchmar",
            "EMAIL": "[email protected]"
        },
        "after": null, // <2>
        "source": { // <3>
            "dbguid": "{9D66A972-A8B9-42E0-8542-82D1DA5F1692}",
            "sequence": 3,
            "filename": "TEST.FDB.journal-000000003",
            "table": "CUSTOMERS",
            "tnx": 258,
            "ts_ms": 1711288254908
        },
        "op": "d", // <4>
        "ts_ms": 1711288256152 // <5>
    }
}

Descriptions of delete event value fields:

Data type mappings

Run Change Data Capture

We will describe in detail the steps required to launch Change Data Capture on your database:

1. Setting up Kafka
2. Setting up Firebird and preparing the database
3. Configuring the fb_streaming service and the kafka_cdc_plugin plugin
4. Launch Kafka
5. Installation and start of the fb_streaming service
6. Start publishing in the database

Setting up Kafka

To test the kafka_cdc_plugin plugin, a configured Kafka installation in docker is used. To do this, use docker-compose.yml with the following content:

version: "2"

services:

  zookeeper:
    image: confluentinc/cp-zookeeper:7.2.1
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-server:7.2.1
    hostname: kafka
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9997:9997"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9997
      KAFKA_JMX_HOSTNAME: kafka

  kafka-ui:
    container_name: kafka-ui
    image: provectuslabs/kafka-ui:latest
    ports:
      - 8080:8080
    environment:
      DYNAMIC_CONFIG_ENABLED: 'true'
    volumes:
      - "d:\\docker\\kafka\\config.yml:/etc/kafkaui/dynamic_config.yaml"

The included file config.yml contains:

auth:
  type: DISABLED
kafka:
  clusters:
  - bootstrapServers: kafka:29092
    name: Kafka CDC Cluster
    properties: {}
    readOnly: false
rbac:
  roles: []
webclient: {}

Setting up Firebird and preparing the database

Now you need to configure asynchronous replication for your database; to do this, you need to add the following lines to the replication.conf file:

database = c:\fbdata\5.0\test.fdb
{
   journal_directory = d:\fbdata\5.0\replication\test\journal
   journal_archive_directory = d:\fbdata\5.0\replication\test\archive
   journal_archive_command = "copy $(pathname) $(archivepathname) && copy $(pathname) d:\fbdata\5.0\replication\test\kafka_source"
   journal_archive_timeout = 10
}

Please note: this involves duplicating log archive files so that logical replication and the task of sending events to Kafka can work simultaneously. This is necessary because log archive files are deleted after processing and cannot be used by another task.

If replication logs are not used for replication itself, but are only needed for Change Data Capture, then the configuration can be simplified:

database = c:\fbdata\5.0\test.fdb
{
   journal_directory = d:\fbdata\5.0\replication\test\journal
   journal_archive_directory = d:\fbdata\5.0\replication\test\kafka_source
   journal_archive_timeout = 10
}

Now you need to include the necessary tables in the publication. For the example above, just add the CUSTOMERS table to the publication. This is done with the following statement:

ALTER DATABASE INCLUDE CUSTOMERS TO PUBLICATION;

or you can include all database tables in the publication at once:

ALTER DATABASE INCLUDE ALL TO PUBLICATION;

Configuring the fb_streaming service and the kafka_cdc_plugin plugin

Next, we’ll configure the fb_streaming.conf configuration so that fb_streaming automatically sends data change events to Kafka.

task = d:\fbdata\5.0\replication\test\kafka_source
{
    database = inet://localhost:3055/test
    username = SYSDBA
    password = masterkey
    deleteProcessedFile = true
    plugin = kafka_cdc_plugin
    dumpBlobs = true
    kafka_brokers = localhost:9092
    kafka_topic_prefix = fb_streaming
    kafka_topic_cdc_event = $[topicPrefix].cdc
    kafka_topic_transaction = $[topicPrefix].transaction
    key_cdc_events_enable = true
    key_converter_schemas_enable = true
    value_converter_schemas_enable = true
    transaction_metadata_enable = true
}

On Linux this configuration would look like this:

task = /mnt/d/fbdata/5.0/replication/test/kafka_source
{
    database = inet://localhost:3055/test
    username = SYSDBA
    password = masterkey
    deleteProcessedFile = true
    plugin = kafka_cdc_plugin
    dumpBlobs = true
    kafka_brokers = localhost:9092
    kafka_topic_prefix = fb_streaming
    kafka_topic_cdc_event = $[topicPrefix].cdc
    kafka_topic_transaction = $[topicPrefix].transaction
    key_cdc_events_enable = true
    key_converter_schemas_enable = true
    value_converter_schemas_enable = true
    transaction_metadata_enable = true
}

The task parameter describes the task to be performed by the fb_streaming service. It specifies the folder where the replication segment files are located for the plugin to process. There may be several such tasks. This parameter is complex and itself describes the configuration of a specific task. Let’s describe the parameters available for the task performed by the kafka_cdc_plugin plugin:

• controlFileDir - the directory in which the control file will be created (by default, the same directory as sourceDir);
• database - database connection string (required);
• username - user name for connecting to the database;
• password - password for connecting to the database;
• plugin - plugin that processes events that occur during the analysis of the replication log (required);
• deleteProcessedFile - whether to delete the log file after processing (true by default). It is useful to set this parameter to false for debugging, when the same logs need to be processed multiple times without deleting them;
• warn_tnx_not_found - generate a warning instead of an error if the transaction is not found in the replication segments. If this parameter is set to true, then an appropriate warning will be written to the fb_streaming log, the contents of the lost transaction will be ignored, and the plugin will continue to work. Default is false;
• errorTimeout - timeout after an error, in seconds. After this timeout expires, the segments will be re-scanned and the task will be restarted. Default is 60 seconds;
• include_tables - a regular expression that defines the names of the tables for which events need to be tracked;
• exclude_tables - a regular expression that defines the names of tables for which events do not need to be tracked;
• dumpBlobs - whether to publish new BLOB field values (default false);
• charset_for_none_fields - character set used for converting data from string fields with the character set type NONE. Typically, data in string fields with the NONE character set is stored “as is,” meaning in the original encoding used by the application when saving it. Firebird does not re-encode such strings. The JSON format requires all strings to be in UTF-8 encoding. To correctly convert strings from NONE fields to UTF8, the system needs to know the original encoding of the data. This is precisely what the charset_for_none_fields parameter specifies;
• kafka_brokers - addresses of Kafka brokers. You can specify multiple addresses. Addresses are separated by commas;
• kafka_topic_prefix – Kafka topic prefix. It specifies a macro substitution $[topicPrefix] that can be used for topic names;
• kafka_topic_cdc_event – the name of the Kafka topic(s) in which data change events are stored. Macro substitutions $[topicPrefix] and $[tableName] can be used;
• kafka_topic_transaction is the name of the Kafka topic in which transaction metadata is stored. Macro substitution $[topicPrefix] can be used;
• async_producer - if you set this parameter to true, then messages in Kafka will be sent asynchronously, that is, fb_streaming will not wait for confirmation that the message was successfully sent, but will immediately proceed to processing the next message. Messages are sent asynchronously until the internal buffer of unsent messages is full. By default, the size of this buffer is 500;
• key_cdc_events_enable - if this option is set to true, each data change event contains key information, otherwise the event key will be null. This can be useful because Kafka uses the event key for partitioning.
• key_converter_schemas_enable - whether to include a schema in the data update event key;
• value_converter_schemas_enable - whether to include the schema in the data update event value;
• transaction_metadata_enable – whether to send transaction start and end events (transaction metadata).

The following describes the parameters of the LibRdKafka library used by this plugin. Each parameter is prefixed with kafka_. Dots inside the parameter are replaced with underscores _. See: LibRdKafka CONFIGURATION.

• kafka_security_protocol - Protocol used to communicate with brokers. Valid values are: plaintext, ssl, sasl_plaintext, sasl_ssl. In the LibRdKafka library, this parameter has the name security.protocol. Default value: plaintext.
• kafka_ssl_cipher_suites - A list of cipher suites. This is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol. By default all the available cipher suites are supported. In the LibRdKafka library, this parameter has the name ssl.cipher.suites.
• kafka_ssl_curves_list - The supported-curves extension in the TLS ClientHello message specifies the curves (standard/named, or explicit GF(2^k) or GF(p)) the client is willing to have the server use. See manual page for SSL_CTX_set1_curves_list(3). OpenSSL >= 1.0.2 required. In the LibRdKafka library, this parameter has the name ssl.curves.list.
• kafka_ssl_sigalgs_list - The client uses the TLS ClientHello signature_algorithms extension to indicate to the server which signature/hash algorithm pairs may be used in digital signatures. See manual page for SSL_CTX_set1_sigalgs_list(3). OpenSSL >= 1.0.2 required. In the LibRdKafka library, this parameter has the name ssl.sigalgs.list.
• kafka_ssl_key_location - Path to client’s private key (PEM) used for authentication. In the LibRdKafka library, this parameter has the name ssl.key.location.
• kafka_ssl_key_password - Private key passphrase (for use with ssl.key.location and set_ssl_cert()). In the LibRdKafka library, this parameter has the name ssl.key.password.
• kafka_ssl_certificate_location - Path to client’s public key (PEM) used for authentication. In the LibRdKafka library, this parameter has the name ssl.certificate.location.
• kafka_ssl_ca_location - File or directory path to CA certificate(s) for verifying the broker’s key. Defaults: On Windows the system’s CA certificates are automatically looked up in the Windows Root certificate store. On Mac OSX this configuration defaults to probe. It is recommended to install openssl using Homebrew, to provide CA certificates. On Linux install the distribution’s ca-certificates package. If OpenSSL is statically linked or ssl.ca.location is set to probe a list of standard paths will be probed and the first one found will be used as the default CA certificate location path. If OpenSSL is dynamically linked the OpenSSL library’s default path will be used (see OPENSSLDIR in openssl version -a). In the LibRdKafka library, this parameter has the name ssl.ca.location.
• kafka_ssl_ca_certificate_stores - Comma-separated list of Windows Certificate stores to load CA certificates from. Certificates will be loaded in the same order as stores are specified. If no certificates can be loaded from any of the specified stores an error is logged and the OpenSSL library’s default CA location is used instead. Store names are typically one or more of: MY, Root, Trust, CA. In the LibRdKafka library, this parameter has the name ssl.ca.certificate.stores. Default value: Root.
• kafka_ssl_crl_location - Path to CRL for verifying broker’s certificate validity. In the LibRdKafka library, this parameter has the name ssl.crl.location.
• kafka_ssl_keystore_location - Path to client’s keystore (PKCS#12) used for authentication. In the LibRdKafka library, this parameter has the name ssl.keystore.location.
• kafka_ssl_keystore_password - Client’s keystore (PKCS#12) password. In the LibRdKafka library, this parameter has the name ssl.keystore.password.
• kafka_ssl_providers - Comma-separated list of OpenSSL 3.0.x implementation providers. E.g., default,legacy. In the LibRdKafka library, this parameter has the name ssl.providers.
• kafka_ssl_engine_id - OpenSSL engine id is the name used for loading engine. In the LibRdKafka library, this parameter has the name ssl.engine.id. Default value: dynamic.
• kafka_enable_ssl_certificate_verification - Enable OpenSSL’s builtin broker (server) certificate verification. Valid values are: true, false. In the LibRdKafka library, this parameter has the name enable.ssl.certificate.verification. Default value: true.
• kafka_ssl_endpoint_identification_algorithm - Endpoint identification algorithm to validate broker hostname using broker certificate. Valid values are: none, https. https - Server (broker) hostname verification as specified in RFC2818. none - No endpoint verification. OpenSSL >= 1.0.2 required. In the LibRdKafka library, this parameter has the name ssl.endpoint.identification.algorithm. Default value: https.
• kafka_sasl_mechanism - SASL mechanism to use for authentication. Supported: GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER. NOTE: Despite the name only one mechanism must be configured. In the LibRdKafka library, this parameter has the name sasl.mechanism. Default value: GSSAPI.
• kafka_sasl_username - SASL username for use with the PLAIN and SASL-SCRAM-.. mechanisms. In the LibRdKafka library, this parameter has the name sasl.username.
• kafka_sasl_password - SASL password for use with the PLAIN and SASL-SCRAM-.. mechanisms. In the LibRdKafka library, this parameter has the name sasl.password.
• kafka_sasl_kerberos_service_name - Kerberos principal name that Kafka runs as, not including /hostname@REALM. In the LibRdKafka library, this parameter has the name sasl.kerberos.service.name. Default value: kafka.
• kafka_sasl_kerberos_principal - This client’s Kerberos principal name. (Not supported on Windows, will use the logon user’s principal). In the LibRdKafka library, this parameter has the name sasl.kerberos.principal. Default value: kafkaclient.
• kafka_sasl_oauthbearer_config - SASL/OAUTHBEARER configuration. The format is implementation-dependent and must be parsed accordingly. The default unsecured token implementation (see https://tools.ietf.org/html/rfc7515#appendix-A.5) recognizes space-separated name=value pairs with valid names including principalClaimName, principal, scopeClaimName, scope, and lifeSeconds. The default value for principalClaimName is sub, the default value for scopeClaimName is scope, and the default value for lifeSeconds is 3600. The scope value is CSV format with the default value being no/empty scope. For example: principalClaimName=azp principal=admin scopeClaimName=roles scope=role1,role2 lifeSeconds=600. In addition, SASL extensions can be communicated to the broker via extension_NAME=value. For example: principal=admin extension_traceId=123. In the LibRdKafka library, this parameter has the name sasl.oauthbearer.config.
• kafka_sasl_oauthbearer_method - Set to default or oidc to control which login method to be used. If set to oidc, the following properties must also be be specified: kafka_sasl_oauthbearer_client_id, kafka_sasl_oauthbearer_client_secret and kafka_sasl_oauthbearer_token_endpoint_url (sasl.oauthbearer.client.id, sasl.oauthbearer.client.secret, and sasl.oauthbearer.token.endpoint.url). In the LibRdKafka library, this parameter has the name sasl.oauthbearer.method. Default value: default.
• kafka_sasl_oauthbearer_client_id - Public identifier for the application. Must be unique across all clients that the authorization server handles. Only used when kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) is set to oidc. In the LibRdKafka library, this parameter has the name sasl.oauthbearer.client.id.
• kafka_sasl_oauthbearer_client_secret - Client secret only known to the application and the authorization server. This should be a sufficiently random string that is not guessable. Only used when kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) is set to oidc. In the LibRdKafka library, this parameter has the name sasl.oauthbearer.client.secret.
• kafka_sasl_oauthbearer_scope - Client use this to specify the scope of the access request to the broker. Only used when kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) is set to oidc. In the LibRdKafka library, this parameter has the name sasl.oauthbearer.scope.
• kafka_sasl_oauthbearer_extensions - Allow additional information to be provided to the broker. Comma-separated list of key=value pairs. E.g., supportFeatureX=true,organizationId=sales-emea. Only used when kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) is set to oidc. In the LibRdKafka library, this parameter has the name sasl.oauthbearer.extensions.
• kafka_sasl_oauthbearer_token_endpoint_url - OAuth/OIDC issuer token endpoint HTTP(S) URI used to retrieve token. Only used when kafka_sasl_oauthbearer_method (sasl.oauthbearer.method) is set to oidc. In the LibRdKafka library, this parameter has the name sasl.oauthbearer.token.endpoint.url.

Launch Kafka

Now you can run docker with a Kafka container:

docker-compose up -d

To stop docker run:

docker-compose down

Installing and starting the fb_streaming service

The next step is to install and start the fb_streaming service.

In Windows, this is done with the following commands (Administrator privileges are required):

fb_streaming install
fb_streaming start

On Linux:

sudo systemctl enable fb_streaming

sudo systemctl start fb_streaming

[!NOTE] To test how fb_streaming works without installing the service, simply type the fb_streaming command without any arguments. fb_streaming will be launched as an application and terminated after pressing Enter.

Start publication in the database

Once you have everything set up and running, you need to enable publishing to your database. This is done with the following SQL query:

ALTER DATABASE ENABLE PUBLICATION;

From this moment on, the fb_streaming service will monitor changes in the specified tables and publish them in the fb_streaming.cdc topic, on the servers specified in kafka_brokers.