[hotfix] [tests] Remove unfruitful MigrationTestUtil class

That utility class had a single helper method, restoreFromSnapshot,
which accepts the target snapshot's Flink version. This was useful
before, because the way to restore snapshots was a bit different for
Flink <= 1.1 and newer versions.

Since we now no longer support compatibility for 1.1 versions and
below, this helper method is simply forwarding the restore operation
to the test harness.

This commit refactors this have equivalent behaviour directly in the
AbstractStreamOperatorTestHarness class.

flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSinkMigrationTest.java

@@ -28,7 +28,6 @@
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OperatorSnapshotUtil;
-import org.apache.flink.streaming.util.migration.MigrationTestUtil;
 import org.apache.flink.streaming.util.migration.MigrationVersion;
 import org.apache.flink.util.OperatingSystem;
 
@@ -167,11 +166,9 @@ public void testRestore() throws Exception {
  new StreamSink<>(sink), 10, 1, 0);
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "bucketing-sink-migration-test-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "bucketing-sink-migration-test-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java

@@ -36,7 +36,6 @@
 import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicsDescriptor;
 import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OperatorSnapshotUtil;
-import org.apache.flink.streaming.util.migration.MigrationTestUtil;
 import org.apache.flink.streaming.util.migration.MigrationVersion;
 import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
 import org.apache.flink.util.SerializedValue;
@@ -207,11 +206,9 @@ public void testRestoreFromEmptyStateNoPartitions() throws Exception {
  testHarness.setup();
 
  // restore state from binary snapshot file
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "kafka-consumer-migration-test-flink" + testMigrateVersion + "-empty-state-snapshot"),
- testMigrateVersion);
+ "kafka-consumer-migration-test-flink" + testMigrateVersion + "-empty-state-snapshot"));
 
  testHarness.open();
 
@@ -248,11 +245,9 @@ public void testRestoreFromEmptyStateWithPartitions() throws Exception {
  testHarness.setup();
 
  // restore state from binary snapshot file
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "kafka-consumer-migration-test-flink" + testMigrateVersion + "-empty-state-snapshot"),
- testMigrateVersion);
+ "kafka-consumer-migration-test-flink" + testMigrateVersion + "-empty-state-snapshot"));
 
  testHarness.open();
 
@@ -302,11 +297,9 @@ public void testRestore() throws Exception {
  testHarness.setup();
 
  // restore state from binary snapshot file
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "kafka-consumer-migration-test-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "kafka-consumer-migration-test-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 
@@ -349,11 +342,9 @@ public void testRestoreFailsWithNonEmptyPreFlink13StatesIfDiscoveryEnabled() thr
 
  // restore state from binary snapshot file; should fail since discovery is enabled
  try {
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "kafka-consumer-migration-test-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "kafka-consumer-migration-test-flink" + testMigrateVersion + "-snapshot"));
 
  fail("Restore from savepoints from version before Flink 1.3.x should have failed if discovery is enabled.");
  } catch (Exception e) {

flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java

@@ -515,7 +515,7 @@ private void testRescaling(
  testHarnesses[i] = createTestHarness(consumers[i], initialParallelism, i);
 
  // initializeState() is always called, null signals that we didn't restore
- testHarnesses[i].initializeState(null);
+ testHarnesses[i].initializeEmptyState();
  testHarnesses[i].open();
  }
 

flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerMigrationTest.java

@@ -38,7 +38,6 @@
 import org.apache.flink.streaming.connectors.kinesis.testutils.TestUtils;
 import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OperatorSnapshotUtil;
-import org.apache.flink.streaming.util.migration.MigrationTestUtil;
 import org.apache.flink.streaming.util.migration.MigrationVersion;
 
 import com.amazonaws.services.kinesis.model.SequenceNumberRange;
@@ -148,9 +147,9 @@ public void testRestoreWithEmptyState() throws Exception {
  new AbstractStreamOperatorTestHarness<>(consumerOperator, 1, 1, 0);
 
  testHarness.setup();
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
- "src/test/resources/kinesis-consumer-migration-test-flink" + testMigrateVersion + "-empty-snapshot", testMigrateVersion);
+ testHarness.initializeState(
+ OperatorSnapshotUtil.getResourceFilename(
+  "kinesis-consumer-migration-test-flink" + testMigrateVersion + "-empty-snapshot"));
  testHarness.open();
 
  consumerFunction.run(new TestSourceContext<>());
@@ -204,9 +203,9 @@ public void testRestore() throws Exception {
  new AbstractStreamOperatorTestHarness<>(consumerOperator, 1, 1, 0);
 
  testHarness.setup();
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
- "src/test/resources/kinesis-consumer-migration-test-flink" + testMigrateVersion + "-snapshot", testMigrateVersion);
+ testHarness.initializeState(
+ OperatorSnapshotUtil.getResourceFilename(
+  "kinesis-consumer-migration-test-flink" + testMigrateVersion + "-snapshot"));
  testHarness.open();
 
  consumerFunction.run(new TestSourceContext<>());
@@ -285,9 +284,9 @@ public void testRestoreWithReshardedStream() throws Exception {
  new AbstractStreamOperatorTestHarness<>(consumerOperator, 1, 1, 0);
 
  testHarness.setup();
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
- "src/test/resources/kinesis-consumer-migration-test-flink" + testMigrateVersion + "-snapshot", testMigrateVersion);
+ testHarness.initializeState(
+ OperatorSnapshotUtil.getResourceFilename(
+  "kinesis-consumer-migration-test-flink" + testMigrateVersion + "-snapshot"));
  testHarness.open();
 
  consumerFunction.run(new TestSourceContext<>());

flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java

@@ -763,7 +763,7 @@ protected KinesisDataFetcher<String> createFetcher(
  testHarness.setTimeCharacteristic(TimeCharacteristic.EventTime);
  testHarness.getExecutionConfig().setAutoWatermarkInterval(autoWatermarkInterval);
 
- testHarness.initializeState(null);
+ testHarness.initializeEmptyState();
  testHarness.open();
 
  ConcurrentLinkedQueue<Watermark> watermarks = new ConcurrentLinkedQueue<>();

flink-fs-tests/src/test/java/org/apache/flink/hdfstests/ContinuousFileProcessingMigrationTest.java

@@ -40,7 +40,6 @@
 import org.apache.flink.streaming.util.AbstractStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OperatorSnapshotUtil;
-import org.apache.flink.streaming.util.migration.MigrationTestUtil;
 import org.apache.flink.streaming.util.migration.MigrationVersion;
 import org.apache.flink.util.OperatingSystem;
 
@@ -176,11 +175,9 @@ public void testReaderRestore() throws Exception {
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "reader-migration-test-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "reader-migration-test-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 
@@ -307,11 +304,9 @@ public void testMonitoringSourceRestore() throws Exception {
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "monitoring-function-migration-test-" + expectedModTime + "-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "monitoring-function-migration-test-" + expectedModTime + "-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 

flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigrationTest.java

@@ -33,7 +33,6 @@
 import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OperatorSnapshotUtil;
-import org.apache.flink.streaming.util.migration.MigrationTestUtil;
 import org.apache.flink.streaming.util.migration.MigrationVersion;
 
 import org.junit.Ignore;
@@ -158,10 +157,8 @@ public Integer getKey(Event value) throws Exception {
  try {
  harness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- harness,
- OperatorSnapshotUtil.getResourceFilename("cep-migration-after-branching-flink" + migrateVersion + "-snapshot"),
- migrateVersion);
+ harness.initializeState(
+ OperatorSnapshotUtil.getResourceFilename("cep-migration-after-branching-flink" + migrateVersion + "-snapshot"));
 
  harness.open();
 
@@ -320,10 +317,9 @@ public Integer getKey(Event value) throws Exception {
  try {
  harness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- harness,
- OperatorSnapshotUtil.getResourceFilename("cep-migration-starting-new-pattern-flink" + migrateVersion + "-snapshot"),
- migrateVersion);
+ harness.initializeState(
+ OperatorSnapshotUtil.getResourceFilename(
+ "cep-migration-starting-new-pattern-flink" + migrateVersion + "-snapshot"));
 
  harness.open();
 
@@ -486,10 +482,9 @@ public Integer getKey(Event value) throws Exception {
  try {
  harness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- harness,
- OperatorSnapshotUtil.getResourceFilename("cep-migration-single-pattern-afterwards-flink" + migrateVersion + "-snapshot"),
- migrateVersion);
+ harness.initializeState(
+ OperatorSnapshotUtil.getResourceFilename(
+ "cep-migration-single-pattern-afterwards-flink" + migrateVersion + "-snapshot"));
 
  harness.open();
 
@@ -579,10 +574,9 @@ public Integer getKey(Event value) throws Exception {
  try {
  harness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- harness,
- OperatorSnapshotUtil.getResourceFilename("cep-migration-conditions-flink" + migrateVersion + "-snapshot"),
- migrateVersion);
+ harness.initializeState(
+ OperatorSnapshotUtil.getResourceFilename(
+ "cep-migration-conditions-flink" + migrateVersion + "-snapshot"));
 
  harness.open();
 

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java

@@ -53,7 +53,6 @@
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.streaming.util.OperatorSnapshotUtil;
 import org.apache.flink.streaming.util.TestHarnessUtil;
-import org.apache.flink.streaming.util.migration.MigrationTestUtil;
 import org.apache.flink.streaming.util.migration.MigrationVersion;
 import org.apache.flink.util.Collector;
 
@@ -184,11 +183,9 @@ public void testRestoreSessionWindowsWithCountTrigger() throws Exception {
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-session-with-stateful-trigger-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "win-op-migration-test-session-with-stateful-trigger-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 
@@ -278,11 +275,9 @@ public void testRestoreSessionWindowsWithCountTriggerInMintCondition() throws Ex
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-session-with-stateful-trigger-mint-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "win-op-migration-test-session-with-stateful-trigger-mint-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 
@@ -402,11 +397,9 @@ public void testRestoreReducingEventTimeWindows() throws Exception {
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-reduce-event-time-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "win-op-migration-test-reduce-event-time-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 
@@ -513,11 +506,9 @@ public void testRestoreApplyEventTimeWindows() throws Exception {
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-apply-event-time-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "win-op-migration-test-apply-event-time-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 
@@ -620,11 +611,9 @@ public void testRestoreReducingProcessingTimeWindows() throws Exception {
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-reduce-processing-time-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "win-op-migration-test-reduce-processing-time-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 
@@ -719,11 +708,9 @@ public void testRestoreApplyProcessingTimeWindows() throws Exception {
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-apply-processing-time-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "win-op-migration-test-apply-processing-time-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 
@@ -837,11 +824,9 @@ public void testRestoreKryoSerializedKeysWindows() throws Exception {
 
  testHarness.setup();
 
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
+ testHarness.initializeState(
  OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-kryo-serialized-key-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
+ "win-op-migration-test-kryo-serialized-key-flink" + testMigrateVersion + "-snapshot"));
 
  testHarness.open();
 

flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java

@@ -340,6 +340,14 @@ public void initializeState(OperatorSubtaskState operatorStateHandles) throws Ex
  initializeState(operatorStateHandles, null);
  }
 
+ public void initializeState(String operatorStateSnapshotPath) throws Exception {
+ initializeState(OperatorSnapshotUtil.readStateHandle(operatorStateSnapshotPath));
+ }
+
+ public void initializeEmptyState() throws Exception {
+ initializeState((OperatorSubtaskState) null);
+ }
+
  /**
  * Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#initializeState()}.
  * Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#setup(StreamTask, StreamConfig, Output)}
@@ -485,7 +493,7 @@ public static OperatorSubtaskState repackageState(OperatorSubtaskState... handle
  */
  public void open() throws Exception {
  if (!initializeCalled) {
- initializeState(null);
+ initializeEmptyState();
  }
  operator.open();
  }