@Override
    public void configure(
            Configuration config,
            HistoryRecordComparator comparator,
            DatabaseHistoryListener listener,
            boolean useCatalogBeforeSchema) {
        super.configure(config, comparator, listener, useCatalogBeforeSchema);
        this.instanceName = config.getString(DATABASE_HISTORY_INSTANCE_NAME);
        this.schemaRecords = getRegisteredHistoryRecord(instanceName);

        // register the schema changes into state
        // every change should be visible to the source function
        DatabaseHistoryUtil.registerHistory(instanceName, schemaRecords);
    }
Gets the registered HistoryRecords under the given instance name.
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/internal/FlinkDatabaseHistory.java