Search docs.molo17.com
    Configuration guide Connectors

    Set up an Apache Kafka target

    The Kafka agent publishes every change of a source table as one JSON message on a Kafka topic. It does not read from Kafka, and it does not create topics: the topic names, the partitions and the access rules are yours to define. This guide creates a Kafka user that can only write to your topics, creates the target with SASL authentication, copies a PostgreSQL table and follows its live changes, and reads the messages back with a console consumer. It then shows how messages are keyed and partitioned, which producer settings you control, how to switch on Kafka transactions and TLS, and how to fix the errors you are most likely to meet.

    In this guide

    You will create a SCRAM user, give it write access to a topic prefix with ACLs, and create the topics with several partitions. You will add the Kafka target in the Control Plane, choose how a table maps to a topic and to a message key, run a snapshot and live changes, and read inserts, updates, deletes and tombstones with kafka-console-consumer. You will check that the changes of one key stay in order, tune the write settings, add transactions and TLS, and recover from authentication, authorization and missing-topic errors.

    Quick information

    Goal

    Publish the snapshot and the change events of a PostgreSQL table to Kafka topics, and read them back

    Audience

    Data engineers and Kafka administrators who feed stream consumers

    Level

    Intermediate

    Time required

    About 60 minutes

    Area

    Kafka → users, ACLs and topics; Control Plane → Pipelines → Kafka agent → Base settings and Advanced settings; entity → Field mapping and Settings

    Applies to

    Gluesync 2.3

    Before you begin

    Make sure that:

    1. You have an Apache Kafka cluster and you can run the Kafka command line tools (kafka-topics.sh, kafka-configs.sh, kafka-acls.sh, kafka-console-consumer.sh) as an administrator. This guide was tested with a single Apache Kafka 3.9 broker in KRaft mode, with an authorizer, SCRAM-SHA-256 and TLS listeners, and automatic topic creation switched off;

    2. Core Hub can open a connection to the broker address that the broker advertises. The agent connects to the host and port you enter, and the producer then follows the addresses in the broker metadata;

    3. You can sign in to the Control Plane with a role that can create pipelines;

    4. You have a source table with a primary key. This guide uses the PostgreSQL table public.orders (id integer primary key, customer varchar(50), status varchar(20), amount numeric(10,2), created_at timestamp) with 20 rows;

    5. The source database is ready for change data capture. For PostgreSQL that means wal_level set to logical, a user with the REPLICATION attribute that can read the table, and the wal2json plug-in. See the PostgreSQL CDC reference for the exact statements.

    Kafka is a target only. The agent cannot read topics, so it cannot be a source. The options of the agent are listed in the Apache Kafka target reference. The agent overview lists the supported versions.

    Guide steps

    Create a Kafka user, the topics and the ACLs

    CLI

    Gluesync needs to write to topics and to list them. It does not need to create topics, read them, or join consumer groups. This guide puts every topic under the prefix gs., so one prefixed ACL covers all of them. The commands read the broker address and the administrator credentials from a client configuration file named admin.properties. Replace <broker-host> and the passwords.

    1. Create the user. Create one SCRAM mechanism for each command, because Kafka refuses two credentials for the same user in one request:

      kafka-configs.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --alter --entity-type users --entity-name gluesync \
  --add-config 'SCRAM-SHA-256=[password=<a strong password>]'

    2. Allow the user to write to every topic whose name starts with gs., and to describe them. Test and continue lists the topics, and a user that cannot describe a topic does not see it:

      kafka-acls.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --add --allow-principal User:gluesync \
  --operation Write --operation Describe \
  --topic gs. --resource-pattern-type prefixed

    3. Create the topic of the table. The agent never creates it. Use three partitions here, so that step 8 has something to show. Use a replication factor of 3 on a production cluster:

      kafka-topics.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --create --topic gs.orders --partitions 3 --replication-factor 1

    4. Check what you created:

      kafka-topics.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --describe --topic gs.orders
kafka-acls.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --list --principal User:gluesync
    Expected result

    The topic shows PartitionCount: 3. The ACL list shows WRITE and DESCRIBE as ALLOW on the topic pattern gs. with the pattern type PREFIXED.

    You can let the broker create topics instead, with auto.create.topics.enable=true. The topic then gets the broker defaults, which is usually one partition. Production clusters often switch automatic creation off, and then every topic must exist before its entity starts: see the troubleshooting table. If you want each topic to keep only the latest message of each key, add --config cleanup.policy=compact to the --create command.
    Prepare the source table

    SQL

    Run this on the PostgreSQL source. It is optional, but it makes the before image of an update complete. With the default replica identity, PostgreSQL sends only the key of the old row:

    ALTER TABLE public.orders REPLICA IDENTITY FULL;
    Expected result

    PostgreSQL answers ALTER TABLE.

    Create the pipeline and the Kafka target

    UI

    1. In the Control Plane, select Create your first pipeline, or the + button of the pipeline list if you already have pipelines. The Agents setup wizard opens;

    2. In 1. Setup source, choose the PostgreSQL CDC agent tile and fill Base settings: Host name / Address, Port, Database name, Username and Password. Select Test and continue. The source appears with a green check mark. Select Next;

    3. In 2. Setup target(s), search for Kafka and choose the Kafka agent tile under Event Streaming. In Base settings fill:

      • Host name / Address: the address of one broker. The agent finds the other brokers from the cluster metadata;

      • Port: 9092, or the port of the listener you use;

      • Username and Password: the user of step 1;

      • SASL Mechanism: SCRAM-SHA-256, the default, to match step 1;

      • Security Protocol: SASL_PLAINTEXT for a listener without TLS. The default is SASL_SSL, which is for a listener with TLS (see step 10);

    4. Open Advanced settings, select Add property, choose Topic prefix and enter gs. as the value. Select Add property again, choose Include before image and switch it on;

    5. Select Test and continue. The target appears with a green check mark. Select Next, enter a Pipeline name and select Finish.

    Setup target wizard step with the Kafka agent
    Figure 1. The Kafka target connection in the setup wizard. This listener uses SASL without TLS.
    Advanced settings tab of the Kafka target with the properties Topic prefix set to gs. and Include before image switched on
    Figure 2. The two advanced settings of this guide: a topic prefix and the before image.
    Expected result

    The pipeline appears in the HEALTHY group of the pipeline list, with the PostgreSQL and the Kafka icons. Test and continue connects with the administration client, signs in with the user and lists the topics. A green check mark proves the address, the protocol, the mechanism and the credentials. It does not prove that the user can write: step 11 shows what a missing ACL looks like.

    The agent adds no topic prefix to the message key. The prefix is only a part of the topic name, as acme. becomes acme.orders. Type the separator yourself. A prefix may contain letters, digits, ., _ and -, because that is what Kafka allows in a topic name. The Apache Kafka target reference describes every setting and the REST call for the same setup.

    Add the entity and name the topic and the key

    UI

    1. In the pipeline, select Entities. The Add entities page opens with the Object browser. Expand the source schema public and select the table orders. In Selected entities, select the mapping icon of the table. The Field mapping dialog opens;

    2. Look at the Target side. Target Topic is the topic name and it starts as the table name, orders, without the schema. The topic the agent writes to is the prefix plus this name, here gs.orders. Leave it as it is;

    3. Select Confirm, then Confirm and continue.

    Field mapping dialog with the PostgreSQL orders table as source and the Kafka target with Target Topic orders
    Figure 3. The Field mapping dialog of the Kafka target. There is no topic to create here: Target Topic only names it.
    Each message also carries a key, and Kafka uses the key to choose the partition. By default the key is the target name and the primary key, joined by ::, for example orders::1. The prefix of step 3 is not part of the key. To change it, select the gear icon of the table in Selected entities, select kafka under Target, and scroll to Document key. Switch off Default key to define a Prefix, a Separator, the Fields and a Suffix. The preview shows the result, for example `cust

    id`.

    Settings dialog for a table with the Target kafka page open
    Figure 4. A user defined document key. The preview shows the prefix, the separator and the field.
    Expected result

    The entity appears in the pipeline on Hold as public.orders. No topic was touched yet.

    Run the snapshot

    UI

    1. In the pipeline, select the play button of the entity. The Sync changes (CDC) dialog opens. Keep Start from on Full copy: it publishes all rows first and then follows the changes;

    2. Select Speed, then Custom and See or change the details, to see how the agent writes. Write method shows Standard and says This target has no bulk load. Writers during the copy is the number of parallel writers, 2 by default. Leave both as they are;

    3. Select Start entity. The entity becomes Active and Last rows shows 20.

    Sync changes dialog with Full copy selected
    Figure 5. The start dialog with the custom speed settings. Kafka has no bulk load, so only the number of writers can change.
    Pipeline page with the entity public.orders Active and 20 in the Last rows column
    Figure 6. The entity after the snapshot, with 20 rows.

    Check the number of messages on the broker. The three numbers are the end offsets of the three partitions, one for each line, and their sum must be the number of rows:

    kafka-get-offsets.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --topic gs.orders
    Expected result

    The command prints one line for each partition, for example gs.orders:0:8, gs.orders:1:2 and gs.orders:2:10. The offsets add up to 20. The split between partitions differs on your cluster, because the key decides it.

    Read the messages back

    CLI

    A consumer needs its own rights: Read on the topic, and Read on its consumer group. Create a reader that can only do that, so that you do not read with the administrator account:

    kafka-configs.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --alter --entity-type users --entity-name reader \
  --add-config 'SCRAM-SHA-256=[password=<a strong password>]'
kafka-acls.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --add --allow-principal User:reader --operation Read --operation Describe \
  --topic gs. --resource-pattern-type prefixed
kafka-acls.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --add --allow-principal User:reader --operation Read \
  --group console- --resource-pattern-type prefixed

    Put the reader credentials in a file named reader.properties:

    security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="reader" password="<password>";

    Then read three messages from the start of the topic. The print.key and print.partition properties show the record key and the partition before each message:

    kafka-console-consumer.sh --bootstrap-server <broker-host>:9092 \
  --consumer.config reader.properties \
  --topic gs.orders --from-beginning --max-messages 3 \
  --property print.key=true --property print.partition=true \
  --property key.separator=' | '
    Expected result

    Each line is the partition, the record key and the JSON message. The first line looks like this. created_at differs on your system:

    Partition:0 | orders::1 | {"key":"orders::1","value":{"before":{"id":null,"customer":null,"status":null,"amount":null,"created_at":null},"after":{"id":1,"customer":"Customer-1","status":"NEW","amount":"10.50","created_at":"2026-11-09T08:25:53.250112"},"keys":{"id":1},"op":"u","ts_ms":1794213606}}

    The message is a JSON object with the same layout for every table:

    Field Content

    key

    The document key, the same text as the Kafka record key.

    value.after

    The columns of the row after the change. It is missing for a delete.

    value.before

    The columns of the row before the change. It only appears when Include before image is on. Snapshot rows have no previous state, so their before image lists every column with null.

    value.keys

    The primary key columns of the row.

    value.op

    i for an insert, u for an update and d for a delete. Snapshot rows use u.

    value.ts_ms

    The time Gluesync built the message, as Unix time in seconds, in spite of the name.

    value.transaction

    The source transaction id, as a text. Snapshot rows have none.

    Decimal values are written as text ("10.50") and timestamps as ISO text without a time zone. The data type mapping page lists the other types.

    Follow the live changes

    SQL + CLI

    1. Run an insert, an update and a delete on the source:

      INSERT INTO public.orders VALUES (21, 'Customer-21', 'NEW', 99.90, now());
UPDATE public.orders SET status = 'SHIPPED' WHERE id = 21;
DELETE FROM public.orders WHERE id = 21;

    2. Read the whole topic and keep the lines of the key orders::21. Use --from-beginning --timeout-ms 8000, and add | grep orders::21 on the command line:

      kafka-console-consumer.sh --bootstrap-server <broker-host>:9092 \
  --consumer.config reader.properties \
  --topic gs.orders --from-beginning --timeout-ms 8000 \
  --property print.key=true --property print.partition=true \
  --property key.separator=' | ' | grep orders::21
    Expected result

    Four lines, all in the same partition and in the order of the changes. The long transaction text is cut here:

    Partition:1 | orders::21 | {"key":"orders::21","value":{"after":{"id":21,"customer":"Customer-21","status":"NEW","amount":"99.90",...},"keys":{"id":21},"op":"i","ts_ms":1794213645,"transaction":"..."}}
Partition:1 | orders::21 | {"key":"orders::21","value":{"before":{...,"status":"NEW",...},"after":{...,"status":"SHIPPED",...},"keys":{"id":21},"op":"u","ts_ms":1794213645,"transaction":"..."}}
Partition:1 | orders::21 | {"key":"orders::21","value":{"before":{...,"status":"SHIPPED",...},"keys":{"id":21},"op":"d","ts_ms":1794213645,"transaction":"..."}}
Partition:1 | orders::21 | null

    The insert has no before. The update has both. The delete has a before and no after. The last line is the tombstone: the same key and an empty value.

    For every delete the agent writes the delete message and then a tombstone with the same key. A consumer that rebuilds a table from the topic can treat the tombstone as the end of the key. On a compacted topic, Kafka removes the key itself after the topic’s delete retention time.

    Check partitions and ordering

    CLI

    Kafka keeps order inside one partition and gives no order across partitions. The producer sends every message with its record key, and Kafka chooses the partition from the key. Every change of the key orders::21 therefore goes to the same partition, and a consumer sees them in the order of the source.

    1. Read the partition of one key several times. The partition does not change:

      kafka-console-consumer.sh --bootstrap-server <broker-host>:9092 \
  --consumer.config reader.properties \
  --topic gs.orders --from-beginning --timeout-ms 8000 \
  --property print.key=true --property print.partition=true | grep orders::21

    2. Plan the partition count before the first start. If you add partitions later, Kafka maps the same key to a partition by a different rule, so new messages of an existing key can land in another partition, and the order of that key is no longer guaranteed across the change.

    Expected result

    Every line of orders::21 starts with the same Partition: value.

    During the snapshot, the agent writes with several writers (Writers during the copy). The rows of a snapshot therefore reach the topic in no fixed order. Rows have different keys, so this does not affect the order of the changes of one row.

    The agent delivers each message at least once. If a write fails and the entity retries, or if you start the entity with Full copy again, the same row can appear twice. Make consumers idempotent by key: apply the last message of each key, and do not count messages as rows.

    Choose the write settings

    UI

    The producer settings are fixed in the agent. You change how much work each write does with the entity and snapshot settings:

    Setting What it does

    Write page size (entity settings, default 250)

    The number of messages the agent builds and hands to the producer as one chunk.

    Write page multiplier (multiplier / message kb) (entity settings, default 0)

    When it is more than 0, the chunk size is this number divided by the size of the page in kilobytes, and at least 1. Larger pages get smaller chunks.

    Writers during the copy (start dialog, default 2, from 1 to 10)

    The number of parallel writers during the snapshot. More writers copy faster and use more connections to the broker.

    Snapshot writing concurrency (entity Fine tuning)

    The number of writers for the entity. It is the setting behind Writers during the copy.

    To change the page size, open the settings of the table: in Selected entities, select the gear icon, then Entity. The change applies at the next start of the entity.

    Settings dialog for a table with the Entity page open
    Figure 7. The entity settings of the Kafka target.

    The agent publishes with a producer that batches: it waits up to 5 milliseconds, collects up to 256 KB for each partition, compresses with Snappy, and acknowledges with acks=1, the leader only. It never retries by itself: the entity retries a failed write. Because Kafka has no bulk load, the speed of the target is the speed of these batches and of your network and brokers. Read Tune replication throughput for the source side.

    Use Transactions in the advanced settings of the target changes this. Each write then becomes a Kafka transaction, with idempotence and acks=all, so a consumer that reads with isolation.level=read_committed never sees a half-written page. The agent uses one transactional id for each entity, and it starts with gluesync-. The user needs the Write and Describe operations on those ids:

    kafka-acls.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --add --allow-principal User:gluesync \
  --operation Write --operation Describe \
  --transactional-id gluesync- --resource-pattern-type prefixed

    To test it, open Edit target(s) as in step 10, select Advanced settings and Add property, choose Use Transactions, switch it on, select Test and continue, change a row, and read with --isolation-level read_committed in the consumer command.

    Expected result

    The changed row appears with read_committed. Without the ACL on the transactional id, the entity goes to Error with Transactional Id authorization failed.

    Encrypt the connection with TLS

    UI

    A listener with TLS uses the security protocol SASL_SSL (or SSL without SASL). The agent checks the broker certificate against a certificate store that you give it, and it does not check the host name in the certificate.

    1. In the pipeline list, select the Kafka icon under the pipeline name. Edit target(s) opens;

    2. Set Port to the TLS listener port, and Security Protocol to SASL_SSL;

    3. In Security, under Upload 'certificate store' file, select Upload files to upload the file, or Use a path to give a path that Core Hub can read, for example /certs/ca.pem. A PEM file holds the certificate of the CA that signed the broker certificate. Supported file types are pem, jks, json, txt, p8, cer, pfx, p12, key and crt;

    4. For a Java key store, select Add property under Advanced connection properties, choose Certificate Type and set it to JKS. A Certificate store password field appears after you give the path. The default type is PEM;

    5. Select Test and continue.

    Edit target page with port 9093
    Figure 8. The TLS settings: port 9093, SASL_SSL, and a certificate store path.
    Expected result

    The target saves and the entity keeps running. Change a row and read it again to confirm. If the certificate store does not hold the CA of the broker, the target does not save: see the troubleshooting table.

    The Disable auth switch in the same page turns off SASL and uses no credentials. Use it only with a listener that does not ask for a login. The agent also supports Confluent Cloud, which this guide did not test. The Apache Kafka target reference lists the SASL mechanisms and security protocols the agent accepts.

    See what a missing ACL looks like

    CLI + UI

    This step shows you the error that you will meet first if the access rules are wrong. Remove the write ACL, change a row, and watch the entity:

    kafka-acls.sh --bootstrap-server <broker-host>:9092 --command-config admin.properties \
  --remove --allow-principal User:gluesync --operation Write \
  --topic gs. --resource-pattern-type prefixed --force
    Pipeline page with the entity public.orders in Error and a notification titled Invalid configuration saying Not authorized to access topics gs.orders
    Figure 9. The entity goes to Error and the notification names the topic.

    Add the ACL again with the command of step 1. The entity restarts by itself, and the change that failed is published.

    Expected result

    The entity goes to Error. The notification title is Invalid configuration and the message is Not authorized to access topics: [gs.orders]. After you add the ACL, the entity returns to Active without a restart and the message arrives in the topic.

    Troubleshooting

    Problem Possible cause What to do

    Failed to save with SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256

    The user does not exist, the password is wrong, or the user has no credential for the mechanism you chose

    Check the user with kafka-configs.sh --describe --entity-type users --entity-name <user>. It lists the mechanisms the user has. Create the missing one, or change SASL Mechanism

    Test and continue waits about a minute and then fails with TimeoutException: Timed out waiting for a node assignment. Call: listTopics

    The agent cannot talk to the listener: the port is 0 or wrong, the host is not reachable, or Security Protocol does not match the listener, for example SASL_SSL against a listener that has no TLS. The broker log says Failing SASL authentication due to invalid receive size in the last case

    Check the port and the protocol of the listener in the listeners and listener.security.protocol.map settings of the broker, and make the two settings match

    Failed to save with The connection was saved, but agent …​ did not accept the new credentials: org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed

    The broker uses TLS and the agent has no certificate store, or the store does not hold the CA of the broker certificate

    Give the certificate store in Security (step 10). For a Java key store, also set Certificate Type to JKS and enter the Certificate store password

    The entity goes to Error with Invalid configuration and Not authorized to access topics: [<topic>]

    The user has no Write ACL on the topic. This often happens after you change Topic prefix or Target Topic, because the new name is outside the ACL pattern

    Add the ACL for the new name, for example --operation Write --topic <prefix> --resource-pattern-type prefixed. The entity restarts by itself

    The entity stays in Snapshot with no rows, and the Core Hub log says Topic <topic> not present in metadata after 60000 ms.

    The topic does not exist and the broker does not create topics automatically

    Create the topic (step 1) with kafka-topics.sh --create. The next retry of the entity publishes the rows. The agent retries every minute, and a few rows can be published twice

    The entity goes to Error with Transactional Id authorization failed.

    Use Transactions is on, and the user has no Write ACL on the transactional ids

    Add the ACL on the transactional id prefix gluesync- (step 9)

    kafka-configs.sh answers A user credential cannot be altered twice in the same request

    You gave two SCRAM mechanisms in one --add-config for the same user

    Run the command once for each mechanism

    kafka-console-consumer.sh answers Not authorized to access group: console-consumer-<n>

    The reading user has no Read ACL on its consumer group. The console consumer picks a random group name that starts with console-consumer-

    Add Read on the group prefix console-, or pass --group <name> and give the user Read on that group (step 6)

    There are more messages than rows

    The snapshot ran twice, or the entity retried a page after an error. Delivery is at least once

    Expected. Make consumers apply the last message of each key. Recreate the topic before a clean start, and start with Full copy once

    before lists every column with null

    The message comes from the snapshot, which has no previous state, or the source does not send old values

    For updates after the snapshot, run ALTER TABLE …​ REPLICA IDENTITY FULL on PostgreSQL (step 2), and keep Include before image on

    A consumer reads null as a message value

    It is the tombstone that follows every delete

    Expected. Handle an empty value as the end of the key

    Related guides

    Contact MOLO17 Support if you need assistance with a Kafka target.