Repository: apache/flink
File: flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java (Java)

Code before:
        /**
         * Judge and handle the eof record.
         *
         * @return whether the record is the eof record.
         */
        private boolean isEndOfStreamReached(T record) {
            if (isStreamEnd) {
                return true;
            }
            if (eofRecordHandler.apply(record)) {
                isStreamEnd = true;
            }
            return isStreamEnd;
        }
    }

    /**
     * A wrapper around {@link SourceOutput} that counts the number of records emitted.
     *
     * <p>This wrapper is used when rate limiting is enabled to track how many records have been
     * emitted since the last rate limit check, allowing the reader to properly apply backpressure
     * when the rate limit is exceeded.
     *
     * @param <T> The type of records being emitted
     */
    private static final class RecordCountingSourceOutputWrapper<T> implements SourceOutput<T> {
        /** The underlying source output to delegate to. */
        final SourceOutput<T> sourceOutput;

        /** The number of records emitted since the last reset. */
        int recordCount;

        /**
         * Creates a new RecordCountingSourceOutputWrapper.
         *
         * @param sourceOutput The underlying source output to wrap
         */
        public RecordCountingSourceOutputWrapper(SourceOutput<T> sourceOutput) {
            this.sourceOutput = sourceOutput;
            this.recordCount = 0;
        }

        @Override
        public void emitWatermark(Watermark watermark) {
            sourceOutput.emitWatermark(watermark);
        }

        @Override
        public void markIdle() {
            sourceOutput.markIdle();
        }

Review comment (suggestion):
We can rename RecordCountingSourceOutputWrapper => RateLimitingSourceOutputWrapper, which is more Intuitively？

Code after:
    }

    /** This output will stop sending records after receiving the eof record. */
    private static final class SourceOutputWrapper<T> implements SourceOutput<T> {
        final SourceOutput<T> sourceOutput;
        final Function<T, Boolean> eofRecordHandler;

        private boolean isStreamEnd = false;

        public SourceOutputWrapper(
                SourceOutput<T> sourceOutput, Function<T, Boolean> eofRecordHandler) {
            this.sourceOutput = sourceOutput;
            this.eofRecordHandler = eofRecordHandler;
        }

        @Override
        public void emitWatermark(Watermark watermark) {
            sourceOutput.emitWatermark(watermark);
        }

        @Override
        public void markIdle() {
            sourceOutput.markIdle();
        }

        @Override
        public void markActive() {
            sourceOutput.markActive();
        }

        @Override
        public void collect(T record) {
            if (!isEndOfStreamReached(record)) {
                sourceOutput.collect(record);
            }
        }

        @Override
        public void collect(T record, long timestamp) {
            if (!isEndOfStreamReached(record)) {
                sourceOutput.collect(record, timestamp);
            }
        }

        /**
         * Judge and handle the eof record.
         *
         * @return whether the record is the eof record.
         */
        private boolean isEndOfStreamReached(T record) {
            if (isStreamEnd) {