@Override
    public void notifyCheckpointComplete(long checkpointId) {
        if (!debeziumStarted) {
            LOG.debug("notifyCheckpointComplete() called when engine is not started.");
            return;
        }

        final DebeziumChangeFetcher<T> fetcher = this.debeziumChangeFetcher;
        if (fetcher == null) {
            LOG.debug("notifyCheckpointComplete() called on uninitialized source");
            return;
        }

        try {
            final int posInMap = pendingOffsetsToCommit.indexOf(checkpointId);
            if (posInMap == -1) {
                LOG.warn(
                        "Consumer subtask {} received confirmation for unknown checkpoint id {}",
                        getRuntimeContext().getIndexOfThisSubtask(),
                        checkpointId);
                return;
            }

            byte[] serializedOffsets = (byte[]) pendingOffsetsToCommit.remove(posInMap);

            // remove older checkpoints in map
            for (int i = 0; i < posInMap; i++) {
                pendingOffsetsToCommit.remove(0);
            }

            if (serializedOffsets == null || serializedOffsets.length == 0) {
                LOG.debug(
                        "Consumer subtask {} has empty checkpoint state.",
                        getRuntimeContext().getIndexOfThisSubtask());
                return;
            }

            DebeziumOffset offset =
                    DebeziumOffsetSerializer.INSTANCE.deserialize(serializedOffsets);
            changeConsumer.commitOffset(offset);
        } catch (Exception e) {
            // ignore exception if we are no longer running
            LOG.warn("Ignore error when committing offset to database.", e);
        }
    }
Buffer the events from the source and record the errors from the debezium.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/DebeziumSourceFunction.java