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