@Override
    public void configure(
            Configuration config,
            HistoryRecordComparator comparator,
            DatabaseHistoryListener listener,
            boolean useCatalogBeforeSchema) {
        this.instanceName = config.getString(DATABASE_HISTORY_INSTANCE_NAME);
        this.listener = listener;
        this.storeOnlyMonitoredTablesDdl = config.getBoolean(STORE_ONLY_MONITORED_TABLES_DDL);
        this.skipUnparseableDDL = config.getBoolean(SKIP_UNPARSEABLE_DDL_STATEMENTS);
        this.useCatalogBeforeSchema = useCatalogBeforeSchema;

        // recover
        this.latestTables = new ConcurrentHashMap<>();
        for (SchemaRecord schemaRecord : DatabaseHistoryUtil.retrieveHistory(instanceName)) {
            // validate here
            TableChange tableChange =
                    FlinkJsonTableChangeSerializer.fromDocument(
                            schemaRecord.toDocument(), useCatalogBeforeSchema);
            latestTables.put(tableChange.getId(), schemaRecord);
        }
        // register
        DatabaseHistoryUtil.registerHistory(instanceName, latestTables.values());
    }
Determine whether the {@link FlinkDatabaseSchemaHistory} is compatible with the specified
state.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/debezium/internal/FlinkDatabaseSchemaHistory.java