Qlik Replicate and Kafka target: How to add tables name in Kafka message

john_wang — Mon, 11 Sep 2023 05:00:48 GMT

When using Kafka as a target in a Qlik Replicate task, the source table's "Schema Name" & "Table Name" are not included in the Kafka message, this is the default behavior.

In some scenarios, you may want to add some additional information into the Kafka messages.

In this article, we will summarize all available options and weigh out the pros/cons between the different options. We use the "Table Name" as an example in below alternatives:

Table Settings → Transform
Global Rules → Transformation
Table Settings → Message Format
Task Settings → Message Format

Resolution

Table Settings → Transform

We may add a column eg "tableName" in table settings output columns as below:

Cons:
-- No variable is available, so it's not dynamic value but a fixed string. In our sample the expression string is 'kit'
-- Affects the single table only
-- The table name appears in message's data part (rather than headers part)

{
"magic": "atMSG",
"type": "DT",
"headers": null,
"messageSchemaId": null,
"messageSchema": null,
"message": {
"data": {
"ID": "2",
"NAME": "test Kafka",
"tableName": "kit"
},
"beforeData": {
"ID": "2",
"NAME": "ok",
"tableName": "kit"
},
"headers": {
"operation": "UPDATE",
"changeSequence": "20230911032325000000000000000000005",
"timestamp": "2023-09-11T03:23:25.000",
"streamPosition": "00000000.00bb2531.00000001.0000.02.0000:154.6963.16",
"transactionId": "00000000000000000000000000060008",
"changeMask": "02",
"columnMask": "07",
"transactionEventCounter": 1,
"transactionLastEvent": true
}
}
}

Global Rules → Transformation

Global Rules can be used to add table name column to all the tables messages

Prons:
-- Affects all the tables
-- Variables are available, in our sample the variable $AR_M_SOURCE_TABLE_NAME is used.
-- The table name can be customized by combining with other transformation eg adding suffix expression "__QA"
Cons:
-- The table name appears in message's data part (rather than headers part)
If both tables transform and global rules transformation are defined (and their values are different) then tables level transform overwrites the global transformation settings.

{
"magic": "atMSG",
"type": "DT",
"headers": null,
"messageSchemaId": null,
"messageSchema": null,
"message": {
"data": {
"ID": "2",
"NAME": "test Kafka 2",
"tableName": "KIT"
},
"beforeData": {
"ID": "2",
"NAME": "test Kafka",
"tableName": "KIT"
},
"headers": {
"operation": "UPDATE",
"changeSequence": "20230911034827000000000000000000005",
"timestamp": "2023-09-11T03:48:27.000",
"streamPosition": "00000000.00bb28db.00000001.0000.02.0000:154.7632.16",
"transactionId": "00000000000000000000000000170006",
"changeMask": "02",
"columnMask": "07",
"transactionEventCounter": 1,
"transactionLastEvent": true
}
}
}

Table Settings → Message Format

Enable the "Table Name" option will include the header information in Kafka messages.

Cons:
-- Affects the single table only
Prons:
-- This new feature was released in Replicate 2023.5 and above versions
-- The table name appears in message's headers part (rather than data part)

{
"magic": "atMSG",
"type": "DT",
"headers": null,
"messageSchemaId": null,
"messageSchema": null,
"message": {
"data": {
"ID": "2",
"NAME": "test Kafka 3"
},
"beforeData": {
"ID": "2",
"NAME": "test Kafka 2"
},
"headers": {
"operation": "UPDATE",
"changeSequence": "20230911041053000000000000000000005",
"timestamp": "2023-09-11T04:10:53.000",
"streamPosition": "00000000.00bb2c30.00000001.0000.02.0000:154.9378.16",
"transactionId": "00000000000000000000000000060005",
"changeMask": "02",
"columnMask": "03",
"transactionEventCounter": 1,
"transactionLastEvent": true,
"tableName": "KIT"
}
}
}

Task Settings → Message Format

Enable the "Table Name" option will include the header information in Kafka messages.

Prons:
-- Affects all the tables
-- This new feature was released in Replicate 2023.5 and above versions
-- The table name appears in message's headers part (rather than data part)
If both table level and task level "Message Format" are defined (and their values are different) then table level settings overwrites the task settings.

{
"magic": "atMSG",
"type": "DT",
"headers": null,
"messageSchemaId": null,
"messageSchema": null,
"message": {
"data": {
"ID": "2",
"NAME": "test Kafka 4"
},
"beforeData": {
"ID": "2",
"NAME": "test Kafka 3"
},
"headers": {
"operation": "UPDATE",
"changeSequence": "20230911042445000000000000000000005",
"timestamp": "2023-09-11T04:24:45.000",
"streamPosition": "00000000.00bb2e56.00000001.0000.02.0000:154.9799.16",
"transactionId": "00000000000000000000000000080001",
"changeMask": "02",
"columnMask": "03",
"transactionEventCounter": 1,
"transactionLastEvent": true,
"tableName": "KIT"
}
}
}

Environment

Qlik Replicate (versions 2023.5 and above)
Kafka target

article Qlik Replicate and Kafka target: How to add tables name in Kafka message in Official Support Articles

Qlik Replicate and Kafka target: How to add tables name in Kafka message

Resolution

Environment

Related Content: