fix bug for kafka offset

Author: todd5167

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

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.source.kafka.table;
 
@@ -46,6 +46,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         kafka09SourceTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSourceTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
         kafka09SourceTableInfo.setGroupId(MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase())));
         kafka09SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
+        kafka09SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));
         return kafka09SourceTableInfo;
     }
 }

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

@@ -41,6 +41,8 @@ public class KafkaSourceTableInfo extends SourceTableInfo {
 
     public static final String GROUPID_KEY = "groupId";
 
+    public static final String OFFSETRESET_KEY="offsetReset";
+
     private String bootstrapServers;
 
     private String topic;

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

@@ -46,6 +46,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         kafka10SourceTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSourceTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
         kafka10SourceTableInfo.setGroupId(MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase())));
         kafka10SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
+        kafka10SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));
         return kafka10SourceTableInfo;
     }
 }

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

@@ -41,6 +41,8 @@ public class KafkaSourceTableInfo extends SourceTableInfo {
 
     public static final String GROUPID_KEY = "groupId";
 
+    public static final String OFFSETRESET_KEY="offsetReset";
+
     private String bootstrapServers;
 
     private String topic;
@@ -89,7 +91,6 @@ public void setOffsetReset(String offsetReset) {
         if(offsetReset == null){
             return;
         }
-
         this.offsetReset = offsetReset;
     }
 

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

@@ -46,6 +46,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         kafka11SourceTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSourceTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
         kafka11SourceTableInfo.setGroupId(MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase())));
         kafka11SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
+        kafka11SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));
         return kafka11SourceTableInfo;
     }
 }

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

@@ -41,6 +41,8 @@ public class KafkaSourceTableInfo extends SourceTableInfo {
 
     public static final String GROUPID_KEY = "groupId";
 
+    public static final String OFFSETRESET_KEY="offsetReset";
+
     private String bootstrapServers;
 
     private String topic;