Implementation:Apache Flink SourceReaderOptions
| Knowledge Sources | |
|---|---|
| Domains | Connectors, Source_Framework |
| Last Updated | 2026-02-09 00:00 GMT |
Overview
A configuration class that defines and resolves the tunable options for SourceReaderBase, including close timeout and element queue capacity.
Description
SourceReaderOptions is a configuration holder class in Flink's source connector framework that defines the configurable options for SourceReaderBase and its subclasses. It declares two static ConfigOption constants that define the available configuration keys, their types, and default values. The class also provides a constructor that resolves these options from a Configuration object into final instance fields for efficient runtime access.
The two configuration options are:
- source.reader.close.timeout (Long, default: 30000ms) -- The timeout in milliseconds when closing the source reader. This controls how long the framework waits for the reader to shut down gracefully before forcibly terminating.
- source.reader.element.queue.capacity (Integer, default: 2) -- The capacity of the element queue that sits between the fetcher threads and the source reader main thread. This queue is the handover mechanism for RecordsWithSplitIds batches. A small default value (2) is used to limit memory consumption while still allowing pipelining between fetching and processing.
The class is annotated with @PublicEvolving, meaning these options are part of Flink's public API but may evolve across minor versions.
Usage
Connector developers typically do not interact with this class directly. It is used internally by SourceReaderBase to configure runtime behavior. However, connector users may set these options in their Flink configuration to tune source reader performance. For example, increasing source.reader.element.queue.capacity may improve throughput for sources with high fetch latency by allowing more prefetching, at the cost of increased memory usage.
Code Reference
Source Location
- Repository: Apache_Flink
- File: flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderOptions.java
- Lines: 1-51
Signature
@PublicEvolving
public class SourceReaderOptions {
public static final ConfigOption<Long> SOURCE_READER_CLOSE_TIMEOUT =
ConfigOptions.key("source.reader.close.timeout")
.longType()
.defaultValue(30000L)
.withDescription("The timeout when closing the source reader");
public static final ConfigOption<Integer> ELEMENT_QUEUE_CAPACITY =
ConfigOptions.key("source.reader.element.queue.capacity")
.intType()
.defaultValue(2)
.withDescription(
"The capacity of the element queue in the source reader.");
// Final fields resolved from Configuration
public final long sourceReaderCloseTimeout;
public final int elementQueueCapacity;
public SourceReaderOptions(Configuration config);
}
Import
import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| config | Configuration | Yes | A Flink Configuration object from which the source reader options are resolved. |
Outputs
| Name | Type | Description |
|---|---|---|
| sourceReaderCloseTimeout | long | The resolved close timeout in milliseconds (default: 30000). |
| elementQueueCapacity | int | The resolved element queue capacity (default: 2). |
Usage Examples
// Example: Configuring source reader options via Flink Configuration
Configuration config = new Configuration();
// Increase element queue capacity for higher throughput
config.set(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY, 5);
// Increase close timeout for slow-shutting-down sources
config.set(SourceReaderOptions.SOURCE_READER_CLOSE_TIMEOUT, 60000L);
// The options are consumed internally by SourceReaderBase
SourceReaderOptions options = new SourceReaderOptions(config);
// options.elementQueueCapacity == 5
// options.sourceReaderCloseTimeout == 60000L