org.apache.flink.streaming.connectors.kafka.table

Class KafkaDynamicSource

java.lang.Object

org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicSource

All Implemented Interfaces:

SupportsReadingMetadata, SupportsWatermarkPushDown, DynamicTableSource, ScanTableSource

@Internal
public class KafkaDynamicSource
extends Object
implements ScanTableSource, SupportsReadingMetadata, SupportsWatermarkPushDown

A version-agnostic Kafka ScanTableSource.

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.table.connector.source.ScanTableSource

ScanTableSource.ScanContext, ScanTableSource.ScanRuntimeProvider

Nested classes/interfaces inherited from interface org.apache.flink.table.connector.source.DynamicTableSource

DynamicTableSource.Context, DynamicTableSource.DataStructureConverter

Field Summary

Fields

Modifier and Type	Field and Description
protected BoundedMode	boundedMode The bounded mode for the contained consumer (default is an unbounded data stream).
protected long	boundedTimestampMillis The bounded timestamp to locate partition offsets; only relevant when bounded mode is BoundedMode.TIMESTAMP.
protected DecodingFormat<DeserializationSchema<RowData>>	keyDecodingFormat Optional format for decoding keys from Kafka.
protected String	keyPrefix Prefix that needs to be removed from fields when constructing the physical data type.
protected int[]	keyProjection Indices that determine the key fields and the target position in the produced row.
protected List<String>	metadataKeys Metadata that is appended at the end of a physical source row.
protected DataType	physicalDataType Data type to configure the formats.
protected DataType	producedDataType Data type that describes the final output of the source.
protected Properties	properties Properties for the Kafka consumer.
protected Map<KafkaTopicPartition,Long>	specificBoundedOffsets Specific end offsets; only relevant when bounded mode is BoundedMode.SPECIFIC_OFFSETS.
protected Map<KafkaTopicPartition,Long>	specificStartupOffsets Specific startup offsets; only relevant when startup mode is StartupMode.SPECIFIC_OFFSETS.
protected StartupMode	startupMode The startup mode for the contained consumer (default is StartupMode.GROUP_OFFSETS).
protected long	startupTimestampMillis The start timestamp to locate partition offsets; only relevant when startup mode is StartupMode.TIMESTAMP.
protected String	tableIdentifier
protected Pattern	topicPattern The Kafka topic pattern to consume.
protected List<String>	topics The Kafka topics to consume.
protected boolean	upsertMode Flag to determine source mode.
protected DecodingFormat<DeserializationSchema<RowData>>	valueDecodingFormat Format for decoding values from Kafka.
protected int[]	valueProjection Indices that determine the value fields and the target position in the produced row.
protected WatermarkStrategy<RowData>	watermarkStrategy Watermark strategy that is used to generate per-partition watermark.

Constructor Summary

Constructors

Constructor and Description

KafkaDynamicSource(DataType physicalDataType, DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat, DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat, int[] keyProjection, int[] valueProjection, String keyPrefix, List<String> topics, Pattern topicPattern, Properties properties, StartupMode startupMode, Map<KafkaTopicPartition,Long> specificStartupOffsets, long startupTimestampMillis, BoundedMode boundedMode, Map<KafkaTopicPartition,Long> specificBoundedOffsets, long boundedTimestampMillis, boolean upsertMode, String tableIdentifier)

Method Summary

Modifier and Type	Method and Description
void	applyReadableMetadata(List<String> metadataKeys, DataType producedDataType) Provides a list of metadata keys that the produced RowData must contain as appended metadata columns.
void	applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) Provides a WatermarkStrategy which defines how to generate Watermarks in the stream source.
String	asSummaryString() Returns a string that summarizes this source for printing to a console or log.
DynamicTableSource	copy() Creates a copy of this instance during planning.
protected KafkaSource<RowData>	createKafkaSource(DeserializationSchema<RowData> keyDeserialization, DeserializationSchema<RowData> valueDeserialization, TypeInformation<RowData> producedTypeInfo)
boolean	equals(Object o)
ChangelogMode	getChangelogMode() Returns the set of changes that the planner can expect during runtime.
ScanTableSource.ScanRuntimeProvider	getScanRuntimeProvider(ScanTableSource.ScanContext context) Returns a provider of runtime implementation for reading the data.
int	hashCode()
Map<String,DataType>	listReadableMetadata() Returns the map of metadata keys and their corresponding data types that can be produced by this table source for reading.
boolean	supportsMetadataProjection() Defines whether projections can be applied to metadata columns.

Methods inherited from class java.lang.Object

clone, finalize, getClass, notify, notifyAll, toString, wait, wait, wait

Field Detail

producedDataType

protected DataType producedDataType

Data type that describes the final output of the source.

metadataKeys

protected List<String> metadataKeys

Metadata that is appended at the end of a physical source row.

watermarkStrategy

@Nullable
protected WatermarkStrategy<RowData> watermarkStrategy

Watermark strategy that is used to generate per-partition watermark.

physicalDataType

protected final DataType physicalDataType

Data type to configure the formats.

keyDecodingFormat

@Nullable
protected final DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat

Optional format for decoding keys from Kafka.

valueDecodingFormat

protected final DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat

Format for decoding values from Kafka.

keyProjection

protected final int[] keyProjection

Indices that determine the key fields and the target position in the produced row.

valueProjection

protected final int[] valueProjection

Indices that determine the value fields and the target position in the produced row.

keyPrefix

@Nullable
protected final String keyPrefix

Prefix that needs to be removed from fields when constructing the physical data type.

topics

protected final List<String> topics

The Kafka topics to consume.

topicPattern

protected final Pattern topicPattern

The Kafka topic pattern to consume.

properties

protected final Properties properties

Properties for the Kafka consumer.

startupMode

protected final StartupMode startupMode

The startup mode for the contained consumer (default is StartupMode.GROUP_OFFSETS).

specificStartupOffsets

protected final Map<KafkaTopicPartition,Long> specificStartupOffsets

Specific startup offsets; only relevant when startup mode is StartupMode.SPECIFIC_OFFSETS.

startupTimestampMillis

protected final long startupTimestampMillis

The start timestamp to locate partition offsets; only relevant when startup mode is StartupMode.TIMESTAMP.

boundedMode

protected final BoundedMode boundedMode

The bounded mode for the contained consumer (default is an unbounded data stream).

specificBoundedOffsets

protected final Map<KafkaTopicPartition,Long> specificBoundedOffsets

Specific end offsets; only relevant when bounded mode is BoundedMode.SPECIFIC_OFFSETS.

boundedTimestampMillis

protected final long boundedTimestampMillis

The bounded timestamp to locate partition offsets; only relevant when bounded mode is BoundedMode.TIMESTAMP.

upsertMode

protected final boolean upsertMode

Flag to determine source mode. In upsert mode, it will keep the tombstone message. *

tableIdentifier

protected final String tableIdentifier

Constructor Detail

KafkaDynamicSource

public KafkaDynamicSource(DataType physicalDataType,
                          @Nullable
                          DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat,
                          DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat,
                          int[] keyProjection,
                          int[] valueProjection,
                          @Nullable
                          String keyPrefix,
                          @Nullable
                          List<String> topics,
                          @Nullable
                          Pattern topicPattern,
                          Properties properties,
                          StartupMode startupMode,
                          Map<KafkaTopicPartition,Long> specificStartupOffsets,
                          long startupTimestampMillis,
                          BoundedMode boundedMode,
                          Map<KafkaTopicPartition,Long> specificBoundedOffsets,
                          long boundedTimestampMillis,
                          boolean upsertMode,
                          String tableIdentifier)

Method Detail

getChangelogMode

public ChangelogMode getChangelogMode()

Description copied from interface: ScanTableSource

Returns the set of changes that the planner can expect during runtime.

Specified by:

getChangelogMode in interface ScanTableSource

See Also:

RowKind

getScanRuntimeProvider

public ScanTableSource.ScanRuntimeProvider getScanRuntimeProvider(ScanTableSource.ScanContext context)

Description copied from interface: ScanTableSource

Returns a provider of runtime implementation for reading the data.

There might exist different interfaces for runtime implementation which is why ScanTableSource.ScanRuntimeProvider serves as the base interface. Concrete ScanTableSource.ScanRuntimeProvider interfaces might be located in other Flink modules.

Independent of the provider interface, the table runtime expects that a source implementation emits internal data structures (see RowData for more information).

The given ScanTableSource.ScanContext offers utilities by the planner for creating runtime implementation with minimal dependencies to internal data structures.

SourceProvider is the recommended core interface. SourceFunctionProvider in flink-table-api-java-bridge and InputFormatProvider are available for backwards compatibility.

Specified by:

getScanRuntimeProvider in interface ScanTableSource

See Also:

SourceProvider

listReadableMetadata

public Map<String,DataType> listReadableMetadata()

Description copied from interface: SupportsReadingMetadata

Returns the map of metadata keys and their corresponding data types that can be produced by this table source for reading.

The returned map will be used by the planner for validation and insertion of explicit casts (see LogicalTypeCasts.supportsExplicitCast(LogicalType, LogicalType)) if necessary.

The iteration order of the returned map determines the order of metadata keys in the list passed in SupportsReadingMetadata.applyReadableMetadata(List, DataType). Therefore, it might be beneficial to return a LinkedHashMap if a strict metadata column order is required.

If a source forwards metadata from one or more formats, we recommend the following column order for consistency:

 KEY FORMAT METADATA COLUMNS + VALUE FORMAT METADATA COLUMNS + SOURCE METADATA COLUMNS
 

Metadata key names follow the same pattern as mentioned in Factory. In case of duplicate names in format and source keys, format keys shall have higher precedence.

Regardless of the returned DataTypes, a metadata column is always represented using internal data structures (see RowData).

Specified by:

listReadableMetadata in interface SupportsReadingMetadata

See Also:

DecodingFormat.listReadableMetadata()

applyReadableMetadata

public void applyReadableMetadata(List<String> metadataKeys,
                                  DataType producedDataType)

Description copied from interface: SupportsReadingMetadata

Provides a list of metadata keys that the produced RowData must contain as appended metadata columns.

Implementations of this method must be idempotent. The planner might call this method multiple times.

Note: Use the passed data type instead of ResolvedSchema.toPhysicalRowDataType() for describing the final output data type when creating TypeInformation. If the source implements SupportsProjectionPushDown, the projection is already considered in the given output data type, use the producedDataType provided by this method instead of the producedDataType provided by SupportsProjectionPushDown.applyProjection(int[][], DataType).

Specified by:

applyReadableMetadata in interface SupportsReadingMetadata

Parameters:

metadataKeys - a subset of the keys returned by SupportsReadingMetadata.listReadableMetadata(), ordered by the iteration order of returned map

producedDataType - the final output type of the source, it is intended to be only forwarded and the planner will decide on the field names to avoid collisions

See Also:

DecodingFormat.applyReadableMetadata(List)

supportsMetadataProjection

public boolean supportsMetadataProjection()

Description copied from interface: SupportsReadingMetadata

Defines whether projections can be applied to metadata columns.

This method is only called if the source does not implement SupportsProjectionPushDown. By default, the planner will only apply metadata columns which have actually been selected in the query regardless. By returning false instead the source can inform the planner to apply all metadata columns defined in the table's schema.

If the source implements SupportsProjectionPushDown, projections of metadata columns are always considered before calling SupportsReadingMetadata.applyReadableMetadata(List, DataType).

Specified by:

supportsMetadataProjection in interface SupportsReadingMetadata

applyWatermark

public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy)

Description copied from interface: SupportsWatermarkPushDown

Provides a WatermarkStrategy which defines how to generate Watermarks in the stream source.

The WatermarkStrategy is a builder/factory for the actual runtime implementation consisting of TimestampAssigner (assigns the event-time timestamps to each record) and the WatermarkGenerator (generates the watermarks).

Note: If necessary, the watermark strategy will contain required computed column expressions and consider metadata columns (if SupportsReadingMetadata is implemented).

Specified by:

applyWatermark in interface SupportsWatermarkPushDown

copy

public DynamicTableSource copy()

Description copied from interface: DynamicTableSource

Creates a copy of this instance during planning. The copy should be a deep copy of all mutable members.

Specified by:

copy in interface DynamicTableSource

asSummaryString

public String asSummaryString()

Description copied from interface: DynamicTableSource

Returns a string that summarizes this source for printing to a console or log.

Specified by:

asSummaryString in interface DynamicTableSource

equals

public boolean equals(Object o)

Overrides:

equals in class Object

hashCode

public int hashCode()

Overrides:

hashCode in class Object

createKafkaSource

protected KafkaSource<RowData> createKafkaSource(DeserializationSchema<RowData> keyDeserialization,
                                                 DeserializationSchema<RowData> valueDeserialization,
                                                 TypeInformation<RowData> producedTypeInfo)