Cassandra: AbstractBufferedRateExecutor improved logging to clarify whether the Read or Write operations logged

dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java

@@ -124,7 +124,7 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
         if (tenantProfileConfiguration != null &&
                 StringUtils.isNotEmpty(tenantProfileConfiguration.getCassandraQueryTenantRateLimitsConfiguration())) {
             if (task.getTenantId() == null) {
-                log.info("Invalid task received: {}", task);
+                log.info("[{}] Invalid task received: {}", getBufferName(), task);
             } else if (!task.getTenantId().isNullUid()) {
                 TbRateLimits rateLimits = perTenantLimits.computeIfAbsent(
                         task.getTenantId(), id -> new TbRateLimits(tenantProfileConfiguration.getCassandraQueryTenantRateLimitsConfiguration())
@@ -173,7 +173,7 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
     public abstract String getBufferName();
 
     private void dispatch() {
-        log.info("Buffered rate executor thread started");
+        log.info("[{}] Buffered rate executor thread started", getBufferName());
         while (!Thread.interrupted()) {
             int curLvl = concurrencyLevel.get();
             AsyncTaskContext<T, V> taskCtx = null;
@@ -185,7 +185,7 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
                         if (printQueriesIdx.incrementAndGet() >= printQueriesFreq) {
                             printQueriesIdx.set(0);
                             String query = queryToString(finalTaskCtx);
-                            log.info("[{}] Cassandra query: {}", taskCtx.getId(), query);
+                            log.info("[{}][{}] Cassandra query: {}", getBufferName(), taskCtx.getId(), query);
                         }
                     }
                     logTask("Processing", finalTaskCtx);
@@ -238,7 +238,7 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
                 }
             }
         }
-        log.info("Buffered rate executor thread stopped");
+        log.info("[{}] Buffered rate executor thread stopped", getBufferName());
     }
 
     private void logTask(String action, AsyncTaskContext<T, V> taskCtx) {
@@ -314,7 +314,7 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
             statsBuilder.append(CONCURRENCY_LEVEL).append(" = [").append(concurrencyLevel.get()).append("] ");
 
             stats.getStatsCounters().forEach(StatsCounter::clear);
-            log.info("Permits {}", statsBuilder);
+            log.info("[{}] Permits {}", getBufferName(), statsBuilder);
         }
 
         stats.getRateLimitedTenants().entrySet().stream()
@@ -330,13 +330,13 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
                             try {
                                 return entityService.fetchEntityName(TenantId.SYS_TENANT_ID, tenantId).orElse(defaultName);
                             } catch (Exception e) {
-                                log.error("[{}] Failed to get tenant name", tenantId, e);
+                                log.error("[{}][{}] Failed to get tenant name", getBufferName(), tenantId, e);
                                 return defaultName;
                             }
                         });
-                        log.info("[{}][{}] Rate limited requests: {}", tenantId, name, rateLimitedRequests);
+                        log.info("[{}][{}][{}] Rate limited requests: {}", getBufferName(), tenantId, name, rateLimitedRequests);
                     } else {
-                        log.info("[{}] Rate limited requests: {}", tenantId, rateLimitedRequests);
+                        log.info("[{}][{}] Rate limited requests: {}", getBufferName(), tenantId, rateLimitedRequests);
                     }
                 });
     }