@Override
    public void recover(
            Map<String, ?> source, Map<String, ?> position, Tables schema, DdlParser ddlParser) {
        listener.recoveryStarted();
        for (SchemaRecord record : latestTables.values()) {
            TableChange tableChange =
                    FlinkJsonTableChangeSerializer.fromDocument(
                            record.getTableChangeDoc(), useCatalogBeforeSchema);
            schema.overwriteTable(tableChange.getTable());
        }
        listener.recoveryStopped();
    }
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