sourceReader, double recordsPerSecond, int parallelism) {
+ return sourceReader;
+ }
+}
diff --git a/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/source/ratelimit/RateLimitedSourceReaderAdapter.java b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/source/ratelimit/RateLimitedSourceReaderAdapter.java
new file mode 100644
index 00000000000..95fe34d694c
--- /dev/null
+++ b/flink-cdc-flink2-compat/src/main/java/org/apache/flink/api/connector/source/ratelimit/RateLimitedSourceReaderAdapter.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.connector.source.ratelimit;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiter;
+import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
+import org.apache.flink.core.io.InputStatus;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Compatibility adapter for Flink 2.2. Wraps a {@link SourceReader} with rate limiting.
+ *
+ * Unlike the built-in {@link RateLimitedSourceReader}, this adapter correctly delegates all
+ * methods including {@link SourceReader#handleSourceEvents(SourceEvent)}, which is critical for
+ * connectors (e.g. MySQL CDC) that rely on source events for split coordination.
+ */
+@Internal
+public class RateLimitedSourceReaderAdapter {
+
+ /**
+ * Wraps the given source reader with rate limiting.
+ *
+ * @param sourceReader the source reader to wrap
+ * @param recordsPerSecond the desired rate limit in records per second
+ * @param parallelism the current parallelism
+ * @return a rate-limited source reader that delegates all methods including handleSourceEvents
+ */
+ public static SourceReader wrapWithRateLimiter(
+ SourceReader sourceReader, double recordsPerSecond, int parallelism) {
+ RateLimiter rateLimiter =
+ RateLimiterStrategy.perSecond(recordsPerSecond)
+ .createRateLimiter(parallelism);
+ return new DelegatingRateLimitedSourceReader<>(sourceReader, rateLimiter);
+ }
+
+ /**
+ * A rate-limited source reader that delegates all methods to the wrapped reader, including
+ * {@link #handleSourceEvents(SourceEvent)}.
+ */
+ private static class DelegatingRateLimitedSourceReader
+ implements SourceReader {
+
+ private final SourceReader sourceReader;
+ private final RateLimiter rateLimiter;
+ private CompletableFuture availabilityFuture = null;
+
+ DelegatingRateLimitedSourceReader(
+ SourceReader sourceReader, RateLimiter rateLimiter) {
+ this.sourceReader = checkNotNull(sourceReader);
+ this.rateLimiter = checkNotNull(rateLimiter);
+ }
+
+ @Override
+ public void start() {
+ sourceReader.start();
+ }
+
+ @Override
+ public InputStatus pollNext(ReaderOutput output) throws Exception {
+ if (availabilityFuture == null) {
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+ availabilityFuture = null;
+ final InputStatus inputStatus = sourceReader.pollNext(output);
+ if (inputStatus == InputStatus.MORE_AVAILABLE) {
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+ return inputStatus;
+ }
+
+ @Override
+ public CompletableFuture isAvailable() {
+ if (availabilityFuture == null) {
+ availabilityFuture =
+ rateLimiter
+ .acquire()
+ .toCompletableFuture()
+ .thenCombine(sourceReader.isAvailable(), (l, r) -> null);
+ }
+ return availabilityFuture;
+ }
+
+ @Override
+ public void addSplits(List splits) {
+ sourceReader.addSplits(splits);
+ }
+
+ @Override
+ public void notifyNoMoreSplits() {
+ sourceReader.notifyNoMoreSplits();
+ }
+
+ @Override
+ public void handleSourceEvents(SourceEvent sourceEvent) {
+ sourceReader.handleSourceEvents(sourceEvent);
+ }
+
+ @Override
+ public List snapshotState(long checkpointId) {
+ return sourceReader.snapshotState(checkpointId);
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ rateLimiter.notifyCheckpointComplete(checkpointId);
+ sourceReader.notifyCheckpointComplete(checkpointId);
+ }
+
+ @Override
+ public void pauseOrResumeSplits(
+ Collection splitsToPause, Collection splitsToResume) {
+ sourceReader.pauseOrResumeSplits(splitsToPause, splitsToResume);
+ }
+
+ @Override
+ public void close() throws Exception {
+ sourceReader.close();
+ }
+ }
+}
diff --git a/flink-cdc-flink2-compat/src/test/java/org/apache/flink/api/connector/source/ratelimit/RateLimitedSourceReaderAdapterTest.java b/flink-cdc-flink2-compat/src/test/java/org/apache/flink/api/connector/source/ratelimit/RateLimitedSourceReaderAdapterTest.java
new file mode 100644
index 00000000000..e5cb7dc6e77
--- /dev/null
+++ b/flink-cdc-flink2-compat/src/test/java/org/apache/flink/api/connector/source/ratelimit/RateLimitedSourceReaderAdapterTest.java
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.connector.source.ratelimit;
+
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.core.io.InputStatus;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/** Tests for {@link RateLimitedSourceReaderAdapter}. */
+class RateLimitedSourceReaderAdapterTest {
+
+ private TrackingSourceReader delegate;
+ private SourceReader rateLimitedReader;
+
+ @BeforeEach
+ void setUp() {
+ delegate = new TrackingSourceReader();
+ rateLimitedReader = RateLimitedSourceReaderAdapter.wrapWithRateLimiter(delegate, 1000.0, 1);
+ }
+
+ @Test
+ void testWrapWithRateLimiterReturnsDifferentInstance() {
+ Assertions.assertThat(rateLimitedReader).isNotSameAs(delegate);
+ }
+
+ @Test
+ void testStartDelegates() {
+ rateLimitedReader.start();
+ Assertions.assertThat(delegate.startCalled).isTrue();
+ }
+
+ @Test
+ void testAddSplitsDelegates() {
+ List splits = Arrays.asList(new DummySplit("s1"), new DummySplit("s2"));
+ rateLimitedReader.addSplits(splits);
+ Assertions.assertThat(delegate.addedSplits).isEqualTo(splits);
+ }
+
+ @Test
+ void testNotifyNoMoreSplitsDelegates() {
+ rateLimitedReader.notifyNoMoreSplits();
+ Assertions.assertThat(delegate.notifyNoMoreSplitsCalled).isTrue();
+ }
+
+ @Test
+ void testHandleSourceEventsDelegates() {
+ SourceEvent event = new DummySourceEvent();
+ rateLimitedReader.handleSourceEvents(event);
+ Assertions.assertThat(delegate.handledEvents).containsExactly(event);
+ }
+
+ @Test
+ void testSnapshotStateDelegates() throws Exception {
+ delegate.snapshotResult = Arrays.asList(new DummySplit("snap"));
+ List result = rateLimitedReader.snapshotState(42L);
+ Assertions.assertThat(result).isEqualTo(delegate.snapshotResult);
+ Assertions.assertThat(delegate.snapshotCheckpointId).isEqualTo(42L);
+ }
+
+ @Test
+ void testNotifyCheckpointCompleteDelegates() throws Exception {
+ rateLimitedReader.notifyCheckpointComplete(99L);
+ Assertions.assertThat(delegate.completedCheckpointId).isEqualTo(99L);
+ }
+
+ @Test
+ void testPauseOrResumeSplitsDelegates() {
+ Collection toPause = Arrays.asList("s1");
+ Collection toResume = Arrays.asList("s2");
+ rateLimitedReader.pauseOrResumeSplits(toPause, toResume);
+ Assertions.assertThat(delegate.pausedSplits).isEqualTo(toPause);
+ Assertions.assertThat(delegate.resumedSplits).isEqualTo(toResume);
+ }
+
+ @Test
+ void testCloseDelegates() throws Exception {
+ rateLimitedReader.close();
+ Assertions.assertThat(delegate.closeCalled).isTrue();
+ }
+
+ @Test
+ void testIsAvailableReturnsFutureAndDoesNotBlock() {
+ // isAvailable() should return a non-null future (may or may not be done)
+ CompletableFuture future = rateLimitedReader.isAvailable();
+ Assertions.assertThat(future).isNotNull();
+ }
+
+ @Test
+ void testPollNextReturnsNothingAvailableWhenFutureNotCompleted() throws Exception {
+ // Without calling isAvailable first, pollNext should return NOTHING_AVAILABLE
+ InputStatus status = rateLimitedReader.pollNext(new DummyReaderOutput<>());
+ Assertions.assertThat(status).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ }
+
+ // ----------- helpers -----------
+
+ private static class DummySplit implements SourceSplit {
+ private final String id;
+
+ DummySplit(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public String splitId() {
+ return id;
+ }
+ }
+
+ private static class DummySourceEvent implements SourceEvent {}
+
+ private static class DummyReaderOutput implements ReaderOutput {
+ @Override
+ public void collect(T record) {}
+
+ @Override
+ public void collect(T record, long timestamp) {}
+
+ @Override
+ public void emitWatermark(org.apache.flink.api.common.eventtime.Watermark watermark) {}
+
+ @Override
+ public void markIdle() {}
+
+ @Override
+ public void markActive() {}
+
+ @Override
+ public org.apache.flink.api.connector.source.SourceOutput createOutputForSplit(
+ String splitId) {
+ return null;
+ }
+
+ @Override
+ public void releaseOutputForSplit(String splitId) {}
+ }
+
+ private static class TrackingSourceReader implements SourceReader {
+ boolean startCalled = false;
+ boolean notifyNoMoreSplitsCalled = false;
+ boolean closeCalled = false;
+ long snapshotCheckpointId = -1L;
+ long completedCheckpointId = -1L;
+ List addedSplits = new ArrayList<>();
+ List handledEvents = new ArrayList<>();
+ List snapshotResult = new ArrayList<>();
+ Collection pausedSplits = null;
+ Collection resumedSplits = null;
+
+ @Override
+ public void start() {
+ startCalled = true;
+ }
+
+ @Override
+ public InputStatus pollNext(ReaderOutput output) throws Exception {
+ return InputStatus.END_OF_INPUT;
+ }
+
+ @Override
+ public CompletableFuture isAvailable() {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void addSplits(List splits) {
+ addedSplits.addAll(splits);
+ }
+
+ @Override
+ public void notifyNoMoreSplits() {
+ notifyNoMoreSplitsCalled = true;
+ }
+
+ @Override
+ public void handleSourceEvents(SourceEvent sourceEvent) {
+ handledEvents.add(sourceEvent);
+ }
+
+ @Override
+ public List snapshotState(long checkpointId) {
+ snapshotCheckpointId = checkpointId;
+ return snapshotResult;
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ completedCheckpointId = checkpointId;
+ }
+
+ @Override
+ public void pauseOrResumeSplits(
+ Collection splitsToPause, Collection splitsToResume) {
+ pausedSplits = splitsToPause;
+ resumedSplits = splitsToResume;
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeCalled = true;
+ }
+ }
+}