diff --git a/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
index 51d3c8dcd..e1eec2815 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java
@@ -198,7 +198,6 @@ public class HoodieTableSource implements
   @Override
   public ChangelogMode getChangelogMode() {
     return conf.getBoolean(FlinkOptions.READ_AS_STREAMING)
-        && !conf.getBoolean(FlinkOptions.CHANGELOG_ENABLED)
         ? ChangelogModes.FULL
         // when all the changes are persisted or read as batch,
         // use INSERT mode.
diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/RowDataProjection.java b/hudi-flink/src/main/java/org/apache/hudi/util/RowDataProjection.java
index 52f2c58fc..048f286fa 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/util/RowDataProjection.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/util/RowDataProjection.java
@@ -25,10 +25,14 @@ import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.RowType;
 
+import java.io.Serializable;
+
 /**
  * Utilities to project the row data with given positions.
  */
-public class RowDataProjection {
+public class RowDataProjection implements Serializable {
+  private static final long serialVersionUID = 1L;
+
   private final RowData.FieldGetter[] fieldGetters;
 
   private RowDataProjection(LogicalType[] types, int[] positions) {