set lag to 0 if no messages consumed (opensearch-project#19393)

Signed-off-by: Varun Bharadwaj <varunbharadwaj1995@gmail.com>

Author: varunbharadwaj

CHANGELOG.md

@@ -63,6 +63,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Flaky Test] Fix flaky test in SecureReactorNetty4HttpServerTransportTests with reproducible seed ([#19327](https://github.com/opensearch-project/OpenSearch/pull/19327))
 - Remove unnecessary looping in field data cache clear ([#19116](https://github.com/opensearch-project/OpenSearch/pull/19116))
 - [Flaky Test] Fix flaky test IngestFromKinesisIT.testAllActiveIngestion ([#19380](https://github.com/opensearch-project/OpenSearch/pull/19380))
+- Fix lag metric for pull-based ingestion when streaming source is empty ([#19393](https://github.com/opensearch-project/OpenSearch/pull/19393))
 
 ### Dependencies
 - Update to Gradle 9.1.0 ([#19329](https://github.com/opensearch-project/OpenSearch/pull/19329))

server/src/main/java/org/opensearch/indices/pollingingest/DefaultStreamPoller.java

@@ -232,6 +232,8 @@ protected void startPoll() {
 
                 if (results.isEmpty()) {
                     // no new records
+                    setLastPolledMessageTimestamp(0);
+                    Thread.sleep(DEFAULT_POLLER_SLEEP_PERIOD_MS);
                     continue;
                 }
 
@@ -267,7 +269,7 @@ private IngestionShardPointer processRecords(
                 }
                 totalPolledCount.inc();
                 blockingQueueContainer.add(result);
-                lastPolledMessageTimestamp = result.getMessage().getTimestamp() == null ? 0 : result.getMessage().getTimestamp();
+                setLastPolledMessageTimestamp(result.getMessage().getTimestamp() == null ? 0 : result.getMessage().getTimestamp());
                 logger.debug(
                     "Put message {} with pointer {} to the blocking queue",
                     String.valueOf(result.getMessage().getPayload()),
@@ -403,9 +405,19 @@ public PollingIngestStats getStats() {
      * Returns the lag in milliseconds since the last polled message
      */
     private long computeLag() {
+        if (lastPolledMessageTimestamp == 0 || paused) {
+            return 0;
+        }
+
         return System.currentTimeMillis() - lastPolledMessageTimestamp;
     }
 
+    private void setLastPolledMessageTimestamp(long timestamp) {
+        if (lastPolledMessageTimestamp != timestamp) {
+            lastPolledMessageTimestamp = timestamp;
+        }
+    }
+
     public State getState() {
         return this.state;
     }