@Test @SuppressWarnings("unchecked") public void testTableSource() { final TableSchema schema = TableSchema.builder() .field(FRUIT_NAME, Types.STRING()) .field(COUNT, Types.DECIMAL()) .field(EVENT_TIME, Types.SQL_TIMESTAMP()) .field(PROC_TIME, Types.SQL_TIMESTAMP()) .build(); final List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors = Collections.singletonList( new RowtimeAttributeDescriptor(EVENT_TIME, new ExistingField(TIME), new AscendingTimestamps())); final Map<String, String> fieldMapping = new HashMap<>(); fieldMapping.put(FRUIT_NAME, NAME); fieldMapping.put(NAME, NAME); fieldMapping.put(COUNT, COUNT); fieldMapping.put(TIME, TIME); final Map<KafkaTopicPartition, Long> specificOffsets = new HashMap<>(); specificOffsets.put(new KafkaTopicPartition(TOPIC, PARTITION_0), OFFSET_0); specificOffsets.put(new KafkaTopicPartition(TOPIC, PARTITION_1), OFFSET_1); final TestDeserializationSchema deserializationSchema = new TestDeserializationSchema( TableSchema.builder() .field(NAME, Types.STRING()) .field(COUNT, Types.DECIMAL()) .field(TIME, Types.SQL_TIMESTAMP()) .build() .toRowType() ); final KafkaTableSourceBase expected = getExpectedKafkaTableSource( schema, Optional.of(PROC_TIME), rowtimeAttributeDescriptors, fieldMapping, TOPIC, KAFKA_PROPERTIES, deserializationSchema, StartupMode.SPECIFIC_OFFSETS, specificOffsets); TableSourceUtil.validateTableSource(expected); final TestTableDescriptor testDesc = new TestTableDescriptor( new Kafka() .version(getKafkaVersion()) .topic(TOPIC) .properties(KAFKA_PROPERTIES) .sinkPartitionerRoundRobin() .startFromSpecificOffsets(OFFSETS)) .withFormat(new TestTableFormat()) .withSchema( new Schema() .field(FRUIT_NAME, Types.STRING()).from(NAME) .field(COUNT, Types.DECIMAL()) .field(EVENT_TIME, Types.SQL_TIMESTAMP()).rowtime( new Rowtime().timestampsFromField(TIME).watermarksPeriodicAscending()) .field(PROC_TIME, Types.SQL_TIMESTAMP()).proctime()) .inAppendMode(); final Map<String, String> propertiesMap = testDesc.toProperties(); final TableSource<?> actualSource = TableFactoryService.find(StreamTableSourceFactory.class, propertiesMap) .createStreamTableSource(propertiesMap); assertEquals(expected, actualSource); final KafkaTableSourceBase actualKafkaSource = (KafkaTableSourceBase) actualSource; final StreamExecutionEnvironmentMock mock = new StreamExecutionEnvironmentMock(); actualKafkaSource.getDataStream(mock); assertTrue(getExpectedFlinkKafkaConsumer().isAssignableFrom(mock.sourceFunction.getClass())); }
@Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final String topic = descriptorProperties.getString(CONNECTOR_TOPIC); final DeserializationSchema<Row> deserializationSchema = getDeserializationSchema(properties); final StartupOptions startupOptions = getStartupOptions(descriptorProperties, topic); return createKafkaTableSource( descriptorProperties.getTableSchema(SCHEMA()), SchemaValidator.deriveProctimeAttribute(descriptorProperties), SchemaValidator.deriveRowtimeAttributes(descriptorProperties), SchemaValidator.deriveFieldMapping( descriptorProperties, Optional.of(deserializationSchema.getProducedType())), topic, getKafkaProperties(descriptorProperties), deserializationSchema, startupOptions.startupMode, startupOptions.specificOffsets); }
KafkaTableSourceSinkFactoryBase implements
		StreamTableSourceFactory<Row>,
		StreamTableSinkFactory<Row> { @Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final String topic = descriptorProperties.getString(CONNECTOR_TOPIC); final DeserializationSchema<Row> deserializationSchema = getDeserializationSchema(properties); final StartupOptions startupOptions = getStartupOptions(descriptorProperties, topic); return createKafkaTableSource( descriptorProperties.getTableSchema(SCHEMA()), SchemaValidator.deriveProctimeAttribute(descriptorProperties), SchemaValidator.deriveRowtimeAttributes(descriptorProperties), SchemaValidator.deriveFieldMapping( descriptorProperties, Optional.of(deserializationSchema.getProducedType())), topic, getKafkaProperties(descriptorProperties), deserializationSchema, startupOptions.startupMode, startupOptions.specificOffsets); } }
KafkaTableSourceSinkFactoryBase implements
		StreamTableSourceFactory<Row>,
		StreamTableSinkFactory<Row> { @Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final String topic = descriptorProperties.getString(CONNECTOR_TOPIC); final DeserializationSchema<Row> deserializationSchema = getDeserializationSchema(properties); final StartupOptions startupOptions = getStartupOptions(descriptorProperties, topic); return createKafkaTableSource( descriptorProperties.getTableSchema(SCHEMA()), SchemaValidator.deriveProctimeAttribute(descriptorProperties), SchemaValidator.deriveRowtimeAttributes(descriptorProperties), SchemaValidator.deriveFieldMapping( descriptorProperties, Optional.of(deserializationSchema.getProducedType())), topic, getKafkaProperties(descriptorProperties), deserializationSchema, startupOptions.startupMode, startupOptions.specificOffsets); }  }
KafkaTableSourceSinkFactoryBase implements
		StreamTableSourceFactory<Row>,
		StreamTableSinkFactory<Row> { @Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final String topic = descriptorProperties.getString(CONNECTOR_TOPIC); final DeserializationSchema<Row> deserializationSchema = getDeserializationSchema(properties); final StartupOptions startupOptions = getStartupOptions(descriptorProperties, topic); return createKafkaTableSource( descriptorProperties.getTableSchema(SCHEMA()), SchemaValidator.deriveProctimeAttribute(descriptorProperties), SchemaValidator.deriveRowtimeAttributes(descriptorProperties), SchemaValidator.deriveFieldMapping( descriptorProperties, Optional.of(deserializationSchema.getProducedType())), topic, getKafkaProperties(descriptorProperties), deserializationSchema, startupOptions.startupMode, startupOptions.specificOffsets); }  @Override Map<String, String> requiredContext(); @Override List<String> supportedProperties(); @Override StreamTableSource<Row> createStreamTableSource(Map<String, String> properties); @Override StreamTableSink<Row> createStreamTableSink(Map<String, String> properties); }
KafkaTableSourceSinkFactoryBase implements
		StreamTableSourceFactory<Row>,
		StreamTableSinkFactory<Row> { @Override public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final String topic = descriptorProperties.getString(CONNECTOR_TOPIC); final DeserializationSchema<Row> deserializationSchema = getDeserializationSchema(properties); final StartupOptions startupOptions = getStartupOptions(descriptorProperties, topic); return createKafkaTableSource( descriptorProperties.getTableSchema(SCHEMA()), SchemaValidator.deriveProctimeAttribute(descriptorProperties), SchemaValidator.deriveRowtimeAttributes(descriptorProperties), SchemaValidator.deriveFieldMapping( descriptorProperties, Optional.of(deserializationSchema.getProducedType())), topic, getKafkaProperties(descriptorProperties), deserializationSchema, startupOptions.startupMode, startupOptions.specificOffsets); }  @Override Map<String, String> requiredContext(); @Override List<String> supportedProperties(); @Override StreamTableSource<Row> createStreamTableSource(Map<String, String> properties); @Override StreamTableSink<Row> createStreamTableSink(Map<String, String> properties);  }