...
Discussion thread | https://lists.apache.org/thread/b8f509878ptwl3kmmgg95tv8sb1j5987 | ||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|
Vote thread | https://lists.apache.org/thread/t1zff21z440pvv48jyhm8pgtqsyplchn | ||||||||||
JIRA |
| ||||||||||
Release | 1.19 |
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
For FutureCompletingBlockingQueue, as a hindsight, it might be better to not expose it to the Source developers. They are unlikely to use it anywhere other than just constructing it. The reason that FutureCompletingBlockingQueue is currently exposed in the SourceReaderBase constructor is because both the SplitFetcherManager and SourceReaderBase need it. One way to hide the FutureCompletingBlockingQueue from the public API is to make SplitFetcherManager the only owner class of the queue, and expose some of its methods via SplitFetcherManager. This way, the SourceReaderBase can invoke the methods via SplitFetcherManager. I believe this also makes . This will make the code slightly cleaner.
...
- To expose the SplitFetcherManager / SingleThreadFetcheManager as Public, allowing connector developers to easily create their own threading models in the SourceReaderBase.
- To hide the element queue from the connector developers and simplify the SourceReaderBase to consist of only SplitFetcherManager and RecordEmitter as major components. make SplitFetcherManager the only owner class of the queue
Public Interfaces
SplitFetcherManager
- Change SplitFetcherManager from Internal to PublicEvolving.
- Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors as replacements which creates the FutureCompletingBlockingQueue instance internally.
- Add a few new methods to expose the functionality of the internal FutureCompletingBlockingQueue via the SplitFetcherManagerinternal method SplitFetcherManager#getQueue() for SourceReaderBase usage.
Code Block |
---|
@PublicEvolving public abstract class SplitFetcherManager<E, SplitT extends SourceSplit> { @Deprecated public SplitFetcherManagerSingleThreadMultiplexSourceReaderBase( Supplier<SplitReader<E, FutureCompletingBlockingQueue<RecordsWithSplitIds<E>>SplitT>> elementsQueuesplitReaderSupplier, RecordEmitter<E, Supplier<SplitReader<ET, SplitT>>SplitStateT> splitReaderFactoryrecordEmitter, Configuration config, configuration) { this(elementsQueue, splitReaderFactory, configuration,SourceReaderContext (ignorecontext) -> { }); super( } @Deprecated new @VisibleForTestingSingleThreadFetcherManager<>(splitReaderSupplier, config), public SplitFetcherManager( recordEmitter, FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, Supplier<SplitReader<E, SplitT>> splitReaderFactoryconfig, Configuration configuration, context); } @Deprecated Consumer<Collection<String>> splitFinishedHook) { } // todo: provide a new constructor without FutureCompletingBlockingQueue. public SplitFetcherManager(public SplitFetcherManager( FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, Supplier<SplitReader<E, SplitT>> splitReaderFactory, Configuration configuration) { this(elementsQueue, splitReaderFactory, configuration, (ignore) -> { }); } } //@Deprecated todo: provide a new@VisibleForTesting constructor without FutureCompletingBlockingQueue. publicpublic SplitFetcherManager( FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, Supplier<SplitReader<E, SplitT>> splitReaderFactory, Configuration configuration, Consumer<Collection<String>> splitFinishedHook) { } // todo: provide a new constructor without this.elementsQueue = new FutureCompletingBlockingQueue<>FutureCompletingBlockingQueue. public SplitFetcherManager( Supplier<SplitReader<E, SplitT>> splitReaderFactory, configuration.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY)); // ...... } /** * returns the RecordsWithSplitIds produced by SplitReader. **/ public RecordsWithSplitIds<E> poll(){ return elementsQueue.poll(); } /** * Returns the availability future. If the queue is non-empty, then this future will already be * complete. Otherwise the obtained future is guaranteed to get completed the next time the * queue becomes non-empty, or a notification happens via {@link #notifyAvailable()}. * Configuration configuration) { this(splitReaderFactory, configuration, (ignore) -> { }); } // todo: provide a new constructor without FutureCompletingBlockingQueue. public SplitFetcherManager( Supplier<SplitReader<E, SplitT>> splitReaderFactory, Configuration configuration, * <p>It is important that a completed futureConsumer<Collection<String>> issplitFinishedHook) no{ guarantee that the next call to {@link this.elementsQueue = new FutureCompletingBlockingQueue<>( * #poll()} will return a non-null element. If there are concurrent consumer, another consumer configuration.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY)); // ...... } @Internal public FutureCompletingBlockingQueue * may have taken the available element. Or there was no element in the first place, because the * future was completed through a call to {@link #notifyAvailable()}.getQueue(); } |
SingleThreadFetcherManager
- Change SingleThreadFetcherManager from Internal to PublicEvolving.
- Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors without FutureCompletingBlockingQueue as replacements which creates the FutureCompletingBlockingQueue instance by its parent class(SplitFetcherManager) internally.
Code Block |
---|
@PublicEvolving public class SingleThreadFetcherManager<E, SplitT extends SourceSplit> extends SplitFetcherManager<E, SplitT> { @Deprecated public SingleThreadFetcherManager( * * <p>For thatFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> reasonelementsQueue, it is important to call this method (to obtain aSupplier<SplitReader<E, newSplitT>> futuresplitReaderSupplier) every time{ * again after {@link #poll()} returned null and you want to wait for data. this(elementsQueue, splitReaderSupplier, new Configuration()); } @Deprecated */ public CompletableFuture<Void> getAvailabilityFutureSingleThreadFetcherManager(){ return elementsQueue.getAvailabilityFuture(); } /** * Makes sureFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> theelementsQueue, availability future is complete, if it is not complete already.Supplier<SplitReader<E, AllSplitT>> futuressplitReaderSupplier, * returned by previous calls to {@linkConfiguration #getAvailabilityFuture()} are guaranteed to be configuration) { * completed.super(elementsQueue, splitReaderSupplier, configuration); *} @Deprecated * <p>All future calls@VisibleForTesting to the method willpublic returnSingleThreadFetcherManager( a completed future, until the point that the FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, * availability is reset via calls to {@link #poll()} that leave theSupplier<SplitReader<E, queueSplitT>> empty.splitReaderSupplier, */ public void notifyAvailable(){ elementsQueue.notifyAvailable(); } /** Checks whetherConfiguration isconfiguration, no data available. */ public boolean noAvailableElement(){ return elementsQueue.isEmpty(); } } |
SingleThreadFetcherManager
- Change SingleThreadFetcherManager from Internal to PublicEvolving.
- Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors without FutureCompletingBlockingQueue as replacements which creates the FutureCompletingBlockingQueue instance by its parent class(SplitFetcherManager) internally.
Code Block |
---|
@PublicEvolving public class SingleThreadFetcherManager<E, SplitT extends SourceSplit> extends SplitFetcherManager<E, SplitT> { @Deprecated public SingleThreadFetcherManager( Consumer<Collection<String>> splitFinishedHook) { super(elementsQueue, splitReaderSupplier, configuration, splitFinishedHook); } // todo: provide a new constructor without FutureCompletingBlockingQueue. public SingleThreadFetcherManager( Supplier<SplitReader<E, SplitT>> splitReaderSupplier) { super(splitReaderSupplier, new Configuration()); FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, } // todo: provide a new constructor without FutureCompletingBlockingQueue. Supplier<SplitReader<E, SplitT>> splitReaderSupplier)public {SingleThreadFetcherManager( this(elementsQueue, splitReaderSupplier, new Configuration()); Supplier<SplitReader<E, SplitT>> } splitReaderSupplier, @Deprecated public SingleThreadFetcherManager( Configuration configuration) { FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, Supplier<SplitReader<E, SplitT>> splitReaderSupplier, super(splitReaderSupplier, configuration); } // todo: provide a new constructor without FutureCompletingBlockingQueue. public SingleThreadFetcherManager( ConfigurationSupplier<SplitReader<E, configuration)SplitT>> {splitReaderSupplier, super(elementsQueue, splitReaderSupplier, configuration); Configuration }configuration, @Deprecated @VisibleForTesting public SingleThreadFetcherManager( FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,Consumer<Collection<String>> splitFinishedHook) { Supplier<SplitReader<E, SplitT>> splitReaderSupplier, Configuration configuration, Consumer<Collection<String>> splitFinishedHook) { super(elementsQueue, splitReaderSupplier, configuration, splitFinishedHook); } // todo: provide a new constructor without FutureCompletingBlockingQueue. public SingleThreadFetcherManager( Supplier<SplitReader<E, SplitT>> splitReaderSupplier) { super(splitReaderSupplier, configuration, splitFinishedHook); } } |
SplitFetcherTask
Change SplitFetcherTask from Internal to PublicEvolving.
Code Block |
---|
@PublicEvolving
public interface SplitFetcherTask {} |
SplitFetcher
Change SplitFetcher from Internal to PublicEvolving, but still not to expose the construct of SplitFetcher, so it can only created by org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager#createSplitFetcher
Code Block |
---|
@PublicEvolving public class SplitFetcher<E, SplitT extends SourceSplit> implements Runnable { SplitFetcher( this(splitReaderSupplier, new Configuration()); int id, } // todo: provide a new constructor without FutureCompletingBlockingQueue. publicFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> SingleThreadFetcherManager(elementsQueue, Supplier<SplitReader<ESplitReader<E, SplitT>>SplitT> splitReaderSuppliersplitReader, Configuration configuration) {Consumer<Throwable> errorHandler, super(splitReaderSupplier, configuration); Runnable } shutdownHook, // todo: provide a new constructor without FutureCompletingBlockingQueue. publicConsumer<Collection<String>> SingleThreadFetcherManager(splitFinishedHook, boolean allowUnalignedSourceSplits); } |
the constructor of SplitFetcher to the end users?
SourceReaderBase
Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors as replacements which creates the FutureCompletingBlockingQueue instance in SplitFetcherManager internally.
Code Block |
---|
@PublicEvolving public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitStateT> implements SourceReader<T, SplitT> { Supplier<SplitReader<E, SplitT>> splitReaderSupplier, Configuration configuration, @Deprecated public SourceReaderBase( Consumer<Collection<String>> splitFinishedHook) { FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> super(splitReaderSupplierelementsQueue, configuration, splitFinishedHook); } } |
SplitFetcherTask
Change SplitFetcherTask from Internal to PublicEvolving.
Code Block |
---|
@PublicEvolving public interface SplitFetcherTask {} |
SplitFetcher
Change SplitFetcher from Internal to PublicEvolving, but still not to expose the construct of SplitFetcher, so it can only created by org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager#createSplitFetcher
Code Block |
---|
@PublicEvolving public class SplitFetcher<E, SplitT extends SourceSplit> implements Runnable { SplitFetcher( SplitFetcherManager<E, SplitT> splitFetcherManager, RecordEmitter<E, T, SplitStateT> recordEmitter, int id Configuration config, SourceReaderContext FutureCompletingBlockingQueue<RecordsWithSplitIds<E>>context) elementsQueue,{ this(elementsQueue, splitFetcherManager, recordEmitter, null, SplitReader<Econfig, SplitT> splitReader, context); } @Deprecated Consumer<Throwable>public errorHandler,SourceReaderBase( RunnableFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> shutdownHookelementsQueue, SplitFetcherManager<E, Consumer<Collection<String>>SplitT> splitFinishedHooksplitFetcherManager, RecordEmitter<E, T, boolean allowUnalignedSourceSplits); } |
the constructor of SplitFetcher to the end users?
SourceReaderBase
- Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors as replacements which creates the FutureCompletingBlockingQueue instance in SplitFetcherManager internally.
- Remove the direct operations to FutureCompletingBlockingQueue and utilize the functionality of SplitFetcherManager as an alternative.
Code Block |
---|
@PublicEvolving public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitStateT> SplitStateT> recordEmitter, @Nullable RecordEvaluator<T> eofRecordEvaluator, implements SourceReader<T, SplitT> { Configuration @Deprecatedconfig, public SourceReaderBase( SourceReaderContext context) { FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, //this.elementsQueue = elementsQueue; SplitFetcherManager<E, SplitT> this.splitFetcherManager, = splitFetcherManager; this.recordEmitter RecordEmitter<E, T, SplitStateT> recordEmitter,= recordEmitter; this.splitStates = Configuration config,new HashMap<>(); this.options = SourceReaderContext context) {new SourceReaderOptions(config); this(elementsQueue, splitFetcherManager, recordEmitter, null, config, context).config = config; } this.context = @Deprecatedcontext; public SourceReaderBase( this.noMoreSplitsAssignment = false; FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, this.eofRecordEvaluator = eofRecordEvaluator; numRecordsInCounter = context.metricGroup().getIOMetricGroup().getNumRecordsInCounter(); SplitFetcherManager<E,} SplitT> splitFetcherManager, // todo: provide a new constructor without FutureCompletingBlockingQueue RecordEmitter<E, T, SplitStateT>public recordEmitter,SourceReaderBase( @NullableSplitFetcherManager<E, RecordEvaluator<T>SplitT> eofRecordEvaluatorsplitFetcherManager, RecordEmitter<E, T, ConfigurationSplitStateT> configrecordEmitter, SourceReaderContext context) { Configuration config, //this.elementsQueue = elementsQueue;SourceReaderContext context) { this.(splitFetcherManager = splitFetcherManager, recordEmitter, null, config, context); } this.recordEmitter = recordEmitter; this.splitStates = new HashMap<>();// todo: provide a new constructor without FutureCompletingBlockingQueue public SourceReaderBase( this.options = new SourceReaderOptions(config); SplitFetcherManager<E, SplitT> splitFetcherManager, this.config = config; RecordEmitter<E, T, SplitStateT> recordEmitter, this.context = context; @Nullable RecordEvaluator<T> eofRecordEvaluator, this.noMoreSplitsAssignment = false; this.eofRecordEvaluator =Configuration eofRecordEvaluator;config, numRecordsInCounter =SourceReaderContext context.metricGroup().getIOMetricGroup().getNumRecordsInCounter();) { } this.splitFetcherManager = splitFetcherManager; // todo: provide a new constructor without FutureCompletingBlockingQueue this.recordEmitter = recordEmitter; public SourceReaderBase( this.splitStates = new HashMap<>(); SplitFetcherManager<E, SplitT> splitFetcherManager, this.options = new SourceReaderOptions(config); RecordEmitter<E, T, SplitStateT> recordEmitter, this.config = config; this.context = context; Configuration config, this.noMoreSplitsAssignment = false; SourceReaderContext context) {this.eofRecordEvaluator = eofRecordEvaluator; this(splitFetcherManager, recordEmitter, null,numRecordsInCounter config,= context.metricGroup().getIOMetricGroup().getNumRecordsInCounter(); } } |
SingleThreadMultiplexSourceReaderBase
Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors as replacements which creates the FutureCompletingBlockingQueue instance in SplitFetcherManager internally.
Code Block |
---|
@PublicEvolving public abstract class SingleThreadMultiplexSourceReaderBase<E, T, SplitT extends SourceSplit, SplitStateT> extends SourceReaderBase<E, T, SplitT, SplitStateT> { @Depricated public SingleThreadMultiplexSourceReaderBase // todo: provide a new constructor without FutureCompletingBlockingQueue public SourceReaderBase( SplitFetcherManager<E, SplitT> splitFetcherManagerFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, RecordEmitter<ESingleThreadFetcherManager<E, T, SplitStateT> recordEmitterSplitT> splitFetcherManager, @Nullable RecordEvaluator<T> eofRecordEvaluatorRecordEmitter<E, T, SplitStateT> recordEmitter, Configuration config, SourceReaderContext context) { super(elementsQueue, this.splitFetcherManager, = splitFetcherManager;recordEmitter, config, context); } @Depricated public SingleThreadMultiplexSourceReaderBase( this.recordEmitter = recordEmitter; FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, this.splitStates = new HashMap<>(); Supplier<SplitReader<E, SplitT>> splitReaderSupplier, this.options = new SourceReaderOptions(config); RecordEmitter<E, T, SplitStateT> recordEmitter, this.config = config; this.context =Configuration context;config, this.noMoreSplitsAssignment = false; SourceReaderContext context) { this.eofRecordEvaluator = eofRecordEvaluator;super( numRecordsInCounter = context.metricGroup().getIOMetricGroup().getNumRecordsInCounter(); } } |
SingleThreadMultiplexSourceReaderBase
Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors as replacements which creates the FutureCompletingBlockingQueue instance in SplitFetcherManager internally.
Code Block |
---|
@PublicEvolving public abstract class SingleThreadMultiplexSourceReaderBase<E, T, SplitT extends SourceSplit, SplitStateT> extends SourceReaderBase<E, T, SplitT, SplitStateT> { @Depricated public SingleThreadMultiplexSourceReaderBase( elementsQueue, new SingleThreadFetcherManager<>(elementsQueue, splitReaderSupplier, config), FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueuerecordEmitter, SingleThreadFetcherManager<E, SplitT> splitFetcherManagerconfig, RecordEmitter<E, T, SplitStateT> recordEmitter, context); } // todo: Add new constructors without FutureCompletingBlockingQueue public SingleThreadMultiplexSourceReaderBase( Configuration config, Supplier<SplitReader<E, SplitT>> splitReaderSupplier, SourceReaderContext context) { RecordEmitter<E, super(elementsQueueT, splitFetcherManager,SplitStateT> recordEmitter, config, context); } @Depricated public SingleThreadMultiplexSourceReaderBase( Configuration FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueueconfig, Supplier<SplitReader<E,SourceReaderContext SplitT>>context) splitReaderSupplier,{ super( RecordEmitter<E, T, SplitStateT> recordEmitter, new ConfigurationSingleThreadFetcherManager<>(splitReaderSupplier, config),, SourceReaderContext context) { recordEmitter, super( config, elementsQueue, context); } // todo: provide a new SingleThreadFetcherManager<>(elementsQueue, splitReaderSupplier, config), constructor without FutureCompletingBlockingQueue public SingleThreadMultiplexSourceReaderBase( SingleThreadFetcherManager<E, SplitT> recordEmittersplitFetcherManager, RecordEmitter<E, T, SplitStateT> configrecordEmitter, @Nullable RecordEvaluator<T> eofRecordEvaluator, context); } // todo: Add new constructors without FutureCompletingBlockingQueue public SingleThreadMultiplexSourceReaderBase( Configuration config, Supplier<SplitReader<E,SourceReaderContext SplitT>>context) splitReaderSupplier,{ super( RecordEmitter<E, T, SplitStateT> recordEmitter, splitFetcherManager, Configuration config, recordEmitter, SourceReaderContext context) { super( eofRecordEvaluator, new SingleThreadFetcherManager<>(splitReaderSupplier, config),, recordEmitter, config, context); } // todo: provide a new constructor without FutureCompletingBlockingQueue public SingleThreadMultiplexSourceReaderBase( SingleThreadFetcherManager<E, SplitT> splitFetcherManager, RecordEmitter<E, T, SplitStateT> recordEmitter, @Nullable RecordEvaluator<T> eofRecordEvaluator, Configuration config, SourceReaderContext context) { super( splitFetcherManager, recordEmitter, eofRecordEvaluator, config, context); } } |
Proposed Changes
- By exposing the SplitFetcherManager / SingleThreadFetcheManager, connector developers can easily create their own threading models in the SourceReaderBase, by implementing addSplits(), removeSplits() and
maybeShutdownFinishedFetchers() functions.
- Note that the SplitFetcher constructor is package private, so users can only create SplitFetchers via SplitFetcherManager.createSplitFetcher(). This ensures each SplitFetcher is always owned by the SplitFetcherManager.
- This FLIP essentially embedded the element queue (a FutureCompletingBlockingQueue) instance into the SplitFetcherManager. This hides the element queue from the connector developers and simplifies the SourceReaderBase to consist of only SplitFetcherManager and RecordEmitter as major components.
Compatibility, Deprecation, and Migration Plan
The Connectors that utilize constructors (including param elementsQueue) of SingleThreadMultiplexSourceReaderBase, SourceReaderBase, and SingleThreadFetcherManager should be migrated to the new constructor that does not have elementsQueue parameter.
Mark the impacted constructors as deprecated in 1.19, and remove them in release of 2.0.
Test Plan
nothing else to do.
Rejected Alternatives
change SplitFetcher to a public Interface
If SplitFetcherManager becomes PublicEvolving, that also means SplitFetcher needs to be PublicEvolving, because it is returned by the protected method SplitFetcherManager.createSplitFetcher()
However, SplitFetcher requires FutureCompletingBlockingQueue as a constructor parameter. SplitFetcher is a class rather than Interface. Therefore, I want to change SplitFetcher to a public Interface and moving its implementation details to an implement subclass .
Reject Reason
} |
Proposed Changes
- By exposing the SplitFetcherManager / SingleThreadFetcheManager, connector developers can easily create their own threading models in the SourceReaderBase, by implementing addSplits(), removeSplits() and
maybeShutdownFinishedFetchers() functions.
- Note that the SplitFetcher constructor is package private, so users can only create SplitFetchers via SplitFetcherManager.createSplitFetcher(). This ensures each SplitFetcher is always owned by the SplitFetcherManager.
- This FLIP essentially embedded the element queue (a FutureCompletingBlockingQueue) instance into the SplitFetcherManager. This hides the element queue from the connector developers and simplifies the SourceReaderBase to consist of only SplitFetcherManager and RecordEmitter as major components.
Compatibility, Deprecation, and Migration Plan
The Connectors that utilize constructors (including param elementsQueue) of SingleThreadMultiplexSourceReaderBase, SourceReaderBase, and SingleThreadFetcherManager should be migrated to the new constructor that does not have elementsQueue parameter.
Mark the impacted constructors as deprecated in 1.19, and remove them in release of 2.0.
Test Plan
nothing else to do.
Rejected Alternatives
change SplitFetcher to a public Interface
If SplitFetcherManager becomes PublicEvolving, that also means SplitFetcher needs to be PublicEvolving, because it is returned by the protected method SplitFetcherManager.createSplitFetcher()
However, SplitFetcher requires FutureCompletingBlockingQueue as a constructor parameter. SplitFetcher is a class rather than Interface. Therefore, I want to change SplitFetcher to a public Interface and moving its implementation details to an implement subclass .
Reject Reason
The constructor of the SplitFetcher is already package private. So it can only be accessed from the classes in the package org.apache.flink.connector.base.source.reader.fetcher. And apparently, user classes should not be in this package. Therefore, even if we mark the
SplitFetcher class as PublicEvolving, the constructor is not available to the users. Only the public and protected methods are considered public API
in this case. Private / package private methods and fields are still internal.
SplitFetcherManager expose poll / getAvailabilityFuture / notifyAvailable / noAvailableElement
- Change SplitFetcherManager from Internal to PublicEvolving.
- Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors as replacements which creates the FutureCompletingBlockingQueue instance internally.
- Add a new internal method SplitFetcherManager#getQueue() for SourceReaderBase usage.
|
Reject Reason
SplitFetcherManager exposes 4 more methods, poll / getAvailabilityFuture / notifyAvailable / noAvailableElement, which are tightly coupled with the implementation of the elementQueue. The naming of these methods look weird, like what does it mean to "poll from a SplitFetcherManager" / "notify a SplitFetcherManager available"? To clarify these methods we have to explain to developers that "well we hide a queue inside SplitFetcherMamager and the poll method is actually polling from the queue". I'm afraid these methods will implicitly expose the concept and the implementation of the queue to developers.
A cleaner solution would be having a new interface that extracts SplitFetcher creating and managing logic from the current SplitFetcherManager, but having too many concepts might make the entire Source API even harder to understand. To make a compromise, only expose constructors of SplitFetcherManager as public APIs, and adding a new internal method SplitFetcherManager#getQueue() for SourceReaderBase (well it's a bit hacky , more worthy than exposing methods like poll and notifyAvailable on SplitFetcherManager )The constructor of the SplitFetcher is already package private. So it can only be accessed from the classes in the package org.apache.flink.connector.base.source.reader.fetcher. And apparently, user classes should not be in this package. Therefore, even if we mark the
SplitFetcher class as PublicEvolving, the constructor is not available to the users. Only the public and protected methods are considered public API
in this case. Private / package private methods and fields are still internal.