Unable to find schema by definition when using Kafka Connect and MongoDB #143
Comments
|
I actually think this is a Kafka Connect and/or Debezium issue and not a Schema Registry issue. I wrote a Custom Single Message Transformer (SMT) to force our schema and it worked. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hi folks 👋
I am trying to run Kafka Connect (via Debezium) to pull change data capture (CDC) data from MongoDB into Kafka. Here is what my config looks like (some fields redacted due to sensitive info):
{ "config": { "collection.include.list": "<redacted>", "connector.class": "io.debezium.connector.mongodb.MongoDbConnector", "database.include.list": "<redacted>", "field.exclude.list": "<redacted>", "key.converter": "com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter", "key.converter.avroRecordType": "GENERIC_RECORD", "key.converter.compatibility": "FULL", "key.converter.dataFormat": "AVRO", "key.converter.region": "us-west-2", "key.converter.registry.name": "avro-schemas", "key.converter.schemaAutoRegistrationEnabled": false, "key.converter.schemaName": "foo.bar.Key", "key.converter.schemas.enable": false, "key.converter.userAgentApp": "Debezium", "mongodb.hosts": "mongors/localhost:27017", "mongodb.members.auto.discover": false, "mongodb.name": "<redacted>", "mongodb.password": "debezium_pass", "mongodb.ssl.enabled": true, "mongodb.ssl.invalid.hostname.allowed": true, "mongodb.user": "debezium_user", "sanitize.field.names": true, "snapshot.mode": "initial", "tasks.max": "1", "tombstones.on.delete": false, "transforms": "unwrap", "transforms.unwrap.array.encoding": "array", "transforms.unwrap.delete.handling.mode": "none", "transforms.unwrap.flatten.struct": false, "transforms.unwrap.sanitize.field.names": true, "transforms.unwrap.type": "io.debezium.connector.mongodb.transforms.ExtractNewDocumentState", "value.converter": "com.amazonaws.services.schemaregistry.kafkaconnect.AWSKafkaAvroConverter", "value.converter.avroRecordType": "GENERIC_RECORD", "value.converter.compatibility": "FULL", "value.converter.region": "us-west-2", "value.converter.registry.name": "avro-schemas", "value.converter.schemaAutoRegistrationEnabled": false, "value.converter.schemaName": "foo.bar.Value", "value.converter.schemas.enable": false, "value.converter.userAgentApp": "Debezium" }, "name": "test_connector" }I will focus on
foo.bar.Key(although I am sure the problem exists forfoo.bar.Valueas well). My schema is defined as follows (and is registered in AWS):{ "type": "record", "name": "Key", "namespace": "foo.bar", "fields": [ { "name": "id", "type": [ "null", "string" ], "default": null } ] }However when I run the Debezium container I get the following error:
I dug in a bit and tried using
schemaAutoRegistrationEnabled=trueand it worked when I did that! However, here is the Avro schema it generates:{ "type": "record", "name": "ConnectDefault", "namespace": "com.amazonaws.services.schemaregistry.kafkaconnect.avrodata", "fields": [ { "name": "id", "type": [ "null", "string" ], "default": null } ] }From my understanding of the error message, it seems it tries to pull a schema definition from MongoDB and then look that definition up against the schema registry. Since the naming is different, it fails to find the correct schema and crashes (since I have
schemaAutoRegistrationEnabled=false). Is there any way to force the connector to use the schema I have defined (and maybe look it up by name) instead of looking it up by definition?I have already looked into implementing
AWSSchemaNamingStrategy(as done in #126), but I have some additional concerns around that. For example, we have some more complex Mongo collections that havenullfields by default. When the connector sees these, it generates a schema that looks like this:{ "name": "deletedAt", "type": [ "null", "string" ], "default": null }When it should actually look like this:
{ "name": "deletedAt", "type": [ "null", "long" ], "default": null }Let me know if I can provide any additional details. Thank you!
Tl;dr
Is it possible to force a Kafka connector to use the schema already defined in the schema registry instead of looking up by definition or implementing a custom
AWSSchemaNamingStrategy?Versions used
aws-glue-schema-registry- 1.1.8The text was updated successfully, but these errors were encountered: