/** This output counted the amount of record emitted. */
private static final class RateLimitedSourceOutputWrapper<T> implements SourceOutput<T> {
    final SourceOutput<T> sourceOutput;

    final RateLimiter rateLimiter;
 	// The amount of data collected in each recordEmitter.emitRecord call.
    int recordCount;

    public RateLimitedSourceOutputWrapper(
            SourceOutput<T> sourceOutput, RateLimiter rateLimiter) {
        this.sourceOutput = sourceOutput;
        this.rateLimiter = rateLimiter;
        this.recordCount = 0;
    }

    @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) {
        sourceOutput.collect(record);
        recordCount++;
    }

    @Override
    public void collect(T record, long timestamp) {
        sourceOutput.collect(record, timestamp);
        recordCount++;
    }

    /**
     * Gets the number of records emitted.
     *
     * @return the number of records emitted.
     */
    public int getRecordCount() {
        return recordCount;
    }

    /** Resets the record count to 0. */
    public void resetRecordCount() {
        recordCount = 0;
    }
}

@Override
public InputStatus pollNext(ReaderOutput<T> output) throws Exception {
    ......
    // we need to loop here, because we may have to go across splits
    while (true) {
        // check if the previous record count reached the limit of ratelimiter.
        if (rateLimitFuture != null && !rateLimitFuture.toCompletableFuture().isDone()) {
            return trace(InputStatus.MORE_AVAILABLE);
        }
        // Process one record.
        final E record = recordsWithSplitId.nextRecordFromSplit();
        if (record != null) {
            // emit the record.
            numRecordsInCounter.inc(1);
            recordEmitter.emitRecord(record, currentSplitOutput, currentSplitContext.state);
            // Acquire permit from rateLimiter.
            if (rateLimiter != null) {
                RateLimitedSourceOutputWrapper<T> rateLimitedSourceOutputWrapper =
                        (RateLimitedSourceOutputWrapper<T>) currentSplitOutput;
                rateLimitFuture =
                        rateLimiter.acquire(rateLimitedSourceOutputWrapper.getRecordCount());
                rateLimitedSourceOutputWrapper.resetRecordCount();
            }
            LOG.trace("Emitted record: {}", record);

            // We always emit MORE_AVAILABLE here, even though we do not strictly know whether
            // more is available. If nothing more is available, the next invocation will find
            // this out and return the correct status.
            // That means we emit the occasional 'false positive' for availability, but this
            // saves us doing checks for every record. Ultimately, this is cheaper.
            return trace(InputStatus.MORE_AVAILABLE);
        } else if (!moveToNextSplit(recordsWithSplitId, output)) {
            // The fetch is done and we just discovered that and have not emitted anything, yet.
            // We need to move to the next fetch. As a shortcut, we call pollNext() here again,
            // rather than emitting nothing and waiting for the caller to call us again.
            return pollNext(output);
        }
    }
}

public List<SplitT> snapshotState(long checkpointId) {
    List<SplitT> splits = new ArrayList<>();
    splitStates.forEach(
            (id, context) -> {
                splits.add(toSplitType(id, context.state));
                if (context.rateLimiter != null) {
                    context.rateLimiter.notifyCheckpointComplete(checkpointId);
                }
            });
    return splits;
}

    private boolean moveToNextSplit(RecordsWithSplitIds<E> recordsWithSplitIds, ReaderOutput<T> output) {
        String nextSplitId = recordsWithSplitIds.nextSplit();
        if (nextSplitId == null) {
            LOG.trace("Current fetch is finished.");
            this.finishCurrentFetch(recordsWithSplitIds, output);
            return false;
        } else {
            this.currentSplitContext = (SplitContext)this.splitStates.get(nextSplitId);
            Preconditions.checkState(this.currentSplitContext != null, "Have records for a split that was not registered");
            Function<T, Boolean> eofRecordHandler = null;
            if (this.eofRecordEvaluator != null) {
                eofRecordHandler = (record) -> {
                    if (!this.eofRecordEvaluator.isEndOfStream(record)) {
                        return false;
                    } else {
                        SplitT split = this.toSplitType(this.currentSplitContext.splitId, this.currentSplitContext.state);
                        this.splitFetcherManager.removeSplits(Collections.singletonList(split));
                        return true;
                    }
                };
            }
            if (this.rateLimiter != null) {
                rateLimiter.notifyAddingSplit(this.toSplitType(this.currentSplitContext.splitId, this.currentSplitContext.state));
            }

            this.currentSplitOutput = this.currentSplitContext.getOrCreateSplitOutput(output, eofRecordHandler);
            LOG.trace("Emitting records from fetch for split {}", nextSplitId);
            return true;
        }
    }

CREATE TABLE my_source (
 id INT,
 score INT
) WITH (
 'connector' = 'kafka',
 'scan.rate.limit.record-per-second' = '100000'
  ......
);

SourceReader<OUT, KafkaPartitionSplit> createReader(
        SourceReaderContext readerContext, Consumer<Collection<String>> splitFinishedHook)
        throws Exception {
    ......
    return new KafkaSourceReader<>(
            elementsQueue,
            new KafkaSourceFetcherManager(
                    elementsQueue, splitReaderSupplier::get, splitFinishedHook),
            recordEmitter,
            toConfiguration(props),
            readerContext,
            kafkaSourceReaderMetrics,
RateLimiterStrategy.perSecond(readerContext.getConfiguration().get("scan.rate.limit.record-per-second")));
}

public KafkaSourceReader(
        FutureCompletingBlockingQueue<RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>>>
                elementsQueue,
        KafkaSourceFetcherManager kafkaSourceFetcherManager,
        RecordEmitter<ConsumerRecord<byte[], byte[]>, T, KafkaPartitionSplitState>
                recordEmitter,
        Configuration config,
        SourceReaderContext context,
        KafkaSourceReaderMetrics kafkaSourceReaderMetrics,
        @Nullable RateLimiterStrategy rateLimiterStrategy) {
    super(elementsQueue, kafkaSourceFetcherManager, recordEmitter, config, context, rateLimiterStrategy);
    this.offsetsToCommit = Collections.synchronizedSortedMap(new TreeMap<>());
    this.offsetsOfFinishedSplits = new ConcurrentHashMap<>();
    this.kafkaSourceReaderMetrics = kafkaSourceReaderMetrics;
    this.commitOffsetsOnCheckpoint =
            config.get(KafkaSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT);
    if (!commitOffsetsOnCheckpoint) {
        LOG.warn(
                "Offset commit on checkpoint is disabled. "
                        + "Consuming offset will not be reported back to Kafka cluster.");
    }
}

public interface SupportsRateLimiting {
    RateLimiter rateLimiter(int currentParallelism);
}

public interface SourceReaderContext extends Serializable {

    /** @return The metric group this source belongs to. */
    SourceReaderMetricGroup metricGroup();

    /** Gets the configuration with which Flink was started. */
    Configuration getConfiguration();

    ......
    /** @return The RateLimiter to control the reading rate of SourceReader. */
    default RateLimiter getRateLimiter() {
        throw new UnsupportedOperationException();
    }
}

public <T extends StreamOperator<OUT>> T createStreamOperator(
        StreamOperatorParameters<OUT> parameters) {
    ......
    FunctionWithException<Integer, RateLimiter, Exception>
            rateLimiterBuildFunction = currentParallelism -> {
                if (source instanceof SupportsRateLimiting) {
                    return ((SupportsRateLimiting) source).rateLimiter(currentParallelism);
                } else {
                    return null;
                }
            };
    final SourceOperator<OUT, ?> sourceOperator =
            instantiateSourceOperator(
                    parameters,
                    source::createReader,
                    gateway,
                    source.getSplitSerializer(),
                    watermarkStrategy,
                    parameters.getProcessingTimeService(),
                    parameters
                            .getContainingTask()
                            .getEnvironment()
                            .getTaskManagerInfo()
                            .getConfiguration(),
                    parameters
                            .getContainingTask()
                            .getEnvironment()
                            .getTaskManagerInfo()
                            .getTaskManagerExternalAddress(),
                    emitProgressiveWatermarks,
                    parameters.getContainingTask().getCanEmitBatchOfRecords(),
                    rateLimiterBuildFunction);
  ......
}

public void initReader() throws Exception {
    if (sourceReader != null) {
        return;
    }

    final int subtaskIndex = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();

    final SourceReaderContext context = new SourceReaderContext(){.....}

    sourceReader = readerFactory.apply(context);
}