support offsetnum

Author: yanxi0227

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

@@ -68,7 +68,11 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 
         Properties props = new Properties();
         props.setProperty("bootstrap.servers", kafka09SourceTableInfo.getBootstrapServers());
-        props.setProperty("auto.offset.reset", kafka09SourceTableInfo.getOffsetReset());
+        if (DtStringUtil.isJosn(kafka09SourceTableInfo.getOffsetReset())){
+            props.setProperty("auto.offset.reset", "none");
+        } else {
+            props.setProperty("auto.offset.reset", kafka09SourceTableInfo.getOffsetReset());
+        }
         if (StringUtils.isNotBlank(kafka09SourceTableInfo.getGroupId())){
             props.setProperty("group.id", kafka09SourceTableInfo.getGroupId());
         }

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

@@ -68,7 +68,11 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 
 		Properties props = new Properties();
 		props.setProperty("bootstrap.servers", kafka010SourceTableInfo.getBootstrapServers());
-		props.setProperty("auto.offset.reset", kafka010SourceTableInfo.getOffsetReset());
+		if (DtStringUtil.isJosn(kafka010SourceTableInfo.getOffsetReset())){
+			props.setProperty("auto.offset.reset", "none");
+		} else {
+			props.setProperty("auto.offset.reset", kafka010SourceTableInfo.getOffsetReset());
+		}
 		if (StringUtils.isNotBlank(kafka010SourceTableInfo.getGroupId())){
 			props.setProperty("group.id", kafka010SourceTableInfo.getGroupId());
 		}

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

@@ -68,7 +68,11 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 
 		Properties props = new Properties();
 		props.setProperty("bootstrap.servers", kafka011SourceTableInfo.getBootstrapServers());
-		props.setProperty("auto.offset.reset", kafka011SourceTableInfo.getOffsetReset());
+		if (DtStringUtil.isJosn(kafka011SourceTableInfo.getOffsetReset())){
+			props.setProperty("auto.offset.reset", "none");
+		} else {
+			props.setProperty("auto.offset.reset", kafka011SourceTableInfo.getOffsetReset());
+		}
 		if (StringUtils.isNotBlank(kafka011SourceTableInfo.getGroupId())){
 			props.setProperty("group.id", kafka011SourceTableInfo.getGroupId());
 		}