diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java index 4481e52461..fc3ddf22ff 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/AbstractTbQueueConsumerTemplate.java +++ b/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 implements TbQueueConsumer { + public static final long ONE_MILLISECOND_IN_NANOS = TimeUnit.MILLISECONDS.toNanos(1); private volatile boolean subscribed; protected volatile boolean stopped = false; protected volatile Set partitions; @@ -83,7 +84,7 @@ public abstract class AbstractTbQueueConsumerTemplate 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 i List 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 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 {