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

DBZ-7258 Added try-wit-resources; changed connection check method

Browse Source
This commit is contained in:
Jiri Pechanec 2024-03-25 18:28:26 +01:00
parent 54e6c855a2
commit 15e795547d
3 changed files with 70 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
debezium-storage/debezium-storage-jdbc/src/main/java/io/debezium/storage/jdbc/RetriableConnection.java
View File

@ -64,7 +64,7 @@ private void createConnection() throws SQLException {
@Override
public void close() throws SQLException {
if (isConnectionCreated()) {
if (isOpen()) {
try {
conn.close();
}
@ -75,8 +75,15 @@ public void close() throws SQLException {
conn = null;
}
public boolean isConnectionCreated() {
return conn != null;
public boolean isOpen() {
try {
return conn != null && !conn.isClosed();
}
catch (SQLException e) {
LOGGER.warn("Exception while checking connection", e);
conn = null;
}
return false;
}
/**
@ -107,7 +114,7 @@ private synchronized <T> T executeWithRetry(ConnectionFunction<T> func, Connecti
throws SQLException {
int attempt = 1;
while (true) {
if (!isConnectionCreated()) {
if (!isOpen()) {
LOGGER.debug("Trying to reconnect (attempt {}).", attempt);
try {
createConnection();

27
debezium-storage/debezium-storage-jdbc/src/main/java/io/debezium/storage/jdbc/history/JdbcSchemaHistory.java
View File

@ -79,7 +79,7 @@ public void start() {
super.start();
lock.write(() -> {
if (running.compareAndSet(false, true)) {
if (!conn.isConnectionCreated()) {
if (!conn.isOpen()) {
throw new IllegalStateException("Database connection must be set before it is started");
}
try {
@ -117,7 +117,7 @@ protected void storeRecord(HistoryRecord record) throws SchemaHistoryException {
List<String> substrings = split(line, 65000);
int partSeq = 0;
for (String dataPart : substrings) {
PreparedStatement sql = conn.prepareStatement(config.getTableInsert());
try (PreparedStatement sql = conn.prepareStatement(config.getTableInsert())) {
sql.setString(1, UUID.randomUUID().toString());
sql.setString(2, dataPart);
sql.setInt(3, partSeq);
@ -126,6 +126,7 @@ protected void storeRecord(HistoryRecord record) throws SchemaHistoryException {
sql.executeUpdate();
partSeq++;
}
}
conn.commit();
}, "store history record", true);
}
@ -163,9 +164,9 @@ protected synchronized void recoverRecords(Consumer<HistoryRecord> records) {
try {
if (exists()) {
conn.executeWithRetry(conn -> {
try (
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery(config.getTableSelect());
ResultSet rs = stmt.executeQuery(config.getTableSelect())) {
while (rs.next()) {
String historyData = rs.getString("history_data");
@ -178,6 +179,7 @@ protected synchronized void recoverRecords(Consumer<HistoryRecord> records) {
}
}
}
}
}, "recover history records", false);
}
else {
@ -198,12 +200,13 @@ public boolean storageExists() {
DatabaseMetaData dbMeta = conn.getMetaData();
String databaseName = config.getDatabaseName();
ResultSet tableExists = dbMeta.getTables(databaseName,
null, config.getTableName(), null);
try (ResultSet tableExists = dbMeta.getTables(databaseName,
null, config.getTableName(), null)) {
if (tableExists.next()) {
exists = true;
}
return exists;
}
}, "history storage exists", false);
}
catch (SQLException e) {
@ -221,12 +224,14 @@ public boolean exists() {
try {
return conn.executeWithRetry(conn -> {
boolean isExists = false;
try (
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery(config.getTableDataExistsSelect());
ResultSet rs = stmt.executeQuery(config.getTableDataExistsSelect());) {
while (rs.next()) {
isExists = true;
}
return isExists;
}
}, "history records exist check", false);
}
catch (SQLException e) {
@ -249,14 +254,16 @@ public void initializeStorage() {
try {
conn.executeWithRetry(conn -> {
DatabaseMetaData dbMeta = conn.getMetaData();
ResultSet tableExists = dbMeta.getTables(null, null, config.getTableName(), null);
try (ResultSet tableExists = dbMeta.getTables(null, null, config.getTableName(), null)) {
if (tableExists.next()) {
return;
}
LOG.info("Creating table {} to store database history", config.getTableName());
conn.prepareStatement(config.getTableCreate()).execute();
try (var ps = conn.prepareStatement(config.getTableCreate())) {
ps.execute();
LOG.info("Created table in given database...");
}
}
}, "initialize storage", false);
}
catch (SQLException e) {

12
debezium-storage/debezium-storage-jdbc/src/main/java/io/debezium/storage/jdbc/offset/JdbcOffsetBackingStore.java
View File

@ -96,13 +96,15 @@ public synchronized void start() {
private void initializeTable() throws SQLException {
conn.executeWithRetry(conn -> {
DatabaseMetaData dbMeta = conn.getMetaData();
ResultSet tableExists = dbMeta.getTables(null, null, config.getTableName(), null);
try (ResultSet tableExists = dbMeta.getTables(null, null, config.getTableName(), null)) {
if (tableExists.next()) {
return;
}
}
LOGGER.info("Creating table {} to store offset", config.getTableName());
conn.prepareStatement(config.getTableCreate()).execute();
try (var ps = conn.prepareStatement(config.getTableCreate())) {
ps.execute();
}
}, "checking / creating table", false);
}
@ -139,13 +141,15 @@ private void load() {
try {
ConcurrentHashMap<String, String> tmpData = new ConcurrentHashMap<>();
conn.executeWithRetry(conn -> {
try (
Statement stmt = conn.createStatement();
ResultSet rs = stmt.executeQuery(config.getTableSelect());
ResultSet rs = stmt.executeQuery(config.getTableSelect())) {
while (rs.next()) {
String key = rs.getString("offset_key");
String val = rs.getString("offset_val");
tmpData.put(key, val);
}
}
data = tmpData;
}, "loading offset data", false);
}