Skip to content

[improve] Adjust CosFile connector instantiation #9121

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 8 commits into
base: dev
Choose a base branch
from
Open

[improve] Adjust CosFile connector instantiation #9121

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8e86af1
[improve] Adjust the instantiation method of CosFile connector
Apr 8, 2025
776e8b7
[improve] code format
Apr 8, 2025
b99661f
[improve] implements getProducedCatalogTables() method
Apr 8, 2025
a9fc8ca
[improve] modify deprecated method
Apr 9, 2025
c0baac8
Revert "[improve] modify deprecated method"
Apr 9, 2025
98fac43
[bugfix] remove CosFileSource prepare()
Apr 10, 2025
7b8edfc
[improve] remove CosFileSink prepare()
Apr 10, 2025
37f68aa
[improve] skip deprecated method check
Apr 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 10 additions & 36 deletions ...os/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/sink/CosFileSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,56 +17,30 @@

package org.apache.seatunnel.connectors.seatunnel.file.cos.sink;

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.sink.SeaTunnelSink;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
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.connectors.seatunnel.file.config.FileBaseOptions;
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.CosFileSinkOptions;
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
import org.apache.seatunnel.connectors.seatunnel.file.sink.BaseFileSink;

import com.google.auto.service.AutoService;

import java.util.Optional;

@AutoService(SeaTunnelSink.class)
public class CosFileSink extends BaseFileSink {
@Override
public String getPluginName() {
return FileSystemType.COS.getFileSystemPluginName();

private final CatalogTable catalogTable;

public CosFileSink(ReadonlyConfig readonlyConfig, CatalogTable catalogTable) {
this.catalogTable = catalogTable;
hadoopConf = CosConf.buildWithConfig(readonlyConfig.toConfig());
}

@Override
public void prepare(Config pluginConfig) throws PrepareFailException {
super.prepare(pluginConfig);
CheckResult result =
CheckConfigUtil.checkAllExists(
pluginConfig,
FileBaseOptions.FILE_PATH.key(),
CosFileSinkOptions.REGION.key(),
CosFileSinkOptions.SECRET_ID.key(),
CosFileSinkOptions.SECRET_KEY.key(),
CosFileSinkOptions.BUCKET.key());
if (!result.isSuccess()) {
throw new FileConnectorException(
SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
String.format(
"PluginName: %s, PluginType: %s, Message: %s",
getPluginName(), PluginType.SINK, result.getMsg()));
}
hadoopConf = CosConf.buildWithConfig(pluginConfig);
public String getPluginName() {
return FileSystemType.COS.getFileSystemPluginName();
}

@Override
public Optional<CatalogTable> getWriteCatalogTable() {
return super.getWriteCatalogTable();
return Optional.ofNullable(catalogTable);
}
}
7 changes: 7 additions & 0 deletions ...main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/sink/CosFileSinkFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@
package org.apache.seatunnel.connectors.seatunnel.file.cos.sink;

import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.table.connector.TableSink;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSinkFactory;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSinkOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
Expand Down Expand Up @@ -110,4 +112,9 @@ public OptionRule optionRule() {
.optional(FileBaseSinkOptions.TMP_PATH)
.build();
}

@Override
public TableSink createSink(TableSinkFactoryContext context) {
return () -> new CosFileSink(context.getOptions(), context.getCatalogTable());
}
}
51 changes: 18 additions & 33 deletions ...rc/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,56 +19,31 @@

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.source.SeaTunnelSource;
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;
import org.apache.seatunnel.connectors.seatunnel.file.source.reader.ReadStrategyFactory;

import com.google.auto.service.AutoService;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

@AutoService(SeaTunnelSource.class)
public class CosFileSource extends BaseFileSource {
@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()));
}
private final CatalogTable catalogTable;

public CosFileSource(ReadonlyConfig readonlyConfig) {
this.catalogTable = CatalogTableUtil.buildWithConfig(readonlyConfig);
Config pluginConfig = readonlyConfig.toConfig();
String path = pluginConfig.getString(CosFileBaseOptions.FILE_PATH.key());
hadoopConf = CosConf.buildWithConfig(pluginConfig);
readStrategy =
Expand Down Expand Up @@ -130,4 +105,14 @@ public void prepare(Config pluginConfig) throws PrepareFailException {
}
}
}

@Override
public String getPluginName() {
return FileSystemType.COS.getFileSystemPluginName();
}

@Override
public List<CatalogTable> getProducedCatalogTables() {
return Collections.singletonList(catalogTable);
}
}
10 changes: 10 additions & 0 deletions .../java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,11 @@
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.source.SourceSplit;
import org.apache.seatunnel.api.table.connector.TableSource;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
Expand All @@ -30,6 +33,7 @@

import com.google.auto.service.AutoService;

import java.io.Serializable;
import java.util.Arrays;

@AutoService(Factory.class)
Expand Down Expand Up @@ -97,4 +101,10 @@ public OptionRule optionRule() {
public Class<? extends SeaTunnelSource> getSourceClass() {
return CosFileSource.class;
}

@Override
public <T, SplitT extends SourceSplit, StateT extends Serializable>
TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
return () -> (SeaTunnelSource<T, SplitT, StateT>) new CosFileSource(context.getOptions());
Copy link
Preview

Copilot AI Apr 16, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Consider adding an annotation to suppress the unchecked cast warning or handling the cast more explicitly to ensure type safety.

Suggested change
return () -> (SeaTunnelSource<T, SplitT, StateT>) new CosFileSource(context.getOptions());
SeaTunnelSource<?, ?, ?> source = new CosFileSource(context.getOptions());
if (!(source instanceof SeaTunnelSource<T, SplitT, StateT>)) {
throw new ClassCastException("The created source is not compatible with the expected type.");
}
return () -> (SeaTunnelSource<T, SplitT, StateT>) source;

Copilot is powered by AI, so mistakes are possible. Review output carefully before use.

}
}
3 changes: 3 additions & 0 deletions ...ist/src/test/java/org/apache/seatunnel/api/connector/ConnectorSpecificationCheckTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,9 @@ public void testAllConnectorImplementFactoryWithUpToDateMethod() throws ClassNot
List<String> blockList = new ArrayList<>();
blockList.add("HiveSourceFactory");
blockList.add("HiveSinkFactory");
// adjusted the connector implementation,not deal with deprecated method yet
blockList.add("CosFileSourceFactory");
Comment on lines +76 to +77
Copy link
Preview

Copilot AI Apr 16, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[nitpick] Consider adding a space after the comma for clarity, e.g., 'connector implementation, not dealing with deprecated method yet'.

Suggested change
// adjusted the connector implementation,not deal with deprecated method yet
blockList.add("CosFileSourceFactory");
// adjusted the connector implementation, not deal with deprecated method yet

Copilot is powered by AI, so mistakes are possible. Review output carefully before use.

blockList.add("CosFileSinkFactory");

for (TableSourceFactory factory : sourceFactories) {
if (ReflectionUtils.getDeclaredMethod(
Expand Down
Loading