MartijnVisser commented on code in PR #254: URL: https://github.com/apache/flink-connector-jdbc/pull/254#discussion_r4334824310
########## 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: Yes, no reason not to. `JdbcTableSink` now holds the `JdbcOutputFormatBuilder` and builds a format per writer, `JdbcDynamicTableSink.outputFormatBuilder()` is the shared assembly, and the provider test clones the sink. ########## 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: I've added `JdbcTableSinkMigrationITCase`, same shape as `FileSinkMigrationITCase`: a job on the 4.x sink function stops with a savepoint and the Sink V2 sink restores it under the same uids. The nested copy of the old sink function is the price; a committed fixture replaces it once Flink drops the legacy package. The uid-less caveat goes into the release note and a comment on the ticket. ########## 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: I've added both tests, including the untouched default (maxRetries 3, rejected at build time), and drafted the release note line. ########## 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: It's gone with the constructor rework: the exactly-once setup lives in `openXaTransaction`, which checks for an `XaConnectionProvider` with a message instead of the bare cast. ########## 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: `TestSinkInitContext` isn't on the dialect modules' classpath, where this ITCase also runs, and its `createInputSerializer()` returns null too. I've given the double a real serializer, metric group and restored checkpoint id, and the ITCase now also runs with object reuse on. -- 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]
