[Improve][Connector-V2-kafka] Support setting read starting offset or time at startup config

docs/en/connector-v2/source/kafka.md

@@ -18,7 +18,7 @@ Source connector for Apache Kafka.
 ## Options
 
 | name                 | type    | required | default value            |
-| -------------------- | ------- | -------- | ------------------------ |
+|----------------------|---------| -------- |--------------------------|
 | topic                | String  | yes      | -                        |
 | bootstrap.servers    | String  | yes      | -                        |
 | pattern              | Boolean | no       | false                    |
@@ -28,6 +28,9 @@ Source connector for Apache Kafka.
 | common-options       |         | no       | -                        |
 | schema               |         | no       | -                        |
 | format               | String  | no       | json                     |
+| start_mode           | String  | no       | group_offsets            |
+| start_mode.offsets   |         | no       |                          |
+| start_mode.timestamp | Long    | no       |                          |
 
 ### topic [string]
 
@@ -66,6 +69,24 @@ The structure of the data, including field names and field types.
 Data format. The default format is json. Optional text format. The default field separator is ", ".
 If you customize the delimiter, add the "field_delimiter" option.
 
+## start_mode
+The initial consumption pattern of consumers,there are several types:
+[earliest],[group_offsets],[latest],[specific_offsets],[timestamp]
+
+## start_mode.timestamp
+The time required for consumption mode to be timestamp
+
+##  start_mode.offsets
+The offset required for consumption mode to be specific_offsets
+for example:
+```hocon
+   start_mode.offsets = {
+            info-0 = 70
+            info-1 = 10
+            info-2 = 10
+         }
+```
+
 ## Example
 
 ###  Simple
@@ -84,7 +105,7 @@ source {
     format = text
     field_delimiter = "#“
     topic = "topic_1,topic_2,topic_3"
-    bootstrap.server = "localhost:9092"
+    bootstrap.servers = "localhost:9092"
     kafka.max.poll.records = 500
     kafka.client.id = client_1
   }

seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/Config.java

@@ -90,4 +90,19 @@ public class Config {
      */
     public static final String PARTITION_KEY = "partition_key";
 
+    /**
+     * The initial consumption pattern of consumers
+     */
+    public static final String  START_MODE = "start_mode";
+
+    /**
+     * The time required for consumption mode to be timestamp
+     */
+    public static final String  START_MODE_TIMESTAMP  = "start_mode.timestamp";
+
+    /**
+     * The offset required for consumption mode to be specific_offsets
+     */
+    public static final String  START_MODE_OFFSETS  = "start_mode.offsets";
+
 }

seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/StartMode.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.kafka.config;
+
+public enum StartMode {
+
+    EARLIEST("earliest"),
+
+    GROUP_OFFSETS("group_offsets"),
+
+    LATEST("latest"),
+
+    TIMESTAMP("timestamp"),
+
+    SPECIFIC_OFFSETS("specific_offsets");
+
+    private String mode;
+
+    StartMode(String mode) {
+        this.mode = mode;
+    }
+
+    public String getMode() {
+        return mode;
+    }
+
+    @Override
+    public String toString() {
+        return mode;
+    }
+}

seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/ConsumerMetadata.java

@@ -17,7 +17,12 @@
 
 package org.apache.seatunnel.connectors.seatunnel.kafka.source;
 
+import org.apache.seatunnel.connectors.seatunnel.kafka.config.StartMode;
+
+import org.apache.kafka.common.TopicPartition;
+
 import java.io.Serializable;
+import java.util.Map;
 import java.util.Properties;
 
 /**
@@ -31,6 +36,9 @@ public class ConsumerMetadata implements Serializable {
     private Properties properties;
     private String consumerGroup;
     private boolean commitOnCheckpoint = false;
+    private StartMode startMode = StartMode.GROUP_OFFSETS;
+    private Map<TopicPartition, Long> specificStartOffsets;
+    private Long startOffsetsTimestamp;
 
     public boolean isCommitOnCheckpoint() {
         return commitOnCheckpoint;
@@ -79,4 +87,28 @@ public String getConsumerGroup() {
     public void setConsumerGroup(String consumerGroup) {
         this.consumerGroup = consumerGroup;
     }
+
+    public StartMode getStartMode() {
+        return startMode;
+    }
+
+    public void setStartMode(StartMode startMode) {
+        this.startMode = startMode;
+    }
+
+    public Map<TopicPartition, Long> getSpecificStartOffsets() {
+        return specificStartOffsets;
+    }
+
+    public void setSpecificStartOffsets(Map<TopicPartition, Long> specificStartOffsets) {
+        this.specificStartOffsets = specificStartOffsets;
+    }
+
+    public Long getStartOffsetsTimestamp() {
+        return startOffsetsTimestamp;
+    }
+
+    public void setStartOffsetsTimestamp(Long startOffsetsTimestamp) {
+        this.startOffsetsTimestamp = startOffsetsTimestamp;
+    }
 }

seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSource.java

@@ -26,6 +26,9 @@
 import static org.apache.seatunnel.connectors.seatunnel.kafka.config.Config.FORMAT;
 import static org.apache.seatunnel.connectors.seatunnel.kafka.config.Config.PATTERN;
 import static org.apache.seatunnel.connectors.seatunnel.kafka.config.Config.SCHEMA;
+import static org.apache.seatunnel.connectors.seatunnel.kafka.config.Config.START_MODE;
+import static org.apache.seatunnel.connectors.seatunnel.kafka.config.Config.START_MODE_OFFSETS;
+import static org.apache.seatunnel.connectors.seatunnel.kafka.config.Config.START_MODE_TIMESTAMP;
 import static org.apache.seatunnel.connectors.seatunnel.kafka.config.Config.TOPIC;
 
 import org.apache.seatunnel.api.common.JobContext;
@@ -42,15 +45,22 @@
 import org.apache.seatunnel.common.config.TypesafeConfigUtils;
 import org.apache.seatunnel.common.constants.JobMode;
 import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.common.utils.JsonUtils;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
+import org.apache.seatunnel.connectors.seatunnel.kafka.config.StartMode;
 import org.apache.seatunnel.connectors.seatunnel.kafka.state.KafkaSourceState;
 import org.apache.seatunnel.format.json.JsonDeserializationSchema;
 import org.apache.seatunnel.format.text.TextDeserializationSchema;
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigRenderOptions;
 
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.auto.service.AutoService;
+import org.apache.kafka.common.TopicPartition;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Properties;
 
 @AutoService(SeaTunnelSource.class)
@@ -96,6 +106,40 @@ public void prepare(Config config) throws PrepareFailException {
             this.metadata.setCommitOnCheckpoint(config.getBoolean(COMMIT_ON_CHECKPOINT));
         }
 
+        if (config.hasPath(START_MODE)) {
+            StartMode startMode = StartMode.valueOf(config.getString(START_MODE).toUpperCase());
+            this.metadata.setStartMode(startMode);
+            switch (startMode) {
+                case TIMESTAMP:
+                    long startOffsetsTimestamp = config.getLong(START_MODE_TIMESTAMP);
+                    long currentTimestamp = System.currentTimeMillis();
+                    if (startOffsetsTimestamp < 0 || startOffsetsTimestamp > currentTimestamp) {
+                        throw new IllegalArgumentException("start_mode.timestamp The value is smaller than 0 or smaller than the current time");
+                    }
+                    this.metadata.setStartOffsetsTimestamp(startOffsetsTimestamp);
+                    break;
+                case SPECIFIC_OFFSETS:
+                    Config offsets = config.getConfig(START_MODE_OFFSETS);
+                    ConfigRenderOptions options = ConfigRenderOptions.concise();
+                    String offsetsJson = offsets.root().render(options);
+                    if (offsetsJson == null) {
+                        throw new IllegalArgumentException("start mode is " + StartMode.SPECIFIC_OFFSETS + "but no specific offsets were specified.");
+                    }
+                    Map<TopicPartition, Long> specificStartOffsets = new HashMap<>();
+                    ObjectNode jsonNodes = JsonUtils.parseObject(offsetsJson);
+                    jsonNodes.fieldNames().forEachRemaining(key -> {
+                        String[] topicAndPartition = key.split("-");
+                        long offset = jsonNodes.get(key).asLong();
+                        TopicPartition topicPartition = new TopicPartition(topicAndPartition[0], Integer.valueOf(topicAndPartition[1]));
+                        specificStartOffsets.put(topicPartition, offset);
+                    });
+                    this.metadata.setSpecificStartOffsets(specificStartOffsets);
+                    break;
+                default:
+                    break;
+            }
+        }
+
         TypesafeConfigUtils.extractSubConfig(config, "kafka.", false).entrySet().forEach(e -> {
             this.metadata.getProperties().put(e.getKey(), String.valueOf(e.getValue().unwrapped()));
         });
@@ -144,19 +188,19 @@ private void setDeserialization(Config config) {
                     delimiter = config.getString(FIELD_DELIMITER);
                 }
                 deserializationSchema = TextDeserializationSchema.builder()
-                        .seaTunnelRowType(typeInfo)
-                        .delimiter(delimiter)
-                        .build();
+                    .seaTunnelRowType(typeInfo)
+                    .delimiter(delimiter)
+                    .build();
             } else {
                 // TODO: use format SPI
                 throw new UnsupportedOperationException("Unsupported format: " + format);
             }
         } else {
             typeInfo = SeaTunnelSchema.buildSimpleTextSchema();
             this.deserializationSchema = TextDeserializationSchema.builder()
-                    .seaTunnelRowType(typeInfo)
-                    .delimiter(String.valueOf('\002'))
-                    .build();
+                .seaTunnelRowType(typeInfo)
+                .delimiter(String.valueOf('\002'))
+                .build();
         }
     }
 }