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

DBZ-6092 Simplify the code; change logging

Browse Source
This commit is contained in:
Jiri Pechanec 2023-03-16 11:50:33 +01:00
parent cbb785d67a
commit d4c3e23767
2 changed files with 8 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
COPYRIGHT.txt
View File

@ -23,6 +23,7 @@ Alfusainey Jallow
Alisa Houskova
Amit Sela
Aman Garg
Anatolii Popov
Anders Engström
Andrea Cosentino
Andreas Bergmeier

14
debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java
View File

@ -295,12 +295,12 @@ private Lsn parseConfirmedFlushLsn(String slotName, String pluginName, String da
try {
confirmedFlushedLsn = tryParseLsn(slotName, pluginName, database, rs, "confirmed_flush_lsn");
if (confirmedFlushedLsn == null) {
LOGGER.debug("Failed to obtain valid replication slot, confirmed flush lsn is null");
LOGGER.info("Failed to obtain valid replication slot, confirmed flush lsn is null");
AtomicBoolean hasConcurrentTransaction = new AtomicBoolean(false);
int connectionPID = ((PgConnection) connection()).getBackendPID();
query("select * from pg_stat_activity where state like 'idle in transaction' AND pid <> " + connectionPID, rset -> {
if (rset.next()) {
hasConcurrentTransaction.compareAndSet(false, true);
hasConcurrentTransaction.set(true);
}
});
if (!hasConcurrentTransaction.get()) {
@ -321,8 +321,8 @@ private Lsn tryFallbackToRestartLsn(String slotName, String pluginName, String d
try {
confirmedFlushedLsn = tryParseLsn(slotName, pluginName, database, rs, "restart_lsn");
}
catch (SQLException e2) {
throw new ConnectException("Neither confirmed_flush_lsn nor restart_lsn could be found");
catch (SQLException e) {
throw new DebeziumException("Neither confirmed_flush_lsn nor restart_lsn could be found", e);
}
return confirmedFlushedLsn;
}
@ -333,7 +333,7 @@ private Lsn parseRestartLsn(String slotName, String pluginName, String database,
restartLsn = tryParseLsn(slotName, pluginName, database, rs, "restart_lsn");
}
catch (SQLException e) {
throw new ConnectException("restart_lsn could be found");
throw new DebeziumException("restart_lsn could be found");
}
return restartLsn;
@ -350,13 +350,13 @@ private Lsn tryParseLsn(String slotName, String pluginName, String database, Res
lsn = Lsn.valueOf(lsnStr);
}
catch (Exception e) {
throw new ConnectException("Value " + column + " in the pg_replication_slots table for slot = '"
throw new DebeziumException("Value " + column + " in the pg_replication_slots table for slot = '"
+ slotName + "', plugin = '"
+ pluginName + "', database = '"
+ database + "' is not valid. This is an abnormal situation and the database status should be checked.");
}
if (!lsn.isValid()) {
throw new ConnectException("Invalid LSN returned from database");
throw new DebeziumException("Invalid LSN returned from database");
}
return lsn;
}