[FLINK-32695] [Tests] Migrated ChangelogStateBackendLoadingTest to Source V2.

Author: Poorvankbhatia

flink-state-backends/flink-statebackend-changelog/pom.xml

@@ -112,6 +112,13 @@ under the License.
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils-connector</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
 </project>

flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/ChangelogStateBackendLoadingTest.java

@@ -19,6 +19,7 @@
 package org.apache.flink.state.changelog;
 
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.IllegalConfigurationException;
@@ -43,8 +44,8 @@
 import org.apache.flink.state.rocksdb.EmbeddedRocksDBStateBackend;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
 import org.apache.flink.streaming.api.graph.StreamGraph;
+import org.apache.flink.test.util.source.AbstractTestSource;
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.TernaryBoolean;
 
@@ -262,19 +263,12 @@ private void testLoadingStateBackend(
 
     private StreamExecutionEnvironment getEnvironment() {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        SourceFunction<Integer> srcFun =
-                new SourceFunction<Integer>() {
-                    private static final long serialVersionUID = 1L;
-
-                    @Override
-                    public void run(SourceContext<Integer> ctx) throws Exception {}
-
-                    @Override
-                    public void cancel() {}
-                };
 
         SingleOutputStreamOperator<Object> operator =
-                env.addSource(srcFun)
+                env.fromSource(
+                                new EmptyTestSource(),
+                                WatermarkStrategy.noWatermarks(),
+                                "EmptyTestSource")
                         .flatMap(
                                 new FlatMapFunction<Integer, Object>() {
 
@@ -288,6 +282,8 @@ public void flatMap(Integer value, Collector<Object> out)
         return env;
     }
 
+    private static class EmptyTestSource extends AbstractTestSource<Integer> {}
+
     private void assertStateBackendAndChangelogInStreamGraphAndJobGraph(
             StreamExecutionEnvironment env, TernaryBoolean isChangelogEnabled) throws Exception {
         assertEquals(isChangelogEnabled, env.isChangelogStateBackendEnabled());