This is an automated email from the ASF dual-hosted git repository. fpaul pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 439a97fac667eece8a9c6c17775c342a7b3ba329 Author: Fabian Paul <[email protected]> AuthorDate: Wed May 4 10:57:55 2022 +0200 [FLINK-27486][hybrid] Make hybrid source building blocks as PublicEvolving --- .../archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8 | 11 ----------- .../flink/connector/base/source/hybrid/HybridSource.java | 9 +++++++++ 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8 b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8 index 8b4e3ad7aaa..f4ca020c770 100644 --- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8 +++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/5b9eed8a-5fb6-4373-98ac-3be2a71941b8 @@ -51,17 +51,6 @@ org.apache.flink.configuration.ClusterOptions.getSchedulerType(org.apache.flink. org.apache.flink.configuration.ConfigOption.fallbackKeys(): Returned leaf type org.apache.flink.configuration.FallbackKey does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated org.apache.flink.configuration.ConfigOptions.key(java.lang.String): Returned leaf type org.apache.flink.configuration.ConfigOptions$OptionBuilder does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated org.apache.flink.configuration.MemorySize.parse(java.lang.String, org.apache.flink.configuration.MemorySize$MemoryUnit): Argument leaf type org.apache.flink.configuration.MemorySize$MemoryUnit does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.builder(org.apache.flink.api.connector.source.Source): Returned leaf type org.apache.flink.connector.base.source.hybrid.HybridSource$HybridSourceBuilder does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.createEnumerator(org.apache.flink.api.connector.source.SplitEnumeratorContext): Argument leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceSplit does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.createEnumerator(org.apache.flink.api.connector.source.SplitEnumeratorContext): Returned leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceEnumeratorState does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.createEnumerator(org.apache.flink.api.connector.source.SplitEnumeratorContext): Returned leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceSplit does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.createReader(org.apache.flink.api.connector.source.SourceReaderContext): Returned leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceSplit does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.getEnumeratorCheckpointSerializer(): Returned leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceEnumeratorState does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.getSplitSerializer(): Returned leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceSplit does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.restoreEnumerator(org.apache.flink.api.connector.source.SplitEnumeratorContext, org.apache.flink.connector.base.source.hybrid.HybridSourceEnumeratorState): Argument leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceEnumeratorState does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @ [...] -org.apache.flink.connector.base.source.hybrid.HybridSource.restoreEnumerator(org.apache.flink.api.connector.source.SplitEnumeratorContext, org.apache.flink.connector.base.source.hybrid.HybridSourceEnumeratorState): Argument leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceSplit does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated -org.apache.flink.connector.base.source.hybrid.HybridSource.restoreEnumerator(org.apache.flink.api.connector.source.SplitEnumeratorContext, org.apache.flink.connector.base.source.hybrid.HybridSourceEnumeratorState): Returned leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceEnumeratorState does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @ [...] -org.apache.flink.connector.base.source.hybrid.HybridSource.restoreEnumerator(org.apache.flink.api.connector.source.SplitEnumeratorContext, org.apache.flink.connector.base.source.hybrid.HybridSourceEnumeratorState): Returned leaf type org.apache.flink.connector.base.source.hybrid.HybridSourceSplit does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.build(): Returned leaf type org.apache.flink.connector.file.sink.FileSink does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.withOutputFileConfig(org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig): Argument leaf type org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated org.apache.flink.connector.file.src.AbstractFileSource.getAssignerFactory(): Returned leaf type org.apache.flink.connector.file.src.assigners.FileSplitAssigner$Provider does not satisfy: reside outside of package 'org.apache.flink..' or reside in any package ['..shaded..'] or annotated with @Public or annotated with @PublicEvolving or annotated with @Deprecated diff --git a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSource.java b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSource.java index 8df875b8156..8e95614f440 100644 --- a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSource.java +++ b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSource.java @@ -18,6 +18,7 @@ package org.apache.flink.connector.base.source.hybrid; +import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.connector.source.Boundedness; @@ -108,18 +109,21 @@ public class HybridSource<T> implements Source<T, HybridSourceSplit, HybridSourc return sources.get(sources.size() - 1).boundedness; } + @Internal @Override public SourceReader<T, HybridSourceSplit> createReader(SourceReaderContext readerContext) throws Exception { return new HybridSourceReader(readerContext); } + @Internal @Override public SplitEnumerator<HybridSourceSplit, HybridSourceEnumeratorState> createEnumerator( SplitEnumeratorContext<HybridSourceSplit> enumContext) { return new HybridSourceSplitEnumerator(enumContext, sources, 0, null); } + @Internal @Override public SplitEnumerator<HybridSourceSplit, HybridSourceEnumeratorState> restoreEnumerator( SplitEnumeratorContext<HybridSourceSplit> enumContext, @@ -129,11 +133,13 @@ public class HybridSource<T> implements Source<T, HybridSourceSplit, HybridSourc enumContext, sources, checkpoint.getCurrentSourceIndex(), checkpoint); } + @Internal @Override public SimpleVersionedSerializer<HybridSourceSplit> getSplitSerializer() { return new HybridSourceSplitSerializer(); } + @Internal @Override public SimpleVersionedSerializer<HybridSourceEnumeratorState> getEnumeratorCheckpointSerializer() { @@ -151,6 +157,7 @@ public class HybridSource<T> implements Source<T, HybridSourceSplit, HybridSourc * backward compatible extension, i.e. additional information about the previous source can be * supplied in the future. */ + @PublicEvolving public interface SourceSwitchContext<EnumT> { EnumT getPreviousEnumerator(); } @@ -171,6 +178,7 @@ public class HybridSource<T> implements Source<T, HybridSourceSplit, HybridSourc * entry point and simply wrapped into the factory, providing the benefit of validation during * submission. */ + @PublicEvolving @FunctionalInterface public interface SourceFactory< T, SourceT extends Source<T, ?, ?>, FromEnumT extends SplitEnumerator> @@ -210,6 +218,7 @@ public class HybridSource<T> implements Source<T, HybridSourceSplit, HybridSourc } /** Builder for HybridSource. */ + @PublicEvolving public static class HybridSourceBuilder<T, EnumT extends SplitEnumerator> implements Serializable { private final List<SourceListEntry> sources;
