Skip to content

Commit d12e7a4

Browse files
committed
Ensure that end offset estimate can't precede consumed offset
1 parent 3a2cf7c commit d12e7a4

1 file changed

Lines changed: 1 addition & 1 deletion

File tree

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -641,7 +641,7 @@ public ProcessContinuation processElement(
641641

642642
final long estimatedBacklogBytes =
643643
(long)
644-
(BigDecimal.valueOf(latestOffsetEstimator.get())
644+
(BigDecimal.valueOf(Math.max(expectedOffset, latestOffsetEstimator.get()))
645645
.subtract(BigDecimal.valueOf(expectedOffset), MathContext.DECIMAL128)
646646
.doubleValue()
647647
* avgRecordSize.get());

0 commit comments

Comments
 (0)