Try to remove unnecessary change.

Author: lvyanquan

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceRecordEmitter.java

@@ -36,7 +36,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.Serializable;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
@@ -57,17 +56,17 @@
  * emit records rather than emit the records directly.
  */
 public class IncrementalSourceRecordEmitter<T>
-        implements RecordEmitter<SourceRecords, T, SourceSplitState>, Serializable {
+        implements RecordEmitter<SourceRecords, T, SourceSplitState> {
 
     private static final Logger LOG = LoggerFactory.getLogger(IncrementalSourceRecordEmitter.class);
     private static final FlinkJsonTableChangeSerializer TABLE_CHANGE_SERIALIZER =
             new FlinkJsonTableChangeSerializer();
 
-    protected DebeziumDeserializationSchema<T> debeziumDeserializationSchema;
-    protected SourceReaderMetrics sourceReaderMetrics;
-    protected boolean includeSchemaChanges;
-    protected OutputCollector<T> outputCollector;
-    protected OffsetFactory offsetFactory;
+    protected final DebeziumDeserializationSchema<T> debeziumDeserializationSchema;
+    protected final SourceReaderMetrics sourceReaderMetrics;
+    protected final boolean includeSchemaChanges;
+    protected final OutputCollector<T> outputCollector;
+    protected final OffsetFactory offsetFactory;
 
     public IncrementalSourceRecordEmitter(
             DebeziumDeserializationSchema<T> debeziumDeserializationSchema,
@@ -153,10 +152,6 @@ public Offset getOffsetPosition(Map<String, ?> offset) {
         return offsetFactory.newOffset(offsetStrMap);
     }
 
-    public void setSourceReaderMetrics(SourceReaderMetrics sourceReaderMetrics) {
-        this.sourceReaderMetrics = sourceReaderMetrics;
-    }
-
     protected void emitElement(SourceRecord element, SourceOutput<T> output) throws Exception {
         sourceReaderMetrics.markRecord();
         sourceReaderMetrics.updateRecordCounters(element);
@@ -191,7 +186,7 @@ protected void reportMetrics(SourceRecord element) {
     }
 
     /** An adapter between {@link SourceOutput} and {@link Collector}. */
-    protected static class OutputCollector<T> implements Collector<T>, Serializable {
+    protected static class OutputCollector<T> implements Collector<T> {
         public SourceOutput<T> output;
         public Long currentMessageTimestamp;