taoran92 commented on code in PR #4422:
URL: https://github.com/apache/flink-cdc/pull/4422#discussion_r3756156529
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java:
##########
@@ -284,6 +286,31 @@ public Predicate<TableId> getTableFilter() {
return tableId ->
tableFilters.dataCollectionFilter().isIncluded(tableId);
}
+ static Tables.TableFilter createCachedTableFilter(
+ Tables.TableFilter tableFilter, @Nullable Selectors
excludeTableFilter) {
+ 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) {
Review Comment:
Thanks. The cache now stores the complete Boolean result, including false.
I added a test using equal but distinct non-matching TableId instances and
verified that the underlying filter is evaluated only once for that table.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/config/MySqlSourceConfig.java:
##########
@@ -42,6 +46,8 @@
/** 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);
Review Comment:
Thanks, agreed. I removed time-based expiration and kept only a bounded
cache.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
