source-mssql CDC mode fails for tables with spaces in names (message.key.columns regex rejection)
- Vorherrschende Sprache
- Python
- Sterne
- 22.1k
- Forks
- 5.4k
- Ø Merge
- 5 Std.
- Gemergte PRs (30 T.)
- 671
Beschreibung
### Connector Name
source-mssql
### Connector Version
4.3.5
### What step the error happened?
During the sync
### Relevant information
## Bug description
When using `source-mssql` in CDC mode against a database with table names containing spaces, the sync fails
immediately with:
io.debezium.DebeziumException: Connector configuration is not valid.
The 'message.key.columns' value is invalid:
dbo.Company Inc$Location$437dbf0e-84ff-417a-965d-ed2bb9650972:Code
has an invalid format (expecting '^\s*([^\s:]+):([^:\s]+)\s*$')
## How to reproduce
1. Configure a SQL Server source with CDC replication method
2. Select a table whose name contains a space (e.g. `dbo.[Company Inc$Location$437dbf0e-84ff-417a-965d-ed2bb9650972]`)
3. Run a sync
This is common with **Microsoft Dynamics 365 Business Central** databases, where every table follows the
naming convention `[Company Name$TableName$GUID]` — if the company name contains a space, all tables are
affected.
## Root cause
`MsSqlServerDebeziumOperations.buildMessageKeyColumns()` generates a `message.key.columns` property entry
for every CDC-enabled table with a configured primary key. Debezium validates each entry against the regex
`^\s*([^\s:]+):([^:\s]+)\s*$`, which explicitly rejects whitespace and colons in table/schema identifiers.
Any table with a space in its name causes the entire connector to fail.
## Suggested fix
In `buildMessageKeyColumns()`, skip tables whose name or namespace contains characters that Debezium's regex
rejects (whitespace or colons). This is safe because Debezium will fall back to reading the native SQL
Server primary key directly from system tables — `message.key.columns` is an override, not the only source
of PK information.
```kotlin
private fun buildMessageKeyColumns(streams: List): String {
return streams
.filter { it.configuredPrimaryKey?.isNotEmpty() == true }
.filter { stream ->
!stream.name.contains(Regex("""[\s:]""")) &&
stream.namespace?.contains(Regex("""[\s:]""")) != true
}
.joinToString(";") { stream ->
val tableId =
"${escapeSpecialChars(stream.namespace)}.${escapeSpecialChars(stream.name)}"
val keyCols =
stream.configuredPrimaryKey!!.joinToString(",") { escapeSpecialChars(it.id) }
"$tableId:$keyCols"
}
}
### Relevant log output
```shell
2025-XX-XX XX:XX:XX [pool-X-thread-X] ERROR i.a.w.i.VersionedAirbyteStreamFactory(internalLog) -
io.debezium.DebeziumException: Connector configuration is not valid. The 'message.key.columns' value is
invalid:
A value 'dbo.Company Inc$Location$437dbf0e-84ff-417a-965d-ed2bb9650972:Code' has invalid format,
the value should match the following regex: `^\s*([^\s:]+):([^:\s]+)\s*$`
at
io.debezium.embedded.async.AsyncEmbeddedEngine.lambda$startConnector$0(AsyncEmbeddedEngine.java:432)
at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
at java.base/java.lang.Thread.run(Thread.java:1583)
```
### Contribute
- [ ] Yes, I want to contribute
---
**Internal Tracking:** https://github.com/airbytehq/oncall/issues/12162
Beitragsleitfaden
Bewertung
Dieses Issue wurde noch nicht bewertet.