diff --git a/flink-cdc-common/pom.xml b/flink-cdc-common/pom.xml index 80237d26def..99ae43f2e53 100644 --- a/flink-cdc-common/pom.xml +++ b/flink-cdc-common/pom.xml @@ -27,6 +27,28 @@ limitations under the License. flink-cdc-common + + + + mysql + mysql-connector-java + 8.0.27 + test + + + org.testcontainers + mysql + ${testcontainers.version} + test + + + org.testcontainers + junit-jupiter + ${testcontainers.version} + test + + + diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscoverer.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscoverer.java new file mode 100644 index 00000000000..bc1ad2c86a1 --- /dev/null +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscoverer.java @@ -0,0 +1,239 @@ +/* + * 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.common.source.discover; + +import org.apache.flink.cdc.common.configuration.ConfigOption; +import org.apache.flink.cdc.common.configuration.ConfigOptions; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.event.TableId; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.Connection; +import java.sql.Driver; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.LinkedHashSet; +import java.util.Properties; +import java.util.ServiceLoader; +import java.util.Set; + +/** A {@link TableDiscoverer} that reads the list of subscribed tables from a JDBC database. */ +public class JdbcTableDiscoverer implements TableDiscoverer { + + private static final long serialVersionUID = 1L; + + private static final Logger LOG = LoggerFactory.getLogger(JdbcTableDiscoverer.class); + + public static final ConfigOption JDBC_URL = + ConfigOptions.key("table.discoverer.jdbc.url") + .stringType() + .noDefaultValue() + .withDescription("The JDBC connection URL for the table discovery database."); + + public static final ConfigOption USERNAME = + ConfigOptions.key("table.discoverer.jdbc.username") + .stringType() + .noDefaultValue() + .withDescription("The JDBC username for the table discovery database."); + + public static final ConfigOption PASSWORD = + ConfigOptions.key("table.discoverer.jdbc.password") + .stringType() + .noDefaultValue() + .withDescription("The JDBC password for the table discovery database."); + + public static final ConfigOption SUBSCRIBE_QUERY = + ConfigOptions.key("table.discoverer.jdbc.subscribe-query") + .stringType() + .noDefaultValue() + .withDescription( + "Custom SELECT statement used to discover subscribed tables. When set, " + + "this takes priority over the shared-table options. Column #1 of " + + "every row must be a fully-qualified table name."); + + public static final ConfigOption TABLE_NAME = + ConfigOptions.key("table.discoverer.jdbc.table-name") + .stringType() + .noDefaultValue() + .withDescription( + "The shared subscription table that stores subscription entries for " + + "one or more CDC jobs. Required in shared-table mode."); + + public static final ConfigOption COLUMN_NAME = + ConfigOptions.key("table.discoverer.jdbc.column-name") + .stringType() + .defaultValue("subscribe_table_name") + .withDescription( + "The column name in the subscription table that contains the " + + "fully-qualified table names to subscribe to."); + + public static final ConfigOption SUBSCRIBE_ID_COLUMN = + ConfigOptions.key("table.discoverer.jdbc.subscribe-id-column") + .stringType() + .defaultValue("subscribe_id") + .withDescription( + "The column name in the subscription table that holds the " + + "subscription-set identifier used for filtering."); + + public static final ConfigOption SUBSCRIBE_ID = + ConfigOptions.key("table.discoverer.jdbc.subscribe-id") + .stringType() + .noDefaultValue() + .withDescription( + "The current subscription-set identifier. Required in shared-table " + + "mode; rows whose subscribe-id column matches this value are " + + "discovered as subscribed tables."); + + /** Compiled SQL to execute on every {@link #discover()} call. */ + private transient String sql; + + /** When non-null, the discoverer runs in shared-table mode and binds this as parameter #1. */ + private transient String subscribeId; + + private transient Connection connection; + + @Override + public void open(Context context) throws Exception { + Configuration config = context.getConfiguration(); + ClassLoader userCodeClassLoader = context.getUserCodeClassLoader(); + + String jdbcUrl = requireNonEmpty(config, JDBC_URL); + String username = requireNonEmpty(config, USERNAME); + String password = requireNonEmpty(config, PASSWORD); + + String subscribeQuery = config.get(SUBSCRIBE_QUERY); + if (subscribeQuery != null && !subscribeQuery.isEmpty()) { + // Mode A — custom query takes priority. Filter options are intentionally ignored. + this.sql = subscribeQuery; + this.subscribeId = null; + LOG.info( + "JdbcTableDiscoverer running in custom-query mode. URL='{}', query='{}'.", + jdbcUrl, + subscribeQuery); + } else { + // Mode B — shared-table filter; subscribe-id is mandatory. + String tableName = requireNonEmpty(config, TABLE_NAME); + String columnName = config.get(COLUMN_NAME); + String subscribeIdColumn = config.get(SUBSCRIBE_ID_COLUMN); + this.subscribeId = requireNonEmpty(config, SUBSCRIBE_ID); + this.sql = + "SELECT " + + columnName + + " FROM " + + tableName + + " WHERE " + + subscribeIdColumn + + " = ?"; + LOG.info( + "JdbcTableDiscoverer running in shared-table mode. URL='{}', table='{}', " + + "column='{}', subscribeIdColumn='{}', subscribeId='{}'.", + jdbcUrl, + tableName, + columnName, + subscribeIdColumn, + subscribeId); + } + + connection = openConnection(jdbcUrl, username, password, userCodeClassLoader); + } + + private static Connection openConnection( + String jdbcUrl, String username, String password, ClassLoader userCodeClassLoader) + throws SQLException { + Properties properties = new Properties(); + properties.setProperty("user", username); + properties.setProperty("password", password); + + Driver driver = findDriver(jdbcUrl, userCodeClassLoader); + if (driver != null) { + Connection connection = driver.connect(jdbcUrl, properties); + if (connection != null) { + return connection; + } + } + // Fall back to DriverManager for drivers that are already visible on this class' + // classloader (e.g. bundled directly with flink-cdc-common). + return DriverManager.getConnection(jdbcUrl, username, password); + } + + private static Driver findDriver(String jdbcUrl, ClassLoader classLoader) throws SQLException { + for (Driver driver : ServiceLoader.load(Driver.class, classLoader)) { + if (driver.acceptsURL(jdbcUrl)) { + return driver; + } + } + return null; + } + + @Override + public Set discover() throws Exception { + Set result = new LinkedHashSet<>(); + if (subscribeId != null) { + try (PreparedStatement ps = connection.prepareStatement(sql)) { + ps.setString(1, subscribeId); + try (ResultSet rs = ps.executeQuery()) { + collect(rs, result); + } + } + } else { + try (Statement stmt = connection.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { + collect(rs, result); + } + } + LOG.info("JdbcTableDiscoverer discovered {} tables.", result.size()); + return result; + } + + private void collect(ResultSet rs, Set result) throws Exception { + while (rs.next()) { + String value = rs.getString(1); + if (value == null || value.isEmpty()) { + continue; + } + try { + result.add(TableId.parse(value)); + } catch (IllegalArgumentException e) { + LOG.warn( + "Skipping invalid table name '{}' returned by JdbcTableDiscoverer.", value); + } + } + } + + private static String requireNonEmpty(Configuration config, ConfigOption option) { + String value = config.get(option); + if (value == null || value.isEmpty()) { + throw new IllegalArgumentException( + "'" + option.key() + "' is required for JdbcTableDiscoverer."); + } + return value; + } + + @Override + public void close() throws Exception { + if (connection != null && !connection.isClosed()) { + connection.close(); + LOG.info("JdbcTableDiscoverer closed JDBC connection."); + } + } +} diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererFactory.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererFactory.java new file mode 100644 index 00000000000..70aa47877ca --- /dev/null +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererFactory.java @@ -0,0 +1,36 @@ +/* + * 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.common.source.discover; + +/** + * {@link TableDiscovererFactory} for {@link JdbcTableDiscoverer} with factory type {@code jdbc}. + */ +public class JdbcTableDiscovererFactory implements TableDiscovererFactory { + + public static final String TYPE = "jdbc"; + + @Override + public String type() { + return TYPE; + } + + @Override + public TableDiscoverer createDiscoverer() { + return new JdbcTableDiscoverer(); + } +} diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java new file mode 100644 index 00000000000..b92fc1393f9 --- /dev/null +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscoverer.java @@ -0,0 +1,49 @@ +/* + * 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.common.source.discover; + +import org.apache.flink.cdc.common.annotation.PublicEvolving; +import org.apache.flink.cdc.common.configuration.Configuration; + +import java.io.Serializable; +import java.util.Set; + +/** Pluggable abstraction for discovering a set of object identifiers. */ +@PublicEvolving +public interface ObjectIdDiscoverer extends Serializable, AutoCloseable { + + /** Opens this discoverer and initializes any resources needed for discovery. */ + void open(Context context) throws Exception; + + /** Discovers and returns the set of object identifiers selected by the caller configuration. */ + Set discover() throws Exception; + + /** Closes this discoverer and releases any resources. */ + @Override + void close() throws Exception; + + /** Context providing runtime information for the discoverer. */ + interface Context { + + /** Returns the full connector configuration. */ + Configuration getConfiguration(); + + /** Returns the user code class loader of the current session. */ + ClassLoader getUserCodeClassLoader(); + } +} diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscovererFactory.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscovererFactory.java new file mode 100644 index 00000000000..4f9a8492a77 --- /dev/null +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscovererFactory.java @@ -0,0 +1,172 @@ +/* + * 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.common.source.discover; + +import org.apache.flink.cdc.common.annotation.PublicEvolving; +import org.apache.flink.cdc.common.configuration.Configuration; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.ServiceLoader; +import java.util.stream.Collectors; + +/** SPI factory that creates an {@link ObjectIdDiscoverer}. */ +@PublicEvolving +public interface ObjectIdDiscovererFactory { + + /** Storage-side type, such as jdbc, local-file, or remote-store. */ + String type(); + + /** Discovered object id type, used to distinguish TableId/ObjectId/etc. */ + Class objectIdClass(); + + /** Creates a new uninitialized {@link ObjectIdDiscoverer}. */ + ObjectIdDiscoverer createDiscoverer(); + + /** Creates a discoverer context with the given configuration and class loader. */ + static ObjectIdDiscoverer.Context createContext( + Configuration configuration, ClassLoader classLoader) { + return new DefaultObjectIdDiscovererContext(configuration, classLoader); + } + + /** Discovers a factory via SPI and delegates discoverer creation to it. */ + static ObjectIdDiscoverer createDiscoverer( + String type, Class objectIdClass, ClassLoader classLoader) { + return discoverFactory(type, objectIdClass, classLoader).createDiscoverer(); + } + + /** + * Discovers an {@link ObjectIdDiscovererFactory} via SPI whose {@link #type()} and {@link + * #objectIdClass()} match the given arguments. + * + * @throws IllegalArgumentException if no factory matches the given combination. + * @throws IllegalStateException if multiple factories match the given combination. + */ + static ObjectIdDiscovererFactory discoverFactory( + String type, Class objectIdClass, ClassLoader classLoader) { + ClassLoader loader = + classLoader != null ? classLoader : Thread.currentThread().getContextClassLoader(); + return discoverFactory(type, objectIdClass, loadFactories(loader)); + } + + /** + * Discovers an {@link ObjectIdDiscovererFactory} from the given candidates whose {@link + * #type()} and {@link #objectIdClass()} match the given arguments. + * + * @throws IllegalArgumentException if no factory matches the given combination. + * @throws IllegalStateException if multiple factories match the given combination. + */ + static ObjectIdDiscovererFactory discoverFactory( + String type, + Class objectIdClass, + Iterable> factories) { + String normalizedType = normalizeType(type); + List> matched = new ArrayList<>(); + List available = new ArrayList<>(); + + for (ObjectIdDiscovererFactory factory : factories) { + String factoryType = normalizeType(factory.type()); + Class factoryObjectIdClass = factory.objectIdClass(); + available.add(formatFactory(factoryType, factoryObjectIdClass, factory)); + if (factoryType.equals(normalizedType) && factoryObjectIdClass.equals(objectIdClass)) { + matched.add(factory); + } + } + + if (matched.isEmpty()) { + throw new IllegalArgumentException( + "Unsupported object id discoverer factory for type '" + + type + + "' and object id class '" + + objectIdClass.getName() + + "'. Available discoverer factories: " + + available + + "."); + } + + if (matched.size() > 1) { + throw new IllegalStateException( + "Multiple ObjectIdDiscovererFactory implementations found for type '" + + type + + "' and object id class '" + + objectIdClass.getName() + + "': " + + matched.stream() + .map(factory -> factory.getClass().getName()) + .collect(Collectors.joining(", "))); + } + + @SuppressWarnings("unchecked") + ObjectIdDiscovererFactory typedFactory = (ObjectIdDiscovererFactory) matched.get(0); + return typedFactory; + } + + /** + * Loads all {@link ObjectIdDiscovererFactory} implementations from both the {@link + * ObjectIdDiscovererFactory} and the legacy {@link TableDiscovererFactory} SPI service files, + * de-duplicated by implementation class. + */ + private static Iterable> loadFactories(ClassLoader loader) { + Map> factories = new LinkedHashMap<>(); + for (ObjectIdDiscovererFactory factory : + ServiceLoader.load(ObjectIdDiscovererFactory.class, loader)) { + factories.putIfAbsent(factory.getClass().getName(), factory); + } + for (TableDiscovererFactory factory : + ServiceLoader.load(TableDiscovererFactory.class, loader)) { + factories.putIfAbsent(factory.getClass().getName(), factory); + } + return factories.values(); + } + + private static String normalizeType(String type) { + if (type == null || type.trim().isEmpty()) { + throw new IllegalArgumentException("Discoverer factory type must not be empty."); + } + return type.trim().toLowerCase(Locale.ROOT); + } + + private static String formatFactory( + String type, Class objectIdClass, ObjectIdDiscovererFactory factory) { + return type + " + " + objectIdClass.getName() + " (" + factory.getClass().getName() + ")"; + } + + /** Default implementation of {@link ObjectIdDiscoverer.Context}. */ + class DefaultObjectIdDiscovererContext implements ObjectIdDiscoverer.Context { + private final Configuration configuration; + private final ClassLoader classLoader; + + DefaultObjectIdDiscovererContext(Configuration configuration, ClassLoader classLoader) { + this.configuration = configuration; + this.classLoader = classLoader; + } + + @Override + public Configuration getConfiguration() { + return configuration; + } + + @Override + public ClassLoader getUserCodeClassLoader() { + return classLoader; + } + } +} diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/TableDiscoverer.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/TableDiscoverer.java new file mode 100644 index 00000000000..cdb9fda73eb --- /dev/null +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/TableDiscoverer.java @@ -0,0 +1,28 @@ +/* + * 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.common.source.discover; + +import org.apache.flink.cdc.common.annotation.PublicEvolving; +import org.apache.flink.cdc.common.event.TableId; + +/** + * Table-specific specialization of {@link ObjectIdDiscoverer} whose discovered object id type is + * {@link TableId}. + */ +@PublicEvolving +public interface TableDiscoverer extends ObjectIdDiscoverer {} diff --git a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/TableDiscovererFactory.java b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/TableDiscovererFactory.java new file mode 100644 index 00000000000..23d04f1f340 --- /dev/null +++ b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/source/discover/TableDiscovererFactory.java @@ -0,0 +1,53 @@ +/* + * 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.common.source.discover; + +import org.apache.flink.cdc.common.annotation.PublicEvolving; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.event.TableId; + +/** Table-specific SPI factory that creates a {@link TableDiscoverer}. */ +@PublicEvolving +public interface TableDiscovererFactory extends ObjectIdDiscovererFactory { + + /** Compatibility alias for the old single identifier, now meaning storage-side type. */ + default String identifier() { + return type(); + } + + @Override + default Class objectIdClass() { + return TableId.class; + } + + /** Creates a new uninitialized {@link TableDiscoverer}. */ + @Override + TableDiscoverer createDiscoverer(); + + /** Creates a discoverer context with the given configuration and class loader. */ + static ObjectIdDiscoverer.Context createContext( + Configuration configuration, ClassLoader classLoader) { + return ObjectIdDiscovererFactory.createContext(configuration, classLoader); + } + + /** Discovers a table discoverer factory via SPI and delegates discoverer creation to it. */ + static TableDiscoverer createDiscoverer(String type, ClassLoader classLoader) { + return (TableDiscoverer) + ObjectIdDiscovererFactory.createDiscoverer(type, TableId.class, classLoader); + } +} diff --git a/flink-cdc-common/src/main/resources/META-INF/services/org.apache.flink.cdc.common.source.discover.ObjectIdDiscovererFactory b/flink-cdc-common/src/main/resources/META-INF/services/org.apache.flink.cdc.common.source.discover.ObjectIdDiscovererFactory new file mode 100644 index 00000000000..a16bb5ae27d --- /dev/null +++ b/flink-cdc-common/src/main/resources/META-INF/services/org.apache.flink.cdc.common.source.discover.ObjectIdDiscovererFactory @@ -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.common.source.discover.JdbcTableDiscovererFactory diff --git a/flink-cdc-common/src/main/resources/META-INF/services/org.apache.flink.cdc.common.source.discover.TableDiscovererFactory b/flink-cdc-common/src/main/resources/META-INF/services/org.apache.flink.cdc.common.source.discover.TableDiscovererFactory new file mode 100644 index 00000000000..a16bb5ae27d --- /dev/null +++ b/flink-cdc-common/src/main/resources/META-INF/services/org.apache.flink.cdc.common.source.discover.TableDiscovererFactory @@ -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.common.source.discover.JdbcTableDiscovererFactory diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererITCase.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererITCase.java new file mode 100644 index 00000000000..b42aa3e7743 --- /dev/null +++ b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/JdbcTableDiscovererITCase.java @@ -0,0 +1,370 @@ +/* + * 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.common.source.discover; + +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.event.TableId; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.Statement; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Integration tests for {@link JdbcTableDiscoverer} and {@link TableDiscovererFactory} SPI loading. + * Uses a real MySQL container via Testcontainers to verify both the default shared-table mode and + * the advanced custom-query escape hatch. + */ +@Testcontainers +class JdbcTableDiscovererITCase { + + private static final String SUBSCRIBE_ID_ORDERS = "orders-subscription"; + private static final String SUBSCRIBE_ID_ANALYTICS = "analytics-subscription"; + + @Container + private static final MySQLContainer MYSQL = + new MySQLContainer<>("mysql:8.0") + .withDatabaseName("meta_db") + .withUsername("test_user") + .withPassword("test_password"); + + @BeforeAll + static void setupDatabase() throws Exception { + try (Connection conn = + DriverManager.getConnection( + MYSQL.getJdbcUrl(), MYSQL.getUsername(), MYSQL.getPassword()); + Statement stmt = conn.createStatement()) { + // Shared subscription table with default column names. + stmt.execute( + "CREATE TABLE cdc_subscriptions (" + + " subscribe_id VARCHAR(64) NOT NULL," + + " subscribe_table_name VARCHAR(255) NOT NULL," + + " PRIMARY KEY (subscribe_id, subscribe_table_name)" + + ")"); + stmt.execute( + "INSERT INTO cdc_subscriptions VALUES " + + "('orders-subscription', 'source_db.orders')," + + "('orders-subscription', 'source_db.order_items')," + + "('orders-subscription', 'source_db.products')," + + "('analytics-subscription', 'analytics_db.user_events')"); + + // Shared subscription table using non-default column names. + stmt.execute( + "CREATE TABLE custom_subscriptions (" + + " sub_id VARCHAR(64) NOT NULL," + + " table_fqn VARCHAR(255) NOT NULL," + + " PRIMARY KEY (sub_id, table_fqn)" + + ")"); + stmt.execute( + "INSERT INTO custom_subscriptions VALUES " + + "('warehouse-sub', 'warehouse.inventory')," + + "('warehouse-sub', 'warehouse.shipments')," + + "('hr-sub', 'hr.employees')"); + } + } + + // ========================================================================= + // SPI loading + // ========================================================================= + + @Test + void testSpiLoadsJdbcObjectIdFactory() { + ObjectIdDiscoverer discoverer = + ObjectIdDiscovererFactory.createDiscoverer( + "jdbc", TableId.class, Thread.currentThread().getContextClassLoader()); + assertThat(discoverer).isInstanceOf(JdbcTableDiscoverer.class); + } + + @Test + void testSpiLoadsJdbcTableFactoryCompatibilityEntry() { + TableDiscoverer discoverer = + TableDiscovererFactory.createDiscoverer( + "jdbc", Thread.currentThread().getContextClassLoader()); + assertThat(discoverer).isInstanceOf(JdbcTableDiscoverer.class); + } + + @Test + void testSpiCaseInsensitive() { + TableDiscoverer discoverer = + TableDiscovererFactory.createDiscoverer( + "JDBC", Thread.currentThread().getContextClassLoader()); + assertThat(discoverer).isInstanceOf(JdbcTableDiscoverer.class); + } + + @Test + void testSpiFailsForUnknownType() { + assertThatThrownBy( + () -> + TableDiscovererFactory.createDiscoverer( + "unknown-type", + Thread.currentThread().getContextClassLoader())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "Unsupported object id discoverer factory for type 'unknown-type'") + .hasMessageContaining(TableId.class.getName()); + } + + // ========================================================================= + // Default mode — shared subscription table + // ========================================================================= + + @Test + void testDefaultModeWithDefaultColumns() throws Exception { + Configuration config = + sharedTableConfigBuilder() + .table("cdc_subscriptions") + .subscribeId(SUBSCRIBE_ID_ORDERS) + .build(); + + assertThat(runDiscovery(config)) + .containsExactlyInAnyOrder( + TableId.tableId("source_db", "orders"), + TableId.tableId("source_db", "order_items"), + TableId.tableId("source_db", "products")); + } + + @Test + void testDefaultModeFiltersBySubscribeId() throws Exception { + Configuration config = + sharedTableConfigBuilder() + .table("cdc_subscriptions") + .subscribeId(SUBSCRIBE_ID_ANALYTICS) + .build(); + + assertThat(runDiscovery(config)) + .containsExactlyInAnyOrder(TableId.tableId("analytics_db", "user_events")); + } + + @Test + void testDefaultModeWithCustomColumns() throws Exception { + Configuration config = + sharedTableConfigBuilder() + .table("custom_subscriptions") + .columnName("table_fqn") + .subscribeIdColumn("sub_id") + .subscribeId("warehouse-sub") + .build(); + + assertThat(runDiscovery(config)) + .containsExactlyInAnyOrder( + TableId.tableId("warehouse", "inventory"), + TableId.tableId("warehouse", "shipments")); + } + + @Test + void testDefaultModeReflectsDynamicChanges() throws Exception { + Configuration config = + sharedTableConfigBuilder() + .table("cdc_subscriptions") + .subscribeId(SUBSCRIBE_ID_ANALYTICS) + .build(); + + TableDiscoverer discoverer = + TableDiscovererFactory.createDiscoverer( + "jdbc", Thread.currentThread().getContextClassLoader()); + discoverer.open( + TableDiscovererFactory.createContext( + config, Thread.currentThread().getContextClassLoader())); + try { + assertThat(discoverer.discover()).hasSize(1); + + executeSql( + "INSERT INTO cdc_subscriptions VALUES " + + "('analytics-subscription', 'analytics_db.sessions')"); + try { + Set updated = discoverer.discover(); + assertThat(updated) + .containsExactlyInAnyOrder( + TableId.tableId("analytics_db", "user_events"), + TableId.tableId("analytics_db", "sessions")); + } finally { + executeSql( + "DELETE FROM cdc_subscriptions WHERE subscribe_table_name " + + "= 'analytics_db.sessions'"); + } + } finally { + discoverer.close(); + } + } + + // ========================================================================= + // Advanced escape hatch — custom query + // ========================================================================= + + @Test + void testCustomQueryReadsFirstColumn() throws Exception { + Map map = baseConnectionConfig(); + map.put( + "table.discoverer.jdbc.subscribe-query", + "SELECT subscribe_table_name FROM cdc_subscriptions " + + "WHERE subscribe_id = 'orders-subscription' " + + "ORDER BY subscribe_table_name"); + + assertThat(runDiscovery(Configuration.fromMap(map))) + .containsExactlyInAnyOrder( + TableId.tableId("source_db", "order_items"), + TableId.tableId("source_db", "orders"), + TableId.tableId("source_db", "products")); + } + + @Test + void testCustomQueryOverridesDefaultModeOptions() throws Exception { + // Mix both modes: the custom-query option must win and silently ignore the default-mode + // options below (table-name / subscribe-id). + Map map = baseConnectionConfig(); + map.put("table.discoverer.jdbc.table-name", "cdc_subscriptions"); + map.put("table.discoverer.jdbc.subscribe-id", SUBSCRIBE_ID_ANALYTICS); + map.put( + "table.discoverer.jdbc.subscribe-query", + "SELECT subscribe_table_name FROM cdc_subscriptions " + + "WHERE subscribe_id = 'orders-subscription'"); + + assertThat(runDiscovery(Configuration.fromMap(map))) + .containsExactlyInAnyOrder( + TableId.tableId("source_db", "orders"), + TableId.tableId("source_db", "order_items"), + TableId.tableId("source_db", "products")); + } + + // ========================================================================= + // Validation + // ========================================================================= + + @Test + void testMissingJdbcUrlThrows() { + Map map = new HashMap<>(); + map.put("table.discoverer.jdbc.username", "user"); + map.put("table.discoverer.jdbc.password", "pass"); + map.put("table.discoverer.jdbc.table-name", "cdc_subscriptions"); + map.put("table.discoverer.jdbc.subscribe-id", "x"); + + assertThatThrownBy(() -> runDiscovery(Configuration.fromMap(map))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table.discoverer.jdbc.url"); + } + + @Test + void testDefaultModeMissingTableNameThrows() { + Map map = baseConnectionConfig(); + map.put("table.discoverer.jdbc.subscribe-id", SUBSCRIBE_ID_ORDERS); + + assertThatThrownBy(() -> runDiscovery(Configuration.fromMap(map))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table.discoverer.jdbc.table-name"); + } + + @Test + void testDefaultModeMissingSubscribeIdThrows() { + Map map = baseConnectionConfig(); + map.put("table.discoverer.jdbc.table-name", "cdc_subscriptions"); + + assertThatThrownBy(() -> runDiscovery(Configuration.fromMap(map))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table.discoverer.jdbc.subscribe-id"); + } + + // ========================================================================= + // Helpers + // ========================================================================= + + private Set runDiscovery(Configuration config) throws Exception { + TableDiscoverer discoverer = + TableDiscovererFactory.createDiscoverer( + "jdbc", Thread.currentThread().getContextClassLoader()); + discoverer.open( + TableDiscovererFactory.createContext( + config, Thread.currentThread().getContextClassLoader())); + try { + return discoverer.discover(); + } finally { + discoverer.close(); + } + } + + private static Map baseConnectionConfig() { + Map map = new HashMap<>(); + map.put("table.discoverer.jdbc.url", MYSQL.getJdbcUrl()); + map.put("table.discoverer.jdbc.username", MYSQL.getUsername()); + map.put("table.discoverer.jdbc.password", MYSQL.getPassword()); + return map; + } + + private static void executeSql(String sql) throws Exception { + try (Connection conn = + DriverManager.getConnection( + MYSQL.getJdbcUrl(), MYSQL.getUsername(), MYSQL.getPassword()); + Statement stmt = conn.createStatement()) { + stmt.execute(sql); + } + } + + private SharedTableConfigBuilder sharedTableConfigBuilder() { + return new SharedTableConfigBuilder(); + } + + private static final class SharedTableConfigBuilder { + private String table; + private String columnName; + private String subscribeIdColumn; + private String subscribeId; + + SharedTableConfigBuilder table(String table) { + this.table = table; + return this; + } + + SharedTableConfigBuilder columnName(String columnName) { + this.columnName = columnName; + return this; + } + + SharedTableConfigBuilder subscribeIdColumn(String subscribeIdColumn) { + this.subscribeIdColumn = subscribeIdColumn; + return this; + } + + SharedTableConfigBuilder subscribeId(String subscribeId) { + this.subscribeId = subscribeId; + return this; + } + + Configuration build() { + Map map = baseConnectionConfig(); + map.put("table.discoverer.jdbc.table-name", table); + map.put("table.discoverer.jdbc.subscribe-id", subscribeId); + if (columnName != null) { + map.put("table.discoverer.jdbc.column-name", columnName); + } + if (subscribeIdColumn != null) { + map.put("table.discoverer.jdbc.subscribe-id-column", subscribeIdColumn); + } + return Configuration.fromMap(map); + } + } +} diff --git a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscovererFactoryTest.java b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscovererFactoryTest.java new file mode 100644 index 00000000000..c465e7b8d1b --- /dev/null +++ b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/source/discover/ObjectIdDiscovererFactoryTest.java @@ -0,0 +1,144 @@ +/* + * 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.common.source.discover; + +import org.apache.flink.cdc.common.event.TableId; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link ObjectIdDiscovererFactory#discoverFactory}. */ +class ObjectIdDiscovererFactoryTest { + + @Test + void testMatchesByTypeAndObjectIdClass() { + ObjectIdDiscovererFactory factory = + ObjectIdDiscovererFactory.discoverFactory( + "jdbc", + String.class, + Arrays.asList(new TestTableFactory(), new TestStringFactory())); + + assertThat(factory).isInstanceOf(TestStringFactory.class); + } + + @Test + void testFailsForUnknownCombination() { + assertThatThrownBy( + () -> + ObjectIdDiscovererFactory.discoverFactory( + "jdbc", + Long.class, + Arrays.asList( + new TestTableFactory(), new TestStringFactory()))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("jdbc") + .hasMessageContaining(Long.class.getName()) + .hasMessageContaining(TableId.class.getName()) + .hasMessageContaining(String.class.getName()); + } + + @Test + void testFailsForDuplicateCombination() { + assertThatThrownBy( + () -> + ObjectIdDiscovererFactory.discoverFactory( + "jdbc", + TableId.class, + Arrays.asList( + new TestTableFactory(), + new DuplicateTestTableFactory()))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining(TestTableFactory.class.getName()) + .hasMessageContaining(DuplicateTestTableFactory.class.getName()); + } + + private static final class TestTableFactory implements ObjectIdDiscovererFactory { + + @Override + public String type() { + return "jdbc"; + } + + @Override + public Class objectIdClass() { + return TableId.class; + } + + @Override + public ObjectIdDiscoverer createDiscoverer() { + return new EmptyDiscoverer<>(); + } + } + + private static final class DuplicateTestTableFactory + implements ObjectIdDiscovererFactory { + + @Override + public String type() { + return "jdbc"; + } + + @Override + public Class objectIdClass() { + return TableId.class; + } + + @Override + public ObjectIdDiscoverer createDiscoverer() { + return new EmptyDiscoverer<>(); + } + } + + private static final class TestStringFactory implements ObjectIdDiscovererFactory { + + @Override + public String type() { + return "jdbc"; + } + + @Override + public Class objectIdClass() { + return String.class; + } + + @Override + public ObjectIdDiscoverer createDiscoverer() { + return new EmptyDiscoverer<>(); + } + } + + private static final class EmptyDiscoverer implements ObjectIdDiscoverer { + + @Override + public void open(Context context) {} + + @Override + public Set discover() { + return Collections.emptySet(); + } + + @Override + public void close() {} + } +}