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 {
