Add log to track negtive unacked msg.
@@ -138,6 +138,8 @@ public class Consumer {
@Setter
private volatile long consumerEpoch;
+ private long negtiveUnackedMsgsTimestamp;
+
public Consumer(Subscription subscription, SubType subType, String topicName, long consumerId,
int priorityLevel, String consumerName,
boolean isDurable, TransportCnx cnx, String appId,
@@ -952,6 +954,10 @@ public class Consumer {
subscription.addUnAckedMessages(ackedMessages);
unackedMsgs = UNACKED_MESSAGES_UPDATER.addAndGet(consumer, ackedMessages);
}
+ if (unackedMsgs < 0 && System.currentTimeMillis() - negtiveUnackedMsgsTimestamp >= 10_000) {
+ negtiveUnackedMsgsTimestamp = System.currentTimeMillis();
+ log.warn("unackedMsgs is : {}, ackedMessages : {}, consumer : {}", unackedMsgs, ackedMessages, consumer);
+ }
return unackedMsgs;
}
