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

Code before:
            if (isStreamEnd) {
                return true;
            }
            if (eofRecordHandler.apply(record)) {
                isStreamEnd = true;
            }
            return isStreamEnd;
        }
    }

    /**
     * A wrapper around {@link SourceOutput} that counts the number of records during the current
     * rate-limiting window.
     *
     * <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 RateLimitingSourceOutputWrapper<T> implements SourceOutput<T> {
        /** The underlying source output to delegate to. */
        final SourceOutput<T> sourceOutput;

        /** Count of records handled during the current rate-limiting window. */
        int currentWindowRecordCount;

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

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

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

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

Review comment (suggestion):
Could be private ?
```suggestion
        private int currentWindowRecordCount;
```

Code after:
        }

        @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) {
                return true;
            }
            if (eofRecordHandler.apply(record)) {
                isStreamEnd = true;
            }
            return isStreamEnd;
        }
    }

    /**
     * A wrapper around {@link SourceOutput} that counts the number of records during the current
     * rate-limiting window.
     *