alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-5924 respect SQL-Server database.instance config option

Browse Source
This commit is contained in:
Keri Harris 2022-12-10 22:19:33 +01:00 committed by Jiri Pechanec
parent dd96ccc27d
commit 58eb77d7e4
4 changed files with 29 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java
View File

@ -100,9 +100,11 @@ public class SqlServerConnection extends JdbcConnection {
private static final String OPENING_QUOTING_CHARACTER = "[";
private static final String CLOSING_QUOTING_CHARACTER = "]";
private static final String URL_PATTERN = "jdbc:sqlserver://${" + JdbcConfiguration.HOSTNAME + "}:${" + JdbcConfiguration.PORT + "}";
private static final String URL_PATTERN = "jdbc:sqlserver://${" + JdbcConfiguration.HOSTNAME + "}";
private final JdbcConfiguration config;
private final boolean useSingleDatabase;
private final String instanceName;
private final String getAllChangesForTable;
private final int queryFetchSize;
@ -124,8 +126,8 @@ public class SqlServerConnection extends JdbcConnection {
*/
public SqlServerConnection(JdbcConfiguration config, SqlServerValueConverters valueConverters,
Set<Envelope.Operation> skippedOperations,
boolean useSingleDatabase) {
super(config, createConnectionFactory(useSingleDatabase), OPENING_QUOTING_CHARACTER, CLOSING_QUOTING_CHARACTER);
boolean useSingleDatabase, String instanceName) {
super(config, createConnectionFactory(config, useSingleDatabase, instanceName), OPENING_QUOTING_CHARACTER, CLOSING_QUOTING_CHARACTER);
defaultValueConverter = new SqlServerDefaultValueConverter(this::connection, valueConverters);
this.queryFetchSize = config().getInteger(CommonConnectorConfig.QUERY_FETCH_SIZE);
@ -161,7 +163,9 @@ public SqlServerConnection(JdbcConfiguration config, SqlServerValueConverters va
getAllChangesForTable = get_all_changes_for_table.replaceFirst(STATEMENTS_PLACEHOLDER,
Matcher.quoteReplacement(", " + LSN_TIMESTAMP_SELECT_STATEMENT));
this.config = config;
this.useSingleDatabase = useSingleDatabase;
this.instanceName = instanceName;
this.optionRecompile = false;
}
@ -176,8 +180,8 @@ public SqlServerConnection(JdbcConfiguration config, SqlServerValueConverters va
*/
public SqlServerConnection(JdbcConfiguration config, SqlServerValueConverters valueConverters,
Set<Envelope.Operation> skippedOperations, boolean useSingleDatabase,
boolean optionRecompile) {
this(config, valueConverters, skippedOperations, useSingleDatabase);
String instanceName, boolean optionRecompile) {
this(config, valueConverters, skippedOperations, useSingleDatabase, instanceName);
this.optionRecompile = optionRecompile;
}
@ -196,15 +200,24 @@ private boolean hasSkippedOperations(Set<Envelope.Operation> skippedOperations)
return false;
}
private static ConnectionFactory createConnectionFactory(boolean useSingleDatabase) {
return JdbcConnection.patternBasedFactory(createUrlPattern(useSingleDatabase),
private static ConnectionFactory createConnectionFactory(JdbcConfiguration config, boolean useSingleDatabase, String instanceName) {
return JdbcConnection.patternBasedFactory(createUrlPattern(config, useSingleDatabase, instanceName),
SQLServerDriver.class.getName(),
SqlServerConnection.class.getClassLoader(),
JdbcConfiguration.PORT.withDefault(SqlServerConnectorConfig.PORT.defaultValueAsString()));
}
private static String createUrlPattern(boolean useSingleDatabase) {
private static String createUrlPattern(JdbcConfiguration config, boolean useSingleDatabase, String instanceName) {
String pattern = URL_PATTERN;
if (instanceName != null) {
pattern += "\\" + instanceName;
if (config.getPortAsString() != null) {
pattern += ":${" + JdbcConfiguration.PORT + "}";
}
}
else {
pattern += ":${" + JdbcConfiguration.PORT + "}";
}
if (useSingleDatabase) {
pattern += ";databaseName=${" + JdbcConfiguration.DATABASE + "}";
}
@ -218,7 +231,7 @@ private static String createUrlPattern(boolean useSingleDatabase) {
* @return a {@code String} where the variables in {@code urlPattern} are replaced with values from the configuration
*/
public String connectionString() {
return connectionString(createUrlPattern(useSingleDatabase));
return connectionString(createUrlPattern(config, useSingleDatabase, instanceName));
}
@Override

2
debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnector.java
View File

@ -153,6 +153,6 @@ protected Map<String, ConfigValue> validateAllFields(Configuration config) {
private SqlServerConnection connect(SqlServerConnectorConfig sqlServerConfig) {
return new SqlServerConnection(sqlServerConfig.getJdbcConfig(), null, Collections.emptySet(),
sqlServerConfig.useSingleDatabase());
sqlServerConfig.useSingleDatabase(), sqlServerConfig.getInstanceName());
}
}

5
debezium-connector-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnectorTask.java
View File

@ -72,9 +72,10 @@ public ChangeEventSourceCoordinator<SqlServerPartition, SqlServerOffsetContext>
dataConnection = new SqlServerConnection(connectorConfig.getJdbcConfig(), valueConverters,
connectorConfig.getSkippedOperations(), connectorConfig.useSingleDatabase(),
connectorConfig.getOptionRecompile());
connectorConfig.getInstanceName(), connectorConfig.getOptionRecompile());
metadataConnection = new SqlServerConnection(connectorConfig.getJdbcConfig(), valueConverters,
connectorConfig.getSkippedOperations(), connectorConfig.useSingleDatabase());
connectorConfig.getSkippedOperations(), connectorConfig.useSingleDatabase(),
connectorConfig.getInstanceName());
this.schema = new SqlServerDatabaseSchema(connectorConfig, metadataConnection.getDefaultValueConverter(), valueConverters, topicNamingStrategy,
schemaNameAdjuster);

6
debezium-connector-sqlserver/src/test/java/io/debezium/connector/sqlserver/util/TestHelper.java
View File

@ -238,7 +238,7 @@ private static void dropTestDatabase(SqlServerConnection connection, String data
public static SqlServerConnection adminConnection() {
return new SqlServerConnection(TestHelper.defaultJdbcConfig(),
new SqlServerValueConverters(JdbcValueConverters.DecimalMode.PRECISE, TemporalPrecisionMode.ADAPTIVE, null),
Collections.emptySet(), false);
Collections.emptySet(), false, null);
}
public static SqlServerConnection testConnection() {
@ -264,7 +264,7 @@ public static SqlServerConnection testConnection(String databaseName) {
public static SqlServerConnection testConnection(JdbcConfiguration config) {
return new SqlServerConnection(config,
new SqlServerValueConverters(JdbcValueConverters.DecimalMode.PRECISE, TemporalPrecisionMode.ADAPTIVE, null),
Collections.emptySet(), false);
Collections.emptySet(), false, null);
}
public static SqlServerConnection testConnectionWithOptionRecompile() {
@ -275,7 +275,7 @@ public static SqlServerConnection testConnectionWithOptionRecompile() {
return new SqlServerConnection(config,
new SqlServerValueConverters(JdbcValueConverters.DecimalMode.PRECISE, TemporalPrecisionMode.ADAPTIVE, null),
Collections.emptySet(), true);
Collections.emptySet(), true, null, true);
}
/**