Fix code reviews

Author: taoran92

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java

@@ -22,6 +22,10 @@
 import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;
 import org.apache.flink.table.catalog.ObjectPath;
 
+import org.apache.flink.shaded.guava31.com.google.common.cache.CacheBuilder;
+import org.apache.flink.shaded.guava31.com.google.common.cache.CacheLoader;
+import org.apache.flink.shaded.guava31.com.google.common.cache.LoadingCache;
+
 import io.debezium.config.Configuration;
 import io.debezium.connector.mysql.MySqlConnectorConfig;
 import io.debezium.relational.RelationalTableFilters;
@@ -35,14 +39,15 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Predicate;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /** A MySql Source configuration which is used by {@link MySqlSource}. */
 public class MySqlSourceConfig implements Serializable {
     private static final long serialVersionUID = 1L;
+    private static final Duration TABLE_FILTER_CACHE_EXPIRE_DURATION = Duration.ofHours(1);
+    private static final long TABLE_FILTER_CACHE_MAXIMUM_SIZE = 1024;
 
     private final String hostname;
     private final int port;
@@ -283,10 +288,19 @@ public Predicate<TableId> getTableFilter() {
 
     static Tables.TableFilter createCachedTableFilter(
             Tables.TableFilter tableFilter, @Nullable Selectors excludeTableFilter) {
-        Map<TableId, Boolean> tableFilterCache = new ConcurrentHashMap<>();
-        return tableId ->
-                tableFilterCache.computeIfAbsent(
-                        tableId, id -> isTableIncluded(tableFilter, excludeTableFilter, id));
+        LoadingCache<TableId, Boolean> tableFilterCache =
+                CacheBuilder.newBuilder()
+                        .expireAfterAccess(TABLE_FILTER_CACHE_EXPIRE_DURATION)
+                        .maximumSize(TABLE_FILTER_CACHE_MAXIMUM_SIZE)
+                        .build(
+                                new CacheLoader<TableId, Boolean>() {
+                                    @Override
+                                    public Boolean load(TableId tableId) {
+                                        return isTableIncluded(
+                                                tableFilter, excludeTableFilter, tableId);
+                                    }
+                                });
+        return tableFilterCache::getUnchecked;
     }
 
     private static boolean isTableIncluded(

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfigTest.java

@@ -41,12 +41,16 @@ void testCachesTableFilterResults() {
                         null);
 
         TableId includedTable = new TableId("test_db", null, "orders_1");
+        TableId sameIncludedTable = new TableId("test_db", null, "orders_1");
         TableId unmatchedTable = new TableId("test_db", null, "customers");
+        TableId sameUnmatchedTable = new TableId("test_db", null, "customers");
 
         assertThat(cachedTableFilter.isIncluded(includedTable)).isTrue();
         assertThat(cachedTableFilter.isIncluded(includedTable)).isTrue();
+        assertThat(cachedTableFilter.isIncluded(sameIncludedTable)).isTrue();
         assertThat(cachedTableFilter.isIncluded(unmatchedTable)).isFalse();
         assertThat(cachedTableFilter.isIncluded(unmatchedTable)).isFalse();
+        assertThat(cachedTableFilter.isIncluded(sameUnmatchedTable)).isFalse();
         assertThat(filterInvocationCount).hasValue(2);
     }