This is an automated email from the ASF dual-hosted git repository. ron pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit e9eabb16f2ce85f2d039b8c0e94d2a47b199e444 Author: fengli <ldliu...@163.com> AuthorDate: Tue Jun 25 15:17:05 2024 +0800 Revert "[FLINK-35189][connectors/filesystem] Modify the visibility of filesystem connector related methods to protected" This reverts commit e3cda01cac737c97a104feae0e35da1eb91a8751. --- .../apache/flink/connector/file/src/AbstractFileSource.java | 11 +++-------- .../flink/connector/file/table/FileSystemTableFactory.java | 4 ++-- .../flink/connector/file/table/FileSystemTableSource.java | 4 ++-- 3 files changed, 7 insertions(+), 12 deletions(-) diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/AbstractFileSource.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/AbstractFileSource.java index c9133172070..50dedbdeedf 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/AbstractFileSource.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/AbstractFileSource.java @@ -147,7 +147,7 @@ public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit> throw new FlinkRuntimeException("Could not enumerate file splits", e); } - return createSplitEnumerator(getBoundedness(), enumContext, enumerator, splits, null); + return createSplitEnumerator(enumContext, enumerator, splits, null); } @Override @@ -164,11 +164,7 @@ public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit> (Collection<FileSourceSplit>) checkpoint.getSplits(); return createSplitEnumerator( - getBoundedness(), - enumContext, - enumerator, - splits, - checkpoint.getAlreadyProcessedPaths()); + enumContext, enumerator, splits, checkpoint.getAlreadyProcessedPaths()); } @Override @@ -190,7 +186,6 @@ public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit> // ------------------------------------------------------------------------ private SplitEnumerator<SplitT, PendingSplitsCheckpoint<SplitT>> createSplitEnumerator( - Boundedness boundedness, SplitEnumeratorContext<SplitT> context, FileEnumerator enumerator, Collection<FileSourceSplit> splits, @@ -204,7 +199,7 @@ public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit> final FileSplitAssigner splitAssigner = assignerFactory.create(splits); - if (Boundedness.BOUNDED == boundedness) { + if (continuousEnumerationSettings == null) { // bounded case return castGeneric(new StaticFileSplitEnumerator(fileSplitContext, splitAssigner)); } else { diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java index 00db848356b..31f6dddbc69 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java @@ -149,7 +149,7 @@ public class FileSystemTableFactory implements DynamicTableSourceFactory, Dynami .collect(Collectors.toSet()); } - protected void validate(FactoryUtil.TableFactoryHelper helper) { + private void validate(FactoryUtil.TableFactoryHelper helper) { // Except format options, some formats like parquet and orc can not list all supported // options. helper.validateExcept(helper.getOptions().get(FactoryUtil.FORMAT) + "."); @@ -160,7 +160,7 @@ public class FileSystemTableFactory implements DynamicTableSourceFactory, Dynami .get(FileSystemConnectorOptions.SINK_PARTITION_COMMIT_WATERMARK_TIME_ZONE)); } - protected <I, F extends DecodingFormatFactory<I>> DecodingFormat<I> discoverDecodingFormat( + private <I, F extends DecodingFormatFactory<I>> DecodingFormat<I> discoverDecodingFormat( Context context, Class<F> formatFactoryClass) { FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context); if (formatFactoryExists(context, formatFactoryClass)) { diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java index bbaa2a310d5..1e53eb53cfb 100644 --- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java +++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java @@ -264,7 +264,7 @@ public class FileSystemTableSource extends AbstractFileSystemTable return bulkFormat; } - protected SourceProvider createSourceProvider(BulkFormat<RowData, FileSourceSplit> bulkFormat) { + private SourceProvider createSourceProvider(BulkFormat<RowData, FileSourceSplit> bulkFormat) { final FileSource.FileSourceBuilder<RowData> fileSourceBuilder = FileSource.forBulkFileFormat(bulkFormat, paths()); @@ -287,7 +287,7 @@ public class FileSystemTableSource extends AbstractFileSystemTable return SourceProvider.of(fileSourceBuilder.build()); } - protected Path[] paths() { + private Path[] paths() { if (partitionKeys.isEmpty()) { return new Path[] {path}; } else {