title: Kafka Connect How to make namespace agnostic to database name
body: <p><strong>My Environment</strong></p>
  
  <p><strong>MySQL(5.7)</strong>: We have multiple schemas and the naming convention is {application_name}_env. </p>
  
  <p>Example: Consider we have two apps: app1 and app2</p>
  
  <p><em>Dev Environment:</em> The database names would be app1_dev, app2_dev</p>
  
  <p><em>QA Environment:</em> The database names would be app1_qa, app2_qa.</p>
  
  <p><strong>Debezium(0.8.3)</strong>. The plugin is used to CDC MySQL Logs. </p>
  
  <p>Connector Configuration is:</p>
  
  <pre><code>{
  "name": "connector-1",
  "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "tasks.max": "1",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "dbz",
      "database.server.id": "184054",
      "database.server.name": "dbserver1",
      "database.history.kafka.bootstrap.servers": "kafka:9092",
      "database.history.kafka.topic": "schema-changes.inventory",
      "decimal.handling.mode": "double",
      "snapshot.mode": "when_needed",
      "table.whitelist":"{database_name}.account",
      "key.converter": "io.confluent.connect.avro.AvroConverter",
      "value.converter": "io.confluent.connect.avro.AvroConverter",
      "key.converter.schema.registry.url": "http://schema-registry:8081",
      "value.converter.schema.registry.url": "http://schema-registry:8081",
      "transforms" : "setSchema",
      "transforms.setSchema.type" : "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
      "transforms.setSchema.schema.name" : "com.test.Account"
  }
  </code></pre>
  
  <p>}</p>
  
  <p><strong>Spring Java Application</strong> I am using Kafka Consumer(@KafkaListener) to read the changes from the Debezium event. </p>
  
  <p>I provide the avsc files and used gradle avro plugin to generate the classes.</p>
  
  <p><strong>Schema from Dev env</strong></p>
  
  <pre><code>{
  "type":"record",
  "name":"Accounts",
  "namespace":"com.test",
  "fields":[
    {
       "name":"before",
       "type":[
          "null",
          {
             "type":"record",
             "name":"Value",
             "namespace":"dbserver1.app1_dev.account",
             "fields":[
                {
                   "name":"id",
                   "type":"long"
                }
             ],
             "connect.name":"dbserver1.app1_dev.account.Value"
          }
       ],
       "default":null
    },
    {
       "name":"after",
       "type":[
          "null",
          "dbserver1.app1_dev.account.Value"
       ],
       "default":null
    },
    {
       "name":"source",
       "type":{
          "type":"record",
          "name":"Source",
          "namespace":"io.debezium.connector.mysql",
          "fields":[
             {
                "name":"version",
                "type":[
                   "null",
                   "string"
                ],
                "default":null
             },
             {
                "name":"name",
                "type":"string"
             },
             {
                "name":"server_id",
                "type":"long"
             },
             {
                "name":"ts_sec",
                "type":"long"
             },
             {
                "name":"gtid",
                "type":[
                   "null",
                   "string"
                ],
                "default":null
             },
             {
                "name":"file",
                "type":"string"
             },
             {
                "name":"pos",
                "type":"long"
             },
             {
                "name":"row",
                "type":"int"
             },
             {
                "name":"snapshot",
                "type":[
                   {
                      "type":"boolean",
                      "connect.default":false
                   },
                   "null"
                ],
                "default":false
             },
             {
                "name":"thread",
                "type":[
                   "null",
                   "long"
                ],
                "default":null
             },
             {
                "name":"db",
                "type":[
                   "null",
                   "string"
                ],
                "default":null
             },
             {
                "name":"table",
                "type":[
                   "null",
                   "string"
                ],
                "default":null
             },
             {
                "name":"query",
                "type":[
                   "null",
                   "string"
                ],
                "default":null
             }
          ],
          "connect.name":"io.debezium.connector.mysql.Source"
       }
     },
     {
       "name":"op",
       "type":"string"
     },
     {
       "name":"ts_ms",
       "type":[
          "null",
          "long"
       ],
       "default":null
     }
   ],
   "connect.name":"com.test.Account"
   }
  </code></pre>
  
  <p><strong>Issue:</strong>
  Since my database schemas are dynamic i.e they end with env suffix. </p>
  
  <p><em>The Schema generated in each environment has a different namespace.</em></p>
  
  <p>Dev: dev.app1_dev.accounts
  QA: dev.app1_qa.accounts</p>
  
  <p>Because of the different namespace, I am not able to deserialize my dev code in QA. So If used schema generated in Dev, the code won't work in QA.</p>
  
  <p>I want to make sure that namespace is consistent across all the environments.</p>