@Test public void testJDBCSinkProperties() { Map<String, String> properties = getBasicProperties(); properties.put("connector.write.flush.max-rows", "1000"); properties.put("connector.write.flush.interval", "2min"); properties.put("connector.write.max-retries", "5"); final StreamTableSink<?> actual = TableFactoryService.find(StreamTableSinkFactory.class, properties) .createStreamTableSink(properties); final JDBCOptions options = JDBCOptions.builder() .setDBUrl("jdbc:derby:memory:mydb") .setTableName("mytable") .build(); final TableSchema schema = TableSchema.builder() .field("aaa", DataTypes.INT()) .field("bbb", DataTypes.STRING()) .field("ccc", DataTypes.DOUBLE()) .build(); final JDBCUpsertTableSink expected = JDBCUpsertTableSink.builder() .setOptions(options) .setTableSchema(schema) .setFlushMaxSize(1000) .setFlushIntervalMills(120_000) .setMaxRetryTimes(5) .build(); assertEquals(expected, actual); }
@Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final JDBCUpsertTableSink.Builder builder = JDBCUpsertTableSink.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setTableSchema(descriptorProperties.getTableSchema(SCHEMA)); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_FLUSH_MAX_ROWS).ifPresent(builder::setFlushMaxSize); descriptorProperties.getOptionalDuration(CONNECTOR_WRITE_FLUSH_INTERVAL).ifPresent( s -> builder.setFlushIntervalMills(s.toMillis())); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_MAX_RETRIES).ifPresent(builder::setMaxRetryTimes); return builder.build(); }
JDBCTableSourceSinkFactory implements
	StreamTableSourceFactory<Row>,
	StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final JDBCUpsertTableSink.Builder builder = JDBCUpsertTableSink.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setTableSchema(descriptorProperties.getTableSchema(SCHEMA)); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_FLUSH_MAX_ROWS).ifPresent(builder::setFlushMaxSize); descriptorProperties.getOptionalDuration(CONNECTOR_WRITE_FLUSH_INTERVAL).ifPresent( s -> builder.setFlushIntervalMills(s.toMillis())); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_MAX_RETRIES).ifPresent(builder::setMaxRetryTimes); return builder.build(); } }
JDBCTableSourceSinkFactory implements
	StreamTableSourceFactory<Row>,
	StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final JDBCUpsertTableSink.Builder builder = JDBCUpsertTableSink.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setTableSchema(descriptorProperties.getTableSchema(SCHEMA)); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_FLUSH_MAX_ROWS).ifPresent(builder::setFlushMaxSize); descriptorProperties.getOptionalDuration(CONNECTOR_WRITE_FLUSH_INTERVAL).ifPresent( s -> builder.setFlushIntervalMills(s.toMillis())); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_MAX_RETRIES).ifPresent(builder::setMaxRetryTimes); return builder.build(); }  }
JDBCTableSourceSinkFactory implements
	StreamTableSourceFactory<Row>,
	StreamTableSinkFactory<Tuple2<Boolean, Row>> { @Override public StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final JDBCUpsertTableSink.Builder builder = JDBCUpsertTableSink.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setTableSchema(descriptorProperties.getTableSchema(SCHEMA)); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_FLUSH_MAX_ROWS).ifPresent(builder::setFlushMaxSize); descriptorProperties.getOptionalDuration(CONNECTOR_WRITE_FLUSH_INTERVAL).ifPresent( s -> builder.setFlushIntervalMills(s.toMillis())); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_MAX_RETRIES).ifPresent(builder::setMaxRetryTimes); return builder.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 StreamTableSink<Tuple2<Boolean, Row>> createStreamTableSink(Map<String, String> properties) { final DescriptorProperties descriptorProperties = getValidatedProperties(properties); final JDBCUpsertTableSink.Builder builder = JDBCUpsertTableSink.builder() .setOptions(getJDBCOptions(descriptorProperties)) .setTableSchema(descriptorProperties.getTableSchema(SCHEMA)); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_FLUSH_MAX_ROWS).ifPresent(builder::setFlushMaxSize); descriptorProperties.getOptionalDuration(CONNECTOR_WRITE_FLUSH_INTERVAL).ifPresent( s -> builder.setFlushIntervalMills(s.toMillis())); descriptorProperties.getOptionalInt(CONNECTOR_WRITE_MAX_RETRIES).ifPresent(builder::setMaxRetryTimes); return builder.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);  }