queue consumer: not going to sleep after pull if time left less then 1 millisecond. topic added to logs

common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java

@@ -38,6 +38,7 @@ import static java.util.Collections.emptyList;
 @Slf4j
 public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> implements TbQueueConsumer<T> {
 
+    public static final long ONE_MILLISECOND_IN_NANOS = TimeUnit.MILLISECONDS.toNanos(1);
     private volatile boolean subscribed;
     protected volatile boolean stopped = false;
     protected volatile Set<TopicPartitionInfo> partitions;
@@ -83,7 +84,7 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
         }
 
         if (consumerLock.isLocked()) {
-            log.error("poll. consumerLock is locked. will wait with no timeout. it looks like a race conditions or deadlock", new RuntimeException("stacktrace"));
+            log.error("poll. consumerLock is locked. will wait with no timeout. it looks like a race conditions or deadlock topic " + topic, new RuntimeException("stacktrace"));
         }
 
         consumerLock.lock();
@@ -131,9 +132,12 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
     List<T> sleepAndReturnEmpty(final long startNanos, final long durationInMillis) {
         long durationNanos = TimeUnit.MILLISECONDS.toNanos(durationInMillis);
         long spentNanos = System.nanoTime() - startNanos;
-        if (spentNanos < durationNanos) {
+        long nanosLeft = durationNanos - spentNanos;
+        if (nanosLeft >= ONE_MILLISECOND_IN_NANOS) {
             try {
-                Thread.sleep(Math.max(TimeUnit.NANOSECONDS.toMillis(durationNanos - spentNanos), 1));
+                long sleepMs = TimeUnit.NANOSECONDS.toMillis(nanosLeft);
+                log.trace("Going to sleep after poll: topic {} for {}ms", topic, sleepMs);
+                Thread.sleep(sleepMs);
             } catch (InterruptedException e) {
                 if (!stopped) {
                     log.error("Failed to wait", e);
@@ -146,7 +150,7 @@ public abstract class AbstractTbQueueConsumerTemplate<R, T extends TbQueueMsg> i
     @Override
     public void commit() {
         if (consumerLock.isLocked()) {
-            log.error("commit. consumerLock is locked. will wait with no timeout. it looks like a race conditions or deadlock", new RuntimeException("stacktrace"));
+            log.error("commit. consumerLock is locked. will wait with no timeout. it looks like a race conditions or deadlock topic " + topic, new RuntimeException("stacktrace"));
         }
         consumerLock.lock();
         try {