MartijnVisser commented on code in PR #254: URL: https://github.com/apache/flink-connector-jdbc/pull/254#discussion_r4334825820
########## 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: I've added `JdbcOutputSerializer.of(WriterInitContext)`, used by `JdbcSink.restoreWriter` and `JdbcTableSink`. ########## 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: I've done it here: the private constructor takes one nullable `XaTransaction`, the `deliveryGuarantee` field is gone and the mode checks are null checks. The cast stays in the helper, since `JdbcSink` keeps a `JdbcConnectionProvider` for both modes. ########## 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: I've added `JdbcITCaseBase.assertSourceProducesExactly` and use it in both classes. ########## 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: I've moved it to `testutils.TestingWriterContext`; the class is package-private again and the field is `initContext`. ########## 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: I've added `testRecovery` to `BaseJdbcSinkTest` on the `KafkaSinkITCase` pattern, a restart under both guarantees, plus two restore unit tests with real state, a hanging and a prepared branch. 10 of 10 local runs green. ########## 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: I've trimmed it to the externally built format, with `bookSerializer()` shared from the base. -- 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]
