dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TbSqlBlockingQueue failover improvement

Browse Source
This commit is contained in:
Sergey Matvienko 2023-12-07 17:27:34 +01:00
parent d08f0467df
commit 4c8cf69632
1 changed files with 11 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
dao/src/main/java/org/thingsboard/server/dao/sql/TbSqlBlockingQueue.java
View File

@ -54,7 +54,7 @@ public class TbSqlBlockingQueue<E> implements TbSqlQueue<E> {
String logName = params.getLogName(); String logName = params.getLogName();
int batchSize = params.getBatchSize(); int batchSize = params.getBatchSize();
long maxDelay = params.getMaxDelay(); long maxDelay = params.getMaxDelay();
List<TbSqlQueueElement<E>> entities = new ArrayList<>(batchSize); final List<TbSqlQueueElement<E>> entities = new ArrayList<>(batchSize);
while (!Thread.interrupted()) { while (!Thread.interrupted()) {
try { try {
long currentTs = System.currentTimeMillis(); long currentTs = System.currentTimeMillis();
@ -83,19 +83,23 @@ public class TbSqlBlockingQueue<E> implements TbSqlQueue<E> {
Thread.sleep(remainingDelay); Thread.sleep(remainingDelay);
} }
} }
} catch (Exception e) { } catch (Throwable t) {
log.error("[{}] Failed to save {} entities", logName, entities.size(), t);
try {
stats.incrementFailed(entities.size()); stats.incrementFailed(entities.size());
entities.forEach(entityFutureWrapper -> entityFutureWrapper.getFuture().setException(e)); entities.forEach(entityFutureWrapper -> entityFutureWrapper.getFuture().setException(t));
if (e instanceof InterruptedException) { } catch (Throwable th) {
log.error("[{}] Failed to set future exception", logName, th);
}
if (t instanceof InterruptedException) {
log.info("[{}] Queue polling was interrupted", logName); log.info("[{}] Queue polling was interrupted", logName);
break; break;
} else {
log.error("[{}] Failed to save {} entities", logName, entities.size(), e);
} }
} finally { } finally {
entities.clear(); entities.clear();
} }
} }
log.info("[{}] Queue polling completed", logName);
}); });
logExecutor.scheduleAtFixedRate(() -> { logExecutor.scheduleAtFixedRate(() -> {