[
https://issues.apache.org/jira/browse/FLINK-9846?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16552645#comment-16552645
]
ASF GitHub Bot commented on FLINK-9846:
---------------------------------------
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6387#discussion_r204351776
--- Diff:
flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceSinkFactoryBase.java
---
@@ -125,89 +131,47 @@
@Override
public StreamTableSource<Row> createStreamTableSource(Map<String,
String> properties) {
- final DescriptorProperties params = new
DescriptorProperties(true);
- params.putProperties(properties);
+ final DescriptorProperties descriptorProperties =
getValidatedProperties(properties);
- // validate
- // allow Kafka timestamps to be used, watermarks can not be
received from source
- new SchemaValidator(true, supportsKafkaTimestamps(),
false).validate(params);
- new KafkaValidator().validate(params);
-
- // deserialization schema using format discovery
- final DeserializationSchemaFactory<?> formatFactory =
TableFactoryService.find(
- DeserializationSchemaFactory.class,
- properties,
- this.getClass().getClassLoader());
- @SuppressWarnings("unchecked")
- final DeserializationSchema<Row> deserializationSchema =
(DeserializationSchema<Row>) formatFactory
- .createDeserializationSchema(properties);
-
- // schema
- final TableSchema schema = params.getTableSchema(SCHEMA());
-
- // proctime
- final Optional<String> proctimeAttribute =
SchemaValidator.deriveProctimeAttribute(params);
-
- // rowtime
- final List<RowtimeAttributeDescriptor> rowtimeAttributes =
SchemaValidator.deriveRowtimeAttributes(params);
-
- // field mapping
- final Map<String, String> fieldMapping =
SchemaValidator.deriveFieldMapping(params, Optional.of(schema));
+ final TableSchema schema =
descriptorProperties.getTableSchema(SCHEMA());
+ final String topic =
descriptorProperties.getString(CONNECTOR_TOPIC);
+ final Tuple2<StartupMode, Map<KafkaTopicPartition, Long>>
startupOptions =
--- End diff --
nit: using tuples in such places slightly reduces code readability, since
it's loosing named variables. I would personally prefer introducing small Pojo
with named fields:
```
private static class StartupOptions {
private final StartupMode startupMode;
private final Map<...> specificOptions;
}
```
otherwise `specificOptions` name disappears.
> Add a Kafka table sink factory
> ------------------------------
>
> Key: FLINK-9846
> URL: https://issues.apache.org/jira/browse/FLINK-9846
> Project: Flink
> Issue Type: New Feature
> Components: Table API & SQL
> Reporter: Timo Walther
> Assignee: Timo Walther
> Priority: Major
> Labels: pull-request-available
>
> FLINK-8866 implements a unified way of creating sinks and using the format
> discovery for searching for formats (FLINK-8858). It is now possible to add a
> Kafka table sink factory for streaming environment that uses the new
> interfaces.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
