kafka 按照time消费

dapeng · dapeng · commit 41f0c87f5460 · 2020-09-08T17:24:25.000+08:00
diff --git a/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/AbstractKafkaSource.java b/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/AbstractKafkaSource.java
@@ -54,6 +54,8 @@ protected Properties getKafkaProperties(KafkaSourceTableInfo kafkaSourceTableInf
 
         if (DtStringUtil.isJson(kafkaSourceTableInfo.getOffsetReset())) {
             props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, EKafkaOffset.NONE.name().toLowerCase());
+        } else if(StringUtils.equalsIgnoreCase(EKafkaOffset.TIMESTAMP.name().toLowerCase(),  kafkaSourceTableInfo.getOffsetReset())){
+                props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, EKafkaOffset.EARLIEST.name().toLowerCase());
         } else {
             props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaSourceTableInfo.getOffsetReset());
         }
@@ -95,10 +97,12 @@ protected void setParallelism(Integer parallelism, DataStreamSource kafkaSource)
         }
     }
 
-    protected void setStartPosition(String offset, String topicName, FlinkKafkaConsumerBase<Row> kafkaSrc) {
+    protected void setStartPosition(String offset, String topicName, FlinkKafkaConsumerBase<Row> kafkaSrc, Runnable runnable) {
         if (StringUtils.equalsIgnoreCase(offset, EKafkaOffset.EARLIEST.name())) {
             kafkaSrc.setStartFromEarliest();
-        } else if (DtStringUtil.isJson(offset)) {
+        } else if(StringUtils.equalsIgnoreCase(offset, EKafkaOffset.TIMESTAMP.name())) {
+            runnable.run();
+        }else if (DtStringUtil.isJson(offset)) {
             Map<KafkaTopicPartition, Long> specificStartupOffsets = buildOffsetMap(offset, topicName);
             kafkaSrc.setStartFromSpecificOffsets(specificStartupOffsets);
         } else {
diff --git a/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/enums/EKafkaOffset.java b/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/enums/EKafkaOffset.java
@@ -27,5 +27,6 @@ public enum EKafkaOffset {
 
     LATEST,
     EARLIEST,
+    TIMESTAMP,
     NONE
 }
diff --git a/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -56,6 +56,9 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSourceTableInfo.setFieldDelimiter(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CSV_FIELD_DELIMITER_KEY.toLowerCase(), "|")));
         kafkaSourceTableInfo.setSourceDataType(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.SOURCE_DATA_TYPE_KEY.toLowerCase(), FormatType.DT_NEST.name())));
 
+        if(props.containsKey(KafkaSourceTableInfo.TIMESTAMP_OFFSET.toLowerCase())){
+            kafkaSourceTableInfo.setTimestampOffset(MathUtil.getLongVal(props.getOrDefault(KafkaSourceTableInfo.TIMESTAMP_OFFSET.toLowerCase(), System.currentTimeMillis())));
+        }
         Map<String, String> kafkaParams = props.keySet().stream()
                 .filter(key -> !key.isEmpty() && key.startsWith("kafka."))
                 .collect(Collectors.toMap(
diff --git a/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceTableInfo.java b/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceTableInfo.java
@@ -55,6 +55,8 @@ public class KafkaSourceTableInfo extends AbstractSourceTableInfo {
 
     public static final String CHARSET_NAME_KEY = "charsetName";
 
+    public static final String TIMESTAMP_OFFSET = "timestampOffset";
+
     private String bootstrapServers;
 
     private String topic;
@@ -75,6 +77,8 @@ public class KafkaSourceTableInfo extends AbstractSourceTableInfo {
 
     public String charsetName;
 
+    private Long timestampOffset;
+
     public String getBootstrapServers() {
         return bootstrapServers;
     }
@@ -159,6 +163,14 @@ public void setCharsetName(String charsetName) {
 		this.charsetName = charsetName;
 	}
 
+    public Long getTimestampOffset() {
+        return timestampOffset;
+    }
+
+    public void setTimestampOffset(Long timestampOffset) {
+        this.timestampOffset = timestampOffset;
+    }
+
 	@Override
 	public boolean check() {
 		Preconditions.checkNotNull(getType(), "kafka of type is required");
diff --git a/kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -51,7 +51,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
 
         setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
-        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
+        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc, () -> kafkaSrc.setStartFromTimestamp(kafkaSourceTableInfo.getTimestampOffset()));
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");
 
         return tableEnv.fromDataStream(kafkaSource, fields);
diff --git a/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -52,7 +52,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
 
         setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
-        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
+        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc, () -> kafkaSrc.setStartFromLatest());
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");
 
         return tableEnv.fromDataStream(kafkaSource, fields);
diff --git a/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -55,7 +55,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
 
         setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
-        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
+        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc, () -> kafkaSrc.setStartFromTimestamp(kafkaSourceTableInfo.getTimestampOffset()));
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");
 
         return tableEnv.fromDataStream(kafkaSource, fields);
diff --git a/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -54,7 +54,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
 
         setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
-        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
+        setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc, () -> kafkaSrc.setStartFromTimestamp(kafkaSourceTableInfo.getTimestampOffset()));
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");
 
         return tableEnv.fromDataStream(kafkaSource, fields);

Original file line number	Diff line number	Diff line change
`@@ -27,5 +27,6 @@ public enum EKafkaOffset {`
`27`	`27`
`28`	`28`	`LATEST,`
`29`	`29`	`EARLIEST,`
	`30`	`+ TIMESTAMP,`
`30`	`31`	`NONE`
`31`	`32`	`}`