...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Why is it necessary to make SingleThreadFetcherManager PublicEvolving?

1. Though

...

1. the SingleThreadFetcherManager is annotated as Internal, it actually acts as some-degree public API, which is widely used in many connector projects:

...

1. flink-

...

1. connector-

...

1. kafka,  flink-connector-mongodb, flink-cdc-connectors and soon. In flink-connector-kafka, KafkaSourceFetcherManager even extends SingleThreadFetcherManager.

2. More over, even the constructor

...

1. of SingleThreadMultiplexSourceReaderBase

...

1. (which is PublicEvolving) includes the params of SingleThreadFetcherManager 

...

1. and FutureCompletingBlockingQueue

...

1. . That means that the SingleThreadFetcherManager

...

1. and FutureCompletingBlockingQueue

...

1. have already been exposed to users for a long time and are widely used.

...

1. As shown in FLINK-31324, FLINK-28853 used to change the default constructor of SingleThreadFetcherManager.However, it influenced a lot. Finally, the former constructor was added back and marked asDeprecated.

   Code Block
   public SingleThreadMultiplexSourceReaderBase( FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue, SingleThreadFetcherManager<E, SplitT> splitFetcherManager, RecordEmitter<E, T, SplitStateT> recordEmitter, Configuration config, SourceReaderContext context) { super(elementsQueue, splitFetcherManager, recordEmitter, config, context); }

...

1. 
   

2. 
   

   The original design of SplitFetcherManager and its subclasses was to make them public to the Source developers. The

As shown in FLINK-31324, FLINK-28853 used to change the default constructor of SingleThreadFetcherManager.However, it influenced a lot. Finally, the former constructor was added back and marked asDeprecated。

Therefore, why not make SingleThreadFetcherManager PublicEvolving?

Let's go back to more contexts from the origin design: 

...

1. goal is to let us take care of the threading model, while the Source developers can just focus on the SplitReader implementation. Therefore, I think making SplitFetcherManater / SingleThreadFetcherManager public aligns with the original design. That is also why these classes are exposed in the constructor of SourceReaderBase.
   

   
   

2. 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.

In summary, this flip has 2 goals:

• 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 make SplitFetcherManager the only owner class of the queue
• Annotate SingleThreadFetcherManager as PublicEvolving.
• Shield FutureCompletingBlockingQueue from users and limit all operations on FutureCompletingBlockingQueue in FetcherManager.

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 SplitFetcherManager
• internal method SplitFetcherManager#getQueue() for SourceReaderBase usage.

@PublicEvolving
public abstract class SplitFetcherManager<E, SplitT extends SourceSplit> {    
    
	@Deprecated
    public SplitFetcherManagerSingleThreadMultiplexSourceReaderBase(
            FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueueSupplier<SplitReader<E, SplitT>> splitReaderSupplier,
            Supplier<SplitReader<ERecordEmitter<E, T, SplitT>>SplitStateT> splitReaderFactoryrecordEmitter,
            Configuration configuration) {
config,
         this(elementsQueue, splitReaderFactory, configuration, (ignore SourceReaderContext context) -> {
        });super(
    }       

    @Deprecated
    @VisibleForTesting
  new SingleThreadFetcherManager<>(splitReaderSupplier, config),
  public SplitFetcherManager(
            FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueuerecordEmitter,
            Supplier<SplitReader<E,   SplitT>> splitReaderFactoryconfig,
             Configuration configuration,   context);
    }

    @Deprecated
    public SplitFetcherManager(
  Consumer<Collection<String>> splitFinishedHook) {

	}



 // todo: provide a new constructor without FutureCompletingBlockingQueue.     
 public SplitFetcherManager(FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
            Supplier<SplitReader<E, SplitT>> splitReaderFactory,
            Configuration configuration) {
        this(elementsQueue, splitReaderFactory, configuration, (ignore) -> {
        });
    }       

  }

   @Deprecated
    @VisibleForTesting
    public SplitFetcherManager(
            FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
            Supplier<SplitReader<E, SplitT>> splitReaderFactory,
            Configuration configuration,
            Consumer<Collection<String>> splitFinishedHook) {
        this.elementsQueue = new FutureCompletingBlockingQueue<>

	}



 // todo: provide a new constructor without 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();
	}

 	/**
          Configuration configuration) {
        this(splitReaderFactory, configuration, (ignore) -> {
        });
      * Returns the
 availability future.}

 If the// queuetodo: is non-empty, then this future will already beprovide a new constructor without FutureCompletingBlockingQueue.
  public SplitFetcherManager(
     * complete. Otherwise the obtained future is guaranteedSupplier<SplitReader<E, toSplitT>> getsplitReaderFactory,
 completed the next time the
     * queue becomesConfiguration non-emptyconfiguration,
 or a notification happens   via {@link #notifyAvailable()}.
   Consumer<Collection<String>> splitFinishedHook) *{
     * <p>It is importantthis.elementsQueue that= anew completedFutureCompletingBlockingQueue<>(
 future is no guarantee that the next call to {@link
     * #poll()} will return a non-null element. If there are concurrent consumer, another consumer
     * 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()}.configuration.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY));
		// ......
	}

	
	@Internal
	public FutureCompletingBlockingQueue 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.

@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 */SingleThreadFetcherManager(
	public CompletableFuture<Void> getAvailabilityFuture(){
 		return elementsQueue.getAvailabilityFuture();
	}

     /**
     *FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> MakeselementsQueue,
 sure the 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
    @VisibleForTesting
 *  <p>All futurepublic callsSingleThreadFetcherManager(
 to the method will return a completed future, until the point thatFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> theelementsQueue,
     * availability is reset via calls to {@link #poll()} that leave the queue empty.
  Supplier<SplitReader<E, SplitT>> splitReaderSupplier,
           */
	public void notifyAvailable(){
		elementsQueue.notifyAvailable();
	}

Configuration configuration,
   /** Checks whether is no data available. */
 	public booleanConsumer<Collection<String>> noAvailableElement(splitFinishedHook) {
		return elementsQueue.isEmpty(        super(elementsQueue, splitReaderSupplier, configuration, splitFinishedHook);
 	   }
}

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.

@PublicEvolving
public class SingleThreadFetcherManager<E, SplitT extends SourceSplit>


  // todo: provide a new constructor without FutureCompletingBlockingQueue.     
  public SingleThreadFetcherManager(
            extendsSupplier<SplitReader<E, SplitFetcherManager<E,SplitT>> SplitT>splitReaderSupplier) {  

   	@Deprecated
    public SingleThreadFetcherManagersuper(
splitReaderSupplier, new Configuration());
    }  

 // todo: provide  FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,a new constructor without FutureCompletingBlockingQueue. 
     public SingleThreadFetcherManager(
            Supplier<SplitReader<E, SplitT>> splitReaderSupplier,
            Configuration configuration) {
        thissuper(elementsQueue, splitReaderSupplier, new Configuration(configuration));
    }

  	@Deprecated
    public SingleThreadFetcherManager(
     // todo: provide a new constructor without FutureCompletingBlockingQueue. 
     public SingleThreadFetcherManager(
            FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueueSupplier<SplitReader<E, SplitT>> splitReaderSupplier,
            Supplier<SplitReader<E, SplitT>> splitReaderSupplierConfiguration configuration,
            ConfigurationConsumer<Collection<String>> configurationsplitFinishedHook) {
        super(elementsQueuesplitReaderSupplier, splitReaderSupplierconfiguration, configurationsplitFinishedHook);
    }

    @Deprecated
    @VisibleForTesting
    public SingleThreadFetcherManager(
            FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue}

SplitFetcherTask

Change SplitFetcherTask from Internal to PublicEvolving.

@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

@PublicEvolving
 public class SplitFetcher<E, SplitT extends SourceSplit> implements Runnable {    
	SplitFetcher(,
            Supplier<SplitReader<E, SplitT>> splitReaderSupplierint id,
            ConfigurationFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> configurationelementsQueue,
            Consumer<Collection<String>>SplitReader<E, splitFinishedHook)SplitT> {splitReader,
        super(elementsQueue, splitReaderSupplier, configuration, splitFinishedHook);
 Consumer<Throwable> errorHandler,
  }

  // todo: provide a new constructor without FutureCompletingBlockingQueue.    Runnable shutdownHook,
  public SingleThreadFetcherManager(
         Consumer<Collection<String>> splitFinishedHook,
  Supplier<SplitReader<E, SplitT>> splitReaderSupplier) {
        this(splitReaderSupplier, new Configuration()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.

@PublicEvolving
public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitStateT>
    }


    public SingleThreadFetcherManager(
           implements Supplier<SplitReader<E, SplitT>> splitReaderSupplier,SourceReader<T, SplitT> {  
 
 
    @Deprecated
    public SourceReaderBase(
   Configuration configuration) {
       FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> super(splitReaderSupplierelementsQueue, configuration);

         }

   SplitFetcherManager<E, publicSplitT> SingleThreadFetcherManager(splitFetcherManager,
            Supplier<SplitReader<ERecordEmitter<E, T, SplitT>>SplitStateT> splitReaderSupplierrecordEmitter,
            Configuration configurationconfig,
            Consumer<Collection<String>>SourceReaderContext splitFinishedHookcontext) {
        superthis(splitReaderSupplierelementsQueue, configurationsplitFetcherManager, splitFinishedHookrecordEmitter, null, config, context);
    }
}

SplitFetcherTask

Change SplitFetcherTask from Internal to PublicEvolving.

@PublicEvolving
public interface SplitFetcherTask {}

SplitFetcher

Change SplitFetcher from Internal to PublicEvolving, but still not to expose the construct of SplitFetcher.

@PublicEvolving
class FetchTask<E, SplitT extends SourceSplit> implements SplitFetcherTask { 
 
    @Deprecated
    public SourceReaderBase(
    private final SplitReader<E, SplitT> splitReader;
    private final FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue;,
    private final Consumer<Collection<String>> splitFinishedCallback;
    private finalSplitFetcherManager<E, intSplitT> fetcherIndex;splitFetcherManager,
    private volatile RecordsWithSplitIds<E> lastRecords;
    private volatileRecordEmitter<E, boolean wakeup;

    FetchTask(T, SplitStateT> recordEmitter,
            SplitReader<E,@Nullable SplitT>RecordEvaluator<T> splitReadereofRecordEvaluator,
            FutureCompletingBlockingQueue<RecordsWithSplitIds<E>>Configuration elementsQueueconfig,
            Consumer<Collection<String>> splitFinishedCallback,
    SourceReaderContext context) {
        //this.elementsQueue = elementsQueue;
        int fetcherIndex) {this.splitFetcherManager = splitFetcherManager;
        this.splitReaderrecordEmitter = splitReaderrecordEmitter;
        this.elementsQueuesplitStates = new elementsQueueHashMap<>();
        this.splitFinishedCallbackoptions = splitFinishedCallback new SourceReaderOptions(config);
        this.lastRecordsconfig = nullconfig;
        this.fetcherIndexcontext = fetcherIndexcontext;
        this.wakeupnoMoreSplitsAssignment = false;
    }
}

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.

@PublicEvolving
public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitStateT>    this.eofRecordEvaluator = eofRecordEvaluator;
 
        implementsnumRecordsInCounter SourceReader<T, SplitT> {  
 
 
    @Deprecated= context.metricGroup().getIOMetricGroup().getNumRecordsInCounter();
    } 
 
    public SourceReaderBase(
      // todo: provide a new constructor without FutureCompletingBlockingQueue
      FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,public SourceReaderBase(
            SplitFetcherManager<E, SplitT> splitFetcherManager,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            Configuration config,
            SourceReaderContext context) {
        this(elementsQueue, splitFetcherManager, recordEmitter, null, config, context);
    }
 
 
    @Deprecated
 // todo: provide publica SourceReaderBase(
new constructor without FutureCompletingBlockingQueue
     public SourceReaderBase(
     FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
            SplitFetcherManager<E, SplitT> splitFetcherManager,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            @Nullable RecordEvaluator<T> eofRecordEvaluator,
            Configuration config,
            SourceReaderContext context) {
        //this.elementsQueue = elementsQueue;
        this.splitFetcherManager = splitFetcherManager;
        this.recordEmitter = recordEmitter;
        this.splitStates = new HashMap<>();
        this.options = new SourceReaderOptions(config);
        this.config = config;
        this.context = context;
        this.noMoreSplitsAssignment = false;
        this.eofRecordEvaluator = eofRecordEvaluator;
 
        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.

@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<EFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
            SingleThreadFetcherManager<E, SplitT> splitFetcherManager,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            Configuration config,
            SourceReaderContext context) {
        thissuper(elementsQueue, splitFetcherManager, recordEmitter, null, config, context);
    	}
 
 
    	@Depricated
	public SourceReaderBaseSingleThreadMultiplexSourceReaderBase(
            SplitFetcherManager<E, SplitT> splitFetcherManagerFutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
            RecordEmitter<ESupplier<SplitReader<E, T, SplitStateT> recordEmitterSplitT>> splitReaderSupplier,
            @Nullable RecordEvaluator<T> eofRecordEvaluatorRecordEmitter<E, T, SplitStateT> recordEmitter,
            Configuration config,
            SourceReaderContext context) {
        this.splitFetcherManager = splitFetcherManager;
super(
          this.recordEmitter = recordEmitter;
    elementsQueue,
    this.splitStates = new HashMap<>();
        this.options = new SourceReaderOptions(SingleThreadFetcherManager<>(elementsQueue, splitReaderSupplier, config);,
        this.config = config;
        this.context = context;
recordEmitter,
          this.noMoreSplitsAssignment = false;
    config,
    this.eofRecordEvaluator = eofRecordEvaluator;
 
        numRecordsInCounter = context.metricGroup().getIOMetricGroup().getNumRecordsInCounter();
	}


	// todo: Add  }
}

SingleThreadMultiplexSourceReaderBase

Deprecate the old constructor exposing the FutureCompletingBlockingQueue, and add new constructors as replacements which creates the FutureCompletingBlockingQueue instance in SplitFetcherManager internally.

@PublicEvolving
public abstract class SingleThreadMultiplexSourceReaderBase<E, T, SplitT extends SourceSplit, SplitStateT> extends SourceReaderBase<E, T, SplitT, SplitStateT> { 

	@Depricated
	public SingleThreadMultiplexSourceReaderBase(
new constructors without FutureCompletingBlockingQueue
 	public SingleThreadMultiplexSourceReaderBase(
            Supplier<SplitReader<E, SplitT>> splitReaderSupplier,
            RecordEmitter<E, FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueueT,
 SplitStateT> recordEmitter,
          SingleThreadFetcherManager<E,  SplitT>Configuration splitFetcherManagerconfig,
            RecordEmitter<E, T, SplitStateT> recordEmitter,SourceReaderContext context) {
        	super(
    Configuration config,
           new SourceReaderContext context) {
SingleThreadFetcherManager<>(splitReaderSupplier, config),,
          super(elementsQueue, splitFetcherManager, recordEmitter, config, context);
 	}

	@Depricated
	public SingleThreadMultiplexSourceReaderBase( recordEmitter,
            FutureCompletingBlockingQueue<RecordsWithSplitIds<E>>    elementsQueueconfig,
              Supplier<SplitReader<E, SplitT>> splitReaderSupplier,context);
    }

    // todo: provide a RecordEmitter<E,new T,constructor SplitStateT>without recordEmitter,FutureCompletingBlockingQueue
     public SingleThreadMultiplexSourceReaderBase(
        Configuration config,
   SingleThreadFetcherManager<E, SplitT> splitFetcherManager,
       SourceReaderContext context) {
   RecordEmitter<E, T,    super(SplitStateT> recordEmitter,
            @Nullable RecordEvaluator<T>   elementsQueueeofRecordEvaluator,
            Configuration config,
   new SingleThreadFetcherManager<>(elementsQueue, splitReaderSupplier, config),
      SourceReaderContext context) {
        recordEmitter,super(
                configsplitFetcherManager,
                context);
	}


	// todo: Add new constructors without FutureCompletingBlockingQueue
 	public SingleThreadMultiplexSourceReaderBase(
recordEmitter,
                eofRecordEvaluator,
              Supplier<SplitReader<E, SplitT>> splitReaderSupplierconfig,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            Configuration config,
            SourceReaderContext context) {;
        	super(
                new SingleThreadFetcherManager<>(splitReaderSupplier, config),,
                recordEmitter,
                config,
                context);
    }

    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, by implementing addSplits() and removeSplits(), connector developers can easily create their own threading models in the SourceReaderBase.
• 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

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.