[Feature][Connector] add get source method to all source connector

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryUtil.java

@@ -22,6 +22,7 @@
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SourceCommonOptions;
 import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.source.SupportParallelism;
 import org.apache.seatunnel.api.table.catalog.Catalog;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.connector.TableSource;
@@ -163,15 +164,19 @@ public static List<Factory> discoverFactories(ClassLoader classLoader) {
      * This method is called by SeaTunnel Web to get the full option rule of a source.
      * @return
      */
-    public static OptionRule sourceFullOptionRule(@NonNull Factory factory) {
+    public static OptionRule sourceFullOptionRule(@NonNull TableSourceFactory factory) {
         OptionRule sourceOptionRule = factory.optionRule();
         if (sourceOptionRule == null) {
             throw new FactoryException("sourceOptionRule can not be null");
         }
 
-        OptionRule sourceCommonOptionRule =
-            OptionRule.builder().optional(SourceCommonOptions.PARALLELISM).build();
-        sourceOptionRule.getOptionalOptions().addAll(sourceCommonOptionRule.getOptionalOptions());
+        Class<? extends SeaTunnelSource> sourceClass = factory.getSourceClass();
+        if (sourceClass.isAssignableFrom(SupportParallelism.class)) {
+            OptionRule sourceCommonOptionRule =
+                OptionRule.builder().optional(SourceCommonOptions.PARALLELISM).build();
+            sourceOptionRule.getOptionalOptions().addAll(sourceCommonOptionRule.getOptionalOptions());
+        }
+
         return sourceOptionRule;
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/TableSourceFactory.java

@@ -17,6 +17,7 @@
 
 package org.apache.seatunnel.api.table.factory;
 
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SourceSplit;
 import org.apache.seatunnel.api.table.connector.TableSource;
 
@@ -36,4 +37,6 @@ default <T, SplitT extends SourceSplit, StateT extends Serializable> TableSource
         TableFactoryContext context) {
         throw new UnsupportedOperationException("unsupported now");
     }
+
+    Class<? extends SeaTunnelSource> getSourceClass();
 }

seatunnel-connectors-v2/connector-amazondynamodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazondynamodb/source/AmazonDynamoDBSourceFactory.java

@@ -24,6 +24,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.amazondynamodb.config.AmazonDynamoDBConfig.URL;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -42,4 +43,9 @@ public OptionRule optionRule() {
         return OptionRule.builder()
             .required(URL, REGION, ACCESS_KEY_ID, SECRET_ACCESS_KEY, TABLE, SeaTunnelSchema.SCHEMA).build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return AmazonDynamoDBSource.class;
+    }
 }

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.cdc.mysql.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
@@ -51,4 +52,9 @@ public OptionRule optionRule() {
                 JdbcSourceOptions.CONNECTION_POOL_SIZE)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return MySqlIncrementalSource.class;
+    }
 }

seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/source/ClickhouseSourceFactory.java

@@ -24,6 +24,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.clickhouse.config.ClickhouseConfig.USERNAME;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 
@@ -40,4 +41,9 @@ public String factoryIdentifier() {
     public OptionRule optionRule() {
         return OptionRule.builder().required(HOST, DATABASE, SQL, USERNAME, PASSWORD).build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return ClickhouseSource.class;
+    }
 }

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceFactory.java

@@ -26,6 +26,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.elasticsearch.config.SourceConfig.SOURCE;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -44,4 +45,9 @@ public OptionRule optionRule() {
         return OptionRule.builder().required(HOSTS, INDEX).optional(USERNAME, PASSWORD, SCROLL_TIME, SCROLL_SIZE)
             .exclusive(SOURCE, SeaTunnelSchema.SCHEMA).build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return ElasticsearchSource.class;
+    }
 }

seatunnel-connectors-v2/connector-fake/src/main/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeSourceFactory.java

@@ -26,6 +26,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.fake.config.FakeOption.STRING_LENGTH;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -44,4 +45,9 @@ public OptionRule optionRule() {
         return OptionRule.builder().required(SeaTunnelSchema.SCHEMA).optional(ROW_NUM, SPLIT_NUM, SPLIT_READ_INTERVAL, MAP_SIZE,
             ARRAY_SIZE, BYTES_LENGTH, STRING_LENGTH).build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return FakeSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.ftp.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -55,4 +56,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return FtpFileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.hdfs.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -52,4 +53,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return HdfsFileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.local.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -51,4 +52,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return LocalFileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-oss-jindo/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.oss.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -55,4 +56,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return OssFileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.oss.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -55,4 +56,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return OssFileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.s3.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -58,4 +59,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return S3FileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.file.sftp.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -55,4 +56,9 @@ public OptionRule optionRule() {
             .optional(BaseSourceConfig.TIME_FORMAT)
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return SftpFileSource.class;
+    }
 }

seatunnel-connectors-v2/connector-google-sheets/src/main/java/org/apache/seatunnel/connectors/seatunnel/google/sheets/source/SheetsSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.google.sheets.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -42,4 +43,9 @@ public OptionRule optionRule() {
                 .optional(SeaTunnelSchema.SCHEMA)
                 .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return SheetsSource.class;
+    }
 }

seatunnel-connectors-v2/connector-hive/src/main/java/org/apache/seatunnel/connectors/seatunnel/hive/source/HiveSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.hive.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.hive.config.HiveConfig;
@@ -38,4 +39,9 @@ public OptionRule optionRule() {
                 .required(HiveConfig.METASTORE_URI)
                 .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return HiveSource.class;
+    }
 }

seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/source/HttpSourceFactory.java

@@ -18,6 +18,7 @@
 package org.apache.seatunnel.connectors.seatunnel.http.source;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.connectors.seatunnel.common.schema.SeaTunnelSchema;
@@ -55,4 +56,9 @@ public OptionRule.Builder getHttpBuilder() {
                 .optional(HttpConfig.RETRY_BACKOFF_MULTIPLIER_MS)
                 .optional(HttpConfig.RETRY_BACKOFF_MAX_MS);
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return HttpSource.class;
+    }
 }

seatunnel-connectors-v2/connector-hudi/src/main/java/org/apache/seatunnel/connectors/seatunnel/hudi/source/HudiSourceFactory.java

@@ -40,4 +40,9 @@ public OptionRule optionRule() {
                 .conditional(HudiSourceConfig.USE_KERBEROS, true, HudiSourceConfig.KERBEROS_PRINCIPAL, HudiSourceConfig.KERBEROS_PRINCIPAL_FILE)
                 .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return HudiSource.class;
+    }
 }

seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/source/IcebergSourceFactory.java

@@ -34,6 +34,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.iceberg.config.SourceConfig.KEY_USE_SNAPSHOT_TIMESTAMP;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 
@@ -70,4 +71,9 @@ public OptionRule optionRule() {
             )
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return IcebergSource.class;
+    }
 }

seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/sink/InfluxDBSinkFactory.java

@@ -32,12 +32,12 @@
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.table.factory.Factory;
-import org.apache.seatunnel.api.table.factory.TableSourceFactory;
+import org.apache.seatunnel.api.table.factory.TableSinkFactory;
 
 import com.google.auto.service.AutoService;
 
 @AutoService(Factory.class)
-public class InfluxDBSinkFactory implements TableSourceFactory {
+public class InfluxDBSinkFactory implements TableSinkFactory {
 
     @Override
     public String factoryIdentifier() {

seatunnel-connectors-v2/connector-influxdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/influxdb/source/InfluxDBSourceFactory.java

@@ -31,6 +31,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.influxdb.config.SourceConfig.UPPER_BOUND;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 
@@ -60,4 +61,9 @@ public OptionRule optionRule() {
             )
             .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return InfluxDBSource.class;
+    }
 }

seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/source/IoTDBSourceFactory.java

@@ -33,6 +33,7 @@
 import static org.apache.seatunnel.connectors.seatunnel.iotdb.config.SourceConfig.VERSION;
 
 import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.table.factory.Factory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 
@@ -53,4 +54,9 @@ public OptionRule optionRule() {
                         ENABLE_CACHE_LEADER, VERSION, LOWER_BOUND, UPPER_BOUND, NUM_PARTITIONS)
                 .build();
     }
+
+    @Override
+    public Class<? extends SeaTunnelSource> getSourceClass() {
+        return IoTDBSource.class;
+    }
 }