PatrickRen commented on code in PR #24171:
URL: https://github.com/apache/flink/pull/24171#discussion_r1465985818
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManager.java:
##########
@@ -151,6 +157,60 @@ public void accept(Throwable t) {
this.closed = false;
}
+ /**
+ * Create a split fetcher manager.
+ *
+ * @param splitReaderFactory a supplier that could be used to create split
readers.
+ * @param configuration the configuration of this fetcher manager.
+ */
+ public SplitFetcherManager(
Review Comment:
Also `SplitFetcherManager` should be marked as `@PublicEvolving` now
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java:
##########
@@ -100,6 +104,43 @@ public SingleThreadFetcherManager(
super(elementsQueue, splitReaderSupplier, configuration,
splitFinishedHook);
}
+ /**
+ * Creates a new SplitFetcherManager with a single I/O threads.
+ *
+ * @param splitReaderSupplier The factory for the split reader that
connects to the source
+ * system.
+ */
+ public SingleThreadFetcherManager(Supplier<SplitReader<E, SplitT>>
splitReaderSupplier) {
Review Comment:
We need to mark the class `SingleThreadFetcherManager` as @PublicEvolving
##########
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java:
##########
@@ -136,6 +146,35 @@ public SourceReaderBase(
numRecordsInCounter =
context.metricGroup().getIOMetricGroup().getNumRecordsInCounter();
}
+ /**
+ * The primary constructor for the source reader.
+ *
+ * <p>The reader will use a handover queue sized as configured via {@link
+ * SourceReaderOptions#ELEMENT_QUEUE_CAPACITY}.
+ */
+ public SourceReaderBase(
+ SplitFetcherManager<E, SplitT> splitFetcherManager,
+ RecordEmitter<E, T, SplitStateT> recordEmitter,
+ Configuration config,
+ SourceReaderContext context) {
+ this(splitFetcherManager, recordEmitter, null, config, context);
+ }
+
+ public SourceReaderBase(
+ SplitFetcherManager<E, SplitT> splitFetcherManager,
+ RecordEmitter<E, T, SplitStateT> recordEmitter,
+ @Nullable RecordEvaluator<T> eofRecordEvaluator,
+ Configuration config,
+ SourceReaderContext context) {
+ this(
Review Comment:
Looks like this is still using a deprecated constructor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
