eskabetxe commented on code in PR #254: URL: https://github.com/apache/flink-connector-jdbc/pull/254#discussion_r4330165945
########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcTableSink.java: ########## @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.jdbc.core.table.sink; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.connector.sink2.Sink; +import org.apache.flink.api.connector.sink2.SinkWriter; +import org.apache.flink.api.connector.sink2.WriterInitContext; +import org.apache.flink.connector.jdbc.core.datastream.sink.writer.JdbcWriter; +import org.apache.flink.connector.jdbc.internal.JdbcOutputFormat; +import org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer; +import org.apache.flink.streaming.api.lineage.LineageVertex; +import org.apache.flink.streaming.api.lineage.LineageVertexProvider; +import org.apache.flink.table.data.RowData; + +import java.io.IOException; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * The Sink V2 behind {@link JdbcDynamicTableSink}: an at-least-once {@link Sink} that runs a {@link + * JdbcWriter} over the {@link JdbcOutputFormat} that {@link JdbcOutputFormatBuilder} assembled for + * the table. + * + * <p>It implements neither {@code SupportsCommitter} nor {@code SupportsWriterState}: the table + * path has nothing to commit and no state to keep. + */ +@Internal +public final class JdbcTableSink implements Sink<RowData>, LineageVertexProvider { + + private static final long serialVersionUID = 1L; + + private final JdbcOutputFormat<RowData, ?, ?> outputFormat; + + JdbcTableSink(JdbcOutputFormat<RowData, ?, ?> outputFormat) { + this.outputFormat = checkNotNull(outputFormat, "outputFormat must be defined"); + } + + @Override + public SinkWriter<RowData> createWriter(WriterInitContext context) throws IOException { Review Comment: The Sink holds one mutable JdbcOutputFormat that a writer opens and closes, hence the "must not be open yet" note on the new ctor. Nothing actually enforces that (open() never resets the closed flag). It's safe today since Flink re-deserializes per task attempt, but since JdbcOutputFormatBuilder is Serializable, why not keep the builder and build() a fresh format inside createWriter, like JdbcSink does? ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcDynamicTableSink.java: ########## @@ -81,8 +80,7 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) { builder.setJdbcExecutionOptions(executionOptions); builder.setFieldDataTypes( DataType.getFieldDataTypes(physicalRowDataType).toArray(new DataType[0])); - return SinkFunctionProvider.of( - new GenericJdbcSinkFunction<>(builder.build()), jdbcOptions.getParallelism()); + return SinkV2Provider.of(new JdbcTableSink(builder.build()), jdbcOptions.getParallelism()); Review Comment: The sink operator changes identity here (CheckpointedFunction -> SinkWriterOperator). The old one registered no state, so restore should work, but we have zero savepoint fixtures for this. Can you add a savepoint-restore test (or at least a state-compat note on FLINK-37987)? Upgrading running SQL jobs is the risky part of this migration. ########## flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/testutils/TestingInitContext.java: ########## @@ -51,4 +66,39 @@ public JobInfo getJobInfo() { public TaskInfo getTaskInfo() { return taskInfo; } + + @Override + public UserCodeClassLoader getUserCodeClassLoader() { + return SimpleUserCodeClassLoader.create(getClass().getClassLoader()); + } + + @Override + public MailboxExecutor getMailboxExecutor() { + return null; + } + + @Override + public ProcessingTimeService getProcessingTimeService() { + return null; + } + + @Override + public SinkWriterMetricGroup metricGroup() { + return null; + } + + @Override + public SerializationSchema.InitializationContext asSerializationSchemaInitializationContext() { + return null; + } + + @Override + public boolean isObjectReuseEnabled() { + return false; + } + + @Override + public <IN> TypeSerializer<IN> createInputSerializer() { Review Comment: createInputSerializer() returning null only works because isObjectReuseEnabled() is hardcoded false, JdbcTableSink.createWriter calls both, so the javadoc claim "the JDBC writer uses none of them" isn't true, and the first test that turns object reuse on gets a confusing NPE. Btw TestSinkInitContext from the flink-connector-base tests jar is already on our classpath and does all of this properly. Would be better to just extend that? ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/JdbcWriter.java: ########## @@ -75,39 +76,72 @@ public JdbcWriter( JdbcOutputSerializer<IN> outputSerializer, DeliveryGuarantee deliveryGuarantee, Collection<JdbcWriterState> recoveredState, - InitContext initContext) + WriterInitContext initContext) throws IOException { + this( + connectionProvider, + exactlyOnceOptions, + simpleOutputFormat( + checkNotNull(connectionProvider, "connectionProvider must be defined"), + checkNotNull(executionOptions, "executionOptions must be defined"), + checkNotNull(queryStatement, "queryStatement must be defined")), + outputSerializer, + deliveryGuarantee, + recoveredState, + initContext); + } + /** + * An at-least-once writer over a format built elsewhere, for the Table sink. The format must + * not be open yet; the writer opens and closes it. Its executor must accept the writer's input + * type; there is no extraction step in between. + */ + public JdbcWriter( + JdbcOutputFormat<IN, ?, ?> jdbcOutput, + JdbcOutputSerializer<IN> outputSerializer, + WriterInitContext initContext) + throws IOException { + this( + null, + null, + jdbcOutput, + outputSerializer, + DeliveryGuarantee.AT_LEAST_ONCE, + Collections.emptyList(), + initContext); + } + + private JdbcWriter( Review Comment: The mode here is implied by "these two nullables are both null -> at-least-once", which is a bit sneaky. Cleaner: build the XaTransaction in the exactly-once ctor and pass @Nullable XaTransaction down, one nullable instead of two, the cast goes away, and the five deliveryGuarantee == EXACTLY_ONCE checks become jdbcTransaction != null. Fine as a follow-up if you want this PR tight. ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/table/sink/JdbcTableSink.java: ########## @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.jdbc.core.table.sink; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.connector.sink2.Sink; +import org.apache.flink.api.connector.sink2.SinkWriter; +import org.apache.flink.api.connector.sink2.WriterInitContext; +import org.apache.flink.connector.jdbc.core.datastream.sink.writer.JdbcWriter; +import org.apache.flink.connector.jdbc.internal.JdbcOutputFormat; +import org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer; +import org.apache.flink.streaming.api.lineage.LineageVertex; +import org.apache.flink.streaming.api.lineage.LineageVertexProvider; +import org.apache.flink.table.data.RowData; + +import java.io.IOException; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * The Sink V2 behind {@link JdbcDynamicTableSink}: an at-least-once {@link Sink} that runs a {@link + * JdbcWriter} over the {@link JdbcOutputFormat} that {@link JdbcOutputFormatBuilder} assembled for + * the table. + * + * <p>It implements neither {@code SupportsCommitter} nor {@code SupportsWriterState}: the table + * path has nothing to commit and no state to keep. + */ +@Internal +public final class JdbcTableSink implements Sink<RowData>, LineageVertexProvider { + + private static final long serialVersionUID = 1L; + + private final JdbcOutputFormat<RowData, ?, ?> outputFormat; + + JdbcTableSink(JdbcOutputFormat<RowData, ?, ?> outputFormat) { + this.outputFormat = checkNotNull(outputFormat, "outputFormat must be defined"); + } + + @Override + public SinkWriter<RowData> createWriter(WriterInitContext context) throws IOException { + return new JdbcWriter<>( + outputFormat, + JdbcOutputSerializer.of( Review Comment: This line is a copy of JdbcSink.restoreWriter (JdbcSink.java:110-112). A JdbcOutputSerializer.of(WriterInitContext) helper would keep the "how do we get a serializer" convention in one place. ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/JdbcWriter.java: ########## @@ -75,39 +76,72 @@ public JdbcWriter( JdbcOutputSerializer<IN> outputSerializer, DeliveryGuarantee deliveryGuarantee, Collection<JdbcWriterState> recoveredState, - InitContext initContext) + WriterInitContext initContext) throws IOException { + this( + connectionProvider, + exactlyOnceOptions, + simpleOutputFormat( + checkNotNull(connectionProvider, "connectionProvider must be defined"), + checkNotNull(executionOptions, "executionOptions must be defined"), + checkNotNull(queryStatement, "queryStatement must be defined")), + outputSerializer, + deliveryGuarantee, + recoveredState, + initContext); + } + /** + * An at-least-once writer over a format built elsewhere, for the Table sink. The format must + * not be open yet; the writer opens and closes it. Its executor must accept the writer's input + * type; there is no extraction step in between. + */ + public JdbcWriter( + JdbcOutputFormat<IN, ?, ?> jdbcOutput, + JdbcOutputSerializer<IN> outputSerializer, + WriterInitContext initContext) + throws IOException { + this( + null, + null, + jdbcOutput, + outputSerializer, + DeliveryGuarantee.AT_LEAST_ONCE, + Collections.emptyList(), + initContext); + } + + private JdbcWriter( + @Nullable JdbcConnectionProvider connectionProvider, + @Nullable JdbcExactlyOnceOptions exactlyOnceOptions, + JdbcOutputFormat<IN, ?, ?> jdbcOutput, + JdbcOutputSerializer<IN> outputSerializer, + DeliveryGuarantee deliveryGuarantee, + Collection<JdbcWriterState> recoveredState, + WriterInitContext initContext) + throws IOException { this.deliveryGuarantee = checkNotNull(deliveryGuarantee, "deliveryGuarantee must be defined"); - checkNotNull(initContext, "initContext must be defined"); - + this.jdbcOutput = checkNotNull(jdbcOutput, "jdbcOutput must be defined"); pendingRecords = false; this.lastCheckpointId = - initContext.getRestoredCheckpointId().orElse(InitContext.INITIAL_CHECKPOINT_ID - 1); - - checkNotNull(connectionProvider, "connectionProvider must be defined"); - + initContext + .getRestoredCheckpointId() + .orElse(WriterInitContext.INITIAL_CHECKPOINT_ID - 1); if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) { - checkArgument( - executionOptions.getMaxRetries() == 0, - "JDBC XA sink requires maxRetries equal to 0, otherwise it could " - + "cause duplicates. See issue FLINK-22311 for details."); - + checkNotNull(connectionProvider, "connectionProvider must be defined"); Review Comment: This checkNotNull(connectionProvider, ...) can never fire: the datastream ctor already checks it before delegating, and the table ctor hardcodes AT_LEAST_ONCE so we never enter this branch with it. Can be droped. ########## flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceITCase.java: ########## @@ -68,10 +60,11 @@ void testReadWithoutParallelismWithoutParamsProvider() throws Exception { .setDriverName(getMetadata().getDriverClass()) .setResultExtractor(extractor) .build(); - env.fromSource(jdbcSource, WatermarkStrategy.noWatermarks(), "TestSource") - .addSink(new TestingSinkFunction()); - env.execute(); - assertThat(collectedRecords).containsExactlyInAnyOrder(TEST_DATA); + try (CloseableIterator<TestEntry> records = Review Comment: This five-line collect-and-assert block appears 4x here and 4x in DataStreamSourceTest. A tiny assertCollectedExactly(env, source, expected) helper in testutils would kill all the copies. ########## flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/BaseJdbcWriterTest.java: ########## @@ -54,7 +54,7 @@ * org.apache.flink.connector.jdbc.core.datastream.sink.writer.JdbcWriter} and the underlying * classes. */ -abstract class BaseJdbcWriterTest implements DerbyTestBase { +public abstract class BaseJdbcWriterTest implements DerbyTestBase { Review Comment: The whole class went public just so the ITCase can grab TestWriterContext. Since you already moved TestingInitContext to testutils, move the 10-line TestWriterContext there too and keep this class package-private. Also, sinkContext is now a WriterInitContext and production code calls it initContext, let's match. ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/JdbcWriter.java: ########## @@ -75,39 +76,72 @@ public JdbcWriter( JdbcOutputSerializer<IN> outputSerializer, DeliveryGuarantee deliveryGuarantee, Collection<JdbcWriterState> recoveredState, - InitContext initContext) + WriterInitContext initContext) Review Comment: This is the ctor JdbcSink.restoreWriter uses, but nothing tests it with real recovered state or a restored checkpoint id. A restart-based exactly-once test would cover the recovery wiring this PR moved to WriterInitContext. ########## 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: if jdbcOutput.open() throws, nothing closes the format and we leak the connection. The old GenericJdbcSinkFunction never leaked because StreamTask calls close() even when open() failed, so this is a new regression for the table path. A try/catch around the open sequence + closing format/transaction should fix it (the datastream path has the same leak already, feel free to fix both). ########## 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: JdbcTableSink is now serialized job-graph payload carrying the whole built format, one InstantiationUtil.clone(sinkProvider.createSink()) on the next line would pin that contract right where we assert the provider type. ########## flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/sink/JdbcSinkBuilder.java: ########## @@ -106,6 +107,10 @@ private JdbcSink<IN> build( DeliveryGuarantee deliveryGuarantee, JdbcExactlyOnceOptions exactlyOnceOptions, JdbcConnectionProvider connectionProvider) { + if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) { + JdbcWriter.checkExactlyOnceExecutionOptions( Review Comment: Heads up: the default execution options have maxRetries=3, so buildExactlyOnce now blows up at build time unless the user explicitly sets 0. Behaviorally nobody is newly broken (it failed at startup before), but it's a visible change on a @PublicEvolving class that japicmp won't catch. Worth a release-note line. Also, JdbcSinkBuilderTest only covers the reject case, please add the happy path (maxRetries=0 builds fine) and the untouched-default case. ########## flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/sink/writer/AlLeastOnceJdbcWriterTest.java: ########## @@ -102,4 +107,38 @@ void testMultipleFlush() throws Exception { assertThat(TEST_TABLE.selectAllTable(getMetadata().getConnection())).isEqualTo(BOOKS); } + + @Test + void testPrebuiltOutputFormat() throws Exception { + // the Table sink path: a format assembled outside the writer + JdbcOutputFormat<BooksTable.BookEntry, ?, ?> outputFormat = + new JdbcOutputFormat<>( + getConnectionProvider(), + getExecutionOptions(), + () -> + JdbcBatchStatementExecutor.simple( + TEST_TABLE.getInsertIntoQuery(), + TEST_TABLE.getStatementBuilder())); + JdbcWriter<BooksTable.BookEntry> writer = + new JdbcWriter<>( + outputFormat, + JdbcOutputSerializer.of( Review Comment: The serializer setup here is copy-pasted from BaseJdbcWriterTest (95-98) extract a small helper. Also the flush and empty commit/state assertions are already covered by testMultipleFlush/testCheckpoint and the new ITCase; the interesting part of this test is just "externally built format works", so consider trimming the rest. -- 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]
