How to create a variable within Debezium connector?

62 Views Asked by At

How this connector can create a topic if the value of ${routedByValue} is different? I would like the connector to be able to create an Apicurio schema in the Apicurio schema registry for each topic. Is that possible?

{
    "name": "connector-outbox-route",
    "config": {
        "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
        "tasks.max": "1",
        "topic.prefix": "mongodb-outbox",
        "mongodb.connection.string": "mongodb://mongodb2:27017/?replicaSet=rsmongo",
        "mongodb.user": "myUserAdmin",
        "mongodb.password": "abc123",
        "collection.include.list": "test.outboxevent",
        "database.history.kafka.bootstrap.servers": "kafka-1:9092",
        "database.history.kafka.topic": "schema-changes.outboxevent",

        "transforms": "outbox",
        "transforms.outbox.type" : "io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter",
        "transforms.outbox.route.by.field": "type",
        "transforms.outbox.route.topic.replacement" : "${routedByValue}",
        "transforms.outbox.collection.expand.json.payload":"true",

        // "key.converter.apicurio.registry.artifact.group-id": "outbox-registry",
        // "key.converter.apicurio.registry.artifact.artifact-id":"${routedByValue}",
        // "value.converter.apicurio.registry.artifact.group-id": "outbox-registry",
        // "value.converter.apicurio.registry.artifact.artifact-id":"${routedByValue}",

        "key.converter.schemas.enable": "false",
        "key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
        "key.converter.apicurio.registry.converter.deserializer": "io.apicurio.registry.serde.avro.AvroKafkaDeserializer",
        "key.converter.apicurio.registry.converter.serializer": "io.apicurio.registry.serde.avro.AvroKafkaSerializer",
        "key.converter.apicurio.registry.url": "http://apicurio:8080/apis/registry/v2",
        "key.converter.apicurio.registry.auto-register": "true",
        "key.converter.apicurio.registry.find-latest": "true",
        "key.converter.apicurio.registry.use-id": "contentId",
      
        "value.converter.schemas.enable": "false",
        "value.converter": "io.apicurio.registry.utils.converter.AvroConverter",
        "value.converter.apicurio.registry.converter.deserializer": "io.apicurio.registry.serde.avro.AvroKafkaDeserializer",
        "value.converter.apicurio.registry.converter.serializer": "io.apicurio.registry.serde.avro.AvroKafkaSerializer",
        "value.converter.apicurio.registry.url": "http://apicurio:8080/apis/registry/v2",
        "value.converter.apicurio.registry.auto-register": "true",
        "value.converter.apicurio.registry.find-latest": "true",
        "value.converter.apicurio.registry.use-id": "contentId"
    }
}
0

There are 0 best solutions below