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

DBZ-7313: close stream

Browse Source
This commit is contained in:
Jiri Kulhanek 2024-01-09 10:45:59 +01:00 committed by Chris Cranford
parent f7a06d0c96
commit 12a99f33f7
1 changed files with 10 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
debezium-connector-oracle/src/main/java/io/debezium/connector/oracle/logminer/processor/AbstractLogMinerEventProcessor.java
View File

@ -1426,9 +1426,11 @@ public void abandonTransactions(Duration retention) throws InterruptedException
if (entry.getValue().getStartScn().compareTo(thresholdScn) <= 0) { if (entry.getValue().getStartScn().compareTo(thresholdScn) <= 0) {
if (first) { if (first) {
LOGGER.warn("All transactions with SCN <= {} will be abandoned.", thresholdScn); LOGGER.warn("All transactions with SCN <= {} will be abandoned.", thresholdScn);
if(LOGGER.isDebugEnabled()) { if (LOGGER.isDebugEnabled()) {
LOGGER.debug("List of transactions in the cache before transactions being abandoned: [{}]", try (Stream s = getTransactionCache().keySet().stream()) {
getTransactionCache().keySet().stream().collect(Collectors.joining(","))); LOGGER.debug("List of transactions in the cache before transactions being abandoned: [{}]",
s.collect(Collectors.joining(",")));
}
} }
first = false; first = false;
} }
@ -1449,9 +1451,11 @@ public void abandonTransactions(Duration retention) throws InterruptedException
((CloseableIterator<Map.Entry<String, T>>) iterator).close(); ((CloseableIterator<Map.Entry<String, T>>) iterator).close();
} }
} }
if(LOGGER.isDebugEnabled()) { if (LOGGER.isDebugEnabled()) {
LOGGER.debug("List of transactions in the cache after transactions being abandoned: [{}]", try (Stream s = getTransactionCache().keySet().stream()) {
getTransactionCache().keySet().stream().collect(Collectors.joining(","))); LOGGER.debug("List of transactions in the cache after transactions being abandoned: [{}]",
s.collect(Collectors.joining(",")));
}
} }
// Update the oldest scn metric are transaction abandonment // Update the oldest scn metric are transaction abandonment