@Override
        public void execute(ChangeEventSourceContext context, PostgresOffsetContext offsetContext)
                throws InterruptedException {
            this.context = context;
            this.offsetContext = offsetContext;

            LOG.info("Execute StreamSplitReadTask for split: {}", streamSplit);

            offsetContext.setStreamingStoppingLsn(
                    ((PostgresOffset) streamSplit.getEndingOffset()).getLsn());
            super.execute(context, offsetContext);
            if (isBoundedRead()) {

                LOG.info("StreamSplit is bounded read: {}", streamSplit);
                final PostgresOffset currentOffset = PostgresOffset.of(offsetContext.getOffset());

                if (currentOffset.isAtOrAfter(streamSplit.getEndingOffset())) {
                    LOG.info("StreamSplitReadTask finished for {}", streamSplit);

                    try {
                        dispatcher.dispatchWatermarkEvent(
                                offsetContext.getPartition(),
                                streamSplit,
                                currentOffset,
                                WatermarkKind.END);
                    } catch (InterruptedException e) {
                        LOG.error("Send signal event error.", e);
                        errorHandler.setProducerThrowable(
                                new FlinkRuntimeException("Error processing WAL signal event", e));
                    }

                    ((PostgresScanFetchTask.PostgresChangeEventSourceContext) context).finished();
                }
            }
        }
A {@link ChangeEventSource} implementation for Postgres to read streaming changes.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/fetch/PostgresStreamFetchTask.java