Merge pull request #7456 from ViacheslavKlimov/fix/sql-deadlock-info

[3.4.2] SQL batch sort enabled by default; additional deadlock logging

application/src/main/resources/thingsboard.yml

@@ -268,7 +268,7 @@ sql:
   audit_logs:
     partition_size: "${SQL_AUDIT_LOGS_PARTITION_SIZE_HOURS:168}" # Default value - 1 week
   # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks
-  batch_sort: "${SQL_BATCH_SORT:false}"
+  batch_sort: "${SQL_BATCH_SORT:true}"
   # Specify whether to remove null characters from strValue of attributes and timeseries before insert
   remove_null_chars: "${SQL_REMOVE_NULL_CHARS:true}"
   # Specify whether to log database queries and their parameters generated by entity query repository

common/data/src/main/java/org/thingsboard/server/common/data/StringUtils.java

@@ -146,6 +146,16 @@ public class StringUtils {
         return org.apache.commons.lang3.StringUtils.substringAfterLast(str, sep);
     }
 
+    public static boolean containedByAny(String searchString, String... strings) {
+        if (searchString == null) return false;
+        for (String string : strings) {
+            if (string != null && string.contains(searchString)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     public static String randomNumeric(int length) {
         return RandomStringUtils.randomNumeric(length);
     }

dao/src/main/java/org/thingsboard/server/dao/aspect/SqlDaoCallsAspect.java

@@ -21,7 +21,6 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.ArrayUtils;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.aspectj.lang.ProceedingJoinPoint;
 import org.aspectj.lang.annotation.Around;
@@ -29,9 +28,11 @@ import org.aspectj.lang.annotation.Aspect;
 import org.aspectj.lang.reflect.MethodSignature;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.hibernate.exception.JDBCConnectionException;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.id.TenantId;
 
 import java.util.Arrays;
@@ -46,6 +47,8 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
+import static org.apache.commons.lang3.StringUtils.join;
+
 @Aspect
 @ConditionalOnProperty(prefix = "sql", value = "log_tenant_stats", havingValue = "true")
 @Component
@@ -55,6 +58,12 @@ public class SqlDaoCallsAspect {
     private final Set<String> invalidTenantDbCallMethods = ConcurrentHashMap.newKeySet();
     private final ConcurrentMap<TenantId, DbCallStats> statsMap = new ConcurrentHashMap<>();
 
+    @Value("${sql.batch_sort:true}")
+    private boolean batchSortEnabled;
+
+    private static final String DEADLOCK_DETECTED_ERROR = "deadlock detected";
+
+
     @Scheduled(initialDelayString = "${sql.log_tenant_stats_interval_ms:60000}",
             fixedDelayString = "${sql.log_tenant_stats_interval_ms:60000}")
     public void printStats() {
@@ -135,7 +144,7 @@ public class SqlDaoCallsAspect {
 
     @SuppressWarnings({"rawtypes", "unchecked"})
     @Around("@within(org.thingsboard.server.dao.util.SqlDao)")
-    public Object logExecutionTime(ProceedingJoinPoint joinPoint) throws Throwable {
+    public Object handleSqlCall(ProceedingJoinPoint joinPoint) throws Throwable {
         MethodSignature signature = (MethodSignature) joinPoint.getSignature();
         var methodName = signature.toShortString();
         if (invalidTenantDbCallMethods.contains(methodName)) {
@@ -155,29 +164,47 @@ public class SqlDaoCallsAspect {
                         new FutureCallback<>() {
                             @Override
                             public void onSuccess(@Nullable Object result) {
-                                logTenantMethodExecution(tenantId, methodName, true, startTime, null);
+                                reportSuccessfulMethodExecution(tenantId, methodName, startTime);
                             }
 
                             @Override
                             public void onFailure(Throwable t) {
-                                logTenantMethodExecution(tenantId, methodName, false, startTime, t);
+                                reportFailedMethodExecution(tenantId, methodName, startTime, t, joinPoint);
                             }
                         },
                         MoreExecutors.directExecutor());
             } else {
-                logTenantMethodExecution(tenantId, methodName, true, startTime, null);
+                reportSuccessfulMethodExecution(tenantId, methodName, startTime);
             }
             return result;
         } catch (Throwable t) {
-            logTenantMethodExecution(tenantId, methodName, false, startTime, t);
+            reportFailedMethodExecution(tenantId, methodName, startTime, t, joinPoint);
             throw t;
         }
     }
 
-    private void logTenantMethodExecution(TenantId tenantId, String method, boolean success, long startTime, Throwable t) {
-        if (!success && ExceptionUtils.indexOfThrowable(t, JDBCConnectionException.class) >= 0) {
-            return;
+    private void reportFailedMethodExecution(TenantId tenantId, String method, long startTime, Throwable t, ProceedingJoinPoint joinPoint) {
+        if (t != null) {
+            if (ExceptionUtils.indexOfThrowable(t, JDBCConnectionException.class) >= 0) {
+                return;
+            }
+            if (StringUtils.containedByAny(DEADLOCK_DETECTED_ERROR, ExceptionUtils.getRootCauseMessage(t), ExceptionUtils.getMessage(t))) {
+                if (!batchSortEnabled) {
+                    log.warn("Deadlock was detected for method {} (tenant: {}). You might need to enable 'sql.batch_sort' option.", method, tenantId);
+                } else {
+                    log.error("Deadlock was detected for method {} (tenant: {}). Arguments passed: \n{}\n The error: ",
+                            method, tenantId, join(joinPoint.getArgs(), System.lineSeparator()), t);
+                }
+            }
         }
+        reportMethodExecution(tenantId, method, false, startTime);
+    }
+
+    private void reportSuccessfulMethodExecution(TenantId tenantId, String method, long startTime) {
+        reportMethodExecution(tenantId, method, true, startTime);
+    }
+
+    private void reportMethodExecution(TenantId tenantId, String method, boolean success, long startTime) {
         statsMap.computeIfAbsent(tenantId, DbCallStats::new)
                 .onMethodCall(method, success, System.currentTimeMillis() - startTime);
     }

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java

@@ -25,6 +25,7 @@ import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.support.TransactionCallbackWithoutResult;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.thingsboard.server.dao.model.sql.AttributeKvEntity;
+import org.thingsboard.server.dao.util.SqlDao;
 
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
@@ -35,6 +36,7 @@ import java.util.regex.Pattern;
 
 @Repository
 @Slf4j
+@SqlDao
 public abstract class AttributeKvInsertRepository {
 
     private static final ThreadLocal<Pattern> PATTERN_THREAD_LOCAL = ThreadLocal.withInitial(() -> Pattern.compile(String.valueOf(Character.MIN_VALUE)));
@@ -58,7 +60,7 @@ public abstract class AttributeKvInsertRepository {
     @Value("${sql.remove_null_chars:true}")
     private boolean removeNullChars;
 
-    protected void saveOrUpdate(List<AttributeKvEntity> entities) {
+    public void saveOrUpdate(List<AttributeKvEntity> entities) {
         transactionTemplate.execute(new TransactionCallbackWithoutResult() {
             @Override
             protected void doInTransactionWithoutResult(TransactionStatus status) {

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java

@@ -78,7 +78,7 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl
     @Value("${sql.attributes.batch_threads:4}")
     private int batchThreads;
 
-    @Value("${sql.batch_sort:false}")
+    @Value("${sql.batch_sort:true}")
     private boolean batchSortEnabled;
 
     private TbSqlBlockingQueueWrapper<AttributeKvEntity> queue;

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/SqlAttributesInsertRepository.java

@@ -17,9 +17,11 @@ package org.thingsboard.server.dao.sql.attributes;
 
 import org.springframework.stereotype.Repository;
 import org.springframework.transaction.annotation.Transactional;
+import org.thingsboard.server.dao.util.SqlDao;
 
 @Repository
 @Transactional
+@SqlDao
 public class SqlAttributesInsertRepository extends AttributeKvInsertRepository {
 
 }

dao/src/main/java/org/thingsboard/server/dao/sql/event/EventInsertRepository.java

@@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.event.LifecycleEvent;
 import org.thingsboard.server.common.data.event.RuleChainDebugEvent;
 import org.thingsboard.server.common.data.event.RuleNodeDebugEvent;
 import org.thingsboard.server.common.data.event.StatisticsEvent;
+import org.thingsboard.server.dao.util.SqlDao;
 
 import javax.annotation.PostConstruct;
 import java.sql.PreparedStatement;
@@ -45,6 +46,7 @@ import java.util.stream.Collectors;
 
 @Repository
 @Transactional
+@SqlDao
 public class EventInsertRepository {
 
     private static final ThreadLocal<Pattern> PATTERN_THREAD_LOCAL = ThreadLocal.withInitial(() -> Pattern.compile(String.valueOf(Character.MIN_VALUE)));
@@ -81,7 +83,7 @@ public class EventInsertRepository {
                 "VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING;");
     }
 
-    protected void save(List<Event> entities) {
+    public void save(List<Event> entities) {
         Map<EventType, List<Event>> eventsByType = entities.stream().collect(Collectors.groupingBy(Event::getType, Collectors.toList()));
         transactionTemplate.execute(new TransactionCallbackWithoutResult() {
             @Override

dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java

@@ -106,7 +106,7 @@ public class JpaBaseEventDao implements EventDao {
     @Value("${sql.events.batch_threads:3}")
     private int batchThreads;
 
-    @Value("${sql.batch_sort:false}")
+    @Value("${sql.batch_sort:true}")
     private boolean batchSortEnabled;
 
     private TbSqlBlockingQueueWrapper<Event> queue;

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java

@@ -63,7 +63,7 @@ public abstract class AbstractSqlTimeseriesDao extends BaseAbstractSqlTimeseries
     @Value("${sql.timescale.batch_threads:4}")
     protected int timescaleBatchThreads;
 
-    @Value("${sql.batch_sort:false}")
+    @Value("${sql.batch_sort:true}")
     protected boolean batchSortEnabled;
 
     @Value("${sql.ttl.ts.ts_key_value_ttl:0}")

dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java

@@ -94,7 +94,7 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme
     @Value("${sql.ts_latest.batch_threads:4}")
     private int tsLatestBatchThreads;
 
-    @Value("${sql.batch_sort:false}")
+    @Value("${sql.batch_sort:true}")
     protected boolean batchSortEnabled;
 
     @Autowired

dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/latest/sql/SqlLatestInsertTsRepository.java

@@ -24,6 +24,7 @@ import org.springframework.transaction.support.TransactionCallbackWithoutResult;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
 import org.thingsboard.server.dao.sqlts.insert.AbstractInsertRepository;
 import org.thingsboard.server.dao.sqlts.insert.latest.InsertLatestTsRepository;
+import org.thingsboard.server.dao.util.SqlDao;
 import org.thingsboard.server.dao.util.SqlTsLatestAnyDao;
 
 import java.sql.PreparedStatement;
@@ -36,6 +37,7 @@ import java.util.List;
 @SqlTsLatestAnyDao
 @Repository
 @Transactional
+@SqlDao
 public class SqlLatestInsertTsRepository extends AbstractInsertRepository implements InsertLatestTsRepository {
 
     @Value("${sql.ts_latest.update_by_latest_ts:true}")