This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new aa42119ef [FLINK-39732][cdc-base] Introduce ObjectIdDiscoverer SPI for
flexible table subscription (with default JdbcTableDiscoverer) (#4409)
aa42119ef is described below
commit aa42119effe6a2ffa1aa85c6ff781d44cc2e18e9
Author: Hongshun Wang <[email protected]>
AuthorDate: Fri Jul 31 11:41:13 2026 +0800
[FLINK-39732][cdc-base] Introduce ObjectIdDiscoverer SPI for flexible table
subscription (with default JdbcTableDiscoverer) (#4409)
---
flink-cdc-common/pom.xml | 22 ++
.../source/discover/JdbcTableDiscoverer.java | 239 +++++++++++++
.../discover/JdbcTableDiscovererFactory.java | 36 ++
.../common/source/discover/ObjectIdDiscoverer.java | 49 +++
.../source/discover/ObjectIdDiscovererFactory.java | 172 ++++++++++
.../common/source/discover/TableDiscoverer.java | 28 ++
.../source/discover/TableDiscovererFactory.java | 53 +++
...ommon.source.discover.ObjectIdDiscovererFactory | 16 +
...c.common.source.discover.TableDiscovererFactory | 16 +
.../source/discover/JdbcTableDiscovererITCase.java | 370 +++++++++++++++++++++
.../discover/ObjectIdDiscovererFactoryTest.java | 144 ++++++++
11 files changed, 1145 insertions(+)
diff --git a/flink-cdc-common/pom.xml b/flink-cdc-common/pom.xml
index 80237d26d..99ae43f2e 100644
--- a/flink-cdc-common/pom.xml
+++ b/flink-cdc-common/pom.xml
@@ -27,6 +27,28 @@ limitations under the License.
<artifactId>flink-cdc-common</artifactId>
+ <dependencies>
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>8.0.27</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>mysql</artifactId>
+ <version>${testcontainers.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${testcontainers.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
<build>
<plugins>
<plugin>
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 000000000..bc1ad2c86
--- /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<String> JDBC_URL =
+ ConfigOptions.key("table.discoverer.jdbc.url")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The JDBC connection URL for the table
discovery database.");
+
+ public static final ConfigOption<String> USERNAME =
+ ConfigOptions.key("table.discoverer.jdbc.username")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The JDBC username for the table
discovery database.");
+
+ public static final ConfigOption<String> PASSWORD =
+ ConfigOptions.key("table.discoverer.jdbc.password")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The JDBC password for the table
discovery database.");
+
+ public static final ConfigOption<String> 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<String> 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<String> 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<String> 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<String> 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<TableId> discover() throws Exception {
+ Set<TableId> 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<TableId> 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<String> 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 000000000..70aa47877
--- /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 000000000..b92fc1393
--- /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<T> 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<T> 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 000000000..4f9a8492a
--- /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<T> {
+
+ /** Storage-side type, such as jdbc, local-file, or remote-store. */
+ String type();
+
+ /** Discovered object id type, used to distinguish TableId/ObjectId/etc. */
+ Class<T> objectIdClass();
+
+ /** Creates a new uninitialized {@link ObjectIdDiscoverer}. */
+ ObjectIdDiscoverer<T> 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 <T> ObjectIdDiscoverer<T> createDiscoverer(
+ String type, Class<T> 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 <T> ObjectIdDiscovererFactory<T> discoverFactory(
+ String type, Class<T> 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 <T> ObjectIdDiscovererFactory<T> discoverFactory(
+ String type,
+ Class<T> objectIdClass,
+ Iterable<? extends ObjectIdDiscovererFactory<?>> factories) {
+ String normalizedType = normalizeType(type);
+ List<ObjectIdDiscovererFactory<?>> matched = new ArrayList<>();
+ List<String> 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<T> typedFactory =
(ObjectIdDiscovererFactory<T>) 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<ObjectIdDiscovererFactory<?>>
loadFactories(ClassLoader loader) {
+ Map<String, ObjectIdDiscovererFactory<?>> 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 000000000..cdb9fda73
--- /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<TableId> {}
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 000000000..23d04f1f3
--- /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<TableId> {
+
+ /** Compatibility alias for the old single identifier, now meaning
storage-side type. */
+ default String identifier() {
+ return type();
+ }
+
+ @Override
+ default Class<TableId> 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 000000000..a16bb5ae2
--- /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 000000000..a16bb5ae2
--- /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 000000000..b42aa3e77
--- /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<TableId> 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<TableId> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<TableId> 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<String, String> baseConnectionConfig() {
+ Map<String, String> 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<String, String> 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 000000000..c465e7b8d
--- /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<String> 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<TableId> {
+
+ @Override
+ public String type() {
+ return "jdbc";
+ }
+
+ @Override
+ public Class<TableId> objectIdClass() {
+ return TableId.class;
+ }
+
+ @Override
+ public ObjectIdDiscoverer<TableId> createDiscoverer() {
+ return new EmptyDiscoverer<>();
+ }
+ }
+
+ private static final class DuplicateTestTableFactory
+ implements ObjectIdDiscovererFactory<TableId> {
+
+ @Override
+ public String type() {
+ return "jdbc";
+ }
+
+ @Override
+ public Class<TableId> objectIdClass() {
+ return TableId.class;
+ }
+
+ @Override
+ public ObjectIdDiscoverer<TableId> createDiscoverer() {
+ return new EmptyDiscoverer<>();
+ }
+ }
+
+ private static final class TestStringFactory implements
ObjectIdDiscovererFactory<String> {
+
+ @Override
+ public String type() {
+ return "jdbc";
+ }
+
+ @Override
+ public Class<String> objectIdClass() {
+ return String.class;
+ }
+
+ @Override
+ public ObjectIdDiscoverer<String> createDiscoverer() {
+ return new EmptyDiscoverer<>();
+ }
+ }
+
+ private static final class EmptyDiscoverer<T> implements
ObjectIdDiscoverer<T> {
+
+ @Override
+ public void open(Context context) {}
+
+ @Override
+ public Set<T> discover() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public void close() {}
+ }
+}
