Implementation:Apache Flink SourceReaderBase

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

Overview

Abstract base implementation of SourceReader that provides the synchronization framework between the mailbox main thread and internal fetcher threads, managing split state, record emission, rate limiting, and checkpoint coordination.

Description

SourceReaderBase is the central abstract class in Flink's connector base source reader framework. It implements the SourceReader interface and provides a complete pipeline for reading records from external systems via split-based parallelism.

The class uses a FutureCompletingBlockingQueue as a hand-off queue between fetcher threads and the main thread. The pollNext() method retrieves RecordsWithSplitIds batches from the queue, iterates through splits and records, and emits them via a RecordEmitter. It maintains per-split state in an internal SplitContext map, keyed by split ID, tracking both the mutable split state and a per-split SourceOutput.

Key features include:

• Split lifecycle management: Initializes split state via initializedState() when splits are added, converts mutable state back to immutable form via toSplitType() for checkpoints, and delegates to onSplitFinished() when splits complete.
• Rate limiting: Optional RateLimiterStrategy support that wraps split outputs in a RateLimitingSourceOutputWrapper to track per-window record counts and acquire permits from the rate limiter.
• EOF record evaluation: Optional RecordEvaluator that wraps split outputs in a SourceOutputWrapper to detect end-of-stream conditions and trigger split removal.
• Metrics: Automatically tracks numRecordsIn counter via the operator IO metric group.
• Checkpoint support: snapshotState() serializes all active split states; notifyCheckpointComplete() propagates checkpoint completion to the rate limiter.

Nearly all Flink connector source readers (Kafka, File, etc.) extend this class, which handles the complex threading model, checkpoint state management, and record emission pipeline.

Usage

Extend SourceReaderBase when building a new Flink source connector that reads from split-based external systems. You must implement the three abstract methods: initializedState() to create mutable state from a split, toSplitType() to convert mutable state back to an immutable split for checkpoints, and onSplitFinished() to handle cleanup when splits complete. A SplitReader implementation and RecordEmitter must be provided to the constructor.

Code Reference

Source Location

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

Signature

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

    // Constructors
    public SourceReaderBase(
            SplitFetcherManager<E, SplitT> splitFetcherManager,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            Configuration config,
            SourceReaderContext context);

    public SourceReaderBase(
            SplitFetcherManager<E, SplitT> splitFetcherManager,
            RecordEmitter<E, T, SplitStateT> recordEmitter,
            @Nullable RecordEvaluator<T> eofRecordEvaluator,
            Configuration config,
            SourceReaderContext context,
            @Nullable RateLimiterStrategy<SplitT> rateLimiterStrategy);

    // SourceReader interface methods
    public void start();
    public InputStatus pollNext(ReaderOutput<T> output) throws Exception;
    public CompletableFuture<Void> isAvailable();
    public List<SplitT> snapshotState(long checkpointId);
    public void notifyCheckpointComplete(long checkpointId) throws Exception;
    public void addSplits(List<SplitT> splits);
    public void notifyNoMoreSplits();
    public void handleSourceEvents(SourceEvent sourceEvent);
    public void pauseOrResumeSplits(
            Collection<String> splitsToPause, Collection<String> splitsToResume);
    public void close() throws Exception;

    // Query method
    public int getNumberOfCurrentlyAssignedSplits();

    // Abstract methods subclasses must implement
    protected abstract void onSplitFinished(Map<String, SplitStateT> finishedSplitIds);
    protected abstract SplitStateT initializedState(SplitT split);
    protected abstract SplitT toSplitType(String splitId, SplitStateT splitState);
}

Import

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

I/O Contract

Inputs

Name	Type	Required	Description
splitFetcherManager	SplitFetcherManager<E, SplitT>	Yes	Manages the fetcher threads that read from external systems and produce RecordsWithSplitIds batches
recordEmitter	RecordEmitter<E, T, SplitStateT>	Yes	Transforms raw elements of type E into output records of type T while updating split state
config	Configuration	Yes	Flink configuration containing SourceReaderOptions such as element queue capacity and close timeout
context	SourceReaderContext	Yes	Provides reader context including parallelism info, metric groups, and event communication with the enumerator
eofRecordEvaluator	RecordEvaluator<T>	No	Evaluates records to detect end-of-stream conditions, triggering automatic split removal
rateLimiterStrategy	RateLimiterStrategy<SplitT>	No	Strategy for creating a rate limiter to throttle record emission throughput
splits (via addSplits)	List<SplitT>	Yes	Source splits assigned by the enumerator that define the data partitions to read

Outputs

Name	Type	Description
pollNext return	InputStatus	Returns MORE_AVAILABLE when records are emitted, NOTHING_AVAILABLE when waiting for data, or END_OF_INPUT when all splits are finished
emitted records	T (via ReaderOutput)	Records emitted to downstream operators through the RecordEmitter and SourceOutput pipeline
snapshotState return	List<SplitT>	Immutable snapshot of all currently assigned split states for checkpoint serialization
numRecordsIn	Counter (metric)	Metric counter incremented for each record emitted

Usage Examples

Basic Usage

// Example: Implementing a custom source reader by extending SourceReaderBase
public class MySourceReader
        extends SourceReaderBase<MyRecord, String, MySplit, MySplitState> {

    public MySourceReader(
            SourceReaderContext context,
            Configuration config) {
        super(
            new SingleThreadFetcherManager<>(
                () -> new MySplitReader(),
                config),
            new MyRecordEmitter(),
            config,
            context);
    }

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

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

    @Override
    protected void onSplitFinished(Map<String, MySplitState> finishedSplitIds) {
        // Handle finished splits, e.g., report to enumerator
        context.sendSplitRequest();
    }
}

Related Pages

• Principle:Apache_Flink_Split_Based_Record_Reading