MartijnVisser commented on code in PR #254:
URL:
https://github.com/apache/flink-connector-jdbc/pull/254#discussion_r4334827511
##########
flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSinkITCase.java:
##########
@@ -519,31 +504,49 @@ protected Map<String, String> getOptions() {
}
@Test
- void testFlushBufferWhenCheckpoint() throws Exception {
- ResolvedSchema schema = checkpointOutputTable.getTableResolvedSchema();
-
- DynamicTableSink tableSink = createTableSink(schema, getOptions());
-
- SinkRuntimeProviderContext context = new
SinkRuntimeProviderContext(false);
- SinkFunctionProvider sinkProvider =
- (SinkFunctionProvider)
tableSink.getSinkRuntimeProvider(context);
- GenericJdbcSinkFunction<RowData> sinkFunction =
- (GenericJdbcSinkFunction<RowData>)
sinkProvider.createSinkFunction();
- sinkFunction.setRuntimeContext(new
TestingStreamingRuntimeContext(true, 1, 0));
- sinkFunction.setInputType(
- TypeInformation.of(GenericRowData.class),
JdbcTestBase.getExecutionConfig(false));
- sinkFunction.open(new DefaultOpenContext());
- sinkFunction.invoke(GenericRowData.of(1L),
SinkContextUtil.forTimestamp(1));
- sinkFunction.invoke(GenericRowData.of(2L),
SinkContextUtil.forTimestamp(1));
-
-
assertThat(checkpointOutputTable.selectAllTable(getMetadata())).isEmpty();
-
- sinkFunction.snapshotState(new StateSnapshotContextSynchronousImpl(1,
1));
-
- assertThat(checkpointOutputTable.selectAllTable(getMetadata()))
- .containsExactlyInAnyOrder(Row.of(1L), Row.of(2L));
+ void testSinkRuntimeProviderIsSinkV2WithTheConfiguredParallelism() {
+ Map<String, String> options = getOptions();
+ options.put("sink.parallelism", "3");
+ DynamicTableSink tableSink =
+
createTableSink(checkpointOutputTable.getTableResolvedSchema(), options);
+
+ DynamicTableSink.SinkRuntimeProvider provider =
+ tableSink.getSinkRuntimeProvider(new
SinkRuntimeProviderContext(false));
+
+ assertThat(provider).isInstanceOf(SinkV2Provider.class);
+ SinkV2Provider sinkProvider = (SinkV2Provider) provider;
+ assertThat(sinkProvider.getParallelism()).hasValue(3);
+
assertThat(sinkProvider.createSink()).isInstanceOf(JdbcTableSink.class);
Review Comment:
I've added the clone, and it also asserts the sink is neither
`SupportsWriterState` nor `SupportsCommitter`.
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/JdbcWriter.java:
##########
@@ -116,19 +150,32 @@ public JdbcWriter(
this.jdbcTransaction.open(state);
this.jdbcTransaction.createTx(lastCheckpointId);
}
-
- checkNotNull(executionOptions, "executionOptions must be defined");
- checkNotNull(queryStatement, "queryStatement must be defined");
- this.jdbcOutput =
- new JdbcOutputFormat<>(
- connectionProvider,
- executionOptions,
- () ->
- JdbcBatchStatementExecutor.simple(
- queryStatement.query(),
queryStatement::statement));
this.jdbcOutput.open(outputSerializer);
Review Comment:
I've fixed both paths in the private constructor: a failed `open()` closes
the format, or rolls the branch back and closes the XA connection. Three tests
pin it and fail on the previous head.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
