DBZ-3621 Changed references to JDBC URI to JDBC URL
This commit is contained in:
parent
646c77f117
commit
3f8b2a8d49
@ -244,7 +244,7 @@ public final class EmbeddedEngine implements DebeziumEngine<SourceRecord> {
|
||||
"Must be bigger than the time it takes two polling loops to finish ({} ms)", ChangeEventSourceCoordinator.SHUTDOWN_WAIT_TIMEOUT.toMillis() * 2));
|
||||
|
||||
// JDBC OFFSET STORAGE FIELDS.
|
||||
public static final Field OFFSET_STORAGE_JDBC_URI = Field.create("offset.storage.jdbc.uri")
|
||||
public static final Field OFFSET_STORAGE_JDBC_URL = Field.create("offset.storage.jdbc.url")
|
||||
.withDescription("URI of the database which will be used to record the database history")
|
||||
.withDefault("jdbc:")
|
||||
.withValidation(Field::isRequired);
|
||||
@ -277,7 +277,7 @@ public final class EmbeddedEngine implements DebeziumEngine<SourceRecord> {
|
||||
OFFSET_FLUSH_INTERVAL_MS, OFFSET_COMMIT_TIMEOUT_MS,
|
||||
ERRORS_MAX_RETRIES, ERRORS_RETRY_DELAY_INITIAL_MS, ERRORS_RETRY_DELAY_MAX_MS,
|
||||
// JDBC OFFSET STORAGE FIELDS
|
||||
OFFSET_STORAGE_JDBC_URI, OFFSET_STORAGE_JDBC_USER, OFFSET_STORAGE_JDBC_PASSWORD, OFFSET_STORAGE_JDBC_TABLE_NAME);
|
||||
OFFSET_STORAGE_JDBC_URL, OFFSET_STORAGE_JDBC_USER, OFFSET_STORAGE_JDBC_PASSWORD, OFFSET_STORAGE_JDBC_TABLE_NAME);
|
||||
|
||||
public static final class BuilderImpl implements Builder {
|
||||
private Configuration config;
|
||||
@ -1230,7 +1230,7 @@ protected static class EmbeddedConfig extends WorkerConfig {
|
||||
Field.group(config, "kafka", OFFSET_STORAGE_KAFKA_TOPIC);
|
||||
Field.group(config, "kafka", OFFSET_STORAGE_KAFKA_PARTITIONS);
|
||||
Field.group(config, "kafka", OFFSET_STORAGE_KAFKA_REPLICATION_FACTOR);
|
||||
Field.group(config, "jdbc", OFFSET_STORAGE_JDBC_URI);
|
||||
Field.group(config, "jdbc", OFFSET_STORAGE_JDBC_URL);
|
||||
Field.group(config, "jdbc", OFFSET_STORAGE_JDBC_USER);
|
||||
Field.group(config, "jdbc", OFFSET_STORAGE_JDBC_PASSWORD);
|
||||
Field.group(config, "jdbc", OFFSET_STORAGE_JDBC_TABLE_NAME);
|
||||
|
@ -6,7 +6,7 @@
|
||||
package io.debezium.storage.jdbc;
|
||||
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_PASSWORD;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_URI;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_URL;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_USER;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_TABLE_NAME;
|
||||
|
||||
@ -20,7 +20,7 @@ public class JdbcConfig extends WorkerConfig {
|
||||
|
||||
static {
|
||||
CONFIG = WorkerConfig.baseConfigDef()
|
||||
.define(OFFSET_STORAGE_JDBC_URI.name(),
|
||||
.define(OFFSET_STORAGE_JDBC_URL.name(),
|
||||
ConfigDef.Type.STRING,
|
||||
ConfigDef.Importance.HIGH,
|
||||
"JDBC database URI")
|
||||
|
@ -42,7 +42,7 @@
|
||||
*/
|
||||
public class JdbcOffsetBackingStore implements OffsetBackingStore {
|
||||
|
||||
public static final Field OFFSET_STORAGE_JDBC_URI = Field.create("offset.storage.jdbc.uri")
|
||||
public static final Field OFFSET_STORAGE_JDBC_URL = Field.create("offset.storage.jdbc.url")
|
||||
.withDescription("URI of the database which will be used to record the database history")
|
||||
.withValidation(Field::isRequired);
|
||||
|
||||
@ -103,7 +103,7 @@ public ByteBuffer toByteBuffer(String data) {
|
||||
public void configure(WorkerConfig config) {
|
||||
|
||||
try {
|
||||
jdbcUri = config.getString("offset.storage.jdbc.uri");
|
||||
jdbcUri = config.getString("offset.storage.jdbc.url");
|
||||
offsetStorageTableName = config.getString(OFFSET_STORAGE_TABLE_NAME.name());
|
||||
conn = DriverManager.getConnection(jdbcUri, config.getString(OFFSET_STORAGE_JDBC_USER.name()), config.getString(OFFSET_STORAGE_JDBC_PASSWORD.name()));
|
||||
conn.setAutoCommit(false);
|
||||
|
@ -52,8 +52,8 @@ public final class JdbcSchemaHistory extends AbstractSchemaHistory {
|
||||
|
||||
private static final Logger LOG = LoggerFactory.getLogger(JdbcSchemaHistory.class);
|
||||
|
||||
public static final Field JDBC_URI = Field.create(CONFIGURATION_FIELD_PREFIX_STRING + "jdbc.uri")
|
||||
.withDescription("URI of the database which will be used to record the database history")
|
||||
public static final Field JDBC_URL = Field.create(CONFIGURATION_FIELD_PREFIX_STRING + "jdbc.url")
|
||||
.withDescription("URL of the database which will be used to record the database history")
|
||||
.withValidation(Field::isRequired);
|
||||
|
||||
public static final Field JDBC_USER = Field.create(CONFIGURATION_FIELD_PREFIX_STRING + "jdbc.user")
|
||||
@ -109,13 +109,13 @@ public void configure(Configuration config, HistoryRecordComparator comparator,
|
||||
}
|
||||
config.validateAndRecord(ALL_FIELDS, LOG::error);
|
||||
if (running.get()) {
|
||||
throw new IllegalStateException("Database history already initialized db: " + config.getString(JDBC_URI));
|
||||
throw new IllegalStateException("Database history already initialized db: " + config.getString(JDBC_URL));
|
||||
}
|
||||
super.configure(config, comparator, listener, useCatalogBeforeSchema);
|
||||
|
||||
try {
|
||||
jdbcUri = config.getString(JDBC_URI.name());
|
||||
conn = DriverManager.getConnection(config.getString(JDBC_URI.name()), config.getString(JDBC_USER.name()), config.getString(JDBC_PASSWORD.name()));
|
||||
jdbcUri = config.getString(JDBC_URL.name());
|
||||
conn = DriverManager.getConnection(config.getString(JDBC_URL.name()), config.getString(JDBC_USER.name()), config.getString(JDBC_PASSWORD.name()));
|
||||
conn.setAutoCommit(false);
|
||||
}
|
||||
catch (SQLException e) {
|
||||
|
@ -16,12 +16,12 @@
|
||||
import static io.debezium.junit.EqualityCheck.LESS_THAN;
|
||||
import io.debezium.junit.SkipWhenDatabaseVersion;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_PASSWORD;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_URI;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_URL;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_USER;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_TABLE_NAME;
|
||||
import io.debezium.storage.jdbc.history.JdbcSchemaHistory;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_PASSWORD;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_URI;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_URL;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_USER;
|
||||
import io.debezium.util.Testing;
|
||||
import org.junit.After;
|
||||
@ -118,7 +118,7 @@ private String topicName() {
|
||||
|
||||
protected Configuration.Builder schemaHistory(Configuration.Builder builder) {
|
||||
return builder
|
||||
.with(JDBC_URI, "jdbc:sqlite:" + SCHEMA_HISTORY_PATH)
|
||||
.with(JDBC_URL, "jdbc:sqlite:" + SCHEMA_HISTORY_PATH)
|
||||
.with(JDBC_USER, "user")
|
||||
.with(JDBC_PASSWORD, "pass");
|
||||
}
|
||||
@ -138,7 +138,7 @@ private Configuration.Builder config(String jdbcUri) {
|
||||
.with(CommonConnectorConfig.TOPIC_PREFIX, TOPIC_PREFIX)
|
||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, MySqlConnectorConfig.SnapshotMode.INITIAL)
|
||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, false)
|
||||
.with(OFFSET_STORAGE_JDBC_URI.name(), jdbcUri)
|
||||
.with(OFFSET_STORAGE_JDBC_URL.name(), jdbcUri)
|
||||
.with(OFFSET_STORAGE_JDBC_USER.name(), "user")
|
||||
.with(OFFSET_STORAGE_JDBC_PASSWORD.name(), "pass")
|
||||
.with(OFFSET_STORAGE_TABLE_NAME.name(), "offsets_jdbc")
|
||||
|
@ -6,7 +6,7 @@
|
||||
package io.debezium.storage.jdbc;
|
||||
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_PASSWORD;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_URI;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_URL;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_USER;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_TABLE_NAME;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
@ -44,7 +44,7 @@ public void setup() throws IOException {
|
||||
dbFile = File.createTempFile("test-", "db");
|
||||
store = new JdbcOffsetBackingStore();
|
||||
props = new HashMap<>();
|
||||
props.put(OFFSET_STORAGE_JDBC_URI.name(), "jdbc:sqlite:" + dbFile.getAbsolutePath());
|
||||
props.put(OFFSET_STORAGE_JDBC_URL.name(), "jdbc:sqlite:" + dbFile.getAbsolutePath());
|
||||
props.put(OFFSET_STORAGE_JDBC_USER.name(), "user");
|
||||
props.put(OFFSET_STORAGE_JDBC_PASSWORD.name(), "pass");
|
||||
props.put(OFFSET_STORAGE_TABLE_NAME.name(), "offsets_jdbc");
|
||||
|
@ -6,11 +6,11 @@
|
||||
package io.debezium.storage.jdbc.history;
|
||||
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_PASSWORD;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_URI;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_URL;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_JDBC_USER;
|
||||
import static io.debezium.storage.jdbc.JdbcOffsetBackingStore.OFFSET_STORAGE_TABLE_NAME;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_PASSWORD;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_URI;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_URL;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_USER;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
@ -112,7 +112,7 @@ private String topicName() {
|
||||
|
||||
protected Configuration.Builder schemaHistory(Configuration.Builder builder) {
|
||||
return builder
|
||||
.with(JDBC_URI, "jdbc:sqlite:" + SCHEMA_HISTORY_PATH)
|
||||
.with(JDBC_URL, "jdbc:sqlite:" + SCHEMA_HISTORY_PATH)
|
||||
.with(JDBC_USER, "user")
|
||||
.with(JDBC_PASSWORD, "pass");
|
||||
}
|
||||
@ -135,7 +135,7 @@ private Configuration.Builder config() throws IOException {
|
||||
.with(CommonConnectorConfig.TOPIC_PREFIX, TOPIC_PREFIX)
|
||||
.with(MySqlConnectorConfig.SNAPSHOT_MODE, SnapshotMode.INITIAL)
|
||||
.with(MySqlConnectorConfig.INCLUDE_SCHEMA_CHANGES, false)
|
||||
.with(OFFSET_STORAGE_JDBC_URI.name(), jdbcUri)
|
||||
.with(OFFSET_STORAGE_JDBC_URL.name(), jdbcUri)
|
||||
.with(OFFSET_STORAGE_JDBC_USER.name(), "user")
|
||||
.with(OFFSET_STORAGE_JDBC_PASSWORD.name(), "pass")
|
||||
.with(OFFSET_STORAGE_TABLE_NAME.name(), "offsets_jdbc");
|
||||
|
@ -6,7 +6,7 @@
|
||||
package io.debezium.storage.jdbc.history;
|
||||
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_PASSWORD;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_URI;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_URL;
|
||||
import static io.debezium.storage.jdbc.history.JdbcSchemaHistory.JDBC_USER;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
@ -100,7 +100,7 @@ public static void beforeClass() {
|
||||
public void beforeEach() {
|
||||
history = new JdbcSchemaHistory();
|
||||
history.configure(Configuration.create()
|
||||
.with(JDBC_URI, "jdbc:sqlite:" + dbFile)
|
||||
.with(JDBC_URL, "jdbc:sqlite:" + dbFile)
|
||||
.with(JDBC_USER, "user")
|
||||
.with(JDBC_PASSWORD, "pass")
|
||||
.build(), null, SchemaHistoryMetrics.NOOP, true);
|
||||
@ -138,7 +138,7 @@ public void shouldRecordChangesAndRecover() throws InterruptedException {
|
||||
// after restart, it should recover history correctly
|
||||
JdbcSchemaHistory history2 = new JdbcSchemaHistory();
|
||||
history2.configure(Configuration.create()
|
||||
.with(JDBC_URI, "jdbc:sqlite:" + dbFile)
|
||||
.with(JDBC_URL, "jdbc:sqlite:" + dbFile)
|
||||
.with(JDBC_USER, "user")
|
||||
.with(JDBC_PASSWORD, "pass")
|
||||
.build(), null, SchemaHistoryMetrics.NOOP, true);
|
||||
|
Loading…
Reference in New Issue
Block a user