ruanhang1993 commented on code in PR #3916: URL: https://github.com/apache/flink-cdc/pull/3916#discussion_r2015557747
########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/src/main/java/org/apache/flink/cdc/connectors/jdbc/config/JdbcSinkConfig.java: ########## @@ -0,0 +1,186 @@ +/* + * 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.cdc.connectors.jdbc.config; + +import java.io.Serializable; +import java.time.Duration; +import java.util.Properties; + +/** Generic configuration class for JDBC-like sinks. */ +public class JdbcSinkConfig implements Serializable { + private final String connUrl; + private final String hostname; + private final int port; Review Comment: `hostname` and `port` are already contained in `connUrl`. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc/pom.xml: ########## @@ -0,0 +1,124 @@ +<?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/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-jdbc-parent</artifactId> + <version>${revision}</version> + </parent> + + <packaging>jar</packaging> + + <artifactId>flink-cdc-pipeline-connector-jdbc</artifactId> + + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-jdbc-core</artifactId> + <version>${project.version}</version> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-jdbc-mysql</artifactId> + <version>${project.version}</version> + </dependency> + + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-test-util</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-composer</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-mysql</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-mysql-cdc</artifactId> + <version>${project.version}</version> + <scope>test</scope> + <type>test-jar</type> + </dependency> + + <dependency> + <groupId>org.testcontainers</groupId> + <artifactId>mysql</artifactId> + <version>${testcontainers.version}</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> Review Comment: We need to exclude some flink dependencies here. For example, flink-core, flink-connector-base. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.cdc.connectors.jdbc.dialect.JdbcSinkDialectFactory: ########## @@ -0,0 +1,16 @@ +# 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.cdc.connectors.jdbc.mysql.factory.JdbcMySqlSinkDialectFactory Review Comment: The services files should be placed at the sub-module. And they will be merged by the transfomers in maven shade plugin. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/pom.xml: ########## @@ -0,0 +1,40 @@ +<?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/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-jdbc-parent</artifactId> + <version>${revision}</version> + </parent> + + <artifactId>flink-cdc-pipeline-connector-jdbc-core</artifactId> + + <properties> + </properties> Review Comment: Remove useless properties setting. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/factory/JdbcDataSinkFactory.java: ########## @@ -0,0 +1,162 @@ +/* + * 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.cdc.connectors.jdbc.factory; + +import org.apache.flink.cdc.common.configuration.ConfigOption; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.factories.DataSinkFactory; +import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.sink.DataSink; +import org.apache.flink.cdc.connectors.base.utils.OptionUtils; +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; +import org.apache.flink.cdc.connectors.jdbc.dialect.JdbcSinkDialectFactory; +import org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions; +import org.apache.flink.cdc.connectors.jdbc.sink.JdbcDataSink; +import org.apache.flink.configuration.ConfigurationUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.ServiceConfigurationError; +import java.util.ServiceLoader; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.CONNECTION_POOL_SIZE; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.CONNECT_MAX_RETRIES; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.CONNECT_TIMEOUT; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.DIALECT; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.DRIVER_CLASS_NAME; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.HOSTNAME; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.JDBC_PROPERTIES_PROP_PREFIX; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.PASSWORD; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.PORT; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.SERVER_TIME_ZONE; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.USERNAME; +import static org.apache.flink.cdc.connectors.jdbc.options.JdbcSinkOptions.WRITE_BATCH_SIZE; + +/** A {@link DataSinkFactory} for creating JDBC sinks. */ +public class JdbcDataSinkFactory implements DataSinkFactory { + + public static final String IDENTIFIER = "jdbc"; + + private static final Logger LOG = LoggerFactory.getLogger(JdbcDataSinkFactory.class); + + @Override + public DataSink createDataSink(Context context) { + FactoryHelper.createFactoryHelper(this, context) + .validateExcept(JDBC_PROPERTIES_PROP_PREFIX); + + String dialect = context.getFactoryConfiguration().get(DIALECT); + + List<JdbcSinkDialectFactory<JdbcSinkConfig>> dialectFactories = + discoverDialectFactories(getClass().getClassLoader()); + + JdbcSinkDialectFactory<JdbcSinkConfig> dialectFactory = + dialectFactories.stream() + .filter(d -> dialect.equalsIgnoreCase(d.identifier())) + .findFirst() + .orElseThrow( + () -> + new IllegalArgumentException( + "Unknown Jdbc sink dialect " + + dialect + + ". Supported dialects are: " + + dialectFactories.stream() + .map( + JdbcSinkDialectFactory + ::identifier) + .collect(Collectors.toList()))); + + final Configuration config = context.getFactoryConfiguration(); + JdbcSinkConfig.Builder<?> builder = new JdbcSinkConfig.Builder<>(); + + config.getOptional(HOSTNAME).ifPresent(builder::hostname); + config.getOptional(PORT).ifPresent(builder::port); + config.getOptional(USERNAME).ifPresent(builder::username); + config.getOptional(PASSWORD).ifPresent(builder::password); + + builder.serverTimeZone(config.getOptional(SERVER_TIME_ZONE).orElse("UTC")); + builder.connectTimeout(config.get(CONNECT_TIMEOUT)); + builder.connectionPoolSize(config.get(CONNECTION_POOL_SIZE)); + builder.connectMaxRetries(config.get(CONNECT_MAX_RETRIES)); + builder.writeBatchSize(config.get(WRITE_BATCH_SIZE)); + builder.driverClassName(config.get(DRIVER_CLASS_NAME)); + + Properties properties = new Properties(); + Map<String, String> jdbcProperties = + JdbcSinkOptions.getPropertiesByPrefix(config, JDBC_PROPERTIES_PROP_PREFIX); + properties.putAll(jdbcProperties); + builder.jdbcProperties(properties); + + if (LOG.isInfoEnabled()) { + OptionUtils.printOptions( + IDENTIFIER, ConfigurationUtils.hideSensitiveValues(config.toMap())); + } + + JdbcSinkConfig jdbcSinkConfig = builder.build(); + return new JdbcDataSink(dialectFactory.createDialect(jdbcSinkConfig), jdbcSinkConfig); + } + + @Override + public String identifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(DIALECT); + options.add(HOSTNAME); + options.add(PORT); Review Comment: Use url instead of dialect, hostname and port. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/sink/JdbcDataSink.java: ########## @@ -0,0 +1,75 @@ +/* + * 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.cdc.connectors.jdbc.sink; + +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.sink.DataSink; +import org.apache.flink.cdc.common.sink.EventSinkProvider; +import org.apache.flink.cdc.common.sink.FlinkSinkProvider; +import org.apache.flink.cdc.common.sink.MetadataApplier; +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; +import org.apache.flink.cdc.connectors.jdbc.dialect.JdbcSinkDialect; +import org.apache.flink.cdc.connectors.jdbc.sink.v2.EventRecordSerializationSchema; +import org.apache.flink.cdc.connectors.jdbc.sink.v2.JdbcSink; +import org.apache.flink.cdc.connectors.jdbc.sink.v2.JdbcSinkBuilder; +import org.apache.flink.connector.jdbc.JdbcConnectionOptions; +import org.apache.flink.connector.jdbc.JdbcExecutionOptions; + +import java.io.Serializable; + +/** Writing data to JDBC-like sinks. */ +public class JdbcDataSink implements DataSink, Serializable { + private final JdbcSinkDialect dialect; + private final JdbcSinkConfig sinkConfig; + + public JdbcDataSink(JdbcSinkDialect dialect, JdbcSinkConfig sinkConfig) { + sinkConfig.setConnUrl(dialect); + this.dialect = dialect; + this.sinkConfig = sinkConfig; + } + + @Override + public EventSinkProvider getEventSinkProvider() { + JdbcConnectionOptions connectionOptions = + new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() + .withUrl(sinkConfig.getConnUrl()) + .withUsername(sinkConfig.getUsername()) + .withPassword(sinkConfig.getPassword()) + .withDriverName(sinkConfig.getDriverClassName()) + .build(); + + JdbcSink<Event> jdbcSink = + new JdbcSinkBuilder<Event>() + .withExecutionOptions( + JdbcExecutionOptions.builder() Review Comment: Please add `JdbcExecutionOptions` options in factory (batchIntervalMs, batchSize, maxRetries). ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/connection/ConnectionPools.java: ########## @@ -0,0 +1,30 @@ +/* + * 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.cdc.connectors.jdbc.connection; + +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; + +/** Factory class for creating connection pools. */ +public interface ConnectionPools<P, C extends JdbcSinkConfig> { + + /** + * Create a new pool if the pool does not exist, or get existing connection pool with given + * identifier. + */ + P getOrCreateConnectionPool(ConnectionPoolId poolId, C sourceConfig); Review Comment: ```suggestion P getOrCreateConnectionPool(ConnectionPoolId poolId, C sinkConfig); ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory: ########## @@ -0,0 +1,16 @@ +# 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.cdc.connectors.jdbc.factory.JdbcDataSinkFactory Review Comment: The services files should be placed at the sub-module. And they will be merged by the transfomers in maven shade plugin. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/connection/ConnectionPools.java: ########## @@ -0,0 +1,30 @@ +/* + * 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.cdc.connectors.jdbc.connection; + +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; + +/** Factory class for creating connection pools. */ +public interface ConnectionPools<P, C extends JdbcSinkConfig> { Review Comment: Does this interface need the close method? ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/connection/JdbcConnectionPoolFactory.java: ########## @@ -0,0 +1,76 @@ +/* + * 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.cdc.connectors.jdbc.connection; + +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; + +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; + +import java.io.Serializable; + +/** A factory class for creating pooled {@link HikariDataSource}. */ +public class JdbcConnectionPoolFactory implements Serializable { + + public static final JdbcConnectionPoolFactory INSTANCE = new JdbcConnectionPoolFactory(); + + public static final String CONNECTION_POOL_PREFIX = "connection-pool-"; + public static final String SERVER_TIMEZONE_KEY = "serverTimezone"; + public static final int MINIMUM_POOL_SIZE = 1; + + public HikariDataSource createPooledDataSource(JdbcSinkConfig sinkConfig) { + final HikariConfig config = new HikariConfig(); + + String hostName = sinkConfig.getHostname(); + int port = sinkConfig.getPort(); + + config.setPoolName(CONNECTION_POOL_PREFIX + hostName + ":" + port); + config.setJdbcUrl(sinkConfig.getConnUrl()); + config.setUsername(sinkConfig.getUsername()); + config.setPassword(sinkConfig.getPassword()); + config.setMinimumIdle(MINIMUM_POOL_SIZE); + config.setMaximumPoolSize(sinkConfig.getConnectionPoolSize()); + config.setConnectionTimeout(sinkConfig.getConnectTimeout().toMillis()); + config.setDriverClassName(sinkConfig.getDriverClassName()); + + // note: the following properties should be optional (only applied to MySQL) + config.addDataSourceProperty(SERVER_TIMEZONE_KEY, sinkConfig.getServerTimeZone()); + // optional optimization configurations for pooled DataSource + config.addDataSourceProperty("cachePrepStmts", "true"); + config.addDataSourceProperty("prepStmtCacheSize", "250"); + config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048"); + + return new HikariDataSource(config); + } + + /** + * The reuse strategy of connection pools. In most situations, connections to different + * databases in same instance (which means same host and port) can reuse same connection pool. + * However, in some situations when different databases in same instance cannot reuse same + * connection pool to connect, such as postgresql, this method should be overridden. + */ + public ConnectionPoolId getPoolId( Review Comment: This method should be static and we do not need to maintain the field for `JdbcConnectionPoolFactory`. And JdbcConnectionPoolFactory do not need to be `Serializable`. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc/pom.xml: ########## @@ -0,0 +1,124 @@ +<?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/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-jdbc-parent</artifactId> + <version>${revision}</version> + </parent> + + <packaging>jar</packaging> + + <artifactId>flink-cdc-pipeline-connector-jdbc</artifactId> + + <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-jdbc-core</artifactId> + <version>${project.version}</version> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-jdbc-mysql</artifactId> + <version>${project.version}</version> + </dependency> + + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-test-util</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-composer</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-cdc-pipeline-connector-mysql</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-mysql-cdc</artifactId> + <version>${project.version}</version> + <scope>test</scope> + <type>test-jar</type> + </dependency> + + <dependency> + <groupId>org.testcontainers</groupId> + <artifactId>mysql</artifactId> + <version>${testcontainers.version}</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> + <filters combine.children="append"> + <filter combine.children="append"> + <artifact>*:*</artifact> + <excludes combine.children="append"> + <exclude>META-INF/*.SF</exclude> + <exclude>META-INF/*.DSA</exclude> + <exclude>META-INF/*.RSA</exclude> + </excludes> + </filter> + </filters> Review Comment: Transformers are needed here. ``` <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"/> </transformers> ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/connection/JdbcConnectionPoolFactory.java: ########## @@ -0,0 +1,76 @@ +/* + * 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.cdc.connectors.jdbc.connection; + +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; + +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; + +import java.io.Serializable; + +/** A factory class for creating pooled {@link HikariDataSource}. */ +public class JdbcConnectionPoolFactory implements Serializable { + + public static final JdbcConnectionPoolFactory INSTANCE = new JdbcConnectionPoolFactory(); + + public static final String CONNECTION_POOL_PREFIX = "connection-pool-"; + public static final String SERVER_TIMEZONE_KEY = "serverTimezone"; + public static final int MINIMUM_POOL_SIZE = 1; + + public HikariDataSource createPooledDataSource(JdbcSinkConfig sinkConfig) { + final HikariConfig config = new HikariConfig(); + + String hostName = sinkConfig.getHostname(); + int port = sinkConfig.getPort(); + + config.setPoolName(CONNECTION_POOL_PREFIX + hostName + ":" + port); + config.setJdbcUrl(sinkConfig.getConnUrl()); + config.setUsername(sinkConfig.getUsername()); + config.setPassword(sinkConfig.getPassword()); + config.setMinimumIdle(MINIMUM_POOL_SIZE); + config.setMaximumPoolSize(sinkConfig.getConnectionPoolSize()); + config.setConnectionTimeout(sinkConfig.getConnectTimeout().toMillis()); Review Comment: Add `DriverManager.getDrivers();` before `setDriverClassName` to avoid deadlock when loading driver classes. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/config/JdbcSinkConfig.java: ########## @@ -0,0 +1,198 @@ +/* + * 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.cdc.connectors.jdbc.config; + +import org.apache.flink.cdc.connectors.jdbc.dialect.JdbcSinkDialect; + +import java.io.Serializable; +import java.time.Duration; +import java.util.Properties; + +/** Generic configuration class for JDBC-like sinks. */ +public class JdbcSinkConfig implements Serializable { Review Comment: Add equals and hashcode methods. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/config/JdbcSinkConfig.java: ########## @@ -0,0 +1,198 @@ +/* + * 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.cdc.connectors.jdbc.config; + +import org.apache.flink.cdc.connectors.jdbc.dialect.JdbcSinkDialect; + +import java.io.Serializable; +import java.time.Duration; +import java.util.Properties; + +/** Generic configuration class for JDBC-like sinks. */ +public class JdbcSinkConfig implements Serializable { + private final String hostname; + private final int port; Review Comment: `hostname` and `port` are already contained in `connUrl`. Delete them. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/dialect/JdbcSinkDialect.java: ########## @@ -0,0 +1,310 @@ +/* + * 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.cdc.connectors.jdbc.dialect; + +import org.apache.flink.cdc.common.event.AddColumnEvent; +import org.apache.flink.cdc.common.event.AlterColumnTypeEvent; +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.DropColumnEvent; +import org.apache.flink.cdc.common.event.DropTableEvent; +import org.apache.flink.cdc.common.event.RenameColumnEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.event.TruncateTableEvent; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.common.utils.Preconditions; +import org.apache.flink.cdc.common.utils.StringUtils; +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; +import org.apache.flink.cdc.connectors.jdbc.connection.JdbcConnectionFactory; +import org.apache.flink.cdc.connectors.jdbc.connection.JdbcConnectionPoolFactory; +import org.apache.flink.cdc.connectors.jdbc.exception.JdbcSinkDialectException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Serializable; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +/** An abstract base class for various JDBC-like sinks. */ +public abstract class JdbcSinkDialect implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(JdbcSinkDialect.class); + + private final String dialectName; + + protected final JdbcConnectionFactory connectionFactory; + + public JdbcSinkDialect( + String dialectName, JdbcSinkConfig sinkConfig, JdbcConnectionPoolFactory poolFactory) { + this.dialectName = dialectName; + this.connectionFactory = new JdbcConnectionFactory(sinkConfig, poolFactory); + } + + // Rewritable methods that could be overwritten by subclasses. + public String identifier() { + return dialectName; + } + + public String getJdbcUrl(JdbcSinkConfig sinkConfig) { + Properties jdbcProperties = sinkConfig.getJdbcProperties(); + StringBuilder url = new StringBuilder("jdbc:" + dialectName + "://"); + url.append(sinkConfig.getHostname()).append(":").append(sinkConfig.getPort()); + url.append("?serverTimezone=").append(sinkConfig.getServerTimeZone()); + + if (jdbcProperties != null && !jdbcProperties.isEmpty()) { + for (String key : jdbcProperties.stringPropertyNames()) { + url.append("&").append(key).append("=").append(jdbcProperties.getProperty(key)); + } + } + return url.toString(); + } + + protected abstract String buildUpsertSql(TableId tableId, Schema schema); Review Comment: Add more java doc about thest methods , refer to `org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialect`. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/connection/JdbcConnectionPools.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.cdc.connectors.jdbc.connection; + +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; +import org.apache.flink.util.FlinkRuntimeException; + +import com.zaxxer.hikari.HikariDataSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.HashMap; +import java.util.Map; + +/** A Jdbc Connection pools implementation. */ +public class JdbcConnectionPools implements ConnectionPools<HikariDataSource, JdbcSinkConfig> { + + private static final Logger LOG = LoggerFactory.getLogger(JdbcConnectionPools.class); + + private static JdbcConnectionPools instance; + private final Map<ConnectionPoolId, HikariDataSource> pools = new HashMap<>(); + private static final Map<String, JdbcConnectionPoolFactory> POOL_FACTORY_MAP = new HashMap<>(); Review Comment: Do we really need this map? ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-jdbc-parent/flink-cdc-pipeline-connector-jdbc-core/src/main/java/org/apache/flink/cdc/connectors/jdbc/dialect/JdbcSinkDialect.java: ########## @@ -0,0 +1,310 @@ +/* + * 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.cdc.connectors.jdbc.dialect; + +import org.apache.flink.cdc.common.event.AddColumnEvent; +import org.apache.flink.cdc.common.event.AlterColumnTypeEvent; +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.DropColumnEvent; +import org.apache.flink.cdc.common.event.DropTableEvent; +import org.apache.flink.cdc.common.event.RenameColumnEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.event.TruncateTableEvent; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.common.utils.Preconditions; +import org.apache.flink.cdc.common.utils.StringUtils; +import org.apache.flink.cdc.connectors.jdbc.config.JdbcSinkConfig; +import org.apache.flink.cdc.connectors.jdbc.connection.JdbcConnectionFactory; +import org.apache.flink.cdc.connectors.jdbc.connection.JdbcConnectionPoolFactory; +import org.apache.flink.cdc.connectors.jdbc.exception.JdbcSinkDialectException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Serializable; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +/** An abstract base class for various JDBC-like sinks. */ +public abstract class JdbcSinkDialect implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(JdbcSinkDialect.class); + + private final String dialectName; + + protected final JdbcConnectionFactory connectionFactory; + + public JdbcSinkDialect( + String dialectName, JdbcSinkConfig sinkConfig, JdbcConnectionPoolFactory poolFactory) { + this.dialectName = dialectName; + this.connectionFactory = new JdbcConnectionFactory(sinkConfig, poolFactory); + } + + // Rewritable methods that could be overwritten by subclasses. + public String identifier() { + return dialectName; + } + + public String getJdbcUrl(JdbcSinkConfig sinkConfig) { + Properties jdbcProperties = sinkConfig.getJdbcProperties(); + StringBuilder url = new StringBuilder("jdbc:" + dialectName + "://"); + url.append(sinkConfig.getHostname()).append(":").append(sinkConfig.getPort()); + url.append("?serverTimezone=").append(sinkConfig.getServerTimeZone()); + + if (jdbcProperties != null && !jdbcProperties.isEmpty()) { + for (String key : jdbcProperties.stringPropertyNames()) { + url.append("&").append(key).append("=").append(jdbcProperties.getProperty(key)); + } + } + return url.toString(); + } + + protected abstract String buildUpsertSql(TableId tableId, Schema schema); + + protected abstract String buildDeleteSql(TableId tableId, Schema schema); + + protected abstract String buildCreateTableSql( + TableId tableId, Schema schema, boolean ignoreIfExists); + + protected abstract String buildAlterAddColumnsSql( + TableId tableId, List<AddColumnEvent.ColumnWithPosition> addedColumns); + + protected abstract String buildRenameColumnSql(TableId tableId, String oldName, String newName); + + protected abstract String buildDropColumnSql(TableId tableId, String column); + + protected abstract String buildAlterColumnTypeSql( + TableId tableId, String columnName, DataType columnType); + + protected abstract String buildTruncateTableSql(TableId tableId); + + protected abstract String buildDropTableSql(TableId tableId, boolean ignoreIfNotExist); + + public abstract void close() throws JdbcSinkDialectException; + + // Final methods that are not meant to be overwritten. + public final void createTable(CreateTableEvent createTableEvent, boolean ignoreIfExists) + throws JdbcSinkDialectException { + TableId tableId = createTableEvent.tableId(); + checkTableIdArguments(tableId); + + Schema schema = createTableEvent.getSchema(); + String createTableSql = buildCreateTableSql(tableId, schema, ignoreIfExists); + + try { + executeUpdate(createTableSql); + LOG.info( + "Successfully created table `{}`.`{}`. Raw SQL: {}", + tableId.getSchemaName(), + tableId.getTableName(), + createTableSql); + } catch (SQLException e) { + throw new JdbcSinkDialectException( + String.format( + "Failed to create table `%s`.`%s`. Raw SQL: %s", + tableId.getSchemaName(), tableId.getTableName(), createTableSql), + e); + } + } + + public final void addColumn(AddColumnEvent addColumnEvent) throws JdbcSinkDialectException { + TableId tableId = addColumnEvent.tableId(); + checkTableIdArguments(tableId); + + Preconditions.checkArgument( + !addColumnEvent.getAddedColumns().isEmpty(), "Added columns should not be empty."); + + String addColumnSql = buildAlterAddColumnsSql(tableId, addColumnEvent.getAddedColumns()); + try { + executeUpdate(addColumnSql); + LOG.info( + "Successfully added column in table `{}`.`{}`. Raw SQL: {}", + tableId.getSchemaName(), + tableId.getTableName(), + addColumnSql); + } catch (Exception e) { + throw new JdbcSinkDialectException( + String.format( + "Failed to add column in table `%s`.`%s`. Raw SQL: %s", + tableId.getSchemaName(), tableId.getTableName(), addColumnSql), + e); + } + } + + public final void dropColumn(DropColumnEvent dropColumnEvent) throws JdbcSinkDialectException { + TableId tableId = dropColumnEvent.tableId(); + checkTableIdArguments(tableId); + + List<String> droppedColumnNames = dropColumnEvent.getDroppedColumnNames(); + Preconditions.checkArgument( + !droppedColumnNames.isEmpty(), "Dropped columns should not be empty."); + + droppedColumnNames.forEach( + column -> { + String dropColumnSql = buildDropColumnSql(tableId, column); + try { + executeUpdate(dropColumnSql); + LOG.info( + "Successfully dropped column in table `{}`.`{}`. Raw SQL: {}", + tableId.getSchemaName(), + tableId.getTableName(), + dropColumnSql); + } catch (Exception e) { + throw new JdbcSinkDialectException( + String.format( + "Failed to drop column in table `%s`.`%s`. Raw SQL: %s", + tableId.getSchemaName(), + tableId.getTableName(), + dropColumnSql), + e); + } + }); + } + + public final void renameColumn(RenameColumnEvent renameColumnEvent) + throws JdbcSinkDialectException { + TableId tableId = renameColumnEvent.tableId(); + checkTableIdArguments(tableId); + + Map<String, String> nameMapping = renameColumnEvent.getNameMapping(); + for (Map.Entry<String, String> entry : nameMapping.entrySet()) { + String renameColumnSql = + buildRenameColumnSql(tableId, entry.getKey(), entry.getValue()); + try { + executeUpdate(renameColumnSql); + LOG.info( + "Successfully renamed column in table `{}`.`{}`. Raw SQL: {}", + tableId.getSchemaName(), + tableId.getTableName(), + renameColumnSql); + } catch (Exception e) { + throw new JdbcSinkDialectException( + String.format( + "Failed to rename column in table `%s`.`%s`. Raw SQL: %s", + tableId.getSchemaName(), tableId.getTableName(), renameColumnSql), + e); + } + } + } + + public final void alterColumnType(AlterColumnTypeEvent alterColumnTypeEvent) + throws JdbcSinkDialectException { + TableId tableId = alterColumnTypeEvent.tableId(); + checkTableIdArguments(tableId); + + Map<String, DataType> typeMapping = alterColumnTypeEvent.getTypeMapping(); + for (Map.Entry<String, DataType> entry : typeMapping.entrySet()) { + String alterColumnTypeSql = + buildAlterColumnTypeSql(tableId, entry.getKey(), entry.getValue()); + + try { + executeUpdate(alterColumnTypeSql); + LOG.info( + "Successfully altered column type in table `{}`.`{}`. Raw SQL: {}", + tableId.getSchemaName(), + tableId.getTableName(), + alterColumnTypeSql); + } catch (Exception e) { + throw new JdbcSinkDialectException( + String.format( + "Failed to alter column type in table `%s`.`%s`. Raw SQL: %s", + tableId.getSchemaName(), + tableId.getTableName(), + alterColumnTypeSql), + e); + } + } + } + + public final void truncateTable(TruncateTableEvent truncateTableEvent) + throws JdbcSinkDialectException { + TableId tableId = truncateTableEvent.tableId(); + checkTableIdArguments(tableId); + + String truncateTableSql = buildTruncateTableSql(tableId); + + try { + executeUpdate(truncateTableSql); + LOG.info( + "Successfully truncated table `{}`.`{}`. Raw SQL: {}", + tableId.getSchemaName(), + tableId.getTableName(), + truncateTableSql); + } catch (SQLException e) { + throw new JdbcSinkDialectException( + String.format( + "Failed to truncate table `%s`.`%s`. Raw SQL: %s", + tableId.getSchemaName(), tableId.getTableName(), truncateTableSql), + e); + } + } + + public final void dropTable(DropTableEvent dropTableEvent, boolean ignoreIfNotExist) + throws JdbcSinkDialectException { + TableId tableId = dropTableEvent.tableId(); + checkTableIdArguments(tableId); + + String dropTableSql = buildDropTableSql(tableId, ignoreIfNotExist); + + try { + executeUpdate(dropTableSql); + LOG.info( + "Successfully dropped table `{}`.`{}`. Raw SQL: {}", + tableId.getSchemaName(), + tableId.getTableName(), + dropTableSql); + } catch (SQLException e) { + throw new JdbcSinkDialectException( + String.format( + "Failed to drop table `%s`.`%s`. Raw SQL: %s", + tableId.getSchemaName(), tableId.getTableName(), dropTableSql), + e); + } + } + + public final String getUpsertStatement(TableId tableId, Schema schema) { Review Comment: What's the difference between `buildUpsertSql`and `getUpsertStatement`? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
