diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java
index 684534434751..ce24893b38b0 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java
@@ -19,8 +19,6 @@
 
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
-import java.math.BigDecimal;
-import java.math.MathContext;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
@@ -65,6 +63,7 @@
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.Closeables;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.UnsignedLong;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -316,7 +315,7 @@ public Consumer<byte[], byte[]> load(
   private transient @MonotonicNonNull LoadingCache<KafkaSourceDescriptor, Consumer<byte[], byte[]>>
       pollConsumerCache;
 
-  // Valid between bundle start and bundle finish.
+  // Initialized in setup() and closed in teardown().
   private transient @Nullable Deserializer<K> keyDeserializerInstance = null;
   private transient @Nullable Deserializer<V> valueDeserializerInstance = null;
   private static final long DEFAULT_KAFKA_POLL_TIMEOUT = 2L;
@@ -447,9 +446,8 @@ public UnsplittableRestrictionTracker<OffsetRange, Long> restrictionTracker(
       return new UnsplittableRestrictionTracker<>(new OffsetRangeTracker(restriction));
     }
 
-    // OffsetEstimators are cached for each topic-partition because they hold a stateful connection,
-    // so we want to minimize the amount of connections that we start and track with Kafka. Another
-    // point is that it has a memoized backlog, and this should make that more reusable estimations.
+    // Offset estimators are cached per topic-partition so that the latest known end offset updated
+    // by processElement() can be reused across restrictionTracker() and getSize() calls.
     final AtomicLong latestOffsetEstimator =
         latestOffsetEstimatorCache.getUnchecked(kafkaSourceDescriptor);
     return new UnsplittableRestrictionTracker<>(
@@ -641,8 +639,8 @@ public ProcessContinuation processElement(
 
         final long estimatedBacklogBytes =
             (long)
-                (BigDecimal.valueOf(Math.max(expectedOffset, latestOffsetEstimator.get()))
-                        .subtract(BigDecimal.valueOf(expectedOffset), MathContext.DECIMAL128)
+                (UnsignedLong.fromLongBits(
+                            Math.max(expectedOffset, latestOffsetEstimator.get()) - expectedOffset)
                         .doubleValue()
                     * avgRecordSize.get());
         backlogBytes.set(estimatedBacklogBytes);
@@ -721,7 +719,7 @@ public void teardown() throws Exception {
         keyDeserializerInstance = null;
       }
     } catch (Exception anyException) {
-      LOG.warn("Fail to close resource during finishing bundle.", anyException);
+      LOG.warn("Failed to close resource during teardown.", anyException);
     }
     if (checkStopReadingFn != null) {
       checkStopReadingFn.teardown();