@Test public void testJDBCValidation() { try { Map<String, String> properties = getBasicProperties(); properties.put("connector.password", "pass"); TableFactoryService.find(StreamTableSourceFactory.class, properties) .createStreamTableSource(properties); fail("exception expected"); } catch (IllegalArgumentException ignored) { } try { Map<String, String> properties = getBasicProperties(); properties.put("connector.read.partition.column", "aaa"); properties.put("connector.read.partition.lower-bound", "-10"); properties.put("connector.read.partition.upper-bound", "100"); TableFactoryService.find(StreamTableSourceFactory.class, properties) .createStreamTableSource(properties); fail("exception expected"); } catch (IllegalArgumentException ignored) { } try { Map<String, String> properties = getBasicProperties(); properties.put("connector.read.partition.column", "aaa"); properties.put("connector.read.partition.lower-bound", "100"); properties.put("connector.read.partition.upper-bound", "-10"); properties.put("connector.read.partition.num", "10"); TableFactoryService.find(StreamTableSourceFactory.class, properties) .createStreamTableSource(properties); fail("exception expected"); } catch (IllegalArgumentException ignored) { } try { Map<String, String> properties = getBasicProperties(); properties.put("connector.lookup.cache.max-rows", "10"); TableFactoryService.find(StreamTableSourceFactory.class, properties) .createStreamTableSource(properties); fail("exception expected"); } catch (IllegalArgumentException ignored) { } try { Map<String, String> properties = getBasicProperties(); properties.put("connector.lookup.cache.ttl", "1s"); TableFactoryService.find(StreamTableSourceFactory.class, properties) .createStreamTableSource(properties); fail("exception expected"); } catch (IllegalArgumentException ignored) { } }
@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(); }
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(); } }
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(); }  }
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(); }  @Override Map<String, String> requiredContext(); @Override List<String> supportedProperties(); @Override StreamTableSource<Row> createStreamTableSource(Map<String, String> properties); @Override StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties); }
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(); }  @Override Map<String, String> requiredContext(); @Override List<String> supportedProperties(); @Override StreamTableSource<Row> createStreamTableSource(Map<String, String> properties); @Override StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties);  }