class JDBCTableSourceSinkFactory implements
	StreamTableSourceFactory<Row>,
	StreamTableSinkFactory<Tuple2<Boolean, Row>> {

	@Override
	public StreamTableSource<Row> createStreamTableSource(Map<String, String> properties) {
		final DescriptorProperties descriptorProperties = getValidatedProperties(properties);

		return JDBCTableSource.builder()
			.setOptions(getJDBCOptions(descriptorProperties))
			.setReadOptions(getJDBCReadOptions(descriptorProperties))
			.setLookupOptions(getJDBCLookupOptions(descriptorProperties))
			.setSchema(descriptorProperties.getTableSchema(SCHEMA))
			.build();
	}

}