diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
index c49ab1ba9f..6a71862b28 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
@@ -98,7 +98,11 @@ public ScanRecords collectFetch(final LogFetchBuffer logFetchBuffer) {
 
                     if (!completedFetch.isInitialized()) {
                         try {
-                            logFetchBuffer.setNextInLineFetch(initialize(completedFetch));
+                            CompletedFetch initialized = initialize(completedFetch);
+                            logFetchBuffer.setNextInLineFetch(initialized);
+                            if (initialized == null) {
+                                completedFetch.drain();
+                            }
                         } catch (Exception e) {
                             // Remove a completedFetch upon a parse with exception if
                             // (1) it contains no records, and