Kafka Connect: How to extract a field

Viewed 2123

I'm using Debezium SQL Server Connector to stream a table into a topic. Thanks to Debezium's ExtractNewRecordState SMT, I'm getting the following message in my topic.

{
   "schema":{
      "type":"struct",
      "fields":[
         {
            "type":"int64",
            "optional":false,
            "field":"id"
         },
         {
            "type":"string",
            "optional":false,
            "field":"customer_code"
         },
         {
            "type":"string",
            "optional":false,
            "field":"topic_name"
         },
         {
            "type":"string",
            "optional":true,
            "field":"payload_key"
         },
         {
            "type":"boolean",
            "optional":false,
            "field":"is_ordered"
         },
         {
            "type":"string",
            "optional":true,
            "field":"headers"
         },
         {
            "type":"string",
            "optional":false,
            "field":"payload"
         },
         {
            "type":"int64",
            "optional":false,
            "name":"io.debezium.time.Timestamp",
            "version":1,
            "field":"created_on"
         }
      ],
      "optional":false,
      "name":"test_server.dbo.kafka_event.Value"
   },
   "payload":{
      "id":129,
      "customer_code":"DVTPRDFT411",
      "topic_name":"DVTPRDFT411",
      "payload_key":null,
      "is_ordered":false,
      "headers":"{\"kafka_timestamp\":1594566354199}",
      "payload":"MSG 18",
      "created_on":1594595154267
   }
}

After adding value.converter.schemas.enable=false, I could get rid of the schema portion and only the payload part is left as shown below.

{
   "id":130,
   "customer_code":"DVTPRDFT411",
   "topic_name":"DVTPRDFT411",
   "payload_key":null,
   "is_ordered":false,
   "headers":"{\"kafka_timestamp\":1594566354199}",
   "payload":"MSG 19",
   "created_on":1594595154280
}

I'd like to go 1 step further and extract only the customer_code field. I tried ExtractField$Value SMT but I keep getting the exception IllegalArgumentException: Unknown field: customer_code.

My configuration is as following

transforms=unwrap,extract
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=true
transforms.unwrap.delete.handling.mode=drop
transforms.extract.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.extract.field=customer_code

I tried a bunch of other SMTs including ExtractField$Key, ValueToKey but I couldn't make it work. I'd be very grateful if you could show me what I've done wrong. According to this tutorial from Confluent, it should work but it didn't.

** UPDATE **

I'm running Kafka Connect using connect-standalone worker.properties sqlserver.properties.

worker.properties

offset.storage.file.filename=C:/development/kafka_2.12-2.5.0/data/kafka/connect/connect.offsets
plugin.path=C:/development/kafka_2.12-2.5.0/plugins
bootstrap.servers=127.0.0.1:9092
offset.flush.interval.ms=10000
rest.port=10082
rest.host.name=127.0.0.1
rest.advertised.port=10082
rest.advertised.host.name=127.0.0.1

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

sqlserver.properties

name=sql-server-connector
connector.class=io.debezium.connector.sqlserver.SqlServerConnector
database.hostname=127.0.0.1
database.port=1433
database.user=sa
database.password=dummypassword
database.dbname=STGCTR
database.history.kafka.bootstrap.servers=127.0.0.1:9092

database.server.name=wfo
table.whitelist=dbo.kafka_event
database.history.kafka.topic=db_schema_history
transforms=unwrap,extract
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=true
transforms.unwrap.delete.handling.mode=drop
transforms.extract.type=org.apache.kafka.connect.transforms.ExtractField$Value
transforms.extract.field=customer_code
1 Answers

The schema and payload fields sound like you're using data that was serialized with a JsonConverter with schemas enabled.

You can just set value.converter.schemas.enable=false to achieve your goal.

Related