You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2025/04/03 07:08:04 UTC
(flink-connector-jdbc) 01/02: [FLINK-36659] Remove deprecated flink-connector-jdbc module.
This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
commit da674747aab6dee520b1aab763800c516464d9d5
Author: lvyanquan <lv...@alibaba-inc.com>
AuthorDate: Tue Mar 25 17:30:48 2025 +0800
[FLINK-36659] Remove deprecated flink-connector-jdbc module.
---
flink-connector-jdbc-architecture/pom.xml | 6 -
flink-connector-jdbc/pom.xml | 214 ----------
.../org/apache/flink/connector/jdbc/JdbcSink.java | 121 ------
.../jdbc/catalog/AbstractJdbcCatalog.java | 39 --
.../flink/connector/jdbc/catalog/JdbcCatalog.java | 172 --------
.../connector/jdbc/catalog/JdbcCatalogUtils.java | 115 -----
.../jdbc/catalog/factory/JdbcCatalogFactory.java | 93 -----
.../jdbc/converter/AbstractJdbcRowConverter.java | 34 --
.../connector/jdbc/converter/JdbcRowConverter.java | 33 --
.../connector/jdbc/dialect/AbstractDialect.java | 38 --
.../flink/connector/jdbc/dialect/JdbcDialect.java | 34 --
.../connector/jdbc/dialect/JdbcDialectFactory.java | 75 ----
.../jdbc/dialect/JdbcDialectTypeMapper.java | 29 --
.../apache/flink/connector/jdbc/sink/JdbcSink.java | 118 ------
.../flink/connector/jdbc/sink/JdbcSinkBuilder.java | 122 ------
.../flink/connector/jdbc/source/JdbcSource.java | 203 ---------
.../connector/jdbc/source/JdbcSourceBuilder.java | 324 --------------
.../source/reader/extractor/ResultExtractor.java | 37 --
.../connector/jdbc/table/JdbcConnectorOptions.java | 31 --
.../jdbc/utils/ContinuousUnBoundingSettings.java | 42 --
.../flink/connector/jdbc/xa/CheckpointAndXid.java | 92 ----
.../jdbc/xa/CheckpointAndXidSerializer.java | 137 ------
.../connector/jdbc/xa/JdbcXaSinkFunction.java | 373 -----------------
.../connector/jdbc/xa/JdbcXaSinkFunctionState.java | 77 ----
.../apache/flink/connector/jdbc/xa/JobSubtask.java | 53 ---
.../connector/jdbc/xa/SemanticXidGenerator.java | 110 -----
.../apache/flink/connector/jdbc/xa/XaFacade.java | 123 ------
.../flink/connector/jdbc/xa/XaFacadeImpl.java | 463 ---------------------
.../connector/jdbc/xa/XaFacadePoolingImpl.java | 194 ---------
.../apache/flink/connector/jdbc/xa/XaGroupOps.java | 99 -----
.../flink/connector/jdbc/xa/XaGroupOpsImpl.java | 143 -------
.../connector/jdbc/xa/XaSinkStateHandler.java | 32 --
.../connector/jdbc/xa/XaSinkStateHandlerImpl.java | 96 -----
.../connector/jdbc/xa/XaSinkStateSerializer.java | 161 -------
.../flink/connector/jdbc/xa/XidGenerator.java | 71 ----
.../flink/connector/jdbc/xa/package-info.java | 24 --
.../org.apache.flink.table.factories.Factory | 16 -
.../jdbc/dialect/JdbcDialectTypeTest.java | 29 --
.../jdbc/xa/CheckpointAndXidSerializersTest.java | 49 ---
.../jdbc/xa/JdbcExactlyOnceSinkE2eTest.java | 400 ------------------
.../connector/jdbc/xa/JdbcXaFacadeImplTest.java | 100 -----
.../connector/jdbc/xa/JdbcXaFacadeTestHelper.java | 107 -----
.../connector/jdbc/xa/JdbcXaSinkDerbyTest.java | 167 --------
.../flink/connector/jdbc/xa/JdbcXaSinkH2Test.java | 81 ----
.../connector/jdbc/xa/JdbcXaSinkMigrationTest.java | 211 ----------
.../jdbc/xa/JdbcXaSinkNoInsertionTest.java | 73 ----
.../connector/jdbc/xa/JdbcXaSinkTestBase.java | 208 ---------
.../connector/jdbc/xa/JdbcXaSinkTestHelper.java | 93 -----
.../jdbc/xa/SemanticXidGeneratorTest.java | 69 ---
tools/releasing/shared | 1 -
50 files changed, 5732 deletions(-)
diff --git a/flink-connector-jdbc-architecture/pom.xml b/flink-connector-jdbc-architecture/pom.xml
index 3e123fc..d916d22 100644
--- a/flink-connector-jdbc-architecture/pom.xml
+++ b/flink-connector-jdbc-architecture/pom.xml
@@ -36,12 +36,6 @@
</dependency>
<!-- Flink Jdbc Modules To Test -->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc</artifactId>
- <version>${project.version}</version>
- <scope>test</scope>
- </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc-core</artifactId>
diff --git a/flink-connector-jdbc/pom.xml b/flink-connector-jdbc/pom.xml
deleted file mode 100644
index 0a6f4cb..0000000
--- a/flink-connector-jdbc/pom.xml
+++ /dev/null
@@ -1,214 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
-
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-parent</artifactId>
- <version>3.3-SNAPSHOT</version>
- </parent>
-
- <artifactId>flink-connector-jdbc</artifactId>
- <name>Flink : Connectors : JDBC : Shaded</name>
-
- <packaging>jar</packaging>
-
- <dependencies>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-core</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-cratedb</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-db2</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-mysql</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-oceanbase</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-oracle</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-postgres</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-sqlserver</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-trino</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-jdbc-core</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-annotations</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-test-utils</artifactId>
- <version>${flink.version}</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-common</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-api-java-bridge</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- <optional>true</optional>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
-
- <!-- Derby tests -->
- <dependency>
- <groupId>org.apache.derby</groupId>
- <artifactId>derby</artifactId>
- <version>10.14.2.0</version>
- <scope>test</scope>
- </dependency>
-
- <!-- H2 tests -->
- <dependency>
- <groupId>com.h2database</groupId>
- <artifactId>h2</artifactId>
- <version>2.2.224</version>
- <scope>test</scope>
- </dependency>
-
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <executions>
- <execution>
- <phase>package</phase>
- <goals>
- <goal>shade</goal>
- </goals>
- <configuration>
- <createDependencyReducedPom>false</createDependencyReducedPom>
- <shadeTestJar>true</shadeTestJar>
- <artifactSet>
- <excludes>
- <exclude>org.apache.flink:flink-annotations</exclude>
- <exclude>org.apache.flink:flink-core</exclude>
- <exclude>org.apache.flink:flink-connector-base</exclude>
- <exclude>org.apache.flink:flink-table-common</exclude>
- <exclude>org.apache.flink:flink-connector-jdbc-core:tests</exclude>
- </excludes>
- </artifactSet>
- <filters combine.children="append">
- <filter combine.children="append">
- <!-- Do not copy the signatures in the META-INF folder. -->
- <!-- Otherwise, this might cause SecurityExceptions when using the JAR. -->
- <artifact>*:*</artifact>
- <excludes combine.children="append">
- <exclude>META-INF/*.SF</exclude>
- <exclude>META-INF/*.DSA</exclude>
- <exclude>META-INF/*.RSA</exclude>
- </excludes>
- </filter>
- </filters>
- <transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"/>
- </transformers>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <executions>
- <execution>
- <goals>
- <goal>test-jar</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-</project>
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcSink.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcSink.java
deleted file mode 100644
index 81bbc83..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/JdbcSink.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * 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;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSinkBuilder;
-import org.apache.flink.connector.jdbc.datasource.connections.SimpleJdbcConnectionProvider;
-import org.apache.flink.connector.jdbc.internal.GenericJdbcSinkFunction;
-import org.apache.flink.connector.jdbc.internal.JdbcOutputFormat;
-import org.apache.flink.connector.jdbc.internal.executor.JdbcBatchStatementExecutor;
-import org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction;
-import org.apache.flink.connector.jdbc.xa.XaFacade;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.util.function.SerializableSupplier;
-
-import javax.sql.XADataSource;
-
-/**
- * Facade to create JDBC {@link SinkFunction sinks}.
- *
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.datastream.Jdbc}
- */
-@PublicEvolving
-@Deprecated
-public class JdbcSink {
-
- /**
- * Create a JDBC sink with the default {@link JdbcExecutionOptions}.
- *
- * @see #sink(String, JdbcStatementBuilder, JdbcExecutionOptions, JdbcConnectionOptions)
- */
- public static <T> SinkFunction<T> sink(
- String sql,
- JdbcStatementBuilder<T> statementBuilder,
- JdbcConnectionOptions connectionOptions) {
- return sink(sql, statementBuilder, JdbcExecutionOptions.defaults(), connectionOptions);
- }
-
- /**
- * Create a JDBC sink.
- *
- * <p>Note: the objects passed to the return sink can be processed in batch and retried.
- * Therefore, objects can not be {@link
- * org.apache.flink.api.common.ExecutionConfig#enableObjectReuse() reused}.
- *
- * @param sql arbitrary DML query (e.g. insert, update, upsert)
- * @param statementBuilder sets parameters on {@link java.sql.PreparedStatement} according to
- * the query
- * @param <T> type of data in {@link
- * org.apache.flink.streaming.runtime.streamrecord.StreamRecord StreamRecord}.
- * @param executionOptions parameters of execution, such as batch size and maximum retries
- * @param connectionOptions parameters of connection, such as JDBC URL
- */
- public static <T> SinkFunction<T> sink(
- String sql,
- JdbcStatementBuilder<T> statementBuilder,
- JdbcExecutionOptions executionOptions,
- JdbcConnectionOptions connectionOptions) {
- return new GenericJdbcSinkFunction<>(
- new JdbcOutputFormat<>(
- new SimpleJdbcConnectionProvider(connectionOptions),
- executionOptions,
- () -> JdbcBatchStatementExecutor.simple(sql, statementBuilder)));
- }
-
- /**
- * Create JDBC sink which provides exactly-once guarantee.
- *
- * <p>Note: the objects passed to the return sink can be processed in batch and retried.
- * Therefore, objects can not be {@link
- * org.apache.flink.api.common.ExecutionConfig#enableObjectReuse() reused}.
- *
- * @param sql arbitrary DML query (e.g. insert, update, upsert)
- * @param statementBuilder sets parameters on {@link java.sql.PreparedStatement} according to
- * the query
- * @param <T> type of data in {@link
- * org.apache.flink.streaming.runtime.streamrecord.StreamRecord StreamRecord}.
- * @param executionOptions parameters of execution, such as batch size and maximum retries
- * @param exactlyOnceOptions exactly-once options. Note: maxRetries setting must be strictly set
- * to 0 for the created sink to work properly and not to produce duplicates. See issue
- * FLINK-22311 for details.
- * @param dataSourceSupplier supplies the {@link XADataSource}
- */
- public static <T> SinkFunction<T> exactlyOnceSink(
- String sql,
- JdbcStatementBuilder<T> statementBuilder,
- JdbcExecutionOptions executionOptions,
- JdbcExactlyOnceOptions exactlyOnceOptions,
- SerializableSupplier<XADataSource> dataSourceSupplier) {
- return new JdbcXaSinkFunction<>(
- sql,
- statementBuilder,
- XaFacade.fromXaDataSourceSupplier(
- dataSourceSupplier,
- exactlyOnceOptions.getTimeoutSec(),
- exactlyOnceOptions.isTransactionPerConnection()),
- executionOptions,
- exactlyOnceOptions);
- }
-
- public static <IN> JdbcSinkBuilder<IN> builder() {
- return org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSink.builder();
- }
-
- private JdbcSink() {}
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
deleted file mode 100644
index c565836..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/AbstractJdbcCatalog.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.catalog;
-
-/**
- * Abstract catalog for any JDBC catalogs.
- *
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.database.catalog.AbstractJdbcCatalog}
- */
-@Deprecated
-public abstract class AbstractJdbcCatalog
- extends org.apache.flink.connector.jdbc.core.database.catalog.AbstractJdbcCatalog {
-
- public AbstractJdbcCatalog(
- ClassLoader userClassLoader,
- String catalogName,
- String defaultDatabase,
- String username,
- String pwd,
- String baseUrl) {
- super(userClassLoader, catalogName, defaultDatabase, username, pwd, baseUrl);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java
deleted file mode 100644
index 08db15d..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalog.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/*
- * 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.catalog;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.connector.jdbc.core.database.catalog.AbstractJdbcCatalog;
-import org.apache.flink.table.catalog.CatalogBaseTable;
-import org.apache.flink.table.catalog.CatalogDatabase;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.exceptions.CatalogException;
-import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
-
-import java.util.List;
-import java.util.Properties;
-
-import static org.apache.flink.connector.jdbc.JdbcConnectionOptions.getBriefAuthProperties;
-
-/**
- * Catalogs for relational databases via JDBC.
- *
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.database.catalog.JdbcCatalog}
- */
-@Deprecated
-@PublicEvolving
-public class JdbcCatalog extends AbstractJdbcCatalog {
-
- private final AbstractJdbcCatalog internal;
-
- @Deprecated
- /**
- * Creates a JdbcCatalog.
- *
- * @deprecated please use {@link JdbcCatalog#JdbcCatalog(ClassLoader, String, String, String,
- * String, Properties)} instead.
- */
- public JdbcCatalog(
- String catalogName,
- String defaultDatabase,
- String username,
- String pwd,
- String baseUrl) {
- this(
- Thread.currentThread().getContextClassLoader(),
- catalogName,
- defaultDatabase,
- baseUrl,
- null,
- getBriefAuthProperties(username, pwd));
- }
-
- @VisibleForTesting
- /**
- * Creates a JdbcCatalog.
- *
- * @param userClassLoader the classloader used to load JDBC driver
- * @param catalogName the registered catalog name
- * @param defaultDatabase the default database name
- * @param username the username used to connect the database
- * @param pwd the password used to connect the database
- * @param baseUrl the base URL of the database, e.g. jdbc:mysql://localhost:3306
- * @param compatibleMode the compatible mode of the database
- */
- public JdbcCatalog(
- ClassLoader userClassLoader,
- String catalogName,
- String defaultDatabase,
- String username,
- String pwd,
- String baseUrl,
- String compatibleMode) {
- this(
- userClassLoader,
- catalogName,
- defaultDatabase,
- baseUrl,
- compatibleMode,
- getBriefAuthProperties(username, pwd));
- }
-
- /**
- * Creates a JdbcCatalog.
- *
- * @param userClassLoader the classloader used to load JDBC driver
- * @param catalogName the registered catalog name
- * @param defaultDatabase the default database name
- * @param connectProperties the properties used to connect the database
- * @param baseUrl the base URL of the database, e.g. jdbc:mysql://localhost:3306
- * @param compatibleMode the compatible mode of the database
- */
- public JdbcCatalog(
- ClassLoader userClassLoader,
- String catalogName,
- String defaultDatabase,
- String baseUrl,
- String compatibleMode,
- Properties connectProperties) {
- super(userClassLoader, catalogName, defaultDatabase, baseUrl, connectProperties);
-
- internal =
- JdbcCatalogUtils.createCatalog(
- userClassLoader,
- catalogName,
- defaultDatabase,
- baseUrl,
- compatibleMode,
- connectProperties);
- }
-
- // ------ databases -----
-
- @Override
- public List<String> listDatabases() throws CatalogException {
- return internal.listDatabases();
- }
-
- @Override
- public CatalogDatabase getDatabase(String databaseName)
- throws DatabaseNotExistException, CatalogException {
- return internal.getDatabase(databaseName);
- }
-
- // ------ tables and views ------
-
- @Override
- public List<String> listTables(String databaseName)
- throws DatabaseNotExistException, CatalogException {
- return internal.listTables(databaseName);
- }
-
- @Override
- public CatalogBaseTable getTable(ObjectPath tablePath)
- throws TableNotExistException, CatalogException {
- return internal.getTable(tablePath);
- }
-
- @Override
- public boolean tableExists(ObjectPath tablePath) throws CatalogException {
- try {
- return databaseExists(tablePath.getDatabaseName())
- && listTables(tablePath.getDatabaseName()).contains(tablePath.getObjectName());
- } catch (DatabaseNotExistException e) {
- return false;
- }
- }
-
- // ------ getters ------
-
- @VisibleForTesting
- @Internal
- public AbstractJdbcCatalog getInternal() {
- return internal;
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java
deleted file mode 100644
index 73b7d44..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/JdbcCatalogUtils.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * 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.catalog;
-
-import org.apache.flink.connector.jdbc.core.database.JdbcFactoryLoader;
-import org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialect;
-import org.apache.flink.connector.jdbc.cratedb.database.catalog.CrateDBCatalog;
-import org.apache.flink.connector.jdbc.cratedb.database.dialect.CrateDBDialect;
-import org.apache.flink.connector.jdbc.mysql.database.catalog.MySqlCatalog;
-import org.apache.flink.connector.jdbc.mysql.database.dialect.MySqlDialect;
-import org.apache.flink.connector.jdbc.postgres.database.catalog.PostgresCatalog;
-import org.apache.flink.connector.jdbc.postgres.database.dialect.PostgresDialect;
-
-import java.util.Properties;
-
-import static org.apache.flink.connector.jdbc.JdbcConnectionOptions.getBriefAuthProperties;
-import static org.apache.flink.util.Preconditions.checkArgument;
-
-/**
- * Utils for {@link JdbcCatalog}.
- *
- * @deprecated
- */
-@Deprecated
-public class JdbcCatalogUtils {
- /**
- * URL has to be without database, like "jdbc:postgresql://localhost:5432/" or
- * "jdbc:postgresql://localhost:5432" rather than "jdbc:postgresql://localhost:5432/db".
- */
- public static void validateJdbcUrl(String url) {
- String[] parts = url.trim().split("\\/+");
-
- checkArgument(parts.length == 2);
- }
-
- @Deprecated
- /** Create catalog instance from given information. */
- public static AbstractJdbcCatalog createCatalog(
- ClassLoader userClassLoader,
- String catalogName,
- String defaultDatabase,
- String username,
- String pwd,
- String baseUrl,
- String compatibleMode) {
- return createCatalog(
- userClassLoader,
- catalogName,
- defaultDatabase,
- baseUrl,
- compatibleMode,
- getBriefAuthProperties(username, pwd));
- }
-
- /** Create catalog instance from given information. */
- public static AbstractJdbcCatalog createCatalog(
- ClassLoader userClassLoader,
- String catalogName,
- String defaultDatabase,
- String baseUrl,
- String compatibleMode,
- Properties connectionProperties) {
-
- JdbcDialect dialect =
- JdbcFactoryLoader.loadDialect(baseUrl, userClassLoader, compatibleMode);
-
- org.apache.flink.connector.jdbc.core.database.catalog.AbstractJdbcCatalog catalog;
- if (dialect instanceof PostgresDialect) {
- catalog =
- new PostgresCatalog(
- userClassLoader,
- catalogName,
- defaultDatabase,
- baseUrl,
- connectionProperties);
- } else if (dialect instanceof CrateDBDialect) {
- catalog =
- new CrateDBCatalog(
- userClassLoader,
- catalogName,
- defaultDatabase,
- baseUrl,
- connectionProperties);
- } else if (dialect instanceof MySqlDialect) {
- catalog =
- new MySqlCatalog(
- userClassLoader,
- catalogName,
- defaultDatabase,
- baseUrl,
- connectionProperties);
- } else {
- throw new UnsupportedOperationException(
- String.format("Catalog for '%s' is not supported yet.", dialect));
- }
-
- return (AbstractJdbcCatalog) catalog;
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
deleted file mode 100644
index 223b91c..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/catalog/factory/JdbcCatalogFactory.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * 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.catalog.factory;
-
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.connector.jdbc.catalog.JdbcCatalog;
-import org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactoryOptions;
-import org.apache.flink.table.catalog.Catalog;
-import org.apache.flink.table.factories.CatalogFactory;
-import org.apache.flink.table.factories.FactoryUtil;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.HashSet;
-import java.util.Set;
-
-import static org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactoryOptions.BASE_URL;
-import static org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactoryOptions.COMPATIBLE_MODE;
-import static org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactoryOptions.DEFAULT_DATABASE;
-import static org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactoryOptions.PASSWORD;
-import static org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactoryOptions.USERNAME;
-import static org.apache.flink.connector.jdbc.utils.JdbcUtils.getConnectionProperties;
-import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
-
-/**
- * Factory for {@link JdbcCatalog}.
- *
- * @deprecated Use {@link
- * org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactory}.
- */
-@Deprecated
-public class JdbcCatalogFactory implements CatalogFactory {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(
- org.apache.flink.connector.jdbc.core.database.catalog.factory.JdbcCatalogFactory
- .class);
-
- @Override
- public String factoryIdentifier() {
- return JdbcCatalogFactoryOptions.IDENTIFIER;
- }
-
- @Override
- public Set<ConfigOption<?>> requiredOptions() {
- final Set<ConfigOption<?>> options = new HashSet<>();
- options.add(DEFAULT_DATABASE);
- options.add(USERNAME);
- options.add(PASSWORD);
- options.add(BASE_URL);
- return options;
- }
-
- @Override
- public Set<ConfigOption<?>> optionalOptions() {
- final Set<ConfigOption<?>> options = new HashSet<>();
- options.add(PROPERTY_VERSION);
- options.add(COMPATIBLE_MODE);
- return options;
- }
-
- @Override
- public Catalog createCatalog(Context context) {
- final FactoryUtil.CatalogFactoryHelper helper =
- FactoryUtil.createCatalogFactoryHelper(this, context);
- helper.validate();
-
- return new JdbcCatalog(
- context.getClassLoader(),
- context.getName(),
- helper.getOptions().get(DEFAULT_DATABASE),
- helper.getOptions().get(BASE_URL),
- helper.getOptions().get(COMPATIBLE_MODE),
- getConnectionProperties(helper.getOptions()));
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java
deleted file mode 100644
index 3f01620..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/AbstractJdbcRowConverter.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.converter;
-
-import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialectConverter;
-import org.apache.flink.table.types.logical.RowType;
-
-/**
- * Base class for all converters that convert between JDBC object and Flink internal object.
- *
- * @deprecated Use {@link AbstractDialectConverter}
- */
-@Deprecated
-public abstract class AbstractJdbcRowConverter extends AbstractDialectConverter {
- public AbstractJdbcRowConverter(RowType rowType) {
- super(rowType);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/JdbcRowConverter.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/JdbcRowConverter.java
deleted file mode 100644
index bd0bc6c..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/converter/JdbcRowConverter.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.converter;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialectConverter;
-import org.apache.flink.table.data.RowData;
-
-/**
- * Converter that is responsible to convert between JDBC object and Flink SQL internal data
- * structure {@link RowData}.
- *
- * @deprecated Use {@link JdbcDialectConverter}
- */
-@Deprecated
-@PublicEvolving
-public interface JdbcRowConverter extends JdbcDialectConverter {}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/AbstractDialect.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/AbstractDialect.java
deleted file mode 100644
index 3a5af41..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/AbstractDialect.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.dialect;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialect;
-
-/**
- * Base class for {@link JdbcDialect JdbcDialects} that implements basic data type validation and
- * the construction of basic {@code INSERT}, {@code UPDATE}, {@code DELETE}, and {@code SELECT}
- * statements.
- *
- * <p>Implementors should be careful to check the default SQL statements are performant for their
- * specific dialect and override them if necessary.
- *
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialect}
- */
-@Deprecated
-@PublicEvolving
-public abstract class AbstractDialect
- extends org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialect
- implements org.apache.flink.connector.jdbc.dialect.JdbcDialect {}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialect.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialect.java
deleted file mode 100644
index b8d2a63..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialect.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.dialect;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.jdbc.core.database.JdbcFactory;
-
-/**
- * Represents a dialect of SQL implemented by a particular JDBC system. Dialects should be immutable
- * and stateless.
- *
- * @see JdbcFactory
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialect}
- */
-@Deprecated
-@PublicEvolving
-public interface JdbcDialect
- extends org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialect {}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectFactory.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectFactory.java
deleted file mode 100644
index de674cd..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectFactory.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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.dialect;
-
-import org.apache.flink.connector.jdbc.core.database.JdbcFactory;
-import org.apache.flink.util.StringUtils;
-
-/**
- * A factory to create a specific {@link org.apache.flink.connector.jdbc.dialect.JdbcDialect}. This
- * factory is used with Java's Service Provider Interfaces (SPI) for discovering.
- *
- * <p>Classes that implement this interface can be added to the
- * "META_INF/services/org.apache.flink.connector.jdbc.JdbcDialectFactory" file of a JAR file in the
- * current classpath to be found.
- *
- * @see org.apache.flink.connector.jdbc.dialect.JdbcDialect
- * @deprecated Use {@link JdbcFactory}
- */
-@Deprecated
-public interface JdbcDialectFactory extends JdbcFactory {
-
- /**
- * @deprecated
- * @return Creates a new instance of the {@link
- * org.apache.flink.connector.jdbc.dialect.JdbcDialect}.
- */
- JdbcDialect create();
-
- /**
- * @deprecated
- * @param compatibleMode the compatible mode of database
- * @return a new instance of {@link org.apache.flink.connector.jdbc.dialect.JdbcDialect}
- */
- default JdbcDialect create(String compatibleMode) {
- if (StringUtils.isNullOrWhitespaceOnly(compatibleMode)) {
- return create();
- }
- throw new UnsupportedOperationException(
- "Not supported option 'compatible-mode' with value: " + compatibleMode);
- }
-
- /**
- * @return a new instance of the {@link org.apache.flink.connector.jdbc.dialect.JdbcDialect}.
- */
- default JdbcDialect createDialect() {
- return create();
- }
-
- /**
- * Creates a new instance of the {@link org.apache.flink.connector.jdbc.dialect.JdbcDialect}
- * based on compatible mode.
- *
- * @param compatibleMode the compatible mode of database
- * @return a new instance of {@link org.apache.flink.connector.jdbc.dialect.JdbcDialect}
- */
- default JdbcDialect createDialect(String compatibleMode) {
- return create(compatibleMode);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
deleted file mode 100644
index b171147..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeMapper.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * 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.dialect;
-
-import org.apache.flink.connector.jdbc.core.database.catalog.JdbcCatalogTypeMapper;
-
-/**
- * Separate the jdbc meta-information type to flink table type into the interface.
- *
- * @deprecated Use {@link JdbcCatalogTypeMapper}
- */
-@Deprecated
-public interface JdbcDialectTypeMapper extends JdbcCatalogTypeMapper {}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/sink/JdbcSink.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/sink/JdbcSink.java
deleted file mode 100644
index 2b1ba89..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/sink/JdbcSink.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * 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.sink;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.connector.sink2.Committer;
-import org.apache.flink.api.connector.sink2.StatefulSink;
-import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
-import org.apache.flink.connector.base.DeliveryGuarantee;
-import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
-import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
-import org.apache.flink.connector.jdbc.core.datastream.sink.committer.JdbcCommitable;
-import org.apache.flink.connector.jdbc.core.datastream.sink.committer.JdbcCommitableSerializer;
-import org.apache.flink.connector.jdbc.core.datastream.sink.committer.JdbcCommitter;
-import org.apache.flink.connector.jdbc.core.datastream.sink.writer.JdbcWriter;
-import org.apache.flink.connector.jdbc.core.datastream.sink.writer.JdbcWriterState;
-import org.apache.flink.connector.jdbc.core.datastream.sink.writer.JdbcWriterStateSerializer;
-import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
-import org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement;
-import org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer;
-import org.apache.flink.core.io.SimpleVersionedSerializer;
-
-import java.io.IOException;
-import java.util.Collection;
-import java.util.Collections;
-
-/**
- * Flink Sink to produce data into a jdbc database.
- *
- * @see JdbcSinkBuilder on how to construct a JdbcSink
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSink}
- */
-@PublicEvolving
-@Deprecated
-public class JdbcSink<IN>
- implements StatefulSink<IN, JdbcWriterState>, TwoPhaseCommittingSink<IN, JdbcCommitable> {
-
- private final DeliveryGuarantee deliveryGuarantee;
- private final JdbcConnectionProvider connectionProvider;
- private final JdbcExecutionOptions executionOptions;
- private final JdbcExactlyOnceOptions exactlyOnceOptions;
- private final JdbcQueryStatement<IN> queryStatement;
-
- public JdbcSink(
- DeliveryGuarantee deliveryGuarantee,
- JdbcConnectionProvider connectionProvider,
- JdbcExecutionOptions executionOptions,
- JdbcExactlyOnceOptions exactlyOnceOptions,
- JdbcQueryStatement<IN> queryStatement) {
- this.deliveryGuarantee = deliveryGuarantee;
- this.connectionProvider = connectionProvider;
- this.executionOptions = executionOptions;
- this.exactlyOnceOptions = exactlyOnceOptions;
- this.queryStatement = queryStatement;
- }
-
- public static <IN> JdbcSinkBuilder<IN> builder() {
- return new JdbcSinkBuilder<>();
- }
-
- @Override
- @Internal
- public JdbcWriter<IN> createWriter(InitContext context) throws IOException {
- return restoreWriter(context, Collections.emptyList());
- }
-
- @Override
- @Internal
- public JdbcWriter<IN> restoreWriter(
- InitContext context, Collection<JdbcWriterState> recoveredState) throws IOException {
- JdbcOutputSerializer<IN> outputSerializer =
- JdbcOutputSerializer.of(
- context.createInputSerializer(), context.isObjectReuseEnabled());
- return new JdbcWriter<>(
- this.connectionProvider,
- this.executionOptions,
- this.exactlyOnceOptions,
- this.queryStatement,
- outputSerializer,
- deliveryGuarantee,
- recoveredState,
- context);
- }
-
- @Override
- @Internal
- public Committer<JdbcCommitable> createCommitter() throws IOException {
- return new JdbcCommitter(deliveryGuarantee, connectionProvider, exactlyOnceOptions);
- }
-
- @Override
- @Internal
- public SimpleVersionedSerializer<JdbcCommitable> getCommittableSerializer() {
- return new JdbcCommitableSerializer();
- }
-
- @Override
- @Internal
- public SimpleVersionedSerializer<JdbcWriterState> getWriterStateSerializer() {
- return new JdbcWriterStateSerializer();
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/sink/JdbcSinkBuilder.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/sink/JdbcSinkBuilder.java
deleted file mode 100644
index da08b99..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/sink/JdbcSinkBuilder.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.sink;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.base.DeliveryGuarantee;
-import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
-import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
-import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
-import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
-import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
-import org.apache.flink.connector.jdbc.datasource.connections.SimpleJdbcConnectionProvider;
-import org.apache.flink.connector.jdbc.datasource.connections.xa.PoolingXaConnectionProvider;
-import org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider;
-import org.apache.flink.connector.jdbc.datasource.connections.xa.XaConnectionProvider;
-import org.apache.flink.connector.jdbc.datasource.statements.JdbcQueryStatement;
-import org.apache.flink.connector.jdbc.datasource.statements.SimpleJdbcQueryStatement;
-import org.apache.flink.util.function.SerializableSupplier;
-
-import javax.sql.XADataSource;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * Builder to construct {@link JdbcSink}.
- *
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSinkBuilder}
- */
-@PublicEvolving
-@Deprecated
-public class JdbcSinkBuilder<IN> {
-
- private JdbcExecutionOptions executionOptions;
- private JdbcQueryStatement<IN> queryStatement;
-
- public JdbcSinkBuilder() {
- this.executionOptions = JdbcExecutionOptions.defaults();
- }
-
- public JdbcSinkBuilder<IN> withExecutionOptions(JdbcExecutionOptions executionOptions) {
- this.executionOptions = checkNotNull(executionOptions, "executionOptions cannot be null");
- return this;
- }
-
- public JdbcSinkBuilder<IN> withQueryStatement(JdbcQueryStatement<IN> queryStatement) {
- this.queryStatement = queryStatement;
- return this;
- }
-
- public JdbcSinkBuilder<IN> withQueryStatement(
- String query, JdbcStatementBuilder<IN> statement) {
- this.queryStatement = new SimpleJdbcQueryStatement<>(query, statement);
- return this;
- }
-
- public JdbcSink<IN> buildAtLeastOnce(JdbcConnectionOptions connectionOptions) {
- checkNotNull(connectionOptions, "connectionOptions cannot be null");
-
- return buildAtLeastOnce(new SimpleJdbcConnectionProvider(connectionOptions));
- }
-
- public JdbcSink<IN> buildAtLeastOnce(JdbcConnectionProvider connectionProvider) {
- checkNotNull(connectionProvider, "connectionProvider cannot be null");
-
- return build(
- DeliveryGuarantee.AT_LEAST_ONCE,
- JdbcExactlyOnceOptions.defaults(),
- checkNotNull(connectionProvider, "connectionProvider cannot be null"));
- }
-
- public JdbcSink<IN> buildExactlyOnce(
- JdbcExactlyOnceOptions exactlyOnceOptions,
- SerializableSupplier<XADataSource> dataSourceSupplier) {
-
- checkNotNull(exactlyOnceOptions, "exactlyOnceOptions cannot be null");
- checkNotNull(dataSourceSupplier, "dataSourceSupplier cannot be null");
- XaConnectionProvider connectionProvider =
- exactlyOnceOptions.isTransactionPerConnection()
- ? PoolingXaConnectionProvider.from(
- dataSourceSupplier, exactlyOnceOptions.getTimeoutSec())
- : SimpleXaConnectionProvider.from(
- dataSourceSupplier, exactlyOnceOptions.getTimeoutSec());
- return buildExactlyOnce(exactlyOnceOptions, connectionProvider);
- }
-
- public JdbcSink<IN> buildExactlyOnce(
- JdbcExactlyOnceOptions exactlyOnceOptions, XaConnectionProvider connectionProvider) {
-
- return build(
- DeliveryGuarantee.EXACTLY_ONCE,
- checkNotNull(exactlyOnceOptions, "exactlyOnceOptions cannot be null"),
- checkNotNull(connectionProvider, "connectionProvider cannot be null"));
- }
-
- private JdbcSink<IN> build(
- DeliveryGuarantee deliveryGuarantee,
- JdbcExactlyOnceOptions exactlyOnceOptions,
- JdbcConnectionProvider connectionProvider) {
-
- return new JdbcSink<>(
- checkNotNull(deliveryGuarantee, "deliveryGuarantee cannot be null"),
- checkNotNull(connectionProvider, "connectionProvider cannot be null"),
- checkNotNull(executionOptions, "executionOptions cannot be null"),
- checkNotNull(exactlyOnceOptions, "exactlyOnceOptions cannot be null"),
- checkNotNull(queryStatement, "queryStatement cannot be null"));
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSource.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSource.java
deleted file mode 100644
index cee58e7..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSource.java
+++ /dev/null
@@ -1,203 +0,0 @@
-/*
- * 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.source;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.connector.source.Boundedness;
-import org.apache.flink.api.connector.source.Source;
-import org.apache.flink.api.connector.source.SourceReader;
-import org.apache.flink.api.connector.source.SourceReaderContext;
-import org.apache.flink.api.connector.source.SplitEnumerator;
-import org.apache.flink.api.connector.source.SplitEnumeratorContext;
-import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.base.DeliveryGuarantee;
-import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSourceEnumStateSerializer;
-import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSourceEnumerator;
-import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSourceEnumeratorState;
-import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSqlSplitEnumeratorBase;
-import org.apache.flink.connector.jdbc.core.datastream.source.reader.JdbcSourceReader;
-import org.apache.flink.connector.jdbc.core.datastream.source.reader.JdbcSourceSplitReader;
-import org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
-import org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
-import org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplitSerializer;
-import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
-import org.apache.flink.connector.jdbc.utils.ContinuousUnBoundingSettings;
-import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.flink.util.Preconditions;
-
-import javax.annotation.Nullable;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Objects;
-
-/**
- * JDBC source.
- *
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.datastream.source.JdbcSource}
- */
-@PublicEvolving
-@Deprecated
-public class JdbcSource<OUT>
- implements Source<OUT, JdbcSourceSplit, JdbcSourceEnumeratorState>,
- ResultTypeQueryable<OUT> {
-
- private final Boundedness boundedness;
- private final TypeInformation<OUT> typeInformation;
- private final @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings;
-
- private final Configuration configuration;
- private final JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> sqlSplitEnumeratorProvider;
-
- protected JdbcConnectionProvider connectionProvider;
- private final ResultExtractor<OUT> resultExtractor;
- private final DeliveryGuarantee deliveryGuarantee;
-
- JdbcSource(
- Configuration configuration,
- JdbcConnectionProvider connectionProvider,
- JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> sqlSplitEnumeratorProvider,
- ResultExtractor<OUT> resultExtractor,
- TypeInformation<OUT> typeInformation,
- @Nullable DeliveryGuarantee deliveryGuarantee,
- @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings) {
- this.configuration = Preconditions.checkNotNull(configuration);
- this.connectionProvider = Preconditions.checkNotNull(connectionProvider);
- this.sqlSplitEnumeratorProvider = Preconditions.checkNotNull(sqlSplitEnumeratorProvider);
- this.resultExtractor = Preconditions.checkNotNull(resultExtractor);
- this.deliveryGuarantee =
- Objects.isNull(deliveryGuarantee) ? DeliveryGuarantee.NONE : deliveryGuarantee;
- this.typeInformation = Preconditions.checkNotNull(typeInformation);
- this.continuousUnBoundingSettings = continuousUnBoundingSettings;
- this.boundedness =
- Objects.isNull(continuousUnBoundingSettings)
- ? Boundedness.BOUNDED
- : Boundedness.CONTINUOUS_UNBOUNDED;
- }
-
- @Override
- public Boundedness getBoundedness() {
- return boundedness;
- }
-
- @Override
- public SourceReader<OUT, JdbcSourceSplit> createReader(SourceReaderContext readerContext)
- throws Exception {
- return new JdbcSourceReader<>(
- () ->
- new JdbcSourceSplitReader<>(
- readerContext,
- configuration,
- typeInformation,
- connectionProvider,
- deliveryGuarantee,
- resultExtractor),
- configuration,
- readerContext);
- }
-
- @Override
- public SplitEnumerator<JdbcSourceSplit, JdbcSourceEnumeratorState> createEnumerator(
- SplitEnumeratorContext<JdbcSourceSplit> enumContext) throws Exception {
- return new JdbcSourceEnumerator(
- enumContext,
- sqlSplitEnumeratorProvider.create(),
- continuousUnBoundingSettings,
- new ArrayList<>());
- }
-
- @Override
- public SplitEnumerator<JdbcSourceSplit, JdbcSourceEnumeratorState> restoreEnumerator(
- SplitEnumeratorContext<JdbcSourceSplit> enumContext,
- JdbcSourceEnumeratorState checkpoint)
- throws Exception {
- Serializable optionalUserDefinedSplitEnumeratorState =
- checkpoint.getOptionalUserDefinedSplitEnumeratorState();
- return new JdbcSourceEnumerator(
- enumContext,
- sqlSplitEnumeratorProvider.restore(optionalUserDefinedSplitEnumeratorState),
- continuousUnBoundingSettings,
- checkpoint.getRemainingSplits());
- }
-
- @Override
- public SimpleVersionedSerializer<JdbcSourceSplit> getSplitSerializer() {
- return new JdbcSourceSplitSerializer();
- }
-
- @Override
- public SimpleVersionedSerializer<JdbcSourceEnumeratorState>
- getEnumeratorCheckpointSerializer() {
- return new JdbcSourceEnumStateSerializer((JdbcSourceSplitSerializer) getSplitSerializer());
- }
-
- @Override
- public TypeInformation<OUT> getProducedType() {
- return typeInformation;
- }
-
- public static <OUT> JdbcSourceBuilder<OUT> builder() {
- return new JdbcSourceBuilder<>();
- }
-
- // ---- Visible for testing methods. ---
-
- @VisibleForTesting
- public JdbcSqlSplitEnumeratorBase.Provider<JdbcSourceSplit> getSqlSplitEnumeratorProvider() {
- return sqlSplitEnumeratorProvider;
- }
-
- @VisibleForTesting
- public TypeInformation<OUT> getTypeInformation() {
- return typeInformation;
- }
-
- @VisibleForTesting
- public Configuration getConfiguration() {
- return configuration;
- }
-
- @VisibleForTesting
- public ResultExtractor<OUT> getResultExtractor() {
- return resultExtractor;
- }
-
- @VisibleForTesting
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- JdbcSource<?> that = (JdbcSource<?>) o;
- return boundedness == that.boundedness
- && Objects.equals(typeInformation, that.typeInformation)
- && Objects.equals(configuration, that.configuration)
- && Objects.equals(sqlSplitEnumeratorProvider, that.sqlSplitEnumeratorProvider)
- && Objects.equals(connectionProvider, that.connectionProvider)
- && Objects.equals(resultExtractor, that.resultExtractor)
- && deliveryGuarantee == that.deliveryGuarantee
- && Objects.equals(continuousUnBoundingSettings, that.continuousUnBoundingSettings);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSourceBuilder.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSourceBuilder.java
deleted file mode 100644
index a193158..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSourceBuilder.java
+++ /dev/null
@@ -1,324 +0,0 @@
-/*
- * 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.source;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.base.DeliveryGuarantee;
-import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
-import org.apache.flink.connector.jdbc.core.datastream.source.JdbcSourceOptions;
-import org.apache.flink.connector.jdbc.core.datastream.source.enumerator.SqlTemplateSplitEnumerator;
-import org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
-import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
-import org.apache.flink.connector.jdbc.datasource.connections.SimpleJdbcConnectionProvider;
-import org.apache.flink.connector.jdbc.split.JdbcParameterValuesProvider;
-import org.apache.flink.connector.jdbc.split.JdbcSlideTimingParameterProvider;
-import org.apache.flink.connector.jdbc.utils.ContinuousUnBoundingSettings;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.StringUtils;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-
-import java.io.Serializable;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.util.Objects;
-
-/**
- * A tool is used to build {@link JdbcSource} quickly.
- *
- * <pre><code>
- * JdbcSource<Row> source = JdbcSource.<Row>builder()
- * .setSql(validSql)
- * .setResultExtractor(new RowResultExtractor())
- * .setDBUrl(dbUrl)
- * .setDriverName(driverName)
- * .setTypeInformation(new TypeHint<Row>() {}.getTypeInfo())
- * .setPassword(password)
- * .setUsername(username)
- * .build();
- * </code></pre>
- *
- * <p>In order to query the JDBC source in parallel, you need to provide a parameterized query
- * template (i.e. a valid {@link PreparedStatement}) and a {@link JdbcParameterValuesProvider} which
- * provides binding values for the query parameters. E.g.:
- *
- * <pre><code>
- *
- * Serializable[][] queryParameters = new String[2][1];
- * queryParameters[0] = new String[]{"Kumar"};
- * queryParameters[1] = new String[]{"Tan Ah Teck"};
- *
- * JdbcSource<Row> jdbcSource = JdbcSource.<Row>builder()
- * .setResultExtractor(new RowResultExtractor())
- * .setTypeInformation(new TypeHint<Row>() {}.getTypeInfo())
- * .setPassword(password)
- * .setUsername(username)
- * .setDriverName("org.apache.derby.jdbc.EmbeddedDriver")
- * .setDBUrl("jdbc:derby:memory:ebookshop")
- * .setSql("select * from books WHERE author = ?")
- * .setJdbcParameterValuesProvider(new JdbcGenericParameterValuesProvider(queryParameters))
- * .build();
- * </code></pre>
- *
- * @see Row
- * @see JdbcParameterValuesProvider
- * @see PreparedStatement
- * @see DriverManager
- * @see JdbcSource
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.datastream.source.JdbcSourceBuilder}
- */
-@PublicEvolving
-@Deprecated
-public class JdbcSourceBuilder<OUT> {
-
- public static final Logger LOG = LoggerFactory.getLogger(JdbcSourceBuilder.class);
-
- public static final String INVALID_CONTINUOUS_SLIDE_TIMING_HINT =
- "The 'jdbcParameterValuesProvider' must be specified with in type of 'JdbcSlideTimingParameterProvider' when using 'continuousUnBoundingSettings'.";
- public static final String INVALID_SLIDE_TIMING_CONTINUOUS_HINT =
- "The 'continuousUnBoundingSettings' must be specified with in type of 'continuousUnBoundingSettings' when using 'jdbcParameterValuesProvider' in type of 'JdbcSlideTimingParameterProvider'.";
-
- private final Configuration configuration;
-
- private int splitReaderFetchBatchSize;
- private int resultSetType;
- private int resultSetConcurrency;
- private int resultSetFetchSize;
- // Boolean to distinguish between default value and explicitly set autoCommit mode.
- private Boolean autoCommit;
-
- private DeliveryGuarantee deliveryGuarantee;
- private @Nullable ContinuousUnBoundingSettings continuousUnBoundingSettings;
-
- private TypeInformation<OUT> typeInformation;
-
- private final JdbcConnectionOptions.JdbcConnectionOptionsBuilder connOptionsBuilder;
- private String sql;
- private JdbcParameterValuesProvider jdbcParameterValuesProvider;
- private @Nullable Serializable optionalSqlSplitEnumeratorState;
- private ResultExtractor<OUT> resultExtractor;
-
- private JdbcConnectionProvider connectionProvider;
-
- JdbcSourceBuilder() {
- this.configuration = new Configuration();
- this.connOptionsBuilder = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder();
- this.splitReaderFetchBatchSize = JdbcSourceOptions.READER_FETCH_BATCH_SIZE.defaultValue();
- this.resultSetType = JdbcSourceOptions.RESULTSET_TYPE.defaultValue();
- this.resultSetConcurrency = JdbcSourceOptions.RESULTSET_CONCURRENCY.defaultValue();
- this.deliveryGuarantee = DeliveryGuarantee.NONE;
- // Boolean to distinguish between default value and explicitly set autoCommit mode.
- this.autoCommit = true;
- }
-
- public JdbcSourceBuilder<OUT> setSql(@Nonnull String sql) {
- Preconditions.checkArgument(
- !StringUtils.isNullOrWhitespaceOnly(sql),
- "It's required to set the 'sql' with non-empty value.");
- this.sql = sql;
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setResultExtractor(ResultExtractor<OUT> resultExtractor) {
- this.resultExtractor =
- Preconditions.checkNotNull(
- resultExtractor, "It's required to set the 'resultExtractor'.");
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setUsername(String username) {
- connOptionsBuilder.withUsername(username);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setPassword(String password) {
- connOptionsBuilder.withPassword(password);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setDriverName(String driverName) {
- Preconditions.checkArgument(
- !StringUtils.isNullOrWhitespaceOnly(driverName),
- "It's required to set the 'driverName'.");
- connOptionsBuilder.withDriverName(driverName);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setDBUrl(String dbURL) {
- Preconditions.checkArgument(
- !StringUtils.isNullOrWhitespaceOnly(dbURL), "It's required to set the 'dbURL'.");
- connOptionsBuilder.withUrl(dbURL);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setTypeInformation(
- @Nonnull TypeInformation<OUT> typeInformation) {
- this.typeInformation =
- Preconditions.checkNotNull(
- typeInformation, "It's required to set the 'typeInformation'.");
- return this;
- }
-
- // ------ Optional ------------------------------------------------------------------
-
- /**
- * The continuousUnBoundingSettings to discovery the next available batch splits. Note: If the
- * value was set, the {@link #jdbcParameterValuesProvider} must specified with the {@link
- * JdbcSlideTimingParameterProvider}.
- */
- public JdbcSourceBuilder<OUT> setContinuousUnBoundingSettings(
- ContinuousUnBoundingSettings continuousUnBoundingSettings) {
- this.continuousUnBoundingSettings = continuousUnBoundingSettings;
- return this;
- }
-
- /**
- * If the value was set as an instance of {@link JdbcSlideTimingParameterProvider}, it's
- * required to specify the {@link #continuousUnBoundingSettings}.
- */
- public JdbcSourceBuilder<OUT> setJdbcParameterValuesProvider(
- @Nonnull JdbcParameterValuesProvider parameterValuesProvider) {
- this.jdbcParameterValuesProvider = Preconditions.checkNotNull(parameterValuesProvider);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setDeliveryGuarantee(DeliveryGuarantee deliveryGuarantee) {
- this.deliveryGuarantee = Preconditions.checkNotNull(deliveryGuarantee);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setConnectionCheckTimeoutSeconds(
- int connectionCheckTimeoutSeconds) {
- connOptionsBuilder.withConnectionCheckTimeoutSeconds(connectionCheckTimeoutSeconds);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setConnectionProperty(String propKey, String propVal) {
- Preconditions.checkNotNull(propKey, "Connection property key mustn't be null");
- Preconditions.checkNotNull(propVal, "Connection property value mustn't be null");
- connOptionsBuilder.withProperty(propKey, propVal);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setSplitReaderFetchBatchSize(int splitReaderFetchBatchSize) {
- Preconditions.checkArgument(
- splitReaderFetchBatchSize > 0,
- "'splitReaderFetchBatchSize' must be in range (0, %s]",
- Integer.MAX_VALUE);
- this.splitReaderFetchBatchSize = splitReaderFetchBatchSize;
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setResultSetType(int resultSetType) {
- this.resultSetType = resultSetType;
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setResultSetConcurrency(int resultSetConcurrency) {
- this.resultSetConcurrency = resultSetConcurrency;
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setAutoCommit(boolean autoCommit) {
- this.autoCommit = autoCommit;
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setResultSetFetchSize(int resultSetFetchSize) {
- Preconditions.checkArgument(
- resultSetFetchSize == Integer.MIN_VALUE || resultSetFetchSize > 0,
- "Illegal value %s for fetchSize, has to be positive or Integer.MIN_VALUE.",
- resultSetFetchSize);
- this.resultSetFetchSize = resultSetFetchSize;
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setConnectionProvider(
- @Nonnull JdbcConnectionProvider connectionProvider) {
- this.connectionProvider = Preconditions.checkNotNull(connectionProvider);
- return this;
- }
-
- public JdbcSourceBuilder<OUT> setOptionalSqlSplitEnumeratorState(
- Serializable optionalSqlSplitEnumeratorState) {
- this.optionalSqlSplitEnumeratorState = optionalSqlSplitEnumeratorState;
- return this;
- }
-
- public JdbcSource<OUT> build() {
- this.connectionProvider = new SimpleJdbcConnectionProvider(connOptionsBuilder.build());
- if (resultSetFetchSize > 0) {
- this.configuration.set(JdbcSourceOptions.RESULTSET_FETCH_SIZE, resultSetFetchSize);
- }
-
- if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
- Preconditions.checkArgument(
- this.resultSetType == ResultSet.TYPE_SCROLL_INSENSITIVE
- || this.resultSetType == ResultSet.CONCUR_READ_ONLY,
- "The 'resultSetType' must be ResultSet.TYPE_SCROLL_INSENSITIVE or ResultSet.CONCUR_READ_ONLY when using %s",
- DeliveryGuarantee.EXACTLY_ONCE);
- }
-
- this.configuration.set(JdbcSourceOptions.RESULTSET_CONCURRENCY, resultSetConcurrency);
- this.configuration.set(JdbcSourceOptions.RESULTSET_TYPE, resultSetType);
- this.configuration.set(
- JdbcSourceOptions.READER_FETCH_BATCH_SIZE, splitReaderFetchBatchSize);
- this.configuration.set(JdbcSourceOptions.AUTO_COMMIT, autoCommit);
-
- Preconditions.checkState(
- !StringUtils.isNullOrWhitespaceOnly(sql), "'sql' mustn't be null or empty.");
- Preconditions.checkNotNull(resultExtractor, "'resultExtractor' mustn't be null.");
- Preconditions.checkNotNull(typeInformation, "'typeInformation' mustn't be null.");
-
- if (Objects.nonNull(continuousUnBoundingSettings)) {
- Preconditions.checkArgument(
- Objects.nonNull(jdbcParameterValuesProvider)
- && jdbcParameterValuesProvider
- instanceof JdbcSlideTimingParameterProvider,
- INVALID_SLIDE_TIMING_CONTINUOUS_HINT);
- }
-
- if (Objects.nonNull(jdbcParameterValuesProvider)
- && jdbcParameterValuesProvider instanceof JdbcSlideTimingParameterProvider) {
- Preconditions.checkArgument(
- Objects.nonNull(continuousUnBoundingSettings),
- INVALID_CONTINUOUS_SLIDE_TIMING_HINT);
- }
-
- return new JdbcSource<>(
- configuration,
- connectionProvider,
- new SqlTemplateSplitEnumerator.TemplateSqlSplitEnumeratorProvider()
- .setOptionalSqlSplitEnumeratorState(optionalSqlSplitEnumeratorState)
- .setSqlTemplate(sql)
- .setParameterValuesProvider(jdbcParameterValuesProvider),
- resultExtractor,
- typeInformation,
- deliveryGuarantee,
- continuousUnBoundingSettings);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/reader/extractor/ResultExtractor.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/reader/extractor/ResultExtractor.java
deleted file mode 100644
index 0a3e0ce..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/reader/extractor/ResultExtractor.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.source.reader.extractor;
-
-import org.apache.flink.annotation.PublicEvolving;
-
-import java.sql.ResultSet;
-
-/**
- * The Extractor to extract the data from {@link ResultSet}.
- *
- * @param <T> The target data type.
- * @deprecated Use {@link
- * org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor}
- */
-@PublicEvolving
-@Deprecated
-public interface ResultExtractor<T>
- extends org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor
- .ResultExtractor<
- T> {}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcConnectorOptions.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcConnectorOptions.java
deleted file mode 100644
index bee0d54..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcConnectorOptions.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.table;
-
-import org.apache.flink.annotation.PublicEvolving;
-
-/**
- * Options for the JDBC connector.
- *
- * @deprecated Use {@link org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions}
- */
-@PublicEvolving
-@Deprecated
-public class JdbcConnectorOptions
- extends org.apache.flink.connector.jdbc.core.table.JdbcConnectorOptions {}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/utils/ContinuousUnBoundingSettings.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/utils/ContinuousUnBoundingSettings.java
deleted file mode 100644
index f25f3d1..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/utils/ContinuousUnBoundingSettings.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.utils;
-
-import org.apache.flink.annotation.PublicEvolving;
-
-import java.time.Duration;
-
-/**
- * Settings describing how to do continuous file discovery and enumeration for the file source's
- * continuous discovery and streaming mode.
- *
- * @deprecated Use {@link
- * org.apache.flink.connector.jdbc.core.datastream.source.config.ContinuousUnBoundingSettings}
- */
-@PublicEvolving
-@Deprecated
-public final class ContinuousUnBoundingSettings
- extends org.apache.flink.connector.jdbc.core.datastream.source.config
- .ContinuousUnBoundingSettings {
-
- public ContinuousUnBoundingSettings(
- Duration initialDiscoveryDelay, Duration discoveryInterval) {
- super(initialDiscoveryDelay, discoveryInterval);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXid.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXid.java
deleted file mode 100644
index 6ce5973..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXid.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.util.Preconditions;
-
-import javax.annotation.concurrent.ThreadSafe;
-import javax.transaction.xa.Xid;
-
-import java.util.Objects;
-
-/**
- * A pair of checkpoint id and {@link Xid} representing a checkpoint and an associated pending
- * (prepared) XA transaction. Thread-safe (assuming immutable {@link Xid} implementation).
- */
-@ThreadSafe
-@Internal
-@Deprecated
-public final class CheckpointAndXid {
- final long checkpointId;
- final Xid xid;
- final int attempts;
- final boolean restored;
-
- public Xid getXid() {
- return xid;
- }
-
- private CheckpointAndXid(long checkpointId, Xid xid, int attempts, boolean restored) {
- this.checkpointId = checkpointId;
- this.xid = Preconditions.checkNotNull(xid);
- this.attempts = attempts;
- this.restored = restored;
- }
-
- @Override
- public String toString() {
- return String.format("checkpointId=%d, xid=%s, restored=%s", checkpointId, xid, restored);
- }
-
- CheckpointAndXid asRestored() {
- return restored ? this : new CheckpointAndXid(checkpointId, xid, attempts, true);
- }
-
- static CheckpointAndXid createRestored(long checkpointId, int attempts, Xid xid) {
- return new CheckpointAndXid(checkpointId, xid, attempts, true);
- }
-
- static CheckpointAndXid createNew(long checkpointId, Xid xid) {
- return new CheckpointAndXid(checkpointId, xid, 0, false);
- }
-
- CheckpointAndXid withAttemptsIncremented() {
- return new CheckpointAndXid(checkpointId, xid, attempts + 1, restored);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof CheckpointAndXid)) {
- return false;
- }
- CheckpointAndXid that = (CheckpointAndXid) o;
- return checkpointId == that.checkpointId
- && attempts == that.attempts
- && restored == that.restored
- && Objects.equals(xid, that.xid);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(checkpointId, xid, attempts, restored);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXidSerializer.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXidSerializer.java
deleted file mode 100644
index a712575..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXidSerializer.java
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
-import org.apache.flink.connector.jdbc.datasource.transactions.xa.xid.XidSerializer;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-
-import javax.transaction.xa.Xid;
-
-import java.io.IOException;
-import java.util.Objects;
-
-/** {@link CheckpointAndXid} serializer. */
-@Internal
-@Deprecated
-public final class CheckpointAndXidSerializer extends TypeSerializer<CheckpointAndXid> {
-
- private static final long serialVersionUID = 1L;
-
- public static final TypeSerializerSnapshot<CheckpointAndXid> SNAPSHOT =
- new CheckpointAndXidSimpleTypeSerializerSnapshot();
-
- private final TypeSerializer<Xid> xidSerializer = new XidSerializer();
-
- @Override
- public boolean isImmutableType() {
- return xidSerializer.isImmutableType();
- }
-
- @Override
- public TypeSerializer<CheckpointAndXid> duplicate() {
- return this;
- }
-
- @Override
- public CheckpointAndXid createInstance() {
- return CheckpointAndXid.createRestored(0L, 0, xidSerializer.createInstance());
- }
-
- @Override
- public CheckpointAndXid copy(CheckpointAndXid from) {
- return CheckpointAndXid.createRestored(
- from.checkpointId, from.attempts, xidSerializer.copy(from.xid));
- }
-
- @Override
- public CheckpointAndXid copy(CheckpointAndXid from, CheckpointAndXid reuse) {
- return from;
- }
-
- @Override
- public int getLength() {
- return -1;
- }
-
- @Override
- public void serialize(CheckpointAndXid record, DataOutputView target) throws IOException {
- target.writeLong(record.checkpointId);
- target.writeInt(record.attempts);
- xidSerializer.serialize(record.xid, target);
- }
-
- @Override
- public CheckpointAndXid deserialize(DataInputView source) throws IOException {
- return CheckpointAndXid.createRestored(
- source.readLong(), source.readInt(), xidSerializer.deserialize(source));
- }
-
- @Override
- public CheckpointAndXid deserialize(CheckpointAndXid reuse, DataInputView source)
- throws IOException {
- return deserialize(source);
- }
-
- @Override
- public void copy(DataInputView source, DataOutputView target) throws IOException {
- serialize(deserialize(source), target);
- }
-
- @Override
- public boolean equals(Object o) {
- return o instanceof CheckpointAndXidSerializer;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(xidSerializer);
- }
-
- @Override
- public TypeSerializerSnapshot<CheckpointAndXid> snapshotConfiguration() {
- return SNAPSHOT;
- }
-
- /** SImple {@link TypeSerializerSnapshot} for {@link CheckpointAndXidSerializer}. */
- public static class CheckpointAndXidSimpleTypeSerializerSnapshot
- extends SimpleTypeSerializerSnapshot<CheckpointAndXid> {
- private static final int VERSION = 1;
-
- public CheckpointAndXidSimpleTypeSerializerSnapshot() {
- super(CheckpointAndXidSerializer::new);
- }
-
- @Override
- public void writeSnapshot(DataOutputView out) throws IOException {
- super.writeSnapshot(out);
- out.writeInt(VERSION);
- }
-
- @Override
- public void readSnapshot(int readVersion, DataInputView in, ClassLoader classLoader)
- throws IOException {
- super.readSnapshot(readVersion, in, classLoader);
- in.readInt();
- }
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkFunction.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkFunction.java
deleted file mode 100644
index b88eddb..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkFunction.java
+++ /dev/null
@@ -1,373 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.state.CheckpointListener;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.InputTypeConfigurable;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
-import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
-import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
-import org.apache.flink.connector.jdbc.internal.JdbcOutputFormat;
-import org.apache.flink.connector.jdbc.internal.JdbcOutputSerializer;
-import org.apache.flink.connector.jdbc.internal.executor.JdbcBatchStatementExecutor;
-import org.apache.flink.connector.jdbc.xa.XaFacade.EmptyXaTransactionException;
-import org.apache.flink.runtime.state.FunctionInitializationContext;
-import org.apache.flink.runtime.state.FunctionSnapshotContext;
-import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.Preconditions;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.transaction.xa.Xid;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Deque;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Optional;
-
-import static org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunctionState.of;
-
-/**
- * JDBC sink function that uses XA transactions to provide exactly once guarantees. That is, if a
- * checkpoint succeeds then all records emitted during it are committed in the database, and rolled
- * back otherwise.
- *
- * <p>Each parallel subtask has it's own transactions, independent from other subtasks. Therefore,
- * consistency is only guaranteed within partitions.
- *
- * <p>XA uses a two-phase commit protocol, which solves the consistency problem, but leaves the
- * following issues:
- *
- * <ol>
- * <li>transactions may be abandoned, holding resources (e.g. locks, versions of rows)
- * <li>abandoned transactions collide with the new transactions if their IDs repeat after recovery
- * <li>commit requests may be repeated after job recovery, resulting in error responses and job
- * failure
- * </ol>
- *
- * <p>The following table summarizes effects of failures during transaction state transitions and
- * ways to mitigate them:
- *
- * <table border="1" style="width:100%;">
- * <col span="1" style="width:15%;">
- * <col span="1" style="width:15%;">
- * <col span="1" style="width:30%;">
- * <col span="1" style="width:40%;">
- * <thead>
- * <tr>
- * <th>Transition</th>
- * <th>Methods</th>
- * <th>What happens if transition lost</th>
- * <th>Ways to mitigate</th>
- * </tr>
- * </thead>
- * <tbody>
- * <tr>
- * <td>none > started, started > ended</td>
- * <td>open(), snapshotState()</td>
- * <td>Database eventually discards these transactions</td>
- * <td><ol>
- * <li>Use globally unique XIDs</li>
- * <li>derive XID from: checkpoint id, subtask id, "job id", "run id" (see {@link SemanticXidGenerator}).</li>
- * </ol></td>
- * </tr>
- * <tr>
- * <td>ended > prepared</td>
- * <td>snapshotState()</td>
- * <td>Database keeps these transactions prepared forever ("in-doubt" state)</td>
- * <td>
- * <ol>
- * <li>store ended transactions in state; rollback on job recovery (still doesn't cover all scenarios)</li>
- * <li>call xa_recover() and xa_rollback() on job recovery; disabled by default in order not to affect transactions of other subtasks and apps</li>
- * <li>setting transaction timeouts (not supported by most databases)</li>
- * <li>manual recovery and rollback</li>
- * </ol>
- * </td>
- * </tr>
- * <tr>
- * <td>prepared > committed</td>
- * <td>open(), notifyCheckpointComplete()</td>
- * <td>
- * Upon job recovery state contains committed transactions; or JM may notifyCheckpointComplete again after recovery.
- * <p>Committing results in {@link javax.transaction.xa.XAException#XAER_NOTA XAER_NOTA} error.</p>
- * </td>
- * <td>
- * Distinguish between transactions created during this run and restored from state and ignore {@link javax.transaction.xa.XAException#XAER_NOTA XAER_NOTA} for the latter.
- * </td>
- * </tr>
- * </tbody>
- * </table>
- *
- * @since 1.13
- */
-@Internal
-@Deprecated
-public class JdbcXaSinkFunction<T> extends AbstractRichFunction
- implements CheckpointedFunction,
- CheckpointListener,
- SinkFunction<T>,
- AutoCloseable,
- InputTypeConfigurable {
-
- private static final Logger LOG = LoggerFactory.getLogger(JdbcXaSinkFunction.class);
-
- private final XaFacade xaFacade;
- private final XaGroupOps xaGroupOps;
- private final XidGenerator xidGenerator;
- private final JdbcOutputFormat<T, T, JdbcBatchStatementExecutor<T>> outputFormat;
- private final XaSinkStateHandler stateHandler;
- private final JdbcExactlyOnceOptions options;
- private JdbcOutputSerializer<T> serializer;
-
- // checkpoints and the corresponding transactions waiting for completion notification from JM
- private transient List<CheckpointAndXid> preparedXids = new ArrayList<>();
- // hanging XIDs - used for cleanup
- // it's a list to support retries and scaling down
- // possible transaction states: active, idle, prepared
- // last element is the current xid
- private transient Deque<Xid> hangingXids = new LinkedList<>();
- private transient Xid currentXid;
-
- /**
- * Creates a {@link JdbcXaSinkFunction}.
- *
- * <p>All parameters must be {@link java.io.Serializable serializable}.
- *
- * <p>Note: {@link JdbcExecutionOptions} maxRetries setting must be strictly set to 0 for this
- * sink to work properly and not to produce duplicates. See issue FLINK-22311 for details.
- *
- * @param xaFacade {@link XaFacade} to manage XA transactions
- */
- public JdbcXaSinkFunction(
- String sql,
- JdbcStatementBuilder<T> statementBuilder,
- XaFacade xaFacade,
- JdbcExecutionOptions executionOptions,
- JdbcExactlyOnceOptions options) {
- this(
- new JdbcOutputFormat<>(
- xaFacade,
- executionOptions,
- () -> JdbcBatchStatementExecutor.simple(sql, statementBuilder)),
- xaFacade,
- XidGenerator.semanticXidGenerator(),
- new XaSinkStateHandlerImpl(),
- options,
- new XaGroupOpsImpl(xaFacade));
- }
-
- /**
- * Creates a {@link JdbcXaSinkFunction}.
- *
- * <p>All parameters must be {@link java.io.Serializable serializable}.
- *
- * @param outputFormat {@link JdbcOutputFormat} to write records with
- * @param xaFacade {@link XaFacade} to manage XA transactions
- * @param xidGenerator {@link XidGenerator} to generate new transaction ids
- */
- public JdbcXaSinkFunction(
- JdbcOutputFormat<T, T, JdbcBatchStatementExecutor<T>> outputFormat,
- XaFacade xaFacade,
- XidGenerator xidGenerator,
- XaSinkStateHandler stateHandler,
- JdbcExactlyOnceOptions options,
- XaGroupOps xaGroupOps) {
-
- Preconditions.checkArgument(
- outputFormat.getExecutionOptions().getMaxRetries() == 0,
- "JDBC XA sink requires maxRetries equal to 0, otherwise it could "
- + "cause duplicates. See issue FLINK-22311 for details.");
-
- this.xaFacade = Preconditions.checkNotNull(xaFacade);
- this.xidGenerator = Preconditions.checkNotNull(xidGenerator);
- this.outputFormat = Preconditions.checkNotNull(outputFormat);
- this.stateHandler = Preconditions.checkNotNull(stateHandler);
- this.options = Preconditions.checkNotNull(options);
- this.xaGroupOps = xaGroupOps;
- }
-
- @Override
- public void initializeState(FunctionInitializationContext context) throws Exception {
- JdbcXaSinkFunctionState state = stateHandler.load(context);
- hangingXids = new LinkedList<>(state.getHanging());
- preparedXids = new ArrayList<>(state.getPrepared());
- LOG.info(
- "initialized state: prepared xids: {}, hanging xids: {}",
- preparedXids.size(),
- hangingXids.size());
- }
-
- @Override
- public void open(Configuration configuration) throws Exception {
- super.open(configuration);
- // Recheck if execution config change
- this.serializer.withObjectReuseEnabled(
- getRuntimeContext().getExecutionConfig().isObjectReuseEnabled());
-
- xidGenerator.open();
- xaFacade.open();
- hangingXids = new LinkedList<>(xaGroupOps.failOrRollback(hangingXids).getForRetry());
- commitUpToCheckpoint(Optional.empty());
- if (options.isDiscoverAndRollbackOnRecovery()) {
- // Pending transactions which are not included into the checkpoint might hold locks and
- // should be rolled back. However, rolling back ALL transactions can cause data loss. So
- // each subtask first commits transactions from its state and then rolls back discovered
- // transactions if they belong to it.
- xaGroupOps.recoverAndRollback(JobSubtask.of(getRuntimeContext()), xidGenerator);
- }
- beginTx(0L);
- // open format only after starting the transaction so it gets a ready to use connection
- outputFormat.open(this.serializer);
- }
-
- @Override
- public void snapshotState(FunctionSnapshotContext context) throws Exception {
- LOG.debug("snapshot state, checkpointId={}", context.getCheckpointId());
- prepareCurrentTx(context.getCheckpointId());
- beginTx(context.getCheckpointId() + 1);
- stateHandler.store(of(preparedXids, hangingXids));
- }
-
- @Override
- public void notifyCheckpointComplete(long checkpointId) {
- commitUpToCheckpoint(Optional.of(checkpointId));
- }
-
- @Override
- public void invoke(T value, Context context) throws IOException {
- Preconditions.checkState(currentXid != null, "current xid must not be null");
- if (LOG.isTraceEnabled()) {
- LOG.trace("invoke, xid: {}, value: {}", currentXid, value);
- }
- outputFormat.writeRecord(value);
- }
-
- @Override
- public void close() throws Exception {
- super.close();
- if (currentXid != null && xaFacade.isOpen()) {
- try {
- LOG.debug("remove current transaction before closing, xid={}", currentXid);
- xaFacade.failAndRollback(currentXid);
- } catch (Exception e) {
- LOG.warn("unable to fail/rollback current transaction, xid={}", currentXid, e);
- }
- }
- xaFacade.close();
- xidGenerator.close();
- // don't format.close(); as we don't want neither to flush nor to close connection here
- currentXid = null;
- hangingXids = null;
- preparedXids = null;
- }
-
- private void prepareCurrentTx(long checkpointId) throws IOException {
- Preconditions.checkState(currentXid != null, "no current xid");
- Preconditions.checkState(
- !hangingXids.isEmpty() && hangingXids.peekLast().equals(currentXid),
- "inconsistent internal state");
- hangingXids.pollLast();
- outputFormat.flush();
- try {
- xaFacade.endAndPrepare(currentXid);
- preparedXids.add(CheckpointAndXid.createNew(checkpointId, currentXid));
- } catch (EmptyXaTransactionException e) {
- LOG.info(
- "empty XA transaction (skip), xid: {}, checkpoint {}",
- currentXid,
- checkpointId);
- } catch (Exception e) {
- ExceptionUtils.rethrowIOException(e);
- }
- currentXid = null;
- }
-
- /** @param checkpointId to associate with the new transaction. */
- private void beginTx(long checkpointId) throws Exception {
- Preconditions.checkState(currentXid == null, "currentXid not null");
- currentXid = xidGenerator.generateXid(JobSubtask.of(getRuntimeContext()), checkpointId);
- hangingXids.offerLast(currentXid);
- xaFacade.start(currentXid);
- if (checkpointId > 0) {
- // associate outputFormat with a new connection that might have been opened in start()
- outputFormat.updateExecutor(false);
- }
- }
-
- private void commitUpToCheckpoint(Optional<Long> checkpointInclusive) {
- Tuple2<List<CheckpointAndXid>, List<CheckpointAndXid>> splittedXids =
- split(preparedXids, checkpointInclusive, true);
- if (splittedXids.f0.isEmpty()) {
- checkpointInclusive.ifPresent(
- cp -> LOG.warn("nothing to commit up to checkpoint: {}", cp));
- } else {
- preparedXids = splittedXids.f1;
- preparedXids.addAll(
- xaGroupOps
- .commit(
- splittedXids.f0,
- options.isAllowOutOfOrderCommits(),
- options.getMaxCommitAttempts())
- .getForRetry());
- }
- }
-
- private Tuple2<List<CheckpointAndXid>, List<CheckpointAndXid>> split(
- List<CheckpointAndXid> list,
- Optional<Long> checkpointInclusive,
- boolean checkpointIntoLo) {
- return checkpointInclusive
- .map(cp -> split(preparedXids, cp, checkpointIntoLo))
- .orElse(new Tuple2<>(list, new ArrayList<>()));
- }
-
- private Tuple2<List<CheckpointAndXid>, List<CheckpointAndXid>> split(
- List<CheckpointAndXid> list, long checkpoint, boolean checkpointIntoLo) {
- List<CheckpointAndXid> lo = new ArrayList<>(list.size() / 2);
- List<CheckpointAndXid> hi = new ArrayList<>(list.size() / 2);
- list.forEach(
- i -> {
- if (i.checkpointId < checkpoint
- || (i.checkpointId == checkpoint && checkpointIntoLo)) {
- lo.add(i);
- } else {
- hi.add(i);
- }
- });
- return new Tuple2<>(lo, hi);
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public void setInputType(TypeInformation<?> type, ExecutionConfig executionConfig) {
- this.serializer =
- JdbcOutputSerializer.of(
- ((TypeInformation<T>) type).createSerializer(executionConfig));
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkFunctionState.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkFunctionState.java
deleted file mode 100644
index e859647..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkFunctionState.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-
-import javax.annotation.concurrent.ThreadSafe;
-import javax.transaction.xa.Xid;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-
-import static java.util.Collections.unmodifiableCollection;
-
-/** Thread-safe (assuming immutable {@link Xid} implementation). */
-@ThreadSafe
-@Internal
-@Deprecated
-class JdbcXaSinkFunctionState {
- private final Collection<CheckpointAndXid> prepared;
- private final Collection<Xid> hanging;
-
- static JdbcXaSinkFunctionState empty() {
- return new JdbcXaSinkFunctionState(Collections.emptyList(), Collections.emptyList());
- }
-
- static JdbcXaSinkFunctionState of(
- Collection<CheckpointAndXid> prepared, Collection<Xid> hanging) {
- return new JdbcXaSinkFunctionState(
- unmodifiableCollection(new ArrayList<>(prepared)),
- unmodifiableCollection(new ArrayList<>(hanging)));
- }
-
- private JdbcXaSinkFunctionState(
- Collection<CheckpointAndXid> prepared, Collection<Xid> hanging) {
- this.prepared = prepared;
- this.hanging = hanging;
- }
-
- /**
- * @return immutable collection of prepared XA transactions to {@link
- * javax.transaction.xa.XAResource#commit commit}.
- */
- public Collection<CheckpointAndXid> getPrepared() {
- return prepared;
- }
-
- /**
- * @return immutable collection of XA transactions to {@link
- * javax.transaction.xa.XAResource#rollback rollback} (if they were prepared) or {@link
- * javax.transaction.xa.XAResource#end end} (if they were only started).
- */
- Collection<Xid> getHanging() {
- return hanging;
- }
-
- @Override
- public String toString() {
- return "prepared=" + prepared + ", hanging=" + hanging;
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JobSubtask.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JobSubtask.java
deleted file mode 100644
index ad335b6..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/JobSubtask.java
+++ /dev/null
@@ -1,53 +0,0 @@
-package org.apache.flink.connector.jdbc.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.connector.sink2.Sink;
-
-import java.io.Serializable;
-
-/** Job identifier. */
-@Internal
-@Deprecated
-class JobSubtask implements Serializable {
-
- private final byte[] jobId;
- private final int subtaskId;
- private final int numberOfSubtasks;
-
- public JobSubtask(byte[] jobId, int subtaskId, int numberOfSubtasks) {
- this.jobId = jobId;
- this.subtaskId = subtaskId;
- this.numberOfSubtasks = numberOfSubtasks;
- }
-
- public byte[] getJobId() {
- return jobId;
- }
-
- public int getSubtaskId() {
- return subtaskId;
- }
-
- public int getNumberOfSubtasks() {
- return numberOfSubtasks;
- }
-
- public static JobSubtask of(RuntimeContext context) {
- return new JobSubtask(
- context.getJobId().getBytes(),
- context.getIndexOfThisSubtask(),
- context.getNumberOfParallelSubtasks());
- }
-
- public static JobSubtask of(Sink.InitContext context) {
- return new JobSubtask(
- context.getJobId().getBytes(),
- context.getSubtaskId(),
- context.getNumberOfParallelSubtasks());
- }
-
- public static JobSubtask empty() {
- return new JobSubtask("".getBytes(), 0, 0);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/SemanticXidGenerator.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/SemanticXidGenerator.java
deleted file mode 100644
index 5acd7af..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/SemanticXidGenerator.java
+++ /dev/null
@@ -1,110 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.connector.jdbc.datasource.transactions.xa.xid.XidImpl;
-
-import javax.transaction.xa.Xid;
-
-import java.security.SecureRandom;
-import java.util.Arrays;
-
-/**
- * Generates {@link Xid} from:
- *
- * <ol>
- * <li>To provide uniqueness over other jobs and apps, and other instances
- * <li>of this job, gtrid consists of
- * <li>job id (16 bytes)
- * <li>subtask index (4 bytes)
- * <li>checkpoint id (4 bytes)
- * <li>bqual consists of 4 random bytes (generated using {@link SecureRandom})
- * </ol>
- *
- * <p>Each {@link SemanticXidGenerator} instance MUST be used for only one Sink (otherwise Xids will
- * collide).
- */
-@Internal
-@Deprecated
-class SemanticXidGenerator implements XidGenerator {
-
- private static final long serialVersionUID = 1L;
-
- private static final SecureRandom SECURE_RANDOM = new SecureRandom();
-
- private static final int FORMAT_ID = 201;
-
- private transient byte[] gtridBuffer;
- private transient byte[] bqualBuffer;
-
- @Override
- public void open() {
- // globalTransactionId = job id + task index + checkpoint id
- gtridBuffer = new byte[JobID.SIZE + Integer.BYTES + Long.BYTES];
- // branchQualifier = random bytes
- bqualBuffer = getRandomBytes(Integer.BYTES);
- }
-
- @Override
- public Xid generateXid(JobSubtask subtaskId, long checkpointId) {
- System.arraycopy(subtaskId.getJobId(), 0, gtridBuffer, 0, JobID.SIZE);
-
- writeNumber(subtaskId.getSubtaskId(), Integer.BYTES, gtridBuffer, JobID.SIZE);
- writeNumber(checkpointId, Long.BYTES, gtridBuffer, JobID.SIZE + Integer.BYTES);
- // relying on arrays copying inside XaXid constructor
- return new XidImpl(FORMAT_ID, gtridBuffer, bqualBuffer);
- }
-
- @Override
- public boolean belongsToSubtask(Xid xid, JobSubtask subtask) {
- if (xid.getFormatId() != FORMAT_ID) {
- return false;
- }
- int subtaskIndex = readNumber(xid.getGlobalTransactionId(), JobID.SIZE, Integer.BYTES);
- if (subtaskIndex != subtask.getSubtaskId()
- && subtaskIndex <= subtask.getNumberOfSubtasks() - 1) {
- return false;
- }
- byte[] jobIdBytes = new byte[JobID.SIZE];
- System.arraycopy(xid.getGlobalTransactionId(), 0, jobIdBytes, 0, JobID.SIZE);
- return Arrays.equals(jobIdBytes, subtask.getJobId());
- }
-
- private static int readNumber(byte[] bytes, int offset, int numBytes) {
- int result = 0;
- for (int i = 0; i < numBytes; i++) {
- result |= (bytes[offset + i] & 0xff) << Byte.SIZE * i;
- }
- return result;
- }
-
- private static void writeNumber(long number, int numBytes, byte[] dst, int dstOffset) {
- for (int i = dstOffset; i < dstOffset + numBytes; i++) {
- dst[i] = (byte) number;
- number >>>= Byte.SIZE;
- }
- }
-
- private byte[] getRandomBytes(int size) {
- byte[] bytes = new byte[size];
- SECURE_RANDOM.nextBytes(bytes);
- return bytes;
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacade.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacade.java
deleted file mode 100644
index b761660..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacade.java
+++ /dev/null
@@ -1,123 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
-import org.apache.flink.util.FlinkRuntimeException;
-
-import javax.sql.XADataSource;
-import javax.transaction.xa.XAException;
-import javax.transaction.xa.Xid;
-
-import java.io.Serializable;
-import java.util.Collection;
-import java.util.function.Supplier;
-
-/**
- * Facade to the XA operations relevant to {@link
- * org.apache.flink.streaming.api.functions.sink.SinkFunction sink}.
- *
- * <p>Typical workflow:
- *
- * <ol>
- * <li>{@link #open}
- * <li>{@link #start} transaction
- * <li>{@link #getConnection}, write some data
- * <li>{@link #endAndPrepare} (or {@link #failAndRollback})
- * <li>{@link #commit} / {@link #rollback}
- * <li>{@link #close}
- * </ol>
- *
- * {@link #recover} can be used to get abandoned prepared transactions for cleanup.
- */
-@Internal
-@Deprecated
-public interface XaFacade extends JdbcConnectionProvider, Serializable, AutoCloseable {
-
- /** @return a non-serializable instance. */
- static XaFacade fromXaDataSourceSupplier(
- Supplier<XADataSource> dataSourceSupplier,
- Integer timeoutSec,
- boolean transactionPerConnection) {
- return transactionPerConnection
- ? new XaFacadePoolingImpl(() -> new XaFacadeImpl(dataSourceSupplier, timeoutSec))
- : new XaFacadeImpl(dataSourceSupplier, timeoutSec);
- }
-
- void open() throws Exception;
-
- boolean isOpen();
-
- /** Start a new transaction. */
- void start(Xid xid) throws Exception;
-
- /** End and then prepare the transaction. Transaction can't be resumed afterwards. */
- void endAndPrepare(Xid xid) throws Exception;
-
- /**
- * Commit previously prepared transaction.
- *
- * @param ignoreUnknown whether to ignore {@link javax.transaction.xa.XAException#XAER_NOTA
- * XAER_NOTA} error.
- */
- void commit(Xid xid, boolean ignoreUnknown) throws TransientXaException;
-
- /** Rollback previously prepared transaction. */
- void rollback(Xid xid) throws TransientXaException;
-
- /**
- * End transaction as {@link javax.transaction.xa.XAResource#TMFAIL failed}; in case of error,
- * try to roll it back.
- */
- void failAndRollback(Xid xid) throws TransientXaException;
-
- /**
- * Note: this can block on some non-MVCC databases if there are ended not prepared transactions.
- */
- Collection<Xid> recover() throws TransientXaException;
-
- /**
- * Thrown by {@link XaFacade} when RM responds with {@link
- * javax.transaction.xa.XAResource#XA_RDONLY XA_RDONLY} indicating that the transaction doesn't
- * include any changes. When such a transaction is committed RM may return an error (usually,
- * {@link javax.transaction.xa.XAException#XAER_NOTA XAER_NOTA}).
- */
- class EmptyXaTransactionException extends FlinkRuntimeException {
- private final Xid xid;
-
- EmptyXaTransactionException(Xid xid) {
- super("end response XA_RDONLY, xid: " + xid);
- this.xid = xid;
- }
-
- public Xid getXid() {
- return xid;
- }
- }
-
- /**
- * Indicates a transient or unknown failure from the resource manager (see {@link
- * XAException#XA_RBTRANSIENT XA_RBTRANSIENT}, {@link XAException#XAER_RMFAIL XAER_RMFAIL}).
- */
- class TransientXaException extends FlinkRuntimeException {
- TransientXaException(XAException cause) {
- super(cause);
- }
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacadeImpl.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacadeImpl.java
deleted file mode 100644
index 606f6c2..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacadeImpl.java
+++ /dev/null
@@ -1,463 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.function.ThrowingRunnable;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.concurrent.NotThreadSafe;
-import javax.sql.XAConnection;
-import javax.sql.XADataSource;
-import javax.transaction.xa.XAException;
-import javax.transaction.xa.XAResource;
-import javax.transaction.xa.Xid;
-
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
-import static java.util.Optional.empty;
-import static java.util.Optional.of;
-import static javax.transaction.xa.XAException.XAER_NOTA;
-import static javax.transaction.xa.XAException.XAER_RMFAIL;
-import static javax.transaction.xa.XAException.XA_HEURCOM;
-import static javax.transaction.xa.XAException.XA_HEURHAZ;
-import static javax.transaction.xa.XAException.XA_HEURMIX;
-import static javax.transaction.xa.XAException.XA_HEURRB;
-import static javax.transaction.xa.XAException.XA_RBBASE;
-import static javax.transaction.xa.XAException.XA_RBTIMEOUT;
-import static javax.transaction.xa.XAException.XA_RBTRANSIENT;
-import static javax.transaction.xa.XAResource.TMENDRSCAN;
-import static javax.transaction.xa.XAResource.TMNOFLAGS;
-import static javax.transaction.xa.XAResource.TMSTARTRSCAN;
-
-/** Default {@link XaFacade} implementation. */
-@NotThreadSafe
-@Internal
-@Deprecated
-class XaFacadeImpl implements XaFacade {
-
- private static final long serialVersionUID = 1L;
-
- private static final Logger LOG = LoggerFactory.getLogger(XaFacadeImpl.class);
- private static final Set<Integer> TRANSIENT_ERR_CODES =
- new HashSet<>(Arrays.asList(XA_RBTRANSIENT, XAER_RMFAIL));
- private static final Set<Integer> HEUR_ERR_CODES =
- new HashSet<>(Arrays.asList(XA_HEURRB, XA_HEURCOM, XA_HEURHAZ, XA_HEURMIX));
- private static final int MAX_RECOVER_CALLS = 100;
-
- private final Supplier<XADataSource> dataSourceSupplier;
- private final Integer timeoutSec;
- private transient XAResource xaResource;
- private transient Connection connection;
- private transient XAConnection xaConnection;
-
- @VisibleForTesting
- static XaFacadeImpl fromXaDataSource(XADataSource ds) {
- return new XaFacadeImpl(() -> ds, null);
- }
-
- XaFacadeImpl(Supplier<XADataSource> dataSourceSupplier, Integer timeoutSec) {
- this.dataSourceSupplier = Preconditions.checkNotNull(dataSourceSupplier);
- this.timeoutSec = timeoutSec;
- }
-
- @Override
- public void open() throws SQLException {
- Preconditions.checkState(!isOpen(), "already connected");
- XADataSource ds = dataSourceSupplier.get();
- xaConnection = ds.getXAConnection();
- xaResource = xaConnection.getXAResource();
- if (timeoutSec != null) {
- try {
- xaResource.setTransactionTimeout(timeoutSec);
- } catch (XAException e) {
- throw new SQLException(e);
- }
- }
- connection = xaConnection.getConnection();
- connection.setReadOnly(false);
- connection.setAutoCommit(false);
- Preconditions.checkState(!connection.getAutoCommit());
- }
-
- @Override
- public void close() throws SQLException {
- if (connection != null) {
- connection.close(); // close connection - likely a wrapper
- connection = null;
- }
- try {
- xaConnection.close(); // close likely a pooled AND the underlying connection
- } catch (SQLException e) {
- // Some databases (e.g. MySQL) rollback changes on normal client disconnect which
- // causes an exception if an XA transaction was prepared. Note that resources are
- // still released in case of an error. Pinning MySQL connections doesn't help as
- // SuspendableXAConnection has the same close() logic.
- // Other DBs don't rollback, e.g. for PgSql the previous connection.close() call
- // disassociates the connection (and that call works because it has a check for XA)
- // and rollback() is not called.
- // In either case, not closing the XA connection here leads to the resource leak.
- LOG.warn("unable to close XA connection", e);
- }
- xaResource = null;
- }
-
- @Override
- public Connection getConnection() {
- Preconditions.checkNotNull(connection);
- return connection;
- }
-
- @Override
- public boolean isConnectionValid() throws SQLException {
- return isOpen() && connection.isValid(connection.getNetworkTimeout());
- }
-
- @Override
- public Connection getOrEstablishConnection() throws SQLException {
- if (!isOpen()) {
- open();
- }
- return connection;
- }
-
- @Override
- public void closeConnection() {
- try {
- close();
- } catch (SQLException e) {
- LOG.warn("Connection close failed.", e);
- }
- }
-
- @Override
- public Connection reestablishConnection() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void start(Xid xid) {
- execute(Command.fromRunnable("start", xid, () -> xaResource.start(xid, TMNOFLAGS)));
- }
-
- @Override
- public void endAndPrepare(Xid xid) {
- execute(Command.fromRunnable("end", xid, () -> xaResource.end(xid, XAResource.TMSUCCESS)));
- int prepResult = execute(new Command<>("prepare", of(xid), () -> xaResource.prepare(xid)));
- if (prepResult == XAResource.XA_RDONLY) {
- throw new EmptyXaTransactionException(xid);
- } else if (prepResult != XAResource.XA_OK) {
- throw new FlinkRuntimeException(
- formatErrorMessage("prepare", of(xid), empty(), "response: " + prepResult));
- }
- }
-
- @Override
- public void failAndRollback(Xid xid) {
- execute(
- Command.fromRunnable(
- "end (fail)",
- xid,
- () -> {
- xaResource.end(xid, XAResource.TMFAIL);
- xaResource.rollback(xid);
- },
- err -> {
- if (err.errorCode >= XA_RBBASE) {
- rollback(xid);
- } else {
- LOG.warn(
- formatErrorMessage(
- "end (fail)", of(xid), of(err.errorCode)));
- }
- }));
- }
-
- @Override
- public void commit(Xid xid, boolean ignoreUnknown) {
- execute(
- Command.fromRunnableRecoverByWarn(
- "commit",
- xid,
- () ->
- xaResource.commit(
- xid,
- false /* not onePhase because the transaction should be prepared already */),
- e -> buildCommitErrorDesc(e, ignoreUnknown)));
- }
-
- @Override
- public void rollback(Xid xid) {
- execute(
- Command.fromRunnableRecoverByWarn(
- "rollback",
- xid,
- () -> xaResource.rollback(xid),
- this::buildRollbackErrorDesc));
- }
-
- private void forget(Xid xid) {
- execute(
- Command.fromRunnableRecoverByWarn(
- "forget",
- xid,
- () -> xaResource.forget(xid),
- e -> of("manual cleanup may be required")));
- }
-
- @Override
- public Collection<Xid> recover() {
- return execute(
- new Command<>(
- "recover",
- empty(),
- () -> {
- List<Xid> list = recover(TMSTARTRSCAN);
- try {
- for (int i = 0; list.addAll(recover(TMNOFLAGS)); i++) {
- // H2 sometimes returns same tx list here - should probably use
- // recover(TMSTARTRSCAN | TMENDRSCAN)
- Preconditions.checkState(
- i < MAX_RECOVER_CALLS, "too many xa_recover() calls");
- }
- } finally {
- recover(TMENDRSCAN);
- }
- return list;
- }));
- }
-
- @Override
- public boolean isOpen() {
- return xaResource != null;
- }
-
- private List<Xid> recover(int flags) throws XAException {
- return Arrays.asList(xaResource.recover(flags));
- }
-
- private <T> T execute(Command<T> cmd) throws FlinkRuntimeException {
- Preconditions.checkState(isOpen(), "not connected");
- LOG.debug("{}, xid={}", cmd.name, cmd.xid);
- try {
- T result = cmd.callable.call();
- LOG.trace("{} succeeded , xid={}", cmd.name, cmd.xid);
- return result;
- } catch (XAException e) {
- if (HEUR_ERR_CODES.contains(e.errorCode)) {
- cmd.xid.ifPresent(this::forget);
- }
- return cmd.recover.apply(e).orElseThrow(() -> wrapException(cmd.name, cmd.xid, e));
- } catch (FlinkRuntimeException e) {
- throw e;
- } catch (Exception e) {
- throw wrapException(cmd.name, cmd.xid, e);
- }
- }
-
- private static class Command<T> {
- private final String name;
- private final Optional<Xid> xid;
- private final Callable<T> callable;
- private final Function<XAException, Optional<T>> recover;
-
- static Command<Object> fromRunnable(
- String action, Xid xid, ThrowingRunnable<XAException> runnable) {
- return fromRunnable(
- action,
- xid,
- runnable,
- e -> {
- throw wrapException(action, of(xid), e);
- });
- }
-
- static Command<Object> fromRunnableRecoverByWarn(
- String action,
- Xid xid,
- ThrowingRunnable<XAException> runnable,
- Function<XAException, Optional<String>> err2msg) {
- return fromRunnable(
- action,
- xid,
- runnable,
- e ->
- LOG.warn(
- formatErrorMessage(
- action,
- of(xid),
- of(e.errorCode),
- err2msg.apply(e)
- .orElseThrow(
- () ->
- wrapException(
- action, of(xid), e)))));
- }
-
- private static Command<Object> fromRunnable(
- String action,
- Xid xid,
- ThrowingRunnable<XAException> runnable,
- Consumer<XAException> recover) {
- return new Command<>(
- action,
- of(xid),
- () -> {
- runnable.run();
- return null;
- },
- e -> {
- recover.accept(e);
- return Optional.of("");
- });
- }
-
- private Command(String name, Optional<Xid> xid, Callable<T> callable) {
- this(name, xid, callable, e -> empty());
- }
-
- private Command(
- String name,
- Optional<Xid> xid,
- Callable<T> callable,
- Function<XAException, Optional<T>> recover) {
- this.name = name;
- this.xid = xid;
- this.callable = callable;
- this.recover = recover;
- }
- }
-
- private static FlinkRuntimeException wrapException(
- String action, Optional<Xid> xid, Exception ex) {
- if (ex instanceof XAException) {
- XAException xa = (XAException) ex;
- if (TRANSIENT_ERR_CODES.contains(xa.errorCode)) {
- throw new TransientXaException(xa);
- } else {
- throw new FlinkRuntimeException(
- formatErrorMessage(action, xid, of(xa.errorCode), xa.getMessage()));
- }
- } else {
- throw new FlinkRuntimeException(
- formatErrorMessage(action, xid, empty(), ex.getMessage()), ex);
- }
- }
-
- private Optional<String> buildCommitErrorDesc(XAException err, boolean ignoreUnknown) {
- if (err.errorCode == XA_HEURCOM) {
- return Optional.of("transaction was heuristically committed earlier");
- } else if (ignoreUnknown && err.errorCode == XAER_NOTA) {
- return Optional.of("transaction is unknown to RM (ignoring)");
- } else {
- return empty();
- }
- }
-
- private Optional<String> buildRollbackErrorDesc(XAException err) {
- if (err.errorCode == XA_HEURRB) {
- return Optional.of("transaction was already heuristically rolled back");
- } else if (err.errorCode >= XA_RBBASE) {
- return Optional.of("transaction was already marked for rollback");
- } else {
- return empty();
- }
- }
-
- private static String formatErrorMessage(
- String action, Optional<Xid> xid, Optional<Integer> errorCode, String... more) {
- return String.format(
- "unable to %s%s%s%s",
- action,
- xid.map(x -> " XA transaction, xid: " + x).orElse(""),
- errorCode
- .map(code -> String.format(", error %d: %s", code, descError(code)))
- .orElse(""),
- more == null || more.length == 0 ? "" : ". " + Arrays.toString(more));
- }
-
- /** @return error description from {@link XAException} javadoc from to ease debug. */
- private static String descError(int code) {
- switch (code) {
- case XA_HEURCOM:
- return "heuristic commit decision was made";
- case XAException.XA_HEURHAZ:
- return "heuristic decision may have been made";
- case XAException.XA_HEURMIX:
- return "heuristic mixed decision was made";
- case XA_HEURRB:
- return "heuristic rollback decision was made";
- case XAException.XA_NOMIGRATE:
- return "the transaction resumption must happen where the suspension occurred";
- case XAException.XA_RBCOMMFAIL:
- return "rollback happened due to a communications failure";
- case XAException.XA_RBDEADLOCK:
- return "rollback happened because deadlock was detected";
- case XAException.XA_RBINTEGRITY:
- return "rollback happened because an internal integrity check failed";
- case XAException.XA_RBOTHER:
- return "rollback happened for some reason not fitting any of the other rollback error codes";
- case XAException.XA_RBPROTO:
- return "rollback happened due to a protocol error in the resource manager";
- case XAException.XA_RBROLLBACK:
- return "rollback happened for an unspecified reason";
- case XA_RBTIMEOUT:
- return "rollback happened because of a timeout";
- case XA_RBTRANSIENT:
- return "rollback happened due to a transient failure";
- case XAException.XA_RDONLY:
- return "the transaction branch was read-only, and has already been committed";
- case XAException.XA_RETRY:
- return "the method invoked returned without having any effect, and that it may be invoked again";
- case XAException.XAER_ASYNC:
- return "an asynchronous operation is outstanding";
- case XAException.XAER_DUPID:
- return "Xid given as an argument is already known to the resource manager";
- case XAException.XAER_INVAL:
- return "invalid arguments were passed";
- case XAER_NOTA:
- return "Xid is not valid";
- case XAException.XAER_OUTSIDE:
- return "the resource manager is doing work outside the global transaction";
- case XAException.XAER_PROTO:
- return "protocol error";
- case XAException.XAER_RMERR:
- return "resource manager error has occurred";
- case XAException.XAER_RMFAIL:
- return "the resource manager has failed and is not available";
- default:
- return "";
- }
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacadePoolingImpl.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacadePoolingImpl.java
deleted file mode 100644
index 2dbb693..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaFacadePoolingImpl.java
+++ /dev/null
@@ -1,194 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.util.function.ThrowingConsumer;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-import javax.transaction.xa.Xid;
-
-import java.io.Serializable;
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.util.Collection;
-import java.util.Deque;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.Map;
-import java.util.function.Supplier;
-
-import static org.apache.flink.util.ExceptionUtils.rethrow;
-import static org.apache.flink.util.Preconditions.checkState;
-
-/**
- * A "pooling" implementation of {@link XaFacade}. Some database implement XA such that one
- * connection is limited to a single transaction. As a workaround, this implementation creates a new
- * XA resource after each xa_start call is made (and associates it with the xid to commit later).
- */
-@Internal
-@Deprecated
-class XaFacadePoolingImpl implements XaFacade {
- private static final long serialVersionUID = 1L;
-
- public interface FacadeSupplier extends Serializable, Supplier<XaFacade> {}
-
- private static final transient Logger LOG = LoggerFactory.getLogger(XaFacadePoolingImpl.class);
- private final FacadeSupplier facadeSupplier;
- private transient XaFacade active;
- private transient Map<Xid, XaFacade> mappedToXids;
- private transient Deque<XaFacade> pooled;
-
- XaFacadePoolingImpl(FacadeSupplier facadeSupplier) {
- this.facadeSupplier = facadeSupplier;
- }
-
- @Override
- public void open() throws Exception {
- checkState(active == null);
- pooled = new LinkedList<>();
- mappedToXids = new HashMap<>();
- }
-
- @Override
- public boolean isOpen() {
- return active != null && active.isOpen();
- }
-
- @Override
- public void start(Xid xid) throws Exception {
- checkState(active == null);
- if (pooled.isEmpty()) {
- active = facadeSupplier.get();
- active.open();
- } else {
- active = pooled.poll();
- }
- active.start(xid);
- mappedToXids.put(xid, active);
- }
-
- /**
- * Must be called after {@link #start(Xid)} with the same {@link Xid}.
- *
- * @see XaFacade#endAndPrepare(Xid)
- */
- @Override
- public void endAndPrepare(Xid xid) throws Exception {
- checkState(active == mappedToXids.get(xid));
- try {
- active.endAndPrepare(xid);
- } finally {
- active = null;
- }
- }
-
- @Override
- public void commit(Xid xid, boolean ignoreUnknown) throws TransientXaException {
- runForXid(xid, facade -> facade.commit(xid, ignoreUnknown));
- }
-
- @Override
- public void rollback(Xid xid) throws TransientXaException {
- runForXid(xid, facade -> facade.rollback(xid));
- }
-
- @Override
- public void failAndRollback(Xid xid) throws TransientXaException {
- runForXid(xid, facade -> facade.failAndRollback(xid));
- }
-
- @Override
- public Collection<Xid> recover() throws TransientXaException {
- return peekPooled().recover();
- }
-
- @Override
- public void close() throws Exception {
- for (XaFacade facade : mappedToXids.values()) {
- facade.close();
- }
- for (XaFacade facade : pooled) {
- facade.close();
- }
- if (active != null && active.isOpen()) {
- active.close();
- }
- }
-
- @Nullable
- @Override
- public Connection getConnection() {
- return active.getConnection();
- }
-
- @Override
- public boolean isConnectionValid() throws SQLException {
- return active.isConnectionValid();
- }
-
- @Override
- public Connection getOrEstablishConnection() throws SQLException, ClassNotFoundException {
- return active.getOrEstablishConnection();
- }
-
- @Override
- public void closeConnection() {
- active.closeConnection();
- }
-
- @Override
- public Connection reestablishConnection() throws SQLException, ClassNotFoundException {
- return active.reestablishConnection();
- }
-
- // WARN: action MUST leave the facade in IDLE state (i.e. not start/end/prepare any tx)
- private void runForXid(Xid xid, ThrowingConsumer<XaFacade, TransientXaException> action) {
- XaFacade mapped = mappedToXids.remove(xid);
- if (mapped == null) {
- // a transaction can be not known during recovery
- LOG.debug("No XA resource found associated with XID: {}", xid);
- action.accept(peekPooled());
- } else {
- LOG.debug("Found mapped XA resource for XID: {} {}", xid, mapped);
- try {
- action.accept(mapped);
- } finally {
- pooled.offer(mapped);
- }
- }
- }
-
- // WARN: the returned facade MUST be left in IDLE state (i.e. not start/end/prepare any tx)
- private XaFacade peekPooled() {
- XaFacade xaFacade = pooled.peek();
- if (xaFacade == null) {
- xaFacade = facadeSupplier.get();
- try {
- xaFacade.open();
- } catch (Exception e) {
- rethrow(e);
- }
- pooled.offer(xaFacade);
- }
- return xaFacade;
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaGroupOps.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaGroupOps.java
deleted file mode 100644
index b355f11..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaGroupOps.java
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.util.FlinkRuntimeException;
-
-import javax.transaction.xa.Xid;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Optional;
-
-@Internal
-@Deprecated
-interface XaGroupOps extends Serializable {
-
- GroupXaOperationResult<CheckpointAndXid> commit(
- List<CheckpointAndXid> xids, boolean allowOutOfOrderCommits, int maxCommitAttempts);
-
- GroupXaOperationResult<Xid> failOrRollback(Collection<Xid> xids);
-
- void recoverAndRollback(JobSubtask subtask, XidGenerator xidGenerator);
-
- class GroupXaOperationResult<T> {
- private final List<T> succeeded = new ArrayList<>();
- private final List<T> failed = new ArrayList<>();
- private final List<T> toRetry = new ArrayList<>();
- private Optional<Exception> failure = Optional.empty();
- private Optional<Exception> transientFailure = Optional.empty();
-
- void failedTransiently(T x, XaFacade.TransientXaException e) {
- toRetry.add(x);
- transientFailure =
- getTransientFailure().isPresent() ? getTransientFailure() : Optional.of(e);
- }
-
- void failed(T x, Exception e) {
- failed.add(x);
- failure = failure.isPresent() ? failure : Optional.of(e);
- }
-
- void succeeded(T x) {
- succeeded.add(x);
- }
-
- private FlinkRuntimeException wrapFailure(
- Exception error, String formatWithCounts, int errCount) {
- return new FlinkRuntimeException(
- String.format(formatWithCounts, errCount, total()), error);
- }
-
- private int total() {
- return succeeded.size() + failed.size() + toRetry.size();
- }
-
- List<T> getForRetry() {
- return toRetry;
- }
-
- Optional<Exception> getTransientFailure() {
- return transientFailure;
- }
-
- boolean hasNoFailures() {
- return !failure.isPresent() && !transientFailure.isPresent();
- }
-
- void throwIfAnyFailed(String action) {
- failure.map(
- f ->
- wrapFailure(
- f,
- "failed to " + action + " %d transactions out of %d",
- toRetry.size() + failed.size()))
- .ifPresent(
- f -> {
- throw f;
- });
- }
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaGroupOpsImpl.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaGroupOpsImpl.java
deleted file mode 100644
index b1ace00..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaGroupOpsImpl.java
+++ /dev/null
@@ -1,143 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.connector.jdbc.xa.XaFacade.TransientXaException;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.transaction.xa.Xid;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
-
-@Internal
-@Deprecated
-class XaGroupOpsImpl implements XaGroupOps {
-
- private static final long serialVersionUID = 1L;
-
- private static final Logger LOG = LoggerFactory.getLogger(XaGroupOpsImpl.class);
-
- private final XaFacade xaFacade;
-
- XaGroupOpsImpl(XaFacade xaFacade) {
- this.xaFacade = xaFacade;
- }
-
- @Override
- public GroupXaOperationResult<CheckpointAndXid> commit(
- List<CheckpointAndXid> xids, boolean allowOutOfOrderCommits, int maxCommitAttempts) {
- GroupXaOperationResult<CheckpointAndXid> result = new GroupXaOperationResult<>();
- int origSize = xids.size();
- LOG.debug("commit {} transactions", origSize);
- for (Iterator<CheckpointAndXid> i = xids.iterator();
- i.hasNext() && (result.hasNoFailures() || allowOutOfOrderCommits); ) {
- CheckpointAndXid x = i.next();
- i.remove();
- try {
- xaFacade.commit(x.xid, x.restored);
- result.succeeded(x);
- } catch (TransientXaException e) {
- result.failedTransiently(x.withAttemptsIncremented(), e);
- } catch (Exception e) {
- result.failed(x, e);
- }
- }
- result.getForRetry().addAll(xids);
- result.throwIfAnyFailed("commit");
- throwIfAnyReachedMaxAttempts(result, maxCommitAttempts);
- result.getTransientFailure()
- .ifPresent(
- f ->
- LOG.warn(
- "failed to commit {} transactions out of {} (keep them to retry later)",
- result.getForRetry().size(),
- origSize,
- f));
- return result;
- }
-
- @Override
- public GroupXaOperationResult<Xid> failOrRollback(Collection<Xid> xids) {
- GroupXaOperationResult<Xid> result = new GroupXaOperationResult<>();
- if (xids.isEmpty()) {
- return result;
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("rolling back {} transactions: {}", xids.size(), xids);
- }
- for (Xid x : xids) {
- try {
- xaFacade.failAndRollback(x);
- result.succeeded(x);
- } catch (TransientXaException e) {
- LOG.info("unable to fail/rollback transaction, xid={}: {}", x, e.getMessage());
- result.failedTransiently(x, e);
- } catch (Exception e) {
- LOG.warn("unable to fail/rollback transaction, xid={}: {}", x, e.getMessage());
- result.failed(x, e);
- }
- }
- if (!result.getForRetry().isEmpty()) {
- LOG.info("failed to roll back {} transactions", result.getForRetry().size());
- }
- return result;
- }
-
- @Override
- public void recoverAndRollback(JobSubtask subtask, XidGenerator xidGenerator) {
- Collection<Xid> recovered = xaFacade.recover();
- if (recovered.isEmpty()) {
- return;
- }
- LOG.warn("rollback {} recovered transactions", recovered.size());
- for (Xid xid : recovered) {
- if (xidGenerator.belongsToSubtask(xid, subtask)) {
- try {
- xaFacade.rollback(xid);
- } catch (Exception e) {
- LOG.info("unable to rollback recovered transaction, xid={}", xid, e);
- }
- }
- }
- }
-
- private static void throwIfAnyReachedMaxAttempts(
- GroupXaOperationResult<CheckpointAndXid> result, int maxAttempts) {
- List<CheckpointAndXid> reached = null;
- for (CheckpointAndXid x : result.getForRetry()) {
- if (x.attempts >= maxAttempts) {
- if (reached == null) {
- reached = new ArrayList<>();
- }
- reached.add(x);
- }
- }
- if (reached != null) {
- throw new RuntimeException(
- String.format(
- "reached max number of commit attempts (%d) for transactions: %s",
- maxAttempts, reached));
- }
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateHandler.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateHandler.java
deleted file mode 100644
index 8c2df29..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateHandler.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.runtime.state.FunctionInitializationContext;
-
-import java.io.Serializable;
-
-@Internal
-@Deprecated
-interface XaSinkStateHandler extends Serializable {
-
- JdbcXaSinkFunctionState load(FunctionInitializationContext context) throws Exception;
-
- void store(JdbcXaSinkFunctionState state) throws Exception;
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateHandlerImpl.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateHandlerImpl.java
deleted file mode 100644
index c293162..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateHandlerImpl.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.state.ListState;
-import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.runtime.state.FunctionInitializationContext;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-import javax.transaction.xa.Xid;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-@Internal
-@Deprecated
-class XaSinkStateHandlerImpl implements XaSinkStateHandler {
-
- private static final long serialVersionUID = 1L;
-
- private static final Logger LOG = LoggerFactory.getLogger(XaSinkStateHandlerImpl.class);
-
- private final TypeSerializer<JdbcXaSinkFunctionState> serializer;
-
- // state could be stored as two separate lists
- // on one hand this would allow more even distribution on re-scale
- // on the other it would lead to more IO calls and less data locality
- private transient ListState<JdbcXaSinkFunctionState> states;
-
- XaSinkStateHandlerImpl() {
- this(new XaSinkStateSerializer());
- }
-
- XaSinkStateHandlerImpl(TypeSerializer<JdbcXaSinkFunctionState> serializer) {
- this.serializer = serializer;
- }
-
- @Override
- public JdbcXaSinkFunctionState load(FunctionInitializationContext context) throws Exception {
- states = getListState(context, serializer, "XaSinkState");
- return context.isRestored() ? merge(states.get()) : JdbcXaSinkFunctionState.empty();
- }
-
- @Override
- public void store(JdbcXaSinkFunctionState state) throws Exception {
- if (LOG.isDebugEnabled()) {
- LOG.debug("store state snapshot: {}", state);
- }
- states.update(Collections.singletonList(state));
- }
-
- private <T> ListState<T> getListState(
- FunctionInitializationContext context, TypeSerializer<T> serializer, String name) {
- try {
- return context.getOperatorStateStore()
- .getListState(new ListStateDescriptor<>(name, serializer));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- private JdbcXaSinkFunctionState merge(@Nullable Iterable<JdbcXaSinkFunctionState> states) {
- if (states == null) {
- return JdbcXaSinkFunctionState.empty();
- }
- List<Xid> hanging = new ArrayList<>();
- List<CheckpointAndXid> prepared = new ArrayList<>();
- states.forEach(
- i -> {
- hanging.addAll(i.getHanging());
- prepared.addAll(i.getPrepared());
- });
- return JdbcXaSinkFunctionState.of(prepared, hanging);
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateSerializer.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateSerializer.java
deleted file mode 100644
index 0111e10..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XaSinkStateSerializer.java
+++ /dev/null
@@ -1,161 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
-import org.apache.flink.connector.jdbc.datasource.transactions.xa.xid.XidSerializer;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-
-import javax.transaction.xa.Xid;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-/** XaSinkStateSerializer. */
-@Internal
-@Deprecated
-public final class XaSinkStateSerializer extends TypeSerializer<JdbcXaSinkFunctionState> {
-
- private static final TypeSerializerSnapshot<JdbcXaSinkFunctionState> SNAPSHOT =
- new XaSinkStateSimpleXaTypeSerializerSnapshot();
- private final TypeSerializer<Xid> xidSerializer;
- private final TypeSerializer<CheckpointAndXid> checkpointAndXidSerializer;
-
- public XaSinkStateSerializer() {
- this(new XidSerializer(), new CheckpointAndXidSerializer());
- }
-
- private XaSinkStateSerializer(
- TypeSerializer<Xid> xidSerializer,
- TypeSerializer<CheckpointAndXid> checkpointAndXidSerializer) {
- this.xidSerializer = xidSerializer;
- this.checkpointAndXidSerializer = checkpointAndXidSerializer;
- }
-
- @Override
- public boolean isImmutableType() {
- return true;
- }
-
- @Override
- public TypeSerializer<JdbcXaSinkFunctionState> duplicate() {
- return this;
- }
-
- @Override
- public JdbcXaSinkFunctionState createInstance() {
- return JdbcXaSinkFunctionState.empty();
- }
-
- @Override
- public JdbcXaSinkFunctionState copy(JdbcXaSinkFunctionState from) {
- return from;
- }
-
- @Override
- public JdbcXaSinkFunctionState copy(
- JdbcXaSinkFunctionState from, JdbcXaSinkFunctionState reuse) {
- return from;
- }
-
- @Override
- public int getLength() {
- return -1;
- }
-
- @Override
- public void serialize(JdbcXaSinkFunctionState state, DataOutputView target) throws IOException {
- target.writeInt(state.getHanging().size());
- for (Xid h : state.getHanging()) {
- xidSerializer.serialize(h, target);
- }
- target.writeInt(state.getPrepared().size());
- for (CheckpointAndXid checkpointAndXid : state.getPrepared()) {
- checkpointAndXidSerializer.serialize(checkpointAndXid, target);
- }
- }
-
- @Override
- public JdbcXaSinkFunctionState deserialize(DataInputView source) throws IOException {
- int hangingSize = source.readInt();
- List<Xid> hanging = new ArrayList<>(hangingSize);
- for (int i = 0; i < hangingSize; i++) {
- hanging.add(xidSerializer.deserialize(source));
- }
- int preparedSize = source.readInt();
- List<CheckpointAndXid> prepared = new ArrayList<>(preparedSize);
- for (int i = 0; i < preparedSize; i++) {
- prepared.add(checkpointAndXidSerializer.deserialize(source));
- }
- return JdbcXaSinkFunctionState.of(prepared, hanging);
- }
-
- @Override
- public JdbcXaSinkFunctionState deserialize(JdbcXaSinkFunctionState reuse, DataInputView source)
- throws IOException {
- return deserialize(source);
- }
-
- @Override
- public void copy(DataInputView source, DataOutputView target) throws IOException {
- serialize(deserialize(source), target);
- }
-
- @Override
- public boolean equals(Object obj) {
- return obj instanceof XaSinkStateSerializer;
- }
-
- @Override
- public int hashCode() {
- return 0;
- }
-
- @Override
- public TypeSerializerSnapshot<JdbcXaSinkFunctionState> snapshotConfiguration() {
- return SNAPSHOT;
- }
-
- /** Simple {@link TypeSerializerSnapshot} for {@link XaSinkStateSerializer}. */
- public static class XaSinkStateSimpleXaTypeSerializerSnapshot
- extends SimpleTypeSerializerSnapshot<JdbcXaSinkFunctionState> {
- private static final int VERSION = 1;
-
- public XaSinkStateSimpleXaTypeSerializerSnapshot() {
- super(XaSinkStateSerializer::new);
- }
-
- @Override
- public void writeSnapshot(DataOutputView out) throws IOException {
- super.writeSnapshot(out);
- out.writeInt(VERSION);
- }
-
- @Override
- public void readSnapshot(int readVersion, DataInputView in, ClassLoader classLoader)
- throws IOException {
- super.readSnapshot(readVersion, in, classLoader);
- in.readInt();
- }
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XidGenerator.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XidGenerator.java
deleted file mode 100644
index e392023..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/XidGenerator.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.annotation.Internal;
-
-import javax.transaction.xa.Xid;
-
-import java.io.Serializable;
-import java.security.SecureRandom;
-
-/** {@link Xid} generator. */
-@Internal
-@Deprecated
-public interface XidGenerator extends Serializable, AutoCloseable {
-
- /**
- * Generate new {@link Xid}. Requirements for generated Xids:
- *
- * <ul>
- * <li>Global Transaction Id MUST be unique across Flink job, and probably across Xids
- * generated by other jobs and applications - depends on the usage of this class
- * <li>SHOULD be immutable
- * <li>SHOULD override {@link Object#hashCode hashCode} and {@link Object#equals equals}
- * </ul>
- *
- * @param subtask can be used for example to derive global transaction id
- * @param checkpointId can be used for example to derive global transaction id
- */
- Xid generateXid(JobSubtask subtask, long checkpointId);
-
- default void open() {}
-
- /** @return true if the provided transaction belongs to this subtask */
- boolean belongsToSubtask(Xid xid, JobSubtask subtask);
-
- @Override
- default void close() {}
-
- /**
- * Creates a {@link XidGenerator} that generates {@link Xid xids} from:
- *
- * <ol>
- * <li>job id
- * <li>subtask index
- * <li>checkpoint id
- * <li>four random bytes generated using {@link SecureRandom})
- * </ol>
- *
- * <p>Each created {@link XidGenerator} instance MUST be used for only one Sink instance
- * (otherwise Xids could collide).
- */
- static XidGenerator semanticXidGenerator() {
- return new SemanticXidGenerator();
- }
-}
diff --git a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/package-info.java b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/package-info.java
deleted file mode 100644
index 41ae010..0000000
--- a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/xa/package-info.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.
- */
-
-/**
- * Provides exactly-once JDBC sink implementation using Java XA transactions API (JTA).
- *
- * @see org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction JdbcXaExactlyOnceSinkFunction
- */
-@Deprecated
-package org.apache.flink.connector.jdbc.xa;
diff --git a/flink-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
deleted file mode 100644
index 44a97bd..0000000
--- a/flink-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactory
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java
deleted file mode 100644
index e67d5d5..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/dialect/JdbcDialectTypeTest.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * 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.dialect;
-
-import org.apache.flink.connector.jdbc.derby.database.DerbyDialectTest;
-
-/**
- * Tests for all DataTypes and Dialects of JDBC connector.
- *
- * @deprecated use org.apache.flink.connector.jdbc.core.table.dialect.JdbcDialectTest
- */
-@Deprecated
-public class JdbcDialectTypeTest extends DerbyDialectTest {}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXidSerializersTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXidSerializersTest.java
deleted file mode 100644
index ef020ea..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/CheckpointAndXidSerializersTest.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.api.common.typeutils.SerializerTestBase;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.connector.jdbc.datasource.transactions.xa.xid.XaXidTest;
-
-/** XaSerializersTest. */
-@Deprecated
-class CheckpointAndXidSerializersTest extends SerializerTestBase<CheckpointAndXid> {
- private static final CheckpointAndXid CHECKPOINT_AND_XID =
- CheckpointAndXid.createRestored(1L, 1, XaXidTest.XID);
-
- @Override
- protected TypeSerializer<CheckpointAndXid> createSerializer() {
- return new CheckpointAndXidSerializer();
- }
-
- @Override
- protected int getLength() {
- return -1;
- }
-
- @Override
- protected Class<CheckpointAndXid> getTypeClass() {
- return CheckpointAndXid.class;
- }
-
- @Override
- protected CheckpointAndXid[] getTestData() {
- return new CheckpointAndXid[] {CHECKPOINT_AND_XID};
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcExactlyOnceSinkE2eTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcExactlyOnceSinkE2eTest.java
deleted file mode 100644
index f27cba5..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcExactlyOnceSinkE2eTest.java
+++ /dev/null
@@ -1,400 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.common.state.CheckpointListener;
-import org.apache.flink.api.common.state.ListState;
-import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
-import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
-import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
-import org.apache.flink.connector.jdbc.testutils.DatabaseTest;
-import org.apache.flink.connector.jdbc.testutils.TableManaged;
-import org.apache.flink.connector.jdbc.testutils.tables.templates.BooksTable;
-import org.apache.flink.runtime.state.FunctionInitializationContext;
-import org.apache.flink.runtime.state.FunctionSnapshotContext;
-import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
-import org.apache.flink.streaming.api.CheckpointingMode;
-import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.function.SerializableSupplier;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.sql.XADataSource;
-
-import java.time.Duration;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Supplier;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-import static java.util.Collections.singletonList;
-import static org.apache.flink.api.common.restartstrategy.RestartStrategies.fixedDelayRestart;
-import static org.apache.flink.configuration.JobManagerOptions.EXECUTION_FAILOVER_STRATEGY;
-import static org.apache.flink.configuration.TaskManagerOptions.TASK_CANCELLATION_TIMEOUT;
-import static org.apache.flink.connector.jdbc.xa.JdbcXaFacadeTestHelper.getInsertedIds;
-import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT;
-import static org.apache.flink.util.Preconditions.checkState;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** A simple end-to-end test for {@link JdbcXaSinkFunction}. */
-@Deprecated
-public abstract class JdbcExactlyOnceSinkE2eTest implements DatabaseTest {
-
- private static final Logger LOG = LoggerFactory.getLogger(JdbcExactlyOnceSinkE2eTest.class);
-
- private static final BooksTable OUTPUT_TABLE = new BooksTable("XaTable");
- protected static final int PARALLELISM = 4;
- protected static final long CHECKPOINT_TIMEOUT_MS = 5_000L;
- protected static final String TASK_CANCELLATION_TIMEOUT_MS = "10000";
-
- @RegisterExtension static final MiniClusterExtension MINI_CLUSTER = createCluster();
-
- private static MiniClusterExtension createCluster() {
- Configuration configuration = new Configuration();
- // single failover region to allow checkpointing even after some sources have finished and
- // restart all tasks if at least one fails
- configuration.set(EXECUTION_FAILOVER_STRATEGY, "full");
- // cancel tasks eagerly to reduce the risk of running out of memory with many restarts
- // TODO: Temporarily using the original configuration settings for CI compatibility
- // We should ensure that the current oldest flink version could process the configuration
- // item correctly.
- // API calling lines will be adapted after the appropriate series versions
- // that the oldest flink version in the CI can process the config options.
- configuration.setString(TASK_CANCELLATION_TIMEOUT.key(), TASK_CANCELLATION_TIMEOUT_MS);
- configuration.set(CHECKPOINTING_TIMEOUT, Duration.ofMillis(CHECKPOINT_TIMEOUT_MS));
-
- return new MiniClusterExtension(
- new MiniClusterResourceConfiguration.Builder()
- .setNumberTaskManagers(PARALLELISM)
- .setConfiguration(configuration)
- .build());
- }
-
- @Override
- public List<TableManaged> getManagedTables() {
- return Collections.singletonList(OUTPUT_TABLE);
- }
-
- // track active sources for:
- // 1. if any cancels, cancel others ASAP
- // 2. wait for others (to participate in checkpointing)
- // not using SharedObjects because we want to explicitly control which tag (attempt) to use
- private static final Map<Integer, CountDownLatch> activeSources = new ConcurrentHashMap<>();
- // track inactive mappers - to start emission only when they are ready (to prevent them from
- // starving for memory)
- // not using SharedObjects because we want to explicitly control which tag (attempt) to use
- private static final Map<Integer, CountDownLatch> inactiveMappers = new ConcurrentHashMap<>();
-
- @AfterEach
- public void after() {
- activeSources.clear();
- inactiveMappers.clear();
- }
-
- @Test
- void testInsert() throws Exception {
- long started = System.currentTimeMillis();
- LOG.info("Test insert for {}", getMetadata().getVersion());
- int elementsPerSource = 50;
- int numElementsPerCheckpoint = 10;
- int expectedFailures = elementsPerSource / numElementsPerCheckpoint;
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(PARALLELISM);
- env.setRestartStrategy(fixedDelayRestart(expectedFailures * 2, Time.milliseconds(100)));
- env.getConfig().setAutoWatermarkInterval(0L);
- env.enableCheckpointing(50, CheckpointingMode.EXACTLY_ONCE);
- // NOTE: keep operator chaining enabled to prevent memory exhaustion by sources while maps
- // are still initializing
- env.addSource(new TestEntrySource(elementsPerSource, numElementsPerCheckpoint))
- .setParallelism(PARALLELISM)
- .map(new FailingMapper(numElementsPerCheckpoint + (numElementsPerCheckpoint / 2)))
- .addSink(
- exactlyOnceSink(
- OUTPUT_TABLE.getInsertIntoQuery(),
- OUTPUT_TABLE.getStatementBuilder(),
- JdbcExecutionOptions.builder().withMaxRetries(0).build(),
- JdbcExactlyOnceOptions.builder()
- .withTransactionPerConnection(true)
- .build(),
- getMetadata().getXaSourceSupplier()));
-
- env.execute();
-
- List<Integer> insertedIds = getInsertedIds(getMetadata(), OUTPUT_TABLE.getTableName());
- List<Integer> expectedIds =
- IntStream.range(0, elementsPerSource * PARALLELISM)
- .boxed()
- .collect(Collectors.toList());
- assertThat(insertedIds)
- .as(insertedIds.toString())
- .containsExactlyInAnyOrderElementsOf(expectedIds);
-
- LOG.info(
- "Test insert for {} finished in {} ms.",
- getMetadata().getVersion(),
- System.currentTimeMillis() - started);
- }
-
- private <T> SinkFunction<T> exactlyOnceSink(
- String sql,
- JdbcStatementBuilder<T> statementBuilder,
- JdbcExecutionOptions executionOptions,
- JdbcExactlyOnceOptions exactlyOnceOptions,
- SerializableSupplier<XADataSource> dataSourceSupplier) {
- return new JdbcXaSinkFunction<>(
- sql,
- statementBuilder,
- XaFacade.fromXaDataSourceSupplier(
- dataSourceSupplier,
- exactlyOnceOptions.getTimeoutSec(),
- exactlyOnceOptions.isTransactionPerConnection()),
- executionOptions,
- exactlyOnceOptions);
- }
-
- /**
- * {@link SourceFunction} emits {@link BooksTable.BookEntry test entries} and waits for the
- * checkpoint.
- */
- private static class TestEntrySource extends RichParallelSourceFunction<BooksTable.BookEntry>
- implements CheckpointListener, CheckpointedFunction {
- private final int numElements;
- private final int numElementsPerCheckpoint;
-
- private transient volatile ListState<SourceRange> ranges;
-
- private volatile long lastCheckpointId = -1L;
- private volatile boolean lastSnapshotConfirmed = false;
- private volatile boolean running = true;
-
- private TestEntrySource(int numElements, int numElementsPerCheckpoint) {
- this.numElements = numElements;
- this.numElementsPerCheckpoint = numElementsPerCheckpoint;
- }
-
- @Override
- public void run(SourceContext<BooksTable.BookEntry> ctx) throws Exception {
- try {
- waitForConsumers();
- for (SourceRange range : ranges.get()) {
- emitRange(range, ctx);
- }
- } finally {
- getActiveSources().countDown();
- }
- waitOtherSources(); // participate in checkpointing
- }
-
- private void waitForConsumers() throws InterruptedException {
- // even though the pipeline is (intended to be) chained, other parallel instances may
- // starve if this source uses all the available memory before they initialize
- sleep(() -> !inactiveMappers.containsKey(getRuntimeContext().getAttemptNumber()));
- inactiveMappers.get(getRuntimeContext().getAttemptNumber()).await();
- }
-
- private void emitRange(SourceRange range, SourceContext<BooksTable.BookEntry> ctx) {
- for (int i = range.from; i < range.to && running; ) {
- int count = Math.min(range.to - i, numElementsPerCheckpoint);
- emit(i, count, range, ctx);
- i += count;
- }
- }
-
- private void emit(
- int start,
- int count,
- SourceRange toAdvance,
- SourceContext<BooksTable.BookEntry> ctx) {
- synchronized (ctx.getCheckpointLock()) {
- lastCheckpointId = -1L;
- lastSnapshotConfirmed = false;
- for (int j = start; j < start + count && running; j++) {
- try {
- ctx.collect(
- new BooksTable.BookEntry(
- j,
- Integer.toString(j),
- Integer.toString(j),
- (double) (j),
- j));
- } catch (Exception e) {
- if (!ExceptionUtils.findThrowable(e, TestException.class).isPresent()) {
- LOG.warn("Exception during record emission", e);
- }
- throw e;
- }
- toAdvance.advance();
- }
- }
- sleep(() -> !lastSnapshotConfirmed);
- }
-
- @Override
- public void cancel() {
- running = false;
- }
-
- @Override
- public void notifyCheckpointComplete(long checkpointId) {
- if (lastCheckpointId > -1L && checkpointId >= this.lastCheckpointId) {
- lastSnapshotConfirmed = true;
- }
- }
-
- @Override
- public void open(Configuration parameters) throws Exception {
- activeSources.putIfAbsent(
- getRuntimeContext().getAttemptNumber(),
- new CountDownLatch(getRuntimeContext().getNumberOfParallelSubtasks()));
- }
-
- @Override
- public void initializeState(FunctionInitializationContext context) throws Exception {
- ranges =
- context.getOperatorStateStore()
- .getListState(
- new ListStateDescriptor<>("SourceState", SourceRange.class));
- if (!context.isRestored()) {
- ranges.update(
- singletonList(
- SourceRange.forSubtask(
- getRuntimeContext().getIndexOfThisSubtask(), numElements)));
- }
- LOG.debug("Source initialized with ranges: {}", ranges.get());
- }
-
- @Override
- public void snapshotState(FunctionSnapshotContext context) {
- lastCheckpointId = context.getCheckpointId();
- }
-
- private void sleep(Supplier<Boolean> condition) {
- while (condition.get()
- && running
- && !Thread.currentThread().isInterrupted()
- && haveActiveSources()) {
- try {
- Thread.sleep(10);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- ExceptionUtils.rethrow(e);
- }
- }
- }
-
- private void waitOtherSources() throws InterruptedException {
- while (running && haveActiveSources()) {
- getActiveSources().await(100, TimeUnit.MILLISECONDS);
- }
- }
-
- private CountDownLatch getActiveSources() {
- return activeSources.get(getRuntimeContext().getAttemptNumber());
- }
-
- private boolean haveActiveSources() {
- return getActiveSources().getCount() > 0;
- }
-
- private static final class SourceRange {
- private int from;
- private final int to;
-
- private SourceRange(int from, int to) {
- this.from = from;
- this.to = to;
- }
-
- public static SourceRange forSubtask(int subtaskIndex, int elementCount) {
- return new SourceRange(
- subtaskIndex * elementCount, (subtaskIndex + 1) * elementCount);
- }
-
- public void advance() {
- checkState(from < to);
- from++;
- }
-
- @Override
- public String toString() {
- return String.format("%d..%d", from, to);
- }
- }
- }
-
- private static class FailingMapper
- extends RichMapFunction<BooksTable.BookEntry, BooksTable.BookEntry> {
- private final int failingMessage;
- private transient AtomicInteger counter;
-
- public FailingMapper(int failingMessage) {
- this.failingMessage = failingMessage;
- }
-
- @Override
- public void open(Configuration parameters) throws Exception {
- inactiveMappers
- .computeIfAbsent(
- getRuntimeContext().getAttemptNumber(),
- u ->
- new CountDownLatch(
- getRuntimeContext().getNumberOfParallelSubtasks()))
- .countDown();
- counter = new AtomicInteger(failingMessage);
- LOG.debug("Mapper will fail after {} records.", failingMessage);
- }
-
- @Override
- public BooksTable.BookEntry map(BooksTable.BookEntry value) throws Exception {
- if (counter.getAndDecrement() <= 0) {
- LOG.debug("Mapper failing intentionally.");
- throw new TestException();
- }
- return value;
- }
- }
-
- private static final class TestException extends Exception {
- public TestException() {
- // use this string to prevent error parsing scripts from failing the build
- // and still have exception type
- super("java.lang.Exception: Artificial failure", null, true, false);
- }
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaFacadeImplTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaFacadeImplTest.java
deleted file mode 100644
index 313c0d5..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaFacadeImplTest.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.connector.jdbc.JdbcTestBase;
-import org.apache.flink.connector.jdbc.JdbcTestFixture;
-
-import org.junit.jupiter.api.Test;
-
-import javax.sql.XAConnection;
-import javax.sql.XADataSource;
-import javax.transaction.xa.Xid;
-
-import java.sql.Connection;
-import java.sql.Statement;
-import java.util.Collection;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-/** {@link XaFacadeImpl} tests. */
-@Deprecated
-class JdbcXaFacadeImplTest extends JdbcTestBase {
-
- private static final Xid XID =
- new Xid() {
- @Override
- public int getFormatId() {
- return 0;
- }
-
- @Override
- public byte[] getGlobalTransactionId() {
- return "01234".getBytes();
- }
-
- @Override
- public byte[] getBranchQualifier() {
- return "56789".getBytes();
- }
- };
-
- @Test
- void testRecover() throws Exception {
- try (XaFacade f = XaFacadeImpl.fromXaDataSource(getMetadata().buildXaDataSource())) {
- f.open();
- assertThat(f.recover()).isEmpty();
- f.start(XID);
- // insert some data to prevent database from ignoring the transaction
- try (Connection c = f.getConnection()) {
- try (Statement s = c.createStatement()) {
- s.executeUpdate(JdbcTestFixture.getInsertQuery());
- }
- }
- f.endAndPrepare(XID);
- }
- try (XaFacade f = XaFacadeImpl.fromXaDataSource(getMetadata().buildXaDataSource())) {
- f.open();
- Collection<Xid> recovered = f.recover();
- recovered.forEach(f::rollback);
- assertThat(recovered).hasSize(1);
- }
- }
-
- @Test
- void testClose() throws Exception {
- // some drivers (derby, H2) close both connection on either
- // connection.close/xaConnection.close() call, so use mocks here to:
- // a) prevent closing XA connection from connection.close()
- // b) verify that both connections were closed
- XADataSource xaDataSource = mock(XADataSource.class);
- XAConnection xaConnection = mock(XAConnection.class);
- Connection connection = mock(Connection.class);
- when(xaDataSource.getXAConnection()).thenReturn(xaConnection);
- when(xaConnection.getConnection()).thenReturn(connection);
-
- try (XaFacade f = XaFacadeImpl.fromXaDataSource(xaDataSource)) {
- f.open();
- }
- verify(connection).close();
- verify(xaConnection).close();
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaFacadeTestHelper.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaFacadeTestHelper.java
deleted file mode 100644
index 7eb232c..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaFacadeTestHelper.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.connector.jdbc.JdbcTestCheckpoint;
-import org.apache.flink.connector.jdbc.testutils.DatabaseMetadata;
-
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA;
-import static org.assertj.core.api.Assertions.assertThat;
-
-@Deprecated
-class JdbcXaFacadeTestHelper implements AutoCloseable {
- private final String table;
- private final DatabaseMetadata metadata;
- private final XaFacade xaFacade;
-
- JdbcXaFacadeTestHelper(DatabaseMetadata metadata, String table) throws Exception {
- this.metadata = metadata;
- this.table = table;
- this.xaFacade = XaFacadeImpl.fromXaDataSource(metadata.buildXaDataSource());
- this.xaFacade.open();
- }
-
- void assertPreparedTxCountEquals(int expected) {
- assertThat(xaFacade.recover()).hasSize(expected);
- }
-
- void assertDbContentsEquals(JdbcTestCheckpoint... checkpoints) throws SQLException {
- assertDbContentsEquals(
- Arrays.stream(checkpoints).flatMapToInt(x -> Arrays.stream(x.dataItemsIdx)));
- }
-
- void assertDbContentsEquals(IntStream dataIdxStream) throws SQLException {
- assertDbContentsEquals(
- dataIdxStream.map(idx -> TEST_DATA[idx].id).boxed().collect(Collectors.toList()));
- }
-
- void assertDbContentsEquals(List<Integer> expected) throws SQLException {
- assertThat(getInsertedIds()).isEqualTo(expected);
- }
-
- private List<Integer> getInsertedIds() throws SQLException {
- return getInsertedIds(metadata, table);
- }
-
- static List<Integer> getInsertedIds(DatabaseMetadata metadata, String table)
- throws SQLException {
- List<Integer> dbContents = new ArrayList<>();
- try (Connection connection = metadata.getConnection()) {
- connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
- connection.setReadOnly(true);
- try (Statement st = connection.createStatement();
- ResultSet rs = st.executeQuery("select id from " + table)) {
- while (rs.next()) {
- dbContents.add(rs.getInt(1));
- }
- }
- }
- return dbContents;
- }
-
- int countInDb() throws SQLException {
- try (Connection connection = metadata.getConnection()) {
- connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
- connection.setReadOnly(true);
- try (Statement st = connection.createStatement();
- ResultSet rs = st.executeQuery("select count(1) from " + table)) {
- rs.next();
- return rs.getInt(1);
- }
- }
- }
-
- @Override
- public void close() throws Exception {
- xaFacade.close();
- }
-
- void cancelAllTx() {
- xaFacade.recover().forEach(xaFacade::rollback);
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkDerbyTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkDerbyTest.java
deleted file mode 100644
index 1a75a5a..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkDerbyTest.java
+++ /dev/null
@@ -1,167 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.connector.jdbc.JdbcTestFixture;
-
-import org.junit.jupiter.api.Test;
-
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/**
- * {@link JdbcXaSinkFunction} tests using Derby DB. Derby supports XA but doesn't use MVCC, so we
- * can't check anything before all transactions are completed.
- */
-@Deprecated
-class JdbcXaSinkDerbyTest extends JdbcXaSinkTestBase {
-
- /**
- * checkpoint > capture state > emit > snapshot > close > init(captured state), open > emit >
- * checkpoint.
- */
- @Test
- void noDuplication() throws Exception {
- sinkHelper.notifyCheckpointComplete(0);
- TestXaSinkStateHandler newState = new TestXaSinkStateHandler();
- newState.store(sinkHelper.getState().load(null));
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- sinkHelper.close(); // todo: test without close
- sinkHelper = buildSinkHelper(newState);
- sinkHelper.emitAndCheckpoint(JdbcTestFixture.CP0);
- xaHelper.assertDbContentsEquals(JdbcTestFixture.CP0);
- }
-
- @Test
- void testTxEndedOnClose() throws Exception {
- sinkHelper.emit(
- TEST_DATA[0]); // don't snapshotState to prevent transaction from being prepared
- sinkHelper.close();
- xaHelper.assertPreparedTxCountEquals(0);
- }
-
- @Test
- void testTxRollbackOnStartup() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- xaHelper.assertPreparedTxCountEquals(1);
- sinkHelper.close();
- xaHelper.assertPreparedTxCountEquals(1);
- TestXaSinkStateHandler state =
- new TestXaSinkStateHandler(); // forget about prepared tx, so it will not commit
- // them and recover and rollback instead
- xaHelper.assertPreparedTxCountEquals(1); // tx should still be there
- buildAndInit(); // should cleanup on startup
- xaHelper.assertPreparedTxCountEquals(0);
- assertThat(xaHelper.countInDb()).isEqualTo(0);
- }
-
- @Test
- void testRestoreWithNotificationMissing() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- sinkHelper.close();
- sinkHelper = buildSinkHelper(sinkHelper.getState());
- sinkHelper.emitAndCheckpoint(JdbcTestFixture.CP1);
- xaHelper.assertDbContentsEquals(JdbcTestFixture.CP0, JdbcTestFixture.CP1);
- }
-
- @Test
- void testCommitUponStart() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- sinkHelper.close();
- buildAndInit(
- 0,
- XaFacadeImpl.fromXaDataSource(getMetadata().buildXaDataSource()),
- sinkHelper.getState());
- xaHelper.assertDbContentsEquals(JdbcTestFixture.CP0);
- }
-
- /** RM may return {@link javax.transaction.xa.XAResource#XA_RDONLY XA_RDONLY} error. */
- @Test
- void testEmptyCheckpoint() throws Exception {
- sinkHelper.snapshotState(0);
- }
-
- @Test
- void testTwoCheckpointsComplete1st() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP1);
- JdbcXaSinkTestHelper sinkHelper = this.sinkHelper;
- long checkpointId = JdbcTestFixture.CP0.id;
- sinkHelper.notifyCheckpointComplete(checkpointId);
- xaHelper.cancelAllTx(); // cancel 2nd tx to prevent the following read from being blocked
- xaHelper.assertDbContentsEquals(JdbcTestFixture.CP0);
- }
-
- @Test
- void testTwoCheckpointsComplete2nd() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- sinkHelper.emitAndCheckpoint(JdbcTestFixture.CP1);
- xaHelper.assertDbContentsEquals(
- JdbcTestFixture.CP0,
- JdbcTestFixture.CP1); // "both records should be inserted after the last snapshot
- // completed."
- }
-
- @Test
- void testTwoCheckpointsCompleteBoth() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP1);
- sinkHelper.notifyCheckpointComplete(JdbcTestFixture.CP0.id);
- sinkHelper.notifyCheckpointComplete(JdbcTestFixture.CP1.id);
- xaHelper.assertDbContentsEquals(
- JdbcTestFixture.CP0,
- JdbcTestFixture.CP1); // "both records should be inserted after the last snapshot
- // completed."
- }
-
- @Test
- void testTwoCheckpointsCompleteBothOutOfOrder() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP1);
- sinkHelper.notifyCheckpointComplete(JdbcTestFixture.CP1.id);
- sinkHelper.notifyCheckpointComplete(JdbcTestFixture.CP0.id);
- xaHelper.assertDbContentsEquals(
- JdbcTestFixture.CP0,
- JdbcTestFixture.CP1); // "both records should be inserted after the last snapshot
- // completed."
- }
-
- @Test
- void testRestore() throws Exception {
- sinkHelper.emitAndCheckpoint(JdbcTestFixture.CP0);
- sinkHelper.close();
- sinkHelper = new JdbcXaSinkTestHelper(buildAndInit(), new TestXaSinkStateHandler());
- sinkHelper.emitAndCheckpoint(JdbcTestFixture.CP1);
- xaHelper.assertDbContentsEquals(JdbcTestFixture.CP0, JdbcTestFixture.CP1);
- }
-
- @Test
- void testFailurePropagation() throws Exception {
- /* big enough flush interval to cause error in snapshotState rather than in invoke*/
- sinkHelper =
- new JdbcXaSinkTestHelper(
- buildAndInit(
- Integer.MAX_VALUE,
- XaFacadeImpl.fromXaDataSource(getMetadata().buildXaDataSource())),
- new TestXaSinkStateHandler());
- sinkHelper.emit(TEST_DATA[0]);
- sinkHelper.emit(TEST_DATA[0]); // duplicate
- assertThatThrownBy(() -> sinkHelper.snapshotState(0)).isInstanceOf(Exception.class);
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkH2Test.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkH2Test.java
deleted file mode 100644
index 6574242..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkH2Test.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.connector.jdbc.JdbcTestFixture;
-import org.apache.flink.connector.jdbc.h2.H2XaTestBase;
-import org.apache.flink.connector.jdbc.h2.testutils.H2XaDatabase;
-import org.apache.flink.connector.jdbc.h2.testutils.xa.H2XaDsWrapper;
-import org.apache.flink.connector.jdbc.testutils.DatabaseMetadata;
-
-import org.junit.jupiter.api.Test;
-
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/**
- * {@link JdbcXaSinkFunction} tests using H2 DB. H2 uses MVCC (so we can e.g. count records while
- * transaction is not yet committed). But XA support isn't full, so for some scenarios {@link
- * H2XaDsWrapper wrapper} is used, and for some - Derby.
- */
-@Deprecated
-class JdbcXaSinkH2Test extends JdbcXaSinkTestBase implements H2XaTestBase {
-
- @Override
- public DatabaseMetadata getMetadata() {
- return H2XaDatabase.getMetadata();
- }
-
- @Test
- void testIgnoreDuplicatedNotification() throws Exception {
- sinkHelper.emitAndCheckpoint(JdbcTestFixture.CP0);
- sinkHelper.notifyCheckpointComplete(JdbcTestFixture.CP0.id);
- }
-
- /** RM may return {@link javax.transaction.xa.XAResource#XA_RDONLY XA_RDONLY} error. */
- @Test
- void testEmptyCheckpoint() throws Exception {
- sinkHelper.snapshotState(0);
- }
-
- @Test
- void testHappyFlow() throws Exception {
- sinkHelper.emit(TEST_DATA[0]);
- assertThat(xaHelper.countInDb())
- .as("record should not be inserted before the checkpoint started.")
- .isEqualTo(0);
-
- sinkHelper.snapshotState(Long.MAX_VALUE);
- assertThat(xaHelper.countInDb())
- .as("record should not be inserted before the checkpoint completed.")
- .isEqualTo(0);
-
- sinkHelper.notifyCheckpointComplete(Long.MAX_VALUE);
- assertThat(xaHelper.countInDb())
- .as("record should be inserted after the checkpoint completed.")
- .isEqualTo(1);
- }
-
- @Test
- void testTwoCheckpointsWithoutData() throws Exception {
- JdbcXaSinkTestHelper sinkHelper = this.sinkHelper;
- sinkHelper.snapshotState(1);
- sinkHelper.snapshotState(2);
- assertThat(xaHelper.countInDb()).isEqualTo(0);
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkMigrationTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkMigrationTest.java
deleted file mode 100644
index 5b4ac29..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkMigrationTest.java
+++ /dev/null
@@ -1,211 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.FlinkVersion;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.jdbc.JdbcTestBase;
-import org.apache.flink.connector.jdbc.JdbcTestFixture;
-import org.apache.flink.connector.jdbc.JdbcTestFixture.TestEntry;
-import org.apache.flink.connector.jdbc.derby.DerbyTestBase;
-import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
-import org.apache.flink.streaming.api.operators.StreamSink;
-import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
-import org.apache.flink.util.Preconditions;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Disabled;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.MethodSource;
-
-import javax.transaction.xa.Xid;
-
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import static java.util.Optional.of;
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.CP0;
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.cleanUpDatabasesStatic;
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.initSchema;
-import static org.apache.flink.connector.jdbc.xa.JdbcXaSinkTestBase.buildInitCtx;
-import static org.apache.flink.streaming.util.OperatorSnapshotUtil.readStateHandle;
-import static org.apache.flink.streaming.util.OperatorSnapshotUtil.writeStateHandle;
-
-/** Tests state migration for {@link JdbcXaSinkFunction}. */
-@Deprecated
-class JdbcXaSinkMigrationTest extends JdbcTestBase implements DerbyTestBase {
-
- // write a snapshot:
- // java <CLASS_NAME> <VERSION>
- // or
- // mvn exec:java -Dexec.mainClass="<CLASS_NAME>" -Dexec.args='<VERSION>'
- // -Dexec.classpathScope=test -Dexec.cleanupDaemonThreads=false
- public static void main(String[] args) throws Exception {
- // new DerbyDatabase();
- JdbcXaSinkMigrationTest test = new JdbcXaSinkMigrationTest();
- test.writeSnapshot(parseVersionArg(args));
- }
-
- public static Collection<FlinkVersion> getReadVersions() {
- return Collections.emptyList();
- }
-
- @ParameterizedTest
- @MethodSource("getReadVersions")
- @Disabled // as getReadVersions is empty and fails
- void testCommitFromSnapshot(FlinkVersion readVersion) throws Exception {
- preparePendingTransaction();
- try (OneInputStreamOperatorTestHarness<TestEntry, Object> harness =
- createHarness(buildSink())) {
- harness.initializeState(readStateHandle(getSnapshotPath(readVersion)));
- harness.open();
- }
- try (JdbcXaFacadeTestHelper h =
- new JdbcXaFacadeTestHelper(getMetadata(), JdbcTestFixture.INPUT_TABLE)) {
- h.assertDbContentsEquals(CP0);
- }
- }
-
- @AfterEach
- void cleanUp() throws Exception {
- cancelAllTx();
- }
-
- private void preparePendingTransaction() throws Exception {
- try (JdbcXaSinkTestHelper sinkHelper =
- new JdbcXaSinkTestHelper(buildSink(), new XaSinkStateHandlerImpl())) {
- sinkHelper.getSinkFunction().initializeState(buildInitCtx(false));
- sinkHelper.getSinkFunction().open(new Configuration());
- sinkHelper.emitAndSnapshot(CP0);
- }
- }
-
- private OperatorSubtaskState captureState() throws Exception {
- try (JdbcXaSinkTestHelper sinkHelper =
- new JdbcXaSinkTestHelper(buildSink(), new XaSinkStateHandlerImpl())) {
- try (OneInputStreamOperatorTestHarness<TestEntry, Object> harness =
- createHarness(sinkHelper.getSinkFunction())) {
- harness.initializeEmptyState();
- harness.open();
- sinkHelper.emit(CP0);
- return harness.snapshot(0L, 0L);
- }
- }
- }
-
- private static XidGenerator getXidGenerator() {
- final AtomicInteger txCounter = new AtomicInteger();
- return new XidGenerator() {
- @Override
- public Xid generateXid(JobSubtask subtask, long checkpointId) {
- return new TestXid(txCounter.incrementAndGet(), 0, 0);
- }
-
- @Override
- public boolean belongsToSubtask(Xid xid, JobSubtask subtask) {
- return false;
- }
- };
- }
-
- private String getSnapshotPath(FlinkVersion version) {
- return String.format(
- "src/test/resources/jdbc-exactly-once-sink-migration-%s-snapshot", version);
- }
-
- private static OneInputStreamOperatorTestHarness<TestEntry, Object> createHarness(
- JdbcXaSinkFunction<TestEntry> sink) throws Exception {
- OneInputStreamOperatorTestHarness<TestEntry, Object> harness =
- new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink));
- harness.setup();
- return harness;
- }
-
- private static FlinkVersion parseVersionArg(String[] args) {
- return (args == null || args.length == 0 ? Optional.<String>empty() : of(args[0]))
- .flatMap(FlinkVersion::byCode)
- .orElseThrow(
- () ->
- new IllegalArgumentException(
- "Please specify a version as a 1st parameter. Valid values are: "
- + Arrays.toString(FlinkVersion.values())));
- }
-
- private JdbcXaSinkFunction<TestEntry> buildSink() {
- return JdbcXaSinkTestBase.buildSink(
- getXidGenerator(),
- XaFacadeImpl.fromXaDataSource(getMetadata().buildXaDataSource()),
- new XaSinkStateHandlerImpl(new XaSinkStateSerializer()),
- 1);
- }
-
- private void cancelAllTx() throws Exception {
- try (JdbcXaFacadeTestHelper xa =
- new JdbcXaFacadeTestHelper(getMetadata(), JdbcTestFixture.INPUT_TABLE)) {
- xa.cancelAllTx();
- }
- }
-
- private void writeSnapshot(FlinkVersion flinkVersion) throws Exception {
- String path = getSnapshotPath(flinkVersion);
-
- // Files.createFile(Paths.get(path));/
- Preconditions.checkArgument(
- !Files.exists(Paths.get(path)),
- String.format("snapshot for version %s already exist: %s", flinkVersion, path));
- initSchema(getMetadata());
- try {
- writeStateHandle(captureState(), path);
- } finally {
- cancelAllTx();
- cleanUpDatabasesStatic(getMetadata());
- }
- }
-
- private static class TestXid implements Xid {
- private final int gtrid;
- private final int bqual;
- private final int format;
-
- private TestXid(int gtrid, int bqual, int format) {
- this.gtrid = gtrid;
- this.bqual = bqual;
- this.format = format;
- }
-
- @Override
- public int getFormatId() {
- return format;
- }
-
- @Override
- public byte[] getGlobalTransactionId() {
- return String.valueOf(gtrid).getBytes();
- }
-
- @Override
- public byte[] getBranchQualifier() {
- return String.valueOf(gtrid).getBytes();
- }
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkNoInsertionTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkNoInsertionTest.java
deleted file mode 100644
index b8e477b..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkNoInsertionTest.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.connector.jdbc.JdbcTestFixture;
-import org.apache.flink.connector.jdbc.h2.H2XaTestBase;
-import org.apache.flink.connector.jdbc.h2.testutils.H2XaDatabase;
-import org.apache.flink.connector.jdbc.testutils.DatabaseMetadata;
-
-import org.junit.jupiter.api.Test;
-
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests that data is not inserted ahead of time. */
-@Deprecated
-class JdbcXaSinkNoInsertionTest extends JdbcXaSinkTestBase implements H2XaTestBase {
-
- @Override
- public DatabaseMetadata getMetadata() {
- return H2XaDatabase.getMetadata();
- }
-
- @Test
- void testNoInsertAfterInvoke() throws Exception {
- sinkHelper.emit(TEST_DATA[0]);
- assertThat(xaHelper.countInDb())
- .as("no records should be inserted for incomplete checkpoints.")
- .isEqualTo(0);
- }
-
- @Test
- void testNoInsertAfterSnapshot() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- assertThat(xaHelper.countInDb())
- .as("no records should be inserted for incomplete checkpoints.")
- .isEqualTo(0);
- }
-
- @Test
- public void testNoInsertAfterSinkClose() throws Exception {
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- sinkHelper.close();
- assertThat(xaHelper.countInDb()).isEqualTo(0);
- }
-
- @Test
- void testNoInsertAfterFacadeClose() throws Exception {
- try (XaFacadeImpl xaFacade =
- XaFacadeImpl.fromXaDataSource(getMetadata().buildXaDataSource())) {
- sinkHelper =
- new JdbcXaSinkTestHelper(
- buildAndInit(0, xaFacade), new TestXaSinkStateHandler());
- sinkHelper.emitAndSnapshot(JdbcTestFixture.CP0);
- }
- assertThat(xaHelper.countInDb()).isEqualTo(0);
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkTestBase.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkTestBase.java
deleted file mode 100644
index c7407b0..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkTestBase.java
+++ /dev/null
@@ -1,208 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
-import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
-import org.apache.flink.connector.jdbc.JdbcTestBase;
-import org.apache.flink.connector.jdbc.JdbcTestFixture.TestEntry;
-import org.apache.flink.connector.jdbc.internal.JdbcOutputFormat;
-import org.apache.flink.connector.jdbc.internal.executor.JdbcBatchStatementExecutor;
-import org.apache.flink.core.fs.CloseableRegistry;
-import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
-import org.apache.flink.runtime.state.FunctionInitializationContext;
-import org.apache.flink.runtime.state.StateInitializationContextImpl;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-
-import javax.transaction.xa.Xid;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.stream.Collectors;
-
-import static org.apache.flink.connector.jdbc.JdbcITCase.TEST_ENTRY_JDBC_STATEMENT_BUILDER;
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.INPUT_TABLE;
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.INSERT_TEMPLATE;
-
-/**
- * // todo: javadoc case Base class for {@link JdbcXaSinkFunction} tests. In addition to {@link
- * JdbcTestBase} init it initializes/closes helpers.
- */
-@Deprecated
-abstract class JdbcXaSinkTestBase extends JdbcTestBase {
-
- JdbcXaFacadeTestHelper xaHelper;
- JdbcXaSinkTestHelper sinkHelper;
-
- @BeforeEach
- void initHelpers() throws Exception {
- xaHelper = new JdbcXaFacadeTestHelper(getMetadata(), INPUT_TABLE);
- sinkHelper = buildSinkHelper(createStateHandler());
- }
-
- private XaSinkStateHandler createStateHandler() {
- return new TestXaSinkStateHandler();
- }
-
- @AfterEach
- void closeHelpers() throws Exception {
- if (sinkHelper != null) {
- sinkHelper.close();
- }
- if (xaHelper != null) {
- xaHelper.close();
- }
- try (JdbcXaFacadeTestHelper xa = new JdbcXaFacadeTestHelper(getMetadata(), INPUT_TABLE)) {
- xa.cancelAllTx();
- }
- }
-
- JdbcXaSinkTestHelper buildSinkHelper(XaSinkStateHandler stateHandler) throws Exception {
- return new JdbcXaSinkTestHelper(buildAndInit(0, getXaFacade(), stateHandler), stateHandler);
- }
-
- private XaFacadeImpl getXaFacade() {
- return XaFacadeImpl.fromXaDataSource(getMetadata().buildXaDataSource());
- }
-
- JdbcXaSinkFunction<TestEntry> buildAndInit() throws Exception {
- return buildAndInit(Integer.MAX_VALUE, getXaFacade());
- }
-
- JdbcXaSinkFunction<TestEntry> buildAndInit(int batchInterval, XaFacade xaFacade)
- throws Exception {
- return buildAndInit(batchInterval, xaFacade, createStateHandler());
- }
-
- static JdbcXaSinkFunction<TestEntry> buildAndInit(
- int batchInterval, XaFacade xaFacade, XaSinkStateHandler state) throws Exception {
- JdbcXaSinkFunction<TestEntry> sink =
- buildSink(new SemanticXidGenerator(), xaFacade, state, batchInterval);
- sink.initializeState(buildInitCtx(false));
- sink.setInputType(TypeInformation.of(TestEntry.class), getExecutionConfig(false));
- sink.open(new Configuration());
- return sink;
- }
-
- static JdbcXaSinkFunction<TestEntry> buildSink(
- XidGenerator xidGenerator,
- XaFacade xaFacade,
- XaSinkStateHandler state,
- int batchInterval) {
- JdbcOutputFormat<TestEntry, TestEntry, JdbcBatchStatementExecutor<TestEntry>> format =
- new JdbcOutputFormat<>(
- xaFacade,
- JdbcExecutionOptions.builder()
- .withBatchIntervalMs(batchInterval)
- .withMaxRetries(0)
- .build(),
- () ->
- JdbcBatchStatementExecutor.simple(
- String.format(INSERT_TEMPLATE, INPUT_TABLE),
- TEST_ENTRY_JDBC_STATEMENT_BUILDER));
- JdbcXaSinkFunction<TestEntry> sink =
- new JdbcXaSinkFunction<>(
- format,
- xaFacade,
- xidGenerator,
- state,
- JdbcExactlyOnceOptions.builder().withRecoveredAndRollback(true).build(),
- new XaGroupOpsImpl(xaFacade));
- sink.setRuntimeContext(TEST_RUNTIME_CONTEXT);
- return sink;
- }
-
- static final RuntimeContext TEST_RUNTIME_CONTEXT = getRuntimeContext(new JobID());
-
- static final SinkFunction.Context TEST_SINK_CONTEXT =
- new SinkFunction.Context() {
- @Override
- public long currentProcessingTime() {
- return 0;
- }
-
- @Override
- public long currentWatermark() {
- return 0;
- }
-
- @Override
- public Long timestamp() {
- return 0L;
- }
- };
-
- static class TestXaSinkStateHandler implements XaSinkStateHandler {
- private static final long serialVersionUID = 1L;
-
- private JdbcXaSinkFunctionState stored;
-
- @Override
- public JdbcXaSinkFunctionState load(FunctionInitializationContext context) {
- List<CheckpointAndXid> prepared =
- stored != null
- ? stored.getPrepared().stream()
- .map(CheckpointAndXid::asRestored)
- .collect(Collectors.toList())
- : Collections.emptyList();
- Collection<Xid> hanging =
- stored != null ? stored.getHanging() : Collections.emptyList();
- return JdbcXaSinkFunctionState.of(prepared, hanging);
- }
-
- @Override
- public void store(JdbcXaSinkFunctionState state) {
- stored = state;
- }
-
- JdbcXaSinkFunctionState get() {
- return stored;
- }
-
- @Override
- public String toString() {
- return stored == null ? null : stored.toString();
- }
- }
-
- static StateInitializationContextImpl buildInitCtx(boolean restored) {
- return new StateInitializationContextImpl(
- restored ? 1L : null,
- new DefaultOperatorStateBackend(
- new ExecutionConfig(),
- new CloseableRegistry(),
- new HashMap<>(),
- new HashMap<>(),
- new HashMap<>(),
- new HashMap<>(),
- null),
- null,
- null,
- null);
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkTestHelper.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkTestHelper.java
deleted file mode 100644
index 81bee7d..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/JdbcXaSinkTestHelper.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.connector.jdbc.JdbcTestCheckpoint;
-import org.apache.flink.connector.jdbc.JdbcTestFixture.TestEntry;
-import org.apache.flink.runtime.state.FunctionSnapshotContext;
-import org.apache.flink.util.Preconditions;
-
-import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA;
-import static org.apache.flink.connector.jdbc.xa.JdbcXaSinkTestBase.TEST_SINK_CONTEXT;
-
-@Deprecated
-class JdbcXaSinkTestHelper implements AutoCloseable {
-
- private final JdbcXaSinkFunction<TestEntry> sink;
- private final XaSinkStateHandler state;
-
- JdbcXaSinkTestHelper(JdbcXaSinkFunction<TestEntry> sink, XaSinkStateHandler stateHandler) {
- this.sink = Preconditions.checkNotNull(sink);
- this.state = Preconditions.checkNotNull(stateHandler);
- }
-
- void emitAndCheckpoint(JdbcTestCheckpoint cp) throws Exception {
- emitAndSnapshot(cp);
- notifyCheckpointComplete(cp.id);
- }
-
- void emitAndSnapshot(JdbcTestCheckpoint checkpoint) throws Exception {
- emit(checkpoint);
- snapshotState(checkpoint.id);
- }
-
- void emit(JdbcTestCheckpoint checkpoint) throws java.io.IOException {
- for (int i = 0; i < checkpoint.dataItemsIdx.length; i++) {
- emit(TEST_DATA[checkpoint.dataItemsIdx[i]]);
- }
- }
-
- void emit(TestEntry entry) throws java.io.IOException {
- sink.invoke(entry, TEST_SINK_CONTEXT);
- }
-
- @Override
- public void close() throws Exception {
- sink.close();
- }
-
- void notifyCheckpointComplete(long checkpointId) {
- sink.notifyCheckpointComplete(checkpointId);
- }
-
- void snapshotState(long id) throws Exception {
- sink.snapshotState(getSnapshotContext(id));
- }
-
- private static FunctionSnapshotContext getSnapshotContext(long id) {
- return new FunctionSnapshotContext() {
- @Override
- public long getCheckpointId() {
- return id;
- }
-
- @Override
- public long getCheckpointTimestamp() {
- return 0;
- }
- };
- }
-
- JdbcXaSinkFunction<TestEntry> getSinkFunction() {
- return sink;
- }
-
- XaSinkStateHandler getState() {
- return state;
- }
-}
diff --git a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/SemanticXidGeneratorTest.java b/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/SemanticXidGeneratorTest.java
deleted file mode 100644
index 70c2631..0000000
--- a/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/xa/SemanticXidGeneratorTest.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.xa;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.connector.jdbc.datasource.transactions.xa.xid.XidImpl;
-
-import org.junit.jupiter.api.Test;
-
-import javax.transaction.xa.Xid;
-
-import java.util.HashSet;
-import java.util.Set;
-import java.util.function.Function;
-
-import static org.apache.flink.connector.jdbc.xa.JdbcXaSinkTestBase.TEST_RUNTIME_CONTEXT;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Simple uniqueness tests for the {@link SemanticXidGenerator}. */
-@Deprecated
-class SemanticXidGeneratorTest {
- private static final int COUNT = 100_000;
-
- @Test
- void testXidsUniqueAmongCheckpoints() {
- SemanticXidGenerator xidGenerator = new SemanticXidGenerator();
- xidGenerator.open();
- checkUniqueness(
- checkpoint ->
- xidGenerator.generateXid(JobSubtask.of(TEST_RUNTIME_CONTEXT), checkpoint));
- }
-
- @Test
- void testXidsUniqueAmongJobs() {
- long checkpointId = 1L;
- SemanticXidGenerator generator = new SemanticXidGenerator();
- checkUniqueness(
- unused -> {
- generator.open();
- RuntimeContext context = JdbcXaSinkTestBase.getRuntimeContext(new JobID());
- return generator.generateXid(JobSubtask.of(context), checkpointId);
- });
- }
-
- private void checkUniqueness(Function<Integer, Xid> generate) {
- Set<Xid> generated = new HashSet<>();
- for (int i = 0; i < COUNT; i++) {
- // We "drop" the branch id because uniqueness of gtrid is important
- generated.add(new XidImpl(0, generate.apply(i).getGlobalTransactionId(), new byte[0]));
- }
- assertThat(generated).hasSize(COUNT);
- }
-}
diff --git a/tools/releasing/shared b/tools/releasing/shared
deleted file mode 160000
index 45bb68f..0000000
--- a/tools/releasing/shared
+++ /dev/null
@@ -1 +0,0 @@
-Subproject commit 45bb68f550874ec398b1da344dd4d82e703b4be1