[improve] modify deprecated method

Author: misi

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/BaseFileSink.java

@@ -75,12 +75,6 @@ public void setJobContext(JobContext jobContext) {
         preCheckConfig();
     }
 
-    @Override
-    public void setTypeInfo(SeaTunnelRowType seaTunnelRowType) {
-        this.seaTunnelRowType = seaTunnelRowType;
-        this.fileSinkConfig = new FileSinkConfig(pluginConfig, seaTunnelRowType);
-    }
-
     @Override
     public SinkWriter<SeaTunnelRow, FileCommitInfo, FileSinkState> restoreWriter(
             SinkWriter.Context context, List<FileSinkState> states) {

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSource.java

@@ -23,7 +23,7 @@
 import org.apache.seatunnel.api.source.SourceSplitEnumerator;
 import org.apache.seatunnel.api.source.SupportColumnProjection;
 import org.apache.seatunnel.api.source.SupportParallelism;
-import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
 import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
@@ -48,11 +48,6 @@ public Boundedness getBoundedness() {
         return Boundedness.BOUNDED;
     }
 
-    @Override
-    public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
-        return rowType;
-    }
-
     @Override
     public SourceReader<SeaTunnelRow, FileSourceSplit> createReader(
             SourceReader.Context readerContext) {
@@ -72,4 +67,9 @@ public SourceSplitEnumerator<FileSourceSplit, FileSourceState> restoreEnumerator
             throws Exception {
         return new FileSourceSplitEnumerator(enumeratorContext, filePaths, checkpointState);
     }
+
+    @Override
+    public List<CatalogTable> getProducedCatalogTables() {
+        return SeaTunnelSource.super.getProducedCatalogTables();
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSource.java

@@ -19,21 +19,15 @@
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
-import org.apache.seatunnel.api.common.PrepareFailException;
-import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.options.ConnectorCommonOptions;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
-import org.apache.seatunnel.common.config.CheckConfigUtil;
-import org.apache.seatunnel.common.config.CheckResult;
-import org.apache.seatunnel.common.constants.PluginType;
 import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
-import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseOptions;
 import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
 import org.apache.seatunnel.connectors.seatunnel.file.config.FileSystemType;
 import org.apache.seatunnel.connectors.seatunnel.file.cos.config.CosConf;
 import org.apache.seatunnel.connectors.seatunnel.file.cos.config.CosFileBaseOptions;
-import org.apache.seatunnel.connectors.seatunnel.file.cos.config.CosFileSourceOptions;
 import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.file.source.BaseFileSource;
@@ -47,33 +41,9 @@ public class CosFileSource extends BaseFileSource {
 
     private final CatalogTable catalogTable;
 
-    public CosFileSource(CatalogTable catalogTable) {
+    public CosFileSource(ReadonlyConfig readonlyConfig, CatalogTable catalogTable) {
         this.catalogTable = catalogTable;
-    }
-
-    @Override
-    public String getPluginName() {
-        return FileSystemType.COS.getFileSystemPluginName();
-    }
-
-    @Override
-    public void prepare(Config pluginConfig) throws PrepareFailException {
-        CheckResult result =
-                CheckConfigUtil.checkAllExists(
-                        pluginConfig,
-                        FileBaseOptions.FILE_PATH.key(),
-                        CosFileSourceOptions.FILE_FORMAT_TYPE.key(),
-                        CosFileSourceOptions.SECRET_ID.key(),
-                        CosFileSourceOptions.SECRET_KEY.key(),
-                        CosFileSourceOptions.REGION.key(),
-                        CosFileSourceOptions.BUCKET.key());
-        if (!result.isSuccess()) {
-            throw new FileConnectorException(
-                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
-                    String.format(
-                            "PluginName: %s, PluginType: %s, Message: %s",
-                            getPluginName(), PluginType.SOURCE, result.getMsg()));
-        }
+        Config pluginConfig = readonlyConfig.toConfig();
         String path = pluginConfig.getString(CosFileBaseOptions.FILE_PATH.key());
         hadoopConf = CosConf.buildWithConfig(pluginConfig);
         readStrategy =
@@ -136,6 +106,11 @@ public void prepare(Config pluginConfig) throws PrepareFailException {
         }
     }
 
+    @Override
+    public String getPluginName() {
+        return FileSystemType.COS.getFileSystemPluginName();
+    }
+
     @Override
     public List<CatalogTable> getProducedCatalogTables() {
         return Collections.singletonList(catalogTable);

seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java

@@ -108,6 +108,8 @@ public Class<? extends SeaTunnelSource> getSourceClass() {
             TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
         return () ->
                 (SeaTunnelSource<T, SplitT, StateT>)
-                        new CosFileSource(CatalogTableUtil.buildWithConfig(context.getOptions()));
+                        new CosFileSource(
+                                context.getOptions(),
+                                CatalogTableUtil.buildWithConfig(context.getOptions()));
     }
 }