Revert "[improve] modify deprecated method"

This reverts commit a9fc8ca.

Author: misi

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

@@ -75,6 +75,12 @@ 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.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 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,6 +48,11 @@ public Boundedness getBoundedness() {
         return Boundedness.BOUNDED;
     }
 
+    @Override
+    public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
+        return rowType;
+    }
+
     @Override
     public SourceReader<SeaTunnelRow, FileSourceSplit> createReader(
             SourceReader.Context readerContext) {
@@ -67,9 +72,4 @@ 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,15 +19,21 @@
 
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
-import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.common.SeaTunnelAPIErrorCode;
 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;
@@ -41,9 +47,33 @@ public class CosFileSource extends BaseFileSource {
 
     private final CatalogTable catalogTable;
 
-    public CosFileSource(ReadonlyConfig readonlyConfig, CatalogTable catalogTable) {
+    public CosFileSource(CatalogTable catalogTable) {
         this.catalogTable = catalogTable;
-        Config pluginConfig = readonlyConfig.toConfig();
+    }
+
+    @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()));
+        }
         String path = pluginConfig.getString(CosFileBaseOptions.FILE_PATH.key());
         hadoopConf = CosConf.buildWithConfig(pluginConfig);
         readStrategy =
@@ -106,11 +136,6 @@ public CosFileSource(ReadonlyConfig readonlyConfig, CatalogTable catalogTable) {
         }
     }
 
-    @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,8 +108,6 @@ public Class<? extends SeaTunnelSource> getSourceClass() {
             TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
         return () ->
                 (SeaTunnelSource<T, SplitT, StateT>)
-                        new CosFileSource(
-                                context.getOptions(),
-                                CatalogTableUtil.buildWithConfig(context.getOptions()));
+                        new CosFileSource(CatalogTableUtil.buildWithConfig(context.getOptions()));
     }
 }