private void restoreHistoryRecordsState() throws Exception {
        DocumentReader reader = DocumentReader.defaultReader();
        ConcurrentLinkedQueue<SchemaRecord> historyRecords = new ConcurrentLinkedQueue<>();
        int recordsCount = 0;
        boolean firstEntry = true;
        for (String record : schemaRecordsState.get()) {
            if (firstEntry) {
                // we store the engine instance name in the first element
                this.engineInstanceName = record;
                firstEntry = false;
            } else {
                // Put the records into the state. The database history should read, reorganize and
                // register the state.
                historyRecords.add(new SchemaRecord(reader.read(record)));
                recordsCount++;
            }
        }
        if (engineInstanceName != null) {
            registerHistory(engineInstanceName, historyRecords);
        }
        LOG.info(
                "Consumer subtask {} restored history records state: {} with {} records.",
                getRuntimeContext().getIndexOfThisSubtask(),
                engineInstanceName,
                recordsCount);
    }
Buffer the events from the source and record the errors from the debezium.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/sqlserver-cdc/src/main/java/org/apache/inlong/sort/cdc/sqlserver/table/DebeziumSourceFunction.java