title: How to get the table-name and database-name in the CDC event received from debezium kafka connect
body: <p><strong>Setup:</strong> I have CDC enabled on MS SQL Server and the CDC events are fed to Kafka using debezium kafka connect(source). Also more than one table CDC events are routed to a single topic in Kafka. </p>
  
  <p><strong>Question:</strong> Since I have more than one table data in the kafka topic, I would like to have the table name and the database name in the CDC data.</p>
  
  <p>I am getting the table name and database name in MySQL CDC but not in MS SQL CDC.</p>
  
  <p>Below is the Debezium source connector for the SQL Server</p>
  
  <pre><code>curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
    "name": "cdc-user_profile-connector",
    "config": {
      "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
      "tasks.max": "1",
      "database.hostname": "&lt;&lt;hostname&gt;&gt;",
      "database.port": "&lt;&lt;port&gt;&gt;",
      "database.user": "&lt;&lt;username&gt;&gt;",
      "database.password": "&lt;&lt;password&gt;&gt;",
      "database.server.name": "test",
      "database.dbname": "testDb",
      "table.whitelist": "schema01.table1,schema01.table2",
      "database.history.kafka.bootstrap.servers": "broker:9092",
      "database.history.kafka.topic": "digital.user_profile.schema.audit",
      "database.history.store.only.monitored.tables.ddl": true,
      "include.schema.changes": false,
      "event.deserialization.failure.handling.mode": "fail",
      "snapshot.mode": "initial_schema_only",
      "snapshot.locking.mode": "none",
      "transforms":"addStaticField,topicRoute",
      "transforms.addStaticField.type":"org.apache.kafka.connect.transforms.InsertField$Value",
      "transforms.addStaticField.static.field":"source_system",
      "transforms.addStaticField.static.value":"source_system_1",
      "transforms.topicRoute.type":"org.apache.kafka.connect.transforms.RegexRouter",
      "transforms.topicRoute.regex":"(.*)",
      "transforms.topicRoute.replacement":"digital.user_profile",
      "errors.tolerance": "none",
      "errors.log.enable": true,
      "errors.log.include.messages": true,
      "errors.retry.delay.max.ms": 60000,
      "errors.retry.timeout": 300000
    }
  }'
  </code></pre>
  
  <p>I am getting the below output (Demo data)</p>
  
  <pre><code>{
    "before": {
      "profile_id": 147,
      "email_address": "test@gmail.com"
    },
    "after": {
      "profile_id": 147,
      "email_address": "test_modified@gmail.com"
    },
    "source": {
      "version": "0.9.4.Final",
      "connector": "sqlserver",
      "name": "test",
      "ts_ms": 1556723528917,
      "change_lsn": "0007cbe5:0000b98c:0002",
      "commit_lsn": "0007cbe5:0000b98c:0003",
      "snapshot": false
    },
    "op": "u",
    "ts_ms": 1556748731417,
    "source_system": "source_system_1"
  }
  </code></pre>
  
  <p>My requirement is to get as below</p>
  
  <pre><code>{
    "before": {
      "profile_id": 147,
      "email_address": "test@gmail.com"
    },
    "after": {
      "profile_id": 147,
      "email_address": "test_modified@gmail.com"
    },
    "source": {
      "version": "0.9.4.Final",
      "connector": "sqlserver",
      "name": "test",
      "ts_ms": 1556723528917,
      "change_lsn": "0007cbe5:0000b98c:0002",
      "commit_lsn": "0007cbe5:0000b98c:0003",
      "snapshot": false,
      "db": "testDb",
      "table": "table1/table2"
    },
    "op": "u",
    "ts_ms": 1556748731417,
    "source_system": "source_system_1"
  }
  </code></pre>