[FLINK-32695] [test-utils] Add flink-test-utils-source module for Source V2 tests.

flink-test-utils-parent/flink-test-utils-source/pom.xml

@@ -0,0 +1,69 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-test-utils-parent</artifactId>
+		<version>2.2-SNAPSHOT</version>
+	</parent>
+
+	<artifactId>flink-test-utils-source</artifactId>
+	<name>Flink : Test utils : Source</name>
+
+	<packaging>jar</packaging>
+
+	<dependencies>
+		<!-- Core dependency for Source V2 APIs -->
+		<dependency>
+			<groupId>org.apache.flink</groupId>
+			<artifactId>flink-core</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+		<!-- Test dependencies -->
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-jar-plugin</artifactId>
+				<executions>
+					<execution>
+						<goals>
+							<goal>test-jar</goal>
+						</goals>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>

flink-test-utils-parent/flink-test-utils-source/src/main/java/org/apache/flink/test/util/source/AbstractTestSource.java

@@ -0,0 +1,41 @@
+/*
+ * 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.test.util.source;
+
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+/**
+ * Abstract base class for test sources that provides default implementations for all Source V2
+ * methods. This class fixes the enumerator checkpoint type to {@code Void} for the common case
+ * where no checkpoint state is needed.
+ *
+ * <p>For test cases that require checkpointing enumerator state, extend {@link
+ * AbstractTestSourceBase} directly with a custom checkpoint type.
+ *
+ * @param <T> The type of records produced by this source
+ */
+public abstract class AbstractTestSource<T> extends AbstractTestSourceBase<T, Void> {
+
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public SimpleVersionedSerializer<Void> getEnumeratorCheckpointSerializer() {
+        return VoidSerializer.INSTANCE;
+    }
+}

flink-test-utils-parent/flink-test-utils-source/src/main/java/org/apache/flink/test/util/source/AbstractTestSourceBase.java

@@ -0,0 +1,110 @@
+/*
+ * 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.test.util.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+/**
+ * Base class for test sources that allows customization of the enumerator checkpoint type.
+ *
+ * <p>Most test cases should extend {@link AbstractTestSource} which fixes the enumerator state to
+ * {@code Void}. Test cases that need checkpointing can create their own extensions of this base
+ * class.
+ *
+ * @param <T> The type of records produced by this source
+ * @param <EnumChkptState> The type of the enumerator checkpoint state
+ */
+@Internal
+public abstract class AbstractTestSourceBase<T, EnumChkptState>
+        implements Source<T, TestSplit, EnumChkptState> {
+
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public Boundedness getBoundedness() {
+        return Boundedness.BOUNDED;
+    }
+
+    /**
+     * Creates a source reader. The default implementation returns an empty reader that immediately
+     * returns END_OF_INPUT. Subclasses can override this method to provide their specific reader
+     * implementation.
+     */
+    @Override
+    public SourceReader<T, TestSplit> createReader(SourceReaderContext readerContext) {
+        return new TestSourceReader<T>(readerContext) {
+            @Override
+            public InputStatus pollNext(ReaderOutput<T> output) {
+                return InputStatus.END_OF_INPUT;
+            }
+        };
+    }
+
+    @Override
+    public SplitEnumerator<TestSplit, EnumChkptState> createEnumerator(
+            SplitEnumeratorContext<TestSplit> enumContext) {
+        return createEnumerator(enumContext, null);
+    }
+
+    /**
+     * Creates a split enumerator. Subclasses can override this to provide custom enumerator
+     * behavior. The default implementation returns a {@link TestSplitEnumerator} that ignores the
+     * checkpoint state.
+     *
+     * @param enumContext The enumerator context
+     * @param checkpoint The checkpoint state to restore from, or null for a fresh start
+     * @return The split enumerator
+     */
+    @SuppressWarnings("unchecked")
+    protected SplitEnumerator<TestSplit, EnumChkptState> createEnumerator(
+            SplitEnumeratorContext<TestSplit> enumContext, EnumChkptState checkpoint) {
+        // Default implementation ignores checkpoint and returns the basic test enumerator
+        // This cast is safe because TestSplitEnumerator implements SplitEnumerator<TestSplit, Void>
+        // and subclasses that need different checkpoint types will override this method
+        return (SplitEnumerator<TestSplit, EnumChkptState>)
+                new TestSplitEnumerator<Void>(enumContext, null);
+    }
+
+    @Override
+    public SplitEnumerator<TestSplit, EnumChkptState> restoreEnumerator(
+            SplitEnumeratorContext<TestSplit> enumContext, EnumChkptState checkpoint) {
+        return createEnumerator(enumContext, checkpoint);
+    }
+
+    @Override
+    public SimpleVersionedSerializer<TestSplit> getSplitSerializer() {
+        return TestSplit.SERIALIZER;
+    }
+
+    /**
+     * Returns the serializer for enumerator checkpoints. Subclasses must implement this to provide
+     * the appropriate serializer for their checkpoint type.
+     */
+    @Override
+    public abstract SimpleVersionedSerializer<EnumChkptState> getEnumeratorCheckpointSerializer();
+}

flink-test-utils-parent/flink-test-utils-source/src/main/java/org/apache/flink/test/util/source/TestReaderOutput.java

@@ -0,0 +1,105 @@
+/*
+ * 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.test.util.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.connector.source.ReaderOutput;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Test utility for capturing output from SourceReader implementations. Provides convenient methods
+ * to verify emitted records without implementing the full ReaderOutput interface.
+ *
+ * @param <T> The type of records emitted by the source
+ */
+@Internal
+public class TestReaderOutput<T> implements ReaderOutput<T> {
+
+    private final List<T> collected = new ArrayList<>();
+
+    @Override
+    public void collect(T record) {
+        collected.add(record);
+    }
+
+    @Override
+    public void collect(T record, long timestamp) {
+        collected.add(record);
+    }
+
+    @Override
+    public void releaseOutputForSplit(String splitId) {
+        // No-op for testing
+    }
+
+    @Override
+    public ReaderOutput<T> createOutputForSplit(String splitId) {
+        return this;
+    }
+
+    @Override
+    public void markIdle() {
+        // No-op for testing
+    }
+
+    @Override
+    public void emitWatermark(Watermark watermark) {
+        // No-op for testing
+    }
+
+    @Override
+    public void markActive() {
+        // No-op for testing
+    }
+
+    /**
+     * Returns all records collected by this output.
+     *
+     * @return List of all collected records
+     */
+    public List<T> getCollected() {
+        return new ArrayList<>(collected);
+    }
+
+    /**
+     * Returns the last emitted record, or null if no records have been emitted.
+     *
+     * @return The most recently emitted record
+     */
+    public T getLastEmitted() {
+        return collected.isEmpty() ? null : collected.get(collected.size() - 1);
+    }
+
+    /**
+     * Returns the number of records collected.
+     *
+     * @return Count of collected records
+     */
+    public int getCollectedCount() {
+        return collected.size();
+    }
+
+    /** Clears all collected records. */
+    public void clear() {
+        collected.clear();
+    }
+}