bugfix serialize

Author: jiemotongxue

kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumerFactory.java

@@ -24,11 +24,13 @@
 import org.apache.flink.types.Row;
 import org.apache.kafka.common.requests.IsolationLevel;
 
+import java.io.Serializable;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
 /**
  * company: www.dtstack.com
+ *
  * @author: toutian
  * create: 2019/12/24
  */
@@ -38,9 +40,9 @@ public class KafkaConsumerFactory extends AbstractKafkaConsumerFactory {
     public FlinkKafkaConsumerBase<Row> createKafkaTableSource(KafkaSourceTableInfo kafkaSourceTableInfo, TypeInformation<Row> typeInformation, Properties props) {
         KafkaConsumer kafkaSrc = null;
         if (kafkaSourceTableInfo.getTopicIsPattern()) {
-            kafkaSrc = new KafkaConsumer(Pattern.compile(kafkaSourceTableInfo.getTopic()), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (subscriptionState, tp) -> subscriptionState.partitionLag(tp, IsolationLevel.READ_UNCOMMITTED)), props);
+            kafkaSrc = new KafkaConsumer(Pattern.compile(kafkaSourceTableInfo.getTopic()), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) (subscriptionState, tp) -> subscriptionState.partitionLag(tp, IsolationLevel.READ_UNCOMMITTED)), props);
         } else {
-            kafkaSrc = new KafkaConsumer(kafkaSourceTableInfo.getTopic(), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (subscriptionState, tp) -> subscriptionState.partitionLag(tp, IsolationLevel.READ_UNCOMMITTED)), props);
+            kafkaSrc = new KafkaConsumer(kafkaSourceTableInfo.getTopic(), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) (subscriptionState, tp) -> subscriptionState.partitionLag(tp, IsolationLevel.READ_UNCOMMITTED)), props);
         }
         return kafkaSrc;
     }

kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer09Factory.java

@@ -23,11 +23,13 @@
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
 import org.apache.flink.types.Row;
 
+import java.io.Serializable;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
 /**
  * company: www.dtstack.com
+ *
  * @author: toutian
  * create: 2019/12/24
  */
@@ -37,9 +39,9 @@ public class KafkaConsumer09Factory extends AbstractKafkaConsumerFactory {
     public FlinkKafkaConsumerBase<Row> createKafkaTableSource(KafkaSourceTableInfo kafkaSourceTableInfo, TypeInformation<Row> typeInformation, Properties props) {
         KafkaConsumer09 kafkaSrc = null;
         if (kafkaSourceTableInfo.getTopicIsPattern()) {
-            kafkaSrc = new KafkaConsumer09(Pattern.compile(kafkaSourceTableInfo.getTopic()), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (subscriptionState, tp) -> 0L), props);
+            kafkaSrc = new KafkaConsumer09(Pattern.compile(kafkaSourceTableInfo.getTopic()), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) (subscriptionState, tp) -> 0L), props);
         } else {
-            kafkaSrc = new KafkaConsumer09(kafkaSourceTableInfo.getTopic(), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (subscriptionState, tp) -> 0L), props);
+            kafkaSrc = new KafkaConsumer09(kafkaSourceTableInfo.getTopic(), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) (subscriptionState, tp) -> 0L), props);
         }
         return kafkaSrc;
     }

kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer010Factory.java

@@ -24,6 +24,7 @@
 import org.apache.flink.types.Row;
 import org.apache.kafka.clients.consumer.internals.SubscriptionState;
 
+import java.io.Serializable;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
@@ -38,9 +39,9 @@ public class KafkaConsumer010Factory extends AbstractKafkaConsumerFactory {
     public FlinkKafkaConsumerBase<Row> createKafkaTableSource(KafkaSourceTableInfo kafkaSourceTableInfo, TypeInformation<Row> typeInformation, Properties props) {
         KafkaConsumer010 kafkaSrc = null;
         if (kafkaSourceTableInfo.getTopicIsPattern()) {
-            kafkaSrc = new KafkaConsumer010(Pattern.compile(kafkaSourceTableInfo.getTopic()), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, SubscriptionState::partitionLag), props);
+            kafkaSrc = new KafkaConsumer010(Pattern.compile(kafkaSourceTableInfo.getTopic()), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) SubscriptionState::partitionLag), props);
         } else {
-            kafkaSrc = new KafkaConsumer010(kafkaSourceTableInfo.getTopic(), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, SubscriptionState::partitionLag), props);
+            kafkaSrc = new KafkaConsumer010(kafkaSourceTableInfo.getTopic(), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) SubscriptionState::partitionLag), props);
         }
         return kafkaSrc;
     }

kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer011Factory.java

@@ -24,6 +24,7 @@
 import org.apache.flink.types.Row;
 import org.apache.kafka.common.requests.IsolationLevel;
 
+import java.io.Serializable;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
@@ -38,9 +39,9 @@ public class KafkaConsumer011Factory extends AbstractKafkaConsumerFactory {
     public FlinkKafkaConsumerBase<Row> createKafkaTableSource(KafkaSourceTableInfo kafkaSourceTableInfo, TypeInformation<Row> typeInformation, Properties props) {
         KafkaConsumer011 kafkaSrc = null;
         if (kafkaSourceTableInfo.getTopicIsPattern()) {
-            kafkaSrc = new KafkaConsumer011(Pattern.compile(kafkaSourceTableInfo.getTopic()), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (subscriptionState, tp) -> subscriptionState.partitionLag(tp, IsolationLevel.READ_UNCOMMITTED)), props);
+            kafkaSrc = new KafkaConsumer011(Pattern.compile(kafkaSourceTableInfo.getTopic()), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) (subscriptionState, tp) -> subscriptionState.partitionLag(tp, IsolationLevel.READ_UNCOMMITTED)), props);
         } else {
-            kafkaSrc = new KafkaConsumer011(kafkaSourceTableInfo.getTopic(), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (subscriptionState, tp) -> subscriptionState.partitionLag(tp, IsolationLevel.READ_UNCOMMITTED)), props);
+            kafkaSrc = new KafkaConsumer011(kafkaSourceTableInfo.getTopic(), createDeserializationMetricWrapper(kafkaSourceTableInfo, typeInformation, (Calculate & Serializable) (subscriptionState, tp) -> subscriptionState.partitionLag(tp, IsolationLevel.READ_UNCOMMITTED)), props);
         }
         return kafkaSrc;
     }