org.apache.flink.streaming.connectors.kafka

Class KafkaTableSourceSinkFactoryBase

java.lang.Object

org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactoryBase

All Implemented Interfaces:

StreamTableSinkFactory<Row>, StreamTableSourceFactory<Row>, TableFactory, TableSinkFactory<Row>, TableSourceFactory<Row>

Direct Known Subclasses:

Kafka010TableSourceSinkFactory, Kafka011TableSourceSinkFactory, KafkaTableSourceSinkFactory

public abstract class KafkaTableSourceSinkFactoryBase
extends Object
implements StreamTableSourceFactory<Row>, StreamTableSinkFactory<Row>

Factory for creating configured instances of KafkaTableSourceBase.

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.table.factories.TableSourceFactory

TableSourceFactory.Context

Nested classes/interfaces inherited from interface org.apache.flink.table.factories.TableSinkFactory

TableSinkFactory.Context

Constructor Summary

Constructors

Constructor and Description
KafkaTableSourceSinkFactoryBase()

Method Summary

Modifier and Type	Method and Description
protected abstract KafkaTableSinkBase	createKafkaTableSink(TableSchema schema, String topic, Properties properties, Optional<FlinkKafkaPartitioner<Row>> partitioner, SerializationSchema<Row> serializationSchema) Constructs the version-specific Kafka table sink.
protected abstract KafkaTableSourceBase	createKafkaTableSource(TableSchema schema, Optional<String> proctimeAttribute, List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors, Map<String,String> fieldMapping, String topic, Properties properties, DeserializationSchema<Row> deserializationSchema, StartupMode startupMode, Map<KafkaTopicPartition,Long> specificStartupOffsets, long startupTimestampMillis) Constructs the version-specific Kafka table source.
StreamTableSink<Row>	createStreamTableSink(Map<String,String> properties) Creates and configures a StreamTableSink using the given properties.
StreamTableSource<Row>	createStreamTableSource(Map<String,String> properties) Creates and configures a StreamTableSource using the given properties.
protected abstract String	kafkaVersion() Returns the Kafka version.
Map<String,String>	requiredContext() Specifies the context that this factory has been implemented for.
List<String>	supportedProperties() List of property keys that this factory can handle.
protected abstract boolean	supportsKafkaTimestamps() True if the Kafka source supports Kafka timestamps, false otherwise.

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.flink.table.factories.StreamTableSourceFactory

createTableSource

Methods inherited from interface org.apache.flink.table.factories.TableSourceFactory

createTableSource, createTableSource

Methods inherited from interface org.apache.flink.table.factories.StreamTableSinkFactory

createTableSink

Methods inherited from interface org.apache.flink.table.factories.TableSinkFactory

createTableSink, createTableSink

Constructor Detail

KafkaTableSourceSinkFactoryBase

public KafkaTableSourceSinkFactoryBase()

Method Detail

requiredContext

public Map<String,String> requiredContext()

Description copied from interface: TableFactory

Specifies the context that this factory has been implemented for. The framework guarantees to only match for this factory if the specified set of properties and values are met.

Typical properties might be: - connector.type - format.type

Specified property versions allow the framework to provide backwards compatible properties in case of string format changes: - connector.property-version - format.property-version

An empty context means that the factory matches for all requests.

Specified by:

requiredContext in interface TableFactory

supportedProperties

public List<String> supportedProperties()

Description copied from interface: TableFactory

List of property keys that this factory can handle. This method will be used for validation. If a property is passed that this factory cannot handle, an exception will be thrown. The list must not contain the keys that are specified by the context.

Example properties might be: - schema.#.type - schema.#.name - connector.topic - format.line-delimiter - format.ignore-parse-errors - format.fields.#.type - format.fields.#.name

Note: Use "#" to denote an array of values where "#" represents one or more digits. Property versions like "format.property-version" must not be part of the supported properties.

In some cases it might be useful to declare wildcards "*". Wildcards can only be declared at the end of a property key.

For example, if an arbitrary format should be supported: - format.*

Note: Wildcards should be used with caution as they might swallow unsupported properties and thus might lead to undesired behavior.

Specified by:

supportedProperties in interface TableFactory

createStreamTableSource

public StreamTableSource<Row> createStreamTableSource(Map<String,String> properties)

Description copied from interface: StreamTableSourceFactory

Creates and configures a StreamTableSource using the given properties.

Specified by:

createStreamTableSource in interface StreamTableSourceFactory<Row>

Parameters:

properties - normalized properties describing a stream table source.

Returns:

the configured stream table source.

createStreamTableSink

public StreamTableSink<Row> createStreamTableSink(Map<String,String> properties)

Description copied from interface: StreamTableSinkFactory

Creates and configures a StreamTableSink using the given properties.

Specified by:

createStreamTableSink in interface StreamTableSinkFactory<Row>

Parameters:

properties - normalized properties describing a table sink.

Returns:

the configured table sink.

kafkaVersion

protected abstract String kafkaVersion()

Returns the Kafka version.

supportsKafkaTimestamps

protected abstract boolean supportsKafkaTimestamps()

True if the Kafka source supports Kafka timestamps, false otherwise.

Returns:

True if the Kafka source supports Kafka timestamps, false otherwise.

createKafkaTableSource

protected abstract KafkaTableSourceBase createKafkaTableSource(TableSchema schema,
                                                               Optional<String> proctimeAttribute,
                                                               List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
                                                               Map<String,String> fieldMapping,
                                                               String topic,
                                                               Properties properties,
                                                               DeserializationSchema<Row> deserializationSchema,
                                                               StartupMode startupMode,
                                                               Map<KafkaTopicPartition,Long> specificStartupOffsets,
                                                               long startupTimestampMillis)

Constructs the version-specific Kafka table source.

Parameters:

schema - Schema of the produced table.

proctimeAttribute - Field name of the processing time attribute.

rowtimeAttributeDescriptors - Descriptor for a rowtime attribute

fieldMapping - Mapping for the fields of the table schema to fields of the physical returned type.

topic - Kafka topic to consume.

properties - Properties for the Kafka consumer.

deserializationSchema - Deserialization schema for decoding records from Kafka.

startupMode - Startup mode for the contained consumer.

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

createKafkaTableSink

protected abstract KafkaTableSinkBase createKafkaTableSink(TableSchema schema,
                                                           String topic,
                                                           Properties properties,
                                                           Optional<FlinkKafkaPartitioner<Row>> partitioner,
                                                           SerializationSchema<Row> serializationSchema)

Constructs the version-specific Kafka table sink.

Parameters:

schema - Schema of the produced table.

topic - Kafka topic to consume.

properties - Properties for the Kafka consumer.

partitioner - Partitioner to select Kafka partition for each item.