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

[john_wang]

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

1. 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
       }
     }
   }

2. 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
       }
     }
   }

3. 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"
       }
     }
   }

4.  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

Related Content:

         Suggestion Kafka JSON value contain schema. Table, scn, etc