Ensure that end offset estimate can't precede consumed offset

Author: sjvanrossum

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

@@ -641,7 +641,7 @@ public ProcessContinuation processElement(
 
         final long estimatedBacklogBytes =
             (long)
-                (BigDecimal.valueOf(latestOffsetEstimator.get())
+                (BigDecimal.valueOf(Math.max(expectedOffset, latestOffsetEstimator.get()))
                         .subtract(BigDecimal.valueOf(expectedOffset), MathContext.DECIMAL128)
                         .doubleValue()
                     * avgRecordSize.get());