Kafkaf connect return different types of value after transformation using mongodb connecotor

Kafka Connector
1

Hi guys, thanks for your time reading my topic
Currently, I’m using kafka connect and mongodb connector to sink data to another storage, and I got few problems with that, this is my file connector configuration

{
  "name": "mongodb-source-connector-topics",
  "config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
    "tasks.max": "1",
    "mongodb.hosts": "mongo:27017",
    "mongodb.name": "backend",
    "database.include.list": "bitu-backend",
    "snapshot.mode": "initial",
    "signal.data.collection": "backend.cdc_signal",
    "collection.include.list": "backend.topics,backend.cdc_signal",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "behavior.on.null.values": "delete",
    "errors.tolerance": "all",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "transforms": "route,unwrap,extractKey,ReplaceField,RenameField",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
    "transforms.route.replacement": "cdc_$3",
    "transforms.unwrap.type": "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState",
    "transforms.unwrap.array.encoding": "array",
    "transforms.unwrap.delete.handling.mode": "none",
    "transforms.unwrap.drop.tombstone": "true",
    "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractKey.field": "id",
    "transforms.ReplaceField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.ReplaceField.include": "_id,level,lang,class_types,title,image,is_active,type,revise_type",
    "transforms.RenameField.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.RenameField.renames": "_id:topic_id"
  }
}

what i expected is

  "topic_id": "6062c8dbbf7efd10e2463146",
  "title": "Why are so many people addicted to social apps?",
  "image": "",
  "is_active": true,
  "type": 2,
  "level": 200,
  "lang": "vi",
  "class_types": [
    1
  ]
}

but sometimes i receive null payload, sometimes i got all structure like transformation without using ExtractNewDocumentState
and sometimes i think it might conflict with other transformations

I got sick because of it, I don’t know how to tackle with that?

Can anyone help me?

2

We do have a Couchbase Kafka Connector.
But for issues with the Mongodb connector, contact Mongodb.

3

Thank u for the reply, I will consider your work.

4

This topic was automatically closed 90 days after the last reply. New replies are no longer allowed.