[Improve]]Connector-V2\[Kafka] Set kafka consumer default group (#4271)

Author: ruanwenjun

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

@@ -71,7 +71,7 @@ public class Config {
     public static final Option<String> CONSUMER_GROUP =
             Options.key("consumer.group")
                     .stringType()
-                    .noDefaultValue()
+                    .defaultValue("SeaTunnel-Consumer-Group")
                     .withDescription(
                             "Kafka consumer group id, used to distinguish different consumer groups.");
 

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

@@ -79,8 +79,6 @@ public class KafkaSource
         implements SeaTunnelSource<SeaTunnelRow, KafkaSourceSplit, KafkaSourceState>,
                 SupportParallelism {
 
-    private static final String DEFAULT_CONSUMER_GROUP = "SeaTunnel-Consumer-Group";
-
     private final ConsumerMetadata metadata = new ConsumerMetadata();
     private DeserializationSchema<SeaTunnelRow> deserializationSchema;
     private SeaTunnelRowType typeInfo;
@@ -113,14 +111,16 @@ public void prepare(Config config) throws PrepareFailException {
         this.metadata.setTopic(config.getString(TOPIC.key()));
         if (config.hasPath(PATTERN.key())) {
             this.metadata.setPattern(config.getBoolean(PATTERN.key()));
+        } else {
+            this.metadata.setPattern(PATTERN.defaultValue());
         }
         this.metadata.setBootstrapServers(config.getString(BOOTSTRAP_SERVERS.key()));
         this.metadata.setProperties(new Properties());
 
         if (config.hasPath(CONSUMER_GROUP.key())) {
             this.metadata.setConsumerGroup(config.getString(CONSUMER_GROUP.key()));
         } else {
-            this.metadata.setConsumerGroup(DEFAULT_CONSUMER_GROUP);
+            this.metadata.setConsumerGroup(CONSUMER_GROUP.defaultValue());
         }
 
         if (config.hasPath(COMMIT_ON_CHECKPOINT.key())) {