Don't return negative offset lags except for -1 when -1001 is the high watermark (#406)

Author: emasab

CHANGELOG.md

@@ -12,7 +12,8 @@ v1.7.0 is a feature release. It is supported for all usage.
    client types (#399).
 4. Fix for at-least-once guarantee not ensured in case a seek happens on one
 partition and there are messages being fetched about other partitions (#393).
-
+5. Avoid returning a negative lag in case there is no cached offset for 
+   the HWM (#406).
 
 # confluent-kafka-javascript 1.6.0
 

lib/kafkajs/_consumer.js

@@ -919,12 +919,18 @@ class Consumer {
       this.#logger.warn(`Could not get watermark offsets for batch: ${e}`, this.#createConsumerBindingMessageMetadata());
     }
 
-    if (Number.isInteger(watermarkOffsets.highOffset)) {
+    /* Keep default values if it's not a real offset (OFFSET_INVALID: -1001). */
+    if (Number.isInteger(watermarkOffsets.highOffset) &&
+        watermarkOffsets.highOffset >= 0) {
       highWatermark = watermarkOffsets.highOffset.toString();
       /* While calculating lag, we subtract 1 from the high offset
        * for compatibility reasons with KafkaJS's API */
       offsetLag_ = (watermarkOffsets.highOffset - 1) - messages[messages.length - 1].offset;
       offsetLagLow_ = (watermarkOffsets.highOffset - 1) - messages[0].offset;
+
+      /* In any case don't return a negative lag but a zero lag instead. */
+      offsetLag_ = offsetLag_ > 0 ? offsetLag_ : 0;
+      offsetLagLow_ = offsetLagLow_ > 0 ? offsetLagLow_ : 0;
     }
 
     const messagesConverted = [];