Report unknown backlog size when latest offset lags behind next offset (#35086)

sjvanrossum · web-flow · commit 4f9d560a9ffd · 2025-07-11T16:29:49.000-04:00
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
@@ -494,11 +494,10 @@ synchronized long approxBacklogInBytes() {
     }
 
     synchronized long backlogMessageCount() {
-      if (latestOffset < 0 || nextOffset < 0) {
+      if (latestOffset < 0 || nextOffset < 0 || latestOffset < nextOffset) {
         return UnboundedReader.BACKLOG_UNKNOWN;
       }
-      double remaining = latestOffset - nextOffset;
-      return Math.max(0, (long) Math.ceil(remaining));
+      return latestOffset - nextOffset;
     }
 
     synchronized TimestampPolicyContext mkTimestampPolicyContext() {

Original file line number	Diff line number	Diff line change
`@@ -494,11 +494,10 @@ synchronized long approxBacklogInBytes() {`
`494`	`494`	`}`
`495`	`495`
`496`	`496`	`synchronized long backlogMessageCount() {`
`497`		`- if (latestOffset < 0 \|\| nextOffset < 0) {`
	`497`	`+ if (latestOffset < 0 \|\| nextOffset < 0 \|\| latestOffset < nextOffset) {`
`498`	`498`	`return UnboundedReader.BACKLOG_UNKNOWN;`
`499`	`499`	`}`
`500`		`- double remaining = latestOffset - nextOffset;`
`501`		`- return Math.max(0, (long) Math.ceil(remaining));`
	`500`	`+ return latestOffset - nextOffset;`
`502`	`501`	`}`
`503`	`502`
`504`	`503`	`synchronized TimestampPolicyContext mkTimestampPolicyContext() {`