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.
---
