Implementation:Apache Flink SingleThreadMultiplexSourceReaderBase

Knowledge Sources	Apache_Flink
Domains	Connectors, Source_Framework
Last Updated	2026-02-09 00:00 GMT

Overview

An abstract base class for source readers that use a single I/O thread to read from multiple splits concurrently or sequentially.

Description

SingleThreadMultiplexSourceReaderBase is an abstract class that extends SourceReaderBase and provides a simplified foundation for building source readers that use a single fetching thread to handle all assigned splits. It automatically creates and manages a SingleThreadFetcherManager internally, relieving the connector developer from manually configuring the fetcher infrastructure.

This pattern supports two primary reading models: (1) sequential reading where splits are processed one after another (e.g., reading files in sequence), and (2) concurrent multiplexed reading where a single client handles multiple splits simultaneously by managing subscriptions (e.g., a single Kafka consumer reading from multiple topic partitions). The class is parameterized with four type parameters: the raw record type (E), the final output type (T), the immutable split type (SplitT), and the mutable split state type (SplitStateT).

The class provides five constructor variants to support different configuration scenarios, including constructors that accept a Supplier<SplitReader> for automatic fetcher management, constructors that accept a pre-built SingleThreadFetcherManager, and constructors that support RecordEvaluator for end-of-stream detection and RateLimiterStrategy for rate limiting.

Usage

Connector developers extend this class when building a source connector that uses a single I/O thread for all splits. The developer must provide: (1) a SplitReader implementation that connects to the external system, (2) a RecordEmitter that transforms raw records and updates split state, (3) implementations of the abstract methods initializedState() and toSplitType() for converting between immutable splits and mutable split state, and (4) implementations of start() and onSplitFinished() for lifecycle management. This class is the recommended base for connectors like file sources, Kafka sources, and other connectors where a single thread can efficiently manage all split reads.

Code Reference

Source Location

• Repository: Apache_Flink
• File: flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SingleThreadMultiplexSourceReaderBase.java
• Lines: 1-153

Signature

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

    // Primary constructor: auto-creates SingleThreadFetcherManager from supplier
    public SingleThreadMultiplexSourceReaderBase(
            Supplier<SplitReader<E, SplitT>> splitReaderSupplier,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            Configuration config,
            SourceReaderContext context);

    // Constructor with rate limiter strategy
    public SingleThreadMultiplexSourceReaderBase(
            Supplier<SplitReader<E, SplitT>> splitReaderSupplier,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            Configuration config,
            SourceReaderContext context,
            @Nullable RateLimiterStrategy<SplitT> rateLimiterStrategy);

    // Constructor with explicit fetcher manager
    public SingleThreadMultiplexSourceReaderBase(
            SingleThreadFetcherManager<E, SplitT> splitFetcherManager,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            Configuration config,
            SourceReaderContext context);

    // Constructor with explicit fetcher manager and record evaluator
    public SingleThreadMultiplexSourceReaderBase(
            SingleThreadFetcherManager<E, SplitT> splitFetcherManager,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            @Nullable RecordEvaluator<T> eofRecordEvaluator,
            Configuration config,
            SourceReaderContext context);

    // Constructor with explicit fetcher manager, record evaluator, and rate limiter
    public SingleThreadMultiplexSourceReaderBase(
            SingleThreadFetcherManager<E, SplitT> splitFetcherManager,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            @Nullable RecordEvaluator<T> eofRecordEvaluator,
            Configuration config,
            SourceReaderContext context,
            @Nullable RateLimiterStrategy<SplitT> rateLimiterStrategy);
}

Import

import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;

I/O Contract

Inputs

Name	Type	Required	Description
splitReaderSupplier	Supplier<SplitReader<E, SplitT>>	Yes (variant 1)	A factory that creates SplitReader instances to connect to the external source system.
splitFetcherManager	SingleThreadFetcherManager<E, SplitT>	Yes (variant 2)	A pre-built fetcher manager for more advanced configuration scenarios.
recordEmitter	RecordEmitter<E, T, SplitStateT>	Yes	The emitter that transforms raw records and updates split state.
eofRecordEvaluator	RecordEvaluator<T>	No	An optional evaluator that determines when a split has reached end-of-stream based on record content.
config	Configuration	Yes	Flink configuration including source reader options such as queue capacity and close timeout.
context	SourceReaderContext	Yes	The context providing access to runtime information such as metric groups and parallelism.
rateLimiterStrategy	RateLimiterStrategy<SplitT>	No	An optional rate limiter to throttle record consumption.

Outputs

Name	Type	Description
(inherited)	-	As an abstract class extending SourceReaderBase, outputs depend on the concrete implementation. Records of type T are emitted via the RecordEmitter to the SourceOutput.

Usage Examples

// Example: A concrete source reader using SingleThreadMultiplexSourceReaderBase
public class MySourceReader
        extends SingleThreadMultiplexSourceReaderBase<
                RawRecord, OutputRecord, MySplit, MySplitState> {

    public MySourceReader(
            SourceReaderContext context,
            Configuration config) {
        super(
                () -> new MySplitReader(),   // SplitReader supplier
                new MyRecordEmitter(),        // RecordEmitter
                config,
                context);
    }

    @Override
    protected MySplitState initializedState(MySplit split) {
        return new MySplitState(split);
    }

    @Override
    protected MySplit toSplitType(String splitId, MySplitState splitState) {
        return splitState.toSplit();
    }

    @Override
    public void start() {
        // Request initial split assignment from the enumerator
        context().sendSplitRequest();
    }

    @Override
    protected void onSplitFinished(Map<String, MySplitState> finishedSplitIds) {
        // Request more splits when current ones are finished
        context().sendSplitRequest();
    }
}

Related Pages

• Principle:Apache_Flink_Split_Based_Record_Reading