# Debezium connector for MS SQL - can we change/set topic name? **URL:** \ **Category:** Self-Managed Connectors\ **Created:** [10 March 2022 22:25 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427 "2022-03-10T22:25:55Z")\ **Posts on this page:** 20\ **Page:** 1
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [10 March 2022 22:25 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/1 "2022-03-10T22:25:55Z")
We are using the Debezium connector and everything is working great…BUT…is there a way to change or set the topic name of the topic that it creates? For example, we create a connector and it creates the topic “Server.Schema.Table” but what if we wanted that topic to be named “server.schema.table”? Thanks ---
**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\ **Post date:** [11 March 2022 09:40 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/2 "2022-03-11T09:40:54Z")
Single Message Transforms are your friend here. See `com.github.jcustenborder.kafka.connect.transform.common.ChangeTopicCase` example shown [here](https://rmoff.net/2020/12/23/twelve-days-of-smt-day-12-community-transformations/#_change_the_topic_case). ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [11 March 2022 14:53 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/3 "2022-03-11T14:53:54Z")
Hmm I really just need everything in lowercase but not sure that is an option here. I will play with it…thanks for the suggestion. ---
**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\ **Post date:** [11 March 2022 15:55 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/4 "2022-03-11T15:55:33Z")
> [@pebriggs](#): > > not sure that is an option here I don’t understand. Using Single Message Transform isn’t an option do you mean? ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [11 March 2022 16:58 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/5 "2022-03-11T16:58:52Z")
Well it mentions camel, underscore, and hyphens…I just need everything to be lowercase. ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [11 March 2022 20:41 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/6 "2022-03-11T20:41:04Z")
Tried multiple options and although it worked, it did not produce what I needed. I don’t need to change hyphens or underscores…I just need to lower or upper case it. ---
**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\ **Post date:** [13 March 2022 09:44 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/7 "2022-03-13T09:44:37Z")
Can you share the config you used and the input/output topics names that you got? ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [14 March 2022 13:58 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/8 "2022-03-14T13:58:40Z")
So first I tried RemoveString, then Reroute, then ChangeTopicCase, some of them did change the name of the resulting topic, but the connector fails with avro format issues and/or schema registry issues. ```auto curl -i -X PUT -H "Content-Type:application/json" \ http://localhost:8083/connectors/peterSourceV3/config \ -d '{ "connector.class":"io.debezium.connector.sqlserver.SqlServerConnector", "database.hostname":"server", "database.connection.url":"jdbc:sqlserver://server;databaseName=db;integratedSecurity=true;authenticationScheme=JavaKerberos", "database.user":"service", "database.password":"", "database.integratedSecurity":"true", "database.authenticationScheme":"JavaKerberos", "database.dbname":"db", "database.server.name":"server", "table.include.list":"dbo.AdvisorUse", "database.history.kafka.bootstrap.servers":"localhost:9092", "database.history.kafka.topic":"peterSource_history", "transforms":"changeTopicCase", "transforms.changeTopicCase.type":"com.github.jcustenborder.kafka.connect.transform.common.ChangeTopicCase", "transforms.changeTopicCase.from":"LOWER_CAMEL", "transforms.changeTopicCase.to":"LOWER_UNDERSCORE" }' ``` ---
**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\ **Post date:** [14 March 2022 14:56 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/9 "2022-03-14T14:56:41Z")
> [@pebriggs](#): > > but the connector fails with avro format issues and/or schema registry issues. What’s the error that you get? ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [14 March 2022 16:02 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/10 "2022-03-14T16:02:25Z")
```auto {"name":"testSourceV3","connector":{"state":"RUNNING","worker_id":"localhost:8083"},"tasks":[{"id":0,"state":"FAILED","worker_id":"localhost:8083","trace":"org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:220) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:142) at org.apache.kafka.connect.runtime.WorkerSourceTask.convertTransformedRecord(WorkerSourceTask.java:330) at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:359) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:272) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:199) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:254) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:750) Caused by: org.apache.kafka.connect.errors.DataException: Failed to serialize Avro data from topic testsourcev3_worked : at io.confluent.connect.avro.AvroConverter.fromConnectData(AvroConverter.java:93) at org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63) at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$3(WorkerSourceTask.java:330) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:166) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:200) ... 11 more Caused by: org.apache.kafka.common.errors.SerializationException: Error registering Avro schema{\"type\":\"record\",\"name\":\"Key\",\"namespace\":\"server.dbo.AdvisorUse\",\"fields\":[{\"name\":\"Id\",\"type\":\"long\"}],\"[connect.name](https://urldefense.com/v3/ __http:/connect.name__ ;!!Cz2fjcuE!iHaFhycNNa7Z-FUbmKrYfzv9rbWGLPPHcLQXdmM6hxMoxDkx6g60FZkGUJXCdAKn_jgLk6-aiWrvUmU2QMdeOiwFNQ$)\":\"server.dbo.AdvisorUse.Key\"} at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.toKafkaException(AbstractKafkaSchemaSerDe.java:259) at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:156) at io.confluent.connect.avro.AvroConverter$Serializer.serialize(AvroConverter.java:153) at io.confluent.connect.avro.AvroConverter.fromConnectData(AvroConverter.java:86) ... 15 more Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema being registered is incompatible with an earlier schema for subject \"testsourcev3_worked-key\"; error code: 409 at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:297) at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:367) at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:544) at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:532) at io.confluent.kafka.schemaregistry.client.rest.RestService.registerSchema(RestService.java:490) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:257) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:366) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:337) at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:115) ... 17 more "}],"type":"source"} ``` ---
**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\ **Post date:** [14 March 2022 16:46 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/11 "2022-03-14T16:46:31Z")
> [@pebriggs](#): > > `Schema being registered is incompatible with an earlier schema for subject ` That’s your problem there. So either target a new topic, or use a schema that is compatible with the existing topic. ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [14 March 2022 17:48 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/12 "2022-03-14T17:48:27Z")
I don’t understand though…this is the initial creation of the topic, there should be no earlier schema. ---
**Author:** ![rmoff](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/rmoff/32/3_2.png) [@rmoff](https://forum.confluent.io/u/rmoff)\ **Post date:** [15 March 2022 09:22 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/13 "2022-03-15T09:22:56Z")
Maybe from a previous test run of the connector? You can use the Schema Registry REST API to delete previous schemas if you want Ref: - [Confluent Schema Registry REST API cheatsheet](https://rmoff.net/2019/01/17/confluent-schema-registry-rest-api-cheatsheet/) - [`DELETE /subjects/`](https://docs.confluent.io/platform/current/schema-registry/develop/api.html#delete--subjects-(string-%20subject)) ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [15 March 2022 14:29 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/14 "2022-03-15T14:29:10Z")
I thought of that too…deleted every instance of -key and -value I could find…same thing ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [15 March 2022 18:51 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/15 "2022-03-15T18:51:32Z")
Debezium works great, I have 75+ tables being synced…but no matter what I do, I cannot set or change the name of the default topic that Debezium creates. What I want to do is simple…I just want to be standard with my topic naming…ALL LOWERCASE. SQL is not and doesn’t matter, but java is so to make things bulletproof, I just need the topic name to be all lowercase regardless of how the SQL table is. Tried numerous transformations and some change the topic name but connector fails with avro issue or schema registry incompatibility. Does anyone have a working example? ---
**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\ **Post date:** [15 March 2022 21:36 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/16 "2022-03-15T21:36:07Z")
> Tried numerous transformations and some change the topic name but connector fails with avro issue or schema registry incompatibility I cannot recall the ordering of the registry client lookup, but assuming it is after the topic name is changed, the upper/mixed-case subject will need to copied manually in the Registry API to a fully-lower cased subject. ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [15 March 2022 22:06 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/17 "2022-03-15T22:06:35Z")
Sorry but I am not quite understanding what you mean. By default, it looks like the Debezium connector creates the topic name with the exact case retrieved from the SQL server using the pieces like Server.Database.Table. The problem is that case doesn’t always match in every environment…so I was hoping to simply lowercase the topic name to make it generic across connectors and environments. I cannot seem to get this working however. ---
**Author:** ![OneCricketeer](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/onecricketeer/32/1551_2.png) [@OneCricketeer](https://forum.confluent.io/u/OneCricketeer)\ **Post date:** [15 March 2022 22:11 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/18 "2022-03-15T22:11:02Z")
1. There is an [external transform library here](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/ChangeCase.html) for changing case 2. [There is an order that transforms get applied](https://www.confluent.io/blog/kafka-connect-single-message-transformation-tutorial-with-examples/), and my point was that if you lowercased the topic in the wrong order of the `AvroConverter`, then the original topic name is what is used for the `/subjects/{name}` API call to lookup and/or register the schema. You’ll want to lowercase before that API call happens. If you cannot control this, you need to `GET /subjects/subjectName-value/versions/X` and `POST /subjects/subjectname-value/versions` for each version of the schema _manually_. ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [15 March 2022 22:19 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/19 "2022-03-15T22:19:36Z")
1. ChangeCase seems like a record transformation but ChangeTopicCase looks promising except there is no simple lower option…I don’t need to add hyphens or underscores, I just need to lowercase. I will give this a try though. 2. Do you think the order of stuff in the config file can resolve? ---
**Author:** ![pebriggs](https://avatars.discourse-cdn.com/v4/letter/p/47e85d/32.png) [@pebriggs](https://forum.confluent.io/u/pebriggs)\ **Post date:** [15 March 2022 22:36 UTC](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427/20 "2022-03-15T22:36:13Z")
Interesting…so I just tried this and well, things are working, which is awesome, but the resulting name of my topic is not what I wanted… Actual: TXI1SQLDEV01.sddev.com.dbo.AdvisorUse Desired: txi1sqldev01.sddev.com.dbo.advisoruse Result: t\_x\_i1\_s\_q\_l\_d\_e\_v01.sddev.com.dbo.\_advisor\_use “transforms.changeTopicCase.from”:“UPPER\_CAMEL”, “transforms.changeTopicCase.to”:“LOWER\_UNDERSCORE”, [Next page](https://forum.confluent.io/t/debezium-connector-for-ms-sql-can-we-change-set-topic-name/4427.md?page=2)