You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2017/08/24 18:22:35 UTC
[04/11] flink git commit: [FLINK-7461] Remove Backwards compatibility
with <= Flink 1.1
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
index 9f19064..d7df479 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorMigrationTest.java
@@ -29,7 +29,6 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.datastream.LegacyWindowOperatorType;
import org.apache.flink.streaming.api.functions.windowing.PassThroughWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
@@ -84,7 +83,7 @@ public class WindowOperatorMigrationTest {
@Parameterized.Parameters(name = "Migration Savepoint: {0}")
public static Collection<MigrationVersion> parameters () {
- return Arrays.asList(MigrationVersion.v1_1, MigrationVersion.v1_2, MigrationVersion.v1_3);
+ return Arrays.asList(MigrationVersion.v1_2, MigrationVersion.v1_3);
}
/**
@@ -753,219 +752,6 @@ public class WindowOperatorMigrationTest {
testHarness.close();
}
- /**
- * Manually run this to write binary snapshot data.
- */
- @Ignore
- @Test
- public void writeAggregatingAlignedProcessingTimeWindowsSnapshot() throws Exception {
- TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
-
- AggregatingProcessingTimeWindowOperator<String, Tuple2<String, Integer>> operator =
- new AggregatingProcessingTimeWindowOperator<>(
- new ReduceFunction<Tuple2<String, Integer>>() {
- private static final long serialVersionUID = -8913160567151867987L;
-
- @Override
- public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
- return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
- }
- },
- new TupleKeySelector(),
- BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
- inputType.createSerializer(new ExecutionConfig()),
- 3000,
- 3000);
-
- OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
-
- testHarness.open();
-
- testHarness.setProcessingTime(3);
-
- // timestamp is ignored in processing time
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), Long.MAX_VALUE));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 7000));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 7000));
-
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
-
- // do a snapshot, close and restore again
- OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
- OperatorSnapshotUtil.writeStateHandle(
- snapshot,
- "src/test/resources/win-op-migration-test-aggr-aligned-flink" + flinkGenerateSavepointVersion + "-snapshot");
- testHarness.close();
- }
-
- @Test
- public void testRestoreAggregatingAlignedProcessingTimeWindows() throws Exception {
- final int windowSize = 3;
-
- TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
-
- ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
- new SumReducer(),
- inputType.createSerializer(new ExecutionConfig()));
-
- WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
- TumblingProcessingTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
- new TimeWindow.Serializer(),
- new TupleKeySelector(),
- BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
- stateDesc,
- new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<String, TimeWindow, Tuple2<String, Integer>>()),
- ProcessingTimeTrigger.create(),
- 0,
- null /* late data output tag */,
- LegacyWindowOperatorType.FAST_AGGREGATING);
-
- OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
-
- ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
-
- testHarness.setup();
-
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
- OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-aggr-aligned-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
-
- testHarness.open();
-
- testHarness.setProcessingTime(5000);
-
- expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 3), 2999));
- expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 2), 2999));
-
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
-
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
-
- testHarness.setProcessingTime(7000);
-
- expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 3), 5999));
-
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
-
- testHarness.close();
- }
-
- /**
- * Manually run this to write binary snapshot data.
- */
- @Ignore
- @Test
- public void writeAlignedProcessingTimeWindowsSnapshot() throws Exception {
- TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
-
- AccumulatingProcessingTimeWindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>> operator =
- new AccumulatingProcessingTimeWindowOperator<>(
- new InternalIterableWindowFunction<>(new WindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>() {
-
- private static final long serialVersionUID = 6551516443265733803L;
-
- @Override
- public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<Tuple2<String, Integer>> out) throws Exception {
- int sum = 0;
- for (Tuple2<String, Integer> anInput : input) {
- sum += anInput.f1;
- }
- out.collect(new Tuple2<>(s, sum));
- }
- }),
- new TupleKeySelector(),
- BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
- inputType.createSerializer(new ExecutionConfig()),
- 3000,
- 3000);
-
- OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
-
- testHarness.open();
-
- testHarness.setProcessingTime(3);
-
- // timestamp is ignored in processing time
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), Long.MAX_VALUE));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 7000));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), 7000));
-
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
-
- // do a snapshot, close and restore again
- OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
- OperatorSnapshotUtil.writeStateHandle(
- snapshot,
- "src/test/resources/win-op-migration-test-accum-aligned-flink" + flinkGenerateSavepointVersion + "-snapshot");
- testHarness.close();
- }
-
- @Test
- public void testRestoreAccumulatingAlignedProcessingTimeWindows() throws Exception {
- final int windowSize = 3;
-
- TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
-
- ReducingStateDescriptor<Tuple2<String, Integer>> stateDesc = new ReducingStateDescriptor<>("window-contents",
- new SumReducer(),
- inputType.createSerializer(new ExecutionConfig()));
-
- WindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>, TimeWindow> operator = new WindowOperator<>(
- TumblingProcessingTimeWindows.of(Time.of(windowSize, TimeUnit.SECONDS)),
- new TimeWindow.Serializer(),
- new TupleKeySelector(),
- BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
- stateDesc,
- new InternalSingleValueWindowFunction<>(new PassThroughWindowFunction<String, TimeWindow, Tuple2<String, Integer>>()),
- ProcessingTimeTrigger.create(),
- 0,
- null /* late data output tag */,
- LegacyWindowOperatorType.FAST_ACCUMULATING);
-
- OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
- new KeyedOneInputStreamOperatorTestHarness<>(operator, new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
-
- ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
-
- testHarness.setup();
-
- MigrationTestUtil.restoreFromSnapshot(
- testHarness,
- OperatorSnapshotUtil.getResourceFilename(
- "win-op-migration-test-accum-aligned-flink" + testMigrateVersion + "-snapshot"),
- testMigrateVersion);
-
- testHarness.open();
-
- testHarness.setProcessingTime(5000);
-
- expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 3), 2999));
- expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 2), 2999));
-
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
-
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
- testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), 7000));
-
- testHarness.setProcessingTime(7000);
-
- expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 3), 5999));
-
- TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new Tuple2ResultSortComparator());
-
- testHarness.close();
- }
-
private static class TupleKeySelector implements KeySelector<Tuple2<String, Integer>, String> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java
index 8748ed4..821438e 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java
@@ -63,7 +63,6 @@ import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.util.Collector;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Test;
import java.util.concurrent.TimeUnit;
@@ -336,32 +335,6 @@ public class WindowTranslationTest {
processElementAndEnsureOutput(winOperator, winOperator.getKeySelector(), BasicTypeInfo.STRING_TYPE_INFO, new Tuple2<>("hello", 1));
}
-
- /**
- * Ignored because we currently don't have the fast processing-time window operator.
- */
- @Test
- @SuppressWarnings("rawtypes")
- @Ignore
- public void testReduceFastProcessingTime() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
-
- DataStream<Tuple2<String, Integer>> source = env.fromElements(Tuple2.of("hello", 1), Tuple2.of("hello", 2));
-
- DataStream<Tuple2<String, Integer>> window = source
- .keyBy(new TupleKeySelector())
- .window(SlidingProcessingTimeWindows.of(Time.of(1, TimeUnit.SECONDS), Time.of(100, TimeUnit.MILLISECONDS)))
- .reduce(new DummyReducer());
-
- OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform =
- (OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>) window.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator = transform.getOperator();
- Assert.assertTrue(operator instanceof AggregatingProcessingTimeWindowOperator);
-
- processElementAndEnsureOutput(operator, null, BasicTypeInfo.STRING_TYPE_INFO, new Tuple2<>("hello", 1));
- }
-
@Test
@SuppressWarnings("rawtypes")
public void testReduceWithWindowFunctionEventTime() throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
index 14ae733..f73499c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
@@ -64,7 +64,6 @@ import org.apache.flink.runtime.taskmanager.TaskManagerActions;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.runtime.util.TestingTaskManagerRuntimeInfo;
import org.apache.flink.streaming.api.TimeCharacteristic;
-import org.apache.flink.streaming.api.checkpoint.Checkpointed;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.graph.StreamConfig;
@@ -75,7 +74,6 @@ import org.junit.Test;
import java.io.EOFException;
import java.io.IOException;
-import java.io.Serializable;
import java.net.URL;
import java.util.Collection;
import java.util.Collections;
@@ -106,12 +104,6 @@ public class InterruptSensitiveRestoreTest {
private static final int OPERATOR_RAW = 1;
private static final int KEYED_MANAGED = 2;
private static final int KEYED_RAW = 3;
- private static final int LEGACY = 4;
-
- @Test
- public void testRestoreWithInterruptLegacy() throws Exception {
- testRestoreWithInterrupt(LEGACY);
- }
@Test
public void testRestoreWithInterruptOperatorManaged() throws Exception {
@@ -137,18 +129,15 @@ public class InterruptSensitiveRestoreTest {
IN_RESTORE_LATCH.reset();
Configuration taskConfig = new Configuration();
- StreamConfig streamConfig = new StreamConfig(taskConfig);
- streamConfig.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);
+ StreamConfig cfg = new StreamConfig(taskConfig);
+ cfg.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);
switch (mode) {
case OPERATOR_MANAGED:
case OPERATOR_RAW:
case KEYED_MANAGED:
case KEYED_RAW:
- streamConfig.setStateKeySerializer(IntSerializer.INSTANCE);
- streamConfig.setStreamOperator(new StreamSource<>(new TestSource()));
- break;
- case LEGACY:
- streamConfig.setStreamOperator(new StreamSource<>(new TestSourceLegacy()));
+ cfg.setStateKeySerializer(IntSerializer.INSTANCE);
+ cfg.setStreamOperator(new StreamSource<>(new TestSource(mode)));
break;
default:
throw new IllegalArgumentException();
@@ -156,7 +145,7 @@ public class InterruptSensitiveRestoreTest {
StreamStateHandle lockingHandle = new InterruptLockingStateHandle();
- Task task = createTask(streamConfig, taskConfig, lockingHandle, mode);
+ Task task = createTask(cfg, taskConfig, lockingHandle, mode);
// start the task and wait until it is in "restore"
task.startTaskThread();
@@ -180,16 +169,15 @@ public class InterruptSensitiveRestoreTest {
// ------------------------------------------------------------------------
private static Task createTask(
- StreamConfig streamConfig,
- Configuration taskConfig,
- StreamStateHandle state,
- int mode) throws IOException {
+ StreamConfig streamConfig,
+ Configuration taskConfig,
+ StreamStateHandle state,
+ int mode) throws IOException {
NetworkEnvironment networkEnvironment = mock(NetworkEnvironment.class);
when(networkEnvironment.createKvStateTaskRegistry(any(JobID.class), any(JobVertexID.class)))
.thenReturn(mock(TaskKvStateRegistry.class));
- StreamStateHandle operatorState = null;
Collection<KeyedStateHandle> keyedStateFromBackend = Collections.emptyList();
Collection<KeyedStateHandle> keyedStateFromStream = Collections.emptyList();
Collection<OperatorStateHandle> operatorStateBackend = Collections.emptyList();
@@ -206,7 +194,7 @@ public class InterruptSensitiveRestoreTest {
Collections.singletonList(new OperatorStateHandle(operatorStateMetadata, state));
List<KeyedStateHandle> keyedStateHandles =
- Collections.<KeyedStateHandle>singletonList(new KeyGroupsStateHandle(keyGroupRangeOffsets, state));
+ Collections.singletonList(new KeyGroupsStateHandle(keyGroupRangeOffsets, state));
switch (mode) {
case OPERATOR_MANAGED:
@@ -221,15 +209,11 @@ public class InterruptSensitiveRestoreTest {
case KEYED_RAW:
keyedStateFromStream = keyedStateHandles;
break;
- case LEGACY:
- operatorState = state;
- break;
default:
throw new IllegalArgumentException();
}
OperatorSubtaskState operatorSubtaskState = new OperatorSubtaskState(
- operatorState,
operatorStateBackend,
operatorStateStream,
keyedStateFromBackend,
@@ -238,14 +222,13 @@ public class InterruptSensitiveRestoreTest {
JobVertexID jobVertexID = new JobVertexID();
OperatorID operatorID = OperatorID.fromJobVertexID(jobVertexID);
streamConfig.setOperatorID(operatorID);
-
TaskStateSnapshot stateSnapshot = new TaskStateSnapshot();
stateSnapshot.putSubtaskStateByOperatorID(operatorID, operatorSubtaskState);
JobInformation jobInformation = new JobInformation(
new JobID(),
"test job name",
new SerializedValue<>(new ExecutionConfig()),
- taskConfig,
+ new Configuration(),
Collections.<BlobKey>emptyList(),
Collections.<URL>emptyList());
@@ -302,11 +285,11 @@ public class InterruptSensitiveRestoreTest {
FSDataInputStream is = new FSDataInputStream() {
@Override
- public void seek(long desired) throws IOException {
+ public void seek(long desired) {
}
@Override
- public long getPos() throws IOException {
+ public long getPos() {
return 0;
}
@@ -358,33 +341,15 @@ public class InterruptSensitiveRestoreTest {
// ------------------------------------------------------------------------
- private static class TestSourceLegacy implements SourceFunction<Object>, Checkpointed<Serializable> {
+ private static class TestSource implements SourceFunction<Object>, CheckpointedFunction {
private static final long serialVersionUID = 1L;
+ private final int testType;
- @Override
- public void run(SourceContext<Object> ctx) throws Exception {
- fail("should never be called");
+ public TestSource(int testType) {
+ this.testType = testType;
}
@Override
- public void cancel() {}
-
- @Override
- public Serializable snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- fail("should never be called");
- return null;
- }
-
- @Override
- public void restoreState(Serializable state) throws Exception {
- fail("should never be called");
- }
- }
-
- private static class TestSource implements SourceFunction<Object>, CheckpointedFunction {
- private static final long serialVersionUID = 1L;
-
- @Override
public void run(SourceContext<Object> ctx) throws Exception {
fail("should never be called");
}
@@ -399,6 +364,8 @@ public class InterruptSensitiveRestoreTest {
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
+ // raw keyed state is already read by timer service, all others to initialize the context...we only need to
+ // trigger this manually.
((StateInitializationContext) context).getRawOperatorStateInputs().iterator().next().getStream().read();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
index 3190620..8d80d66 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
@@ -28,8 +28,6 @@ import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.core.fs.FSDataOutputStream;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
@@ -47,21 +45,18 @@ import org.apache.flink.streaming.api.graph.StreamEdge;
import org.apache.flink.streaming.api.graph.StreamNode;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.operators.StreamCheckpointedOperator;
import org.apache.flink.streaming.api.operators.StreamMap;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.partitioner.BroadcastPartitioner;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.streamstatus.StreamStatus;
import org.apache.flink.streaming.util.TestHarnessUtil;
-import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
import org.junit.Assert;
import org.junit.Test;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -69,7 +64,6 @@ import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
@@ -78,7 +72,6 @@ import scala.concurrent.duration.Deadline;
import scala.concurrent.duration.FiniteDuration;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -93,7 +86,7 @@ import static org.junit.Assert.fail;
public class OneInputStreamTaskTest extends TestLogger {
private static final ListStateDescriptor<Integer> TEST_DESCRIPTOR =
- new ListStateDescriptor<>("test", new IntSerializer());
+ new ListStateDescriptor<>("test", new IntSerializer());
/**
* This test verifies that open() and close() are correctly called. This test also verifies
@@ -129,8 +122,8 @@ public class OneInputStreamTaskTest extends TestLogger {
assertTrue("RichFunction methods where not called.", TestOpenCloseMapFunction.closeCalled);
TestHarnessUtil.assertOutputEquals("Output was not correct.",
- expectedOutput,
- testHarness.getOutput());
+ expectedOutput,
+ testHarness.getOutput());
}
/**
@@ -174,8 +167,8 @@ public class OneInputStreamTaskTest extends TestLogger {
testHarness.waitForInputProcessing();
expectedOutput.add(new Watermark(initialTime));
TestHarnessUtil.assertOutputEquals("Output was not correct.",
- expectedOutput,
- testHarness.getOutput());
+ expectedOutput,
+ testHarness.getOutput());
// contrary to checkpoint barriers these elements are not blocked by watermarks
testHarness.processElement(new StreamRecord<String>("Hello", initialTime));
@@ -215,7 +208,7 @@ public class OneInputStreamTaskTest extends TestLogger {
testHarness.processElement(new Watermark(initialTime + 6), 0, 0);
testHarness.processElement(new Watermark(initialTime + 5), 1, 1); // this watermark should be advanced first
testHarness.processElement(StreamStatus.IDLE, 1, 1); // once this is acknowledged,
- // watermark (initial + 6) should be forwarded
+ // watermark (initial + 6) should be forwarded
testHarness.waitForInputProcessing();
expectedOutput.add(new Watermark(initialTime + 5));
expectedOutput.add(new Watermark(initialTime + 6));
@@ -263,21 +256,16 @@ public class OneInputStreamTaskTest extends TestLogger {
// ------------------ setup the chain ------------------
TriggerableFailOnWatermarkTestOperator headOperator = new TriggerableFailOnWatermarkTestOperator();
- OperatorID headOperatorId = new OperatorID();
-
StreamConfig headOperatorConfig = testHarness.getStreamConfig();
WatermarkGeneratingTestOperator watermarkOperator = new WatermarkGeneratingTestOperator();
- OperatorID watermarkOperatorId = new OperatorID();
-
StreamConfig watermarkOperatorConfig = new StreamConfig(new Configuration());
TriggerableFailOnWatermarkTestOperator tailOperator = new TriggerableFailOnWatermarkTestOperator();
- OperatorID tailOperatorId = new OperatorID();
StreamConfig tailOperatorConfig = new StreamConfig(new Configuration());
headOperatorConfig.setStreamOperator(headOperator);
- headOperatorConfig.setOperatorID(headOperatorId);
+ headOperatorConfig.setOperatorID(new OperatorID(42L, 42L));
headOperatorConfig.setChainStart();
headOperatorConfig.setChainIndex(0);
headOperatorConfig.setChainedOutputs(Collections.singletonList(new StreamEdge(
@@ -290,7 +278,7 @@ public class OneInputStreamTaskTest extends TestLogger {
)));
watermarkOperatorConfig.setStreamOperator(watermarkOperator);
- watermarkOperatorConfig.setOperatorID(watermarkOperatorId);
+ watermarkOperatorConfig.setOperatorID(new OperatorID(4711L, 42L));
watermarkOperatorConfig.setTypeSerializerIn1(StringSerializer.INSTANCE);
watermarkOperatorConfig.setChainIndex(1);
watermarkOperatorConfig.setChainedOutputs(Collections.singletonList(new StreamEdge(
@@ -312,7 +300,7 @@ public class OneInputStreamTaskTest extends TestLogger {
null));
tailOperatorConfig.setStreamOperator(tailOperator);
- tailOperatorConfig.setOperatorID(tailOperatorId);
+ tailOperatorConfig.setOperatorID(new OperatorID(123L, 123L));
tailOperatorConfig.setTypeSerializerIn1(StringSerializer.INSTANCE);
tailOperatorConfig.setBufferTimeout(0);
tailOperatorConfig.setChainIndex(2);
@@ -555,13 +543,11 @@ public class OneInputStreamTaskTest extends TestLogger {
long checkpointId = 1L;
long checkpointTimestamp = 1L;
- long recoveryTimestamp = 3L;
- long seed = 2L;
int numberChainedTasks = 11;
StreamConfig streamConfig = testHarness.getStreamConfig();
- configureChainedTestingStreamOperator(streamConfig, numberChainedTasks, seed, recoveryTimestamp);
+ configureChainedTestingStreamOperator(streamConfig, numberChainedTasks);
AcknowledgeStreamMockEnvironment env = new AcknowledgeStreamMockEnvironment(
testHarness.jobConfig,
@@ -599,7 +585,7 @@ public class OneInputStreamTaskTest extends TestLogger {
StreamConfig restoredTaskStreamConfig = restoredTaskHarness.getStreamConfig();
- configureChainedTestingStreamOperator(restoredTaskStreamConfig, numberChainedTasks, seed, recoveryTimestamp);
+ configureChainedTestingStreamOperator(restoredTaskStreamConfig, numberChainedTasks);
TaskStateSnapshot stateHandles = env.getCheckpointStateHandles();
Assert.assertEquals(numberChainedTasks, stateHandles.getSubtaskStateMappings().size());
@@ -625,16 +611,12 @@ public class OneInputStreamTaskTest extends TestLogger {
private void configureChainedTestingStreamOperator(
StreamConfig streamConfig,
- int numberChainedTasks,
- long seed,
- long recoveryTimestamp) {
+ int numberChainedTasks) {
Preconditions.checkArgument(numberChainedTasks >= 1, "The operator chain must at least " +
"contain one operator.");
- Random random = new Random(seed);
-
- TestingStreamOperator<Integer, Integer> previousOperator = new TestingStreamOperator<>(random.nextLong(), recoveryTimestamp);
+ TestingStreamOperator<Integer, Integer> previousOperator = new TestingStreamOperator<>();
streamConfig.setStreamOperator(previousOperator);
streamConfig.setOperatorID(new OperatorID(0L, 0L));
@@ -643,7 +625,7 @@ public class OneInputStreamTaskTest extends TestLogger {
List<StreamEdge> outputEdges = new ArrayList<>(numberChainedTasks - 1);
for (int chainedIndex = 1; chainedIndex < numberChainedTasks; chainedIndex++) {
- TestingStreamOperator<Integer, Integer> chainedOperator = new TestingStreamOperator<>(random.nextLong(), recoveryTimestamp);
+ TestingStreamOperator<Integer, Integer> chainedOperator = new TestingStreamOperator<>();
StreamConfig chainedConfig = new StreamConfig(new Configuration());
chainedConfig.setStreamOperator(chainedOperator);
chainedConfig.setOperatorID(new OperatorID(0L, chainedIndex));
@@ -702,17 +684,17 @@ public class OneInputStreamTaskTest extends TestLogger {
}
AcknowledgeStreamMockEnvironment(
- Configuration jobConfig, Configuration taskConfig,
- ExecutionConfig executionConfig, long memorySize,
- MockInputSplitProvider inputSplitProvider, int bufferSize) {
+ Configuration jobConfig, Configuration taskConfig,
+ ExecutionConfig executionConfig, long memorySize,
+ MockInputSplitProvider inputSplitProvider, int bufferSize) {
super(jobConfig, taskConfig, executionConfig, memorySize, inputSplitProvider, bufferSize);
}
@Override
public void acknowledgeCheckpoint(
- long checkpointId,
- CheckpointMetrics checkpointMetrics,
- TaskStateSnapshot checkpointStateHandles) {
+ long checkpointId,
+ CheckpointMetrics checkpointMetrics,
+ TaskStateSnapshot checkpointStateHandles) {
this.checkpointId = checkpointId;
this.checkpointStateHandles = checkpointStateHandles;
@@ -729,19 +711,14 @@ public class OneInputStreamTaskTest extends TestLogger {
}
private static class TestingStreamOperator<IN, OUT>
- extends AbstractStreamOperator<OUT>
- implements OneInputStreamOperator<IN, OUT>, StreamCheckpointedOperator {
+ extends AbstractStreamOperator<OUT>
+ implements OneInputStreamOperator<IN, OUT> {
private static final long serialVersionUID = 774614855940397174L;
public static int numberRestoreCalls = 0;
public static int numberSnapshotCalls = 0;
- private final long seed;
- private final long recoveryTimestamp;
-
- private transient Random random;
-
@Override
public void open() throws Exception {
super.open();
@@ -767,7 +744,7 @@ public class OneInputStreamTaskTest extends TestLogger {
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
ListState<Integer> partitionableState =
- getOperatorStateBackend().getListState(TEST_DESCRIPTOR);
+ getOperatorStateBackend().getListState(TEST_DESCRIPTOR);
partitionableState.clear();
partitionableState.add(42);
@@ -778,59 +755,14 @@ public class OneInputStreamTaskTest extends TestLogger {
@Override
public void initializeState(StateInitializationContext context) throws Exception {
-
- }
-
- TestingStreamOperator(long seed, long recoveryTimestamp) {
- this.seed = seed;
- this.recoveryTimestamp = recoveryTimestamp;
- }
-
- @Override
- public void processElement(StreamRecord<IN> element) throws Exception {
-
- }
-
- @Override
- public void snapshotState(FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
- if (random == null) {
- random = new Random(seed);
+ if (context.isRestored()) {
+ ++numberRestoreCalls;
}
-
- Serializable functionState = generateFunctionState();
- Integer operatorState = generateOperatorState();
-
- InstantiationUtil.serializeObject(out, functionState);
- InstantiationUtil.serializeObject(out, operatorState);
}
@Override
- public void restoreState(FSDataInputStream in) throws Exception {
- numberRestoreCalls++;
-
- if (random == null) {
- random = new Random(seed);
- }
-
- assertEquals(this.recoveryTimestamp, recoveryTimestamp);
-
- assertNotNull(in);
-
- ClassLoader cl = Thread.currentThread().getContextClassLoader();
-
- Serializable functionState = InstantiationUtil.deserializeObject(in, cl);
- Integer operatorState = InstantiationUtil.deserializeObject(in, cl);
-
- assertEquals(random.nextInt(), functionState);
- assertEquals(random.nextInt(), (int) operatorState);
- }
-
- private Serializable generateFunctionState() {
- return random.nextInt();
- }
+ public void processElement(StreamRecord<IN> element) throws Exception {
- private Integer generateOperatorState() {
- return random.nextInt();
}
}
@@ -913,8 +845,8 @@ public class OneInputStreamTaskTest extends TestLogger {
* <p>If it receives a watermark when it's not expecting one, it'll throw an exception and fail.
*/
private static class TriggerableFailOnWatermarkTestOperator
- extends AbstractStreamOperator<String>
- implements OneInputStreamOperator<String, String> {
+ extends AbstractStreamOperator<String>
+ implements OneInputStreamOperator<String, String> {
private static final long serialVersionUID = 2048954179291813243L;
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
index 08c3207..a2dc6c4 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
@@ -81,7 +81,6 @@ import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OperatorSnapshotResult;
import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.api.operators.StreamCheckpointedOperator;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.operators.StreamSource;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -139,7 +138,6 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import static org.mockito.Mockito.withSettings;
import static org.powermock.api.mockito.PowerMockito.whenNew;
/**
@@ -307,9 +305,9 @@ public class StreamTaskTest extends TestLogger {
streamTask.setEnvironment(mockEnvironment);
// mock the operators
- StreamOperator<?> streamOperator1 = mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
- StreamOperator<?> streamOperator2 = mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
- StreamOperator<?> streamOperator3 = mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
+ StreamOperator<?> streamOperator1 = mock(StreamOperator.class);
+ StreamOperator<?> streamOperator2 = mock(StreamOperator.class);
+ StreamOperator<?> streamOperator3 = mock(StreamOperator.class);
// mock the returned snapshots
OperatorSnapshotResult operatorSnapshotResult1 = mock(OperatorSnapshotResult.class);
@@ -321,15 +319,6 @@ public class StreamTaskTest extends TestLogger {
when(streamOperator2.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult2);
when(streamOperator3.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenThrow(testException);
- // mock the returned legacy snapshots
- StreamStateHandle streamStateHandle1 = mock(StreamStateHandle.class);
- StreamStateHandle streamStateHandle2 = mock(StreamStateHandle.class);
- StreamStateHandle streamStateHandle3 = mock(StreamStateHandle.class);
-
- when(streamOperator1.snapshotLegacyOperatorState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(streamStateHandle1);
- when(streamOperator2.snapshotLegacyOperatorState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(streamStateHandle2);
- when(streamOperator3.snapshotLegacyOperatorState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(streamStateHandle3);
-
OperatorID operatorID1 = new OperatorID();
OperatorID operatorID2 = new OperatorID();
OperatorID operatorID3 = new OperatorID();
@@ -359,10 +348,6 @@ public class StreamTaskTest extends TestLogger {
verify(operatorSnapshotResult1).cancel();
verify(operatorSnapshotResult2).cancel();
-
- verify(streamStateHandle1).discardState();
- verify(streamStateHandle2).discardState();
- verify(streamStateHandle3).discardState();
}
/**
@@ -384,12 +369,12 @@ public class StreamTaskTest extends TestLogger {
CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, timestamp);
streamTask.setEnvironment(mockEnvironment);
- StreamOperator<?> streamOperator1 = mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
- StreamOperator<?> streamOperator2 = mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
- StreamOperator<?> streamOperator3 = mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
-
- // mock the new state handles / futures
+ // mock the operators
+ StreamOperator<?> streamOperator1 = mock(StreamOperator.class);
+ StreamOperator<?> streamOperator2 = mock(StreamOperator.class);
+ StreamOperator<?> streamOperator3 = mock(StreamOperator.class);
+ // mock the new state operator snapshots
OperatorSnapshotResult operatorSnapshotResult1 = mock(OperatorSnapshotResult.class);
OperatorSnapshotResult operatorSnapshotResult2 = mock(OperatorSnapshotResult.class);
OperatorSnapshotResult operatorSnapshotResult3 = mock(OperatorSnapshotResult.class);
@@ -403,15 +388,6 @@ public class StreamTaskTest extends TestLogger {
when(streamOperator2.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult2);
when(streamOperator3.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult3);
- // mock the legacy state snapshot
- StreamStateHandle streamStateHandle1 = mock(StreamStateHandle.class);
- StreamStateHandle streamStateHandle2 = mock(StreamStateHandle.class);
- StreamStateHandle streamStateHandle3 = mock(StreamStateHandle.class);
-
- when(streamOperator1.snapshotLegacyOperatorState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(streamStateHandle1);
- when(streamOperator2.snapshotLegacyOperatorState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(streamStateHandle2);
- when(streamOperator3.snapshotLegacyOperatorState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(streamStateHandle3);
-
OperatorID operatorID1 = new OperatorID();
OperatorID operatorID2 = new OperatorID();
OperatorID operatorID3 = new OperatorID();
@@ -438,10 +414,6 @@ public class StreamTaskTest extends TestLogger {
verify(operatorSnapshotResult1).cancel();
verify(operatorSnapshotResult2).cancel();
verify(operatorSnapshotResult3).cancel();
-
- verify(streamStateHandle1).discardState();
- verify(streamStateHandle2).discardState();
- verify(streamStateHandle3).discardState();
}
/**
@@ -481,7 +453,7 @@ public class StreamTaskTest extends TestLogger {
CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, timestamp);
streamTask.setEnvironment(mockEnvironment);
- StreamOperator<?> streamOperator = mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
+ StreamOperator<?> streamOperator = mock(StreamOperator.class);
KeyedStateHandle managedKeyedStateHandle = mock(KeyedStateHandle.class);
KeyedStateHandle rawKeyedStateHandle = mock(KeyedStateHandle.class);
@@ -581,7 +553,6 @@ public class StreamTaskTest extends TestLogger {
whenNew(OperatorSubtaskState.class).
withArguments(
- any(StreamStateHandle.class),
anyCollectionOf(OperatorStateHandle.class),
anyCollectionOf(OperatorStateHandle.class),
anyCollectionOf(KeyedStateHandle.class),
@@ -593,11 +564,10 @@ public class StreamTaskTest extends TestLogger {
completeSubtask.await();
Object[] arguments = invocation.getArguments();
return new OperatorSubtaskState(
- (StreamStateHandle) arguments[0],
+ (OperatorStateHandle) arguments[0],
(OperatorStateHandle) arguments[1],
- (OperatorStateHandle) arguments[2],
- (KeyedStateHandle) arguments[3],
- (KeyedStateHandle) arguments[4]
+ (KeyedStateHandle) arguments[2],
+ (KeyedStateHandle) arguments[3]
);
}
});
@@ -606,7 +576,7 @@ public class StreamTaskTest extends TestLogger {
CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, timestamp);
streamTask.setEnvironment(mockEnvironment);
- final StreamOperator<?> streamOperator = mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
+ final StreamOperator<?> streamOperator = mock(StreamOperator.class);
final OperatorID operatorID = new OperatorID();
when(streamOperator.getOperatorID()).thenReturn(operatorID);
@@ -717,7 +687,7 @@ public class StreamTaskTest extends TestLogger {
// mock the operators
StreamOperator<?> statelessOperator =
- mock(StreamOperator.class, withSettings().extraInterfaces(StreamCheckpointedOperator.class));
+ mock(StreamOperator.class);
final OperatorID operatorID = new OperatorID();
when(statelessOperator.getOperatorID()).thenReturn(operatorID);
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
index 720346a..9156f34 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java
@@ -25,11 +25,6 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.CloseableRegistry;
-import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.core.fs.FSDataOutputStream;
-import org.apache.flink.migration.runtime.checkpoint.savepoint.SavepointV0Serializer;
-import org.apache.flink.migration.streaming.runtime.tasks.StreamTaskState;
-import org.apache.flink.migration.util.MigrationInstantiationUtil;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.OperatorStateRepartitioner;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
@@ -41,12 +36,10 @@ import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.KeyGroupRange;
-import org.apache.flink.runtime.state.KeyGroupsStateHandle;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateBackend;
import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.StateBackend;
-import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.graph.StreamConfig;
@@ -54,7 +47,6 @@ import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.AbstractStreamOperatorTest;
import org.apache.flink.streaming.api.operators.OperatorSnapshotResult;
import org.apache.flink.streaming.api.operators.Output;
-import org.apache.flink.streaming.api.operators.StreamCheckpointedOperator;
import org.apache.flink.streaming.api.operators.StreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
@@ -72,7 +64,6 @@ import org.apache.flink.util.Preconditions;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-import java.io.FileInputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -308,36 +299,6 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
setupCalled = true;
}
- public void initializeStateFromLegacyCheckpoint(String checkpointFilename) throws Exception {
-
- FileInputStream fin = new FileInputStream(checkpointFilename);
- StreamTaskState state = MigrationInstantiationUtil.deserializeObject(fin, ClassLoader.getSystemClassLoader());
- fin.close();
-
- if (!setupCalled) {
- setup();
- }
-
- StreamStateHandle stateHandle = SavepointV0Serializer.convertOperatorAndFunctionState(state);
-
- List<KeyedStateHandle> keyGroupStatesList = new ArrayList<>();
- if (state.getKvStates() != null) {
- KeyGroupsStateHandle keyedStateHandle = SavepointV0Serializer.convertKeyedBackendState(
- state.getKvStates(),
- environment.getTaskInfo().getIndexOfThisSubtask(),
- 0);
- keyGroupStatesList.add(keyedStateHandle);
- }
-
- // finally calling the initializeState() with the legacy operatorStateHandles
- initializeState(new OperatorStateHandles(0,
- stateHandle,
- keyGroupStatesList,
- Collections.<KeyedStateHandle>emptyList(),
- Collections.<OperatorStateHandle>emptyList(),
- Collections.<OperatorStateHandle>emptyList()));
- }
-
/**
* Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#initializeState(OperatorSubtaskState)}.
* Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#setup(StreamTask, StreamConfig, Output)}
@@ -397,7 +358,6 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
numSubtasks).get(subtaskIndex);
OperatorSubtaskState massagedOperatorStateHandles = new OperatorSubtaskState(
- operatorStateHandles.getLegacyOperatorState(),
nullToEmptyCollection(localManagedOperatorState),
nullToEmptyCollection(localRawOperatorState),
nullToEmptyCollection(localManagedKeyGroupState),
@@ -473,7 +433,6 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
return new OperatorStateHandles(
0,
- null,
mergedManagedKeyedState,
mergedRawKeyedState,
mergedManagedOperatorState,
@@ -497,8 +456,6 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
*/
public OperatorStateHandles snapshot(long checkpointId, long timestamp) throws Exception {
- CheckpointStreamFactory streamFactory = stateBackend.createStreamFactory(new JobID(), "test_op");
-
OperatorSnapshotResult operatorStateResult = operator.snapshotState(
checkpointId,
timestamp,
@@ -510,21 +467,8 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
OperatorStateHandle opManaged = FutureUtil.runIfNotDoneAndGet(operatorStateResult.getOperatorStateManagedFuture());
OperatorStateHandle opRaw = FutureUtil.runIfNotDoneAndGet(operatorStateResult.getOperatorStateRawFuture());
- // also snapshot legacy state, if any
- StreamStateHandle legacyStateHandle = null;
-
- if (operator instanceof StreamCheckpointedOperator) {
-
- final CheckpointStreamFactory.CheckpointStateOutputStream outStream =
- streamFactory.createCheckpointStateOutputStream(checkpointId, timestamp);
-
- ((StreamCheckpointedOperator) operator).snapshotState(outStream, checkpointId, timestamp);
- legacyStateHandle = outStream.closeAndGetHandle();
- }
-
return new OperatorStateHandles(
0,
- legacyStateHandle,
keyedManaged != null ? Collections.singletonList(keyedManaged) : null,
keyedRaw != null ? Collections.singletonList(keyedRaw) : null,
opManaged != null ? Collections.singletonList(opManaged) : null,
@@ -532,24 +476,6 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
}
/**
- * Calls {@link StreamCheckpointedOperator#snapshotState(FSDataOutputStream, long, long)} if
- * the operator implements this interface.
- */
- @Deprecated
- public StreamStateHandle snapshotLegacy(long checkpointId, long timestamp) throws Exception {
-
- CheckpointStreamFactory.CheckpointStateOutputStream outStream = stateBackend.createStreamFactory(
- new JobID(),
- "test_op").createCheckpointStateOutputStream(checkpointId, timestamp);
- if (operator instanceof StreamCheckpointedOperator) {
- ((StreamCheckpointedOperator) operator).snapshotState(outStream, checkpointId, timestamp);
- return outStream.closeAndGetHandle();
- } else {
- throw new RuntimeException("Operator is not StreamCheckpointedOperator");
- }
- }
-
- /**
* Calls {@link org.apache.flink.streaming.api.operators.StreamOperator#notifyOfCompletedCheckpoint(long)} ()}.
*/
public void notifyOfCompletedCheckpoint(long checkpointId) throws Exception {
@@ -557,22 +483,6 @@ public class AbstractStreamOperatorTestHarness<OUT> implements AutoCloseable {
}
/**
- * Calls {@link StreamCheckpointedOperator#restoreState(FSDataInputStream)} if
- * the operator implements this interface.
- */
- @Deprecated
- @SuppressWarnings("deprecation")
- public void restore(StreamStateHandle snapshot) throws Exception {
- if (operator instanceof StreamCheckpointedOperator) {
- try (FSDataInputStream in = snapshot.openInputStream()) {
- ((StreamCheckpointedOperator) operator).restoreState(in);
- }
- } else {
- throw new RuntimeException("Operator is not StreamCheckpointedOperator");
- }
- }
-
- /**
* Calls close and dispose on the operator.
*/
public void close() throws Exception {
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/KeyedOneInputStreamOperatorTestHarness.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/KeyedOneInputStreamOperatorTestHarness.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/KeyedOneInputStreamOperatorTestHarness.java
index 0d42d9f..c2ec63a 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/KeyedOneInputStreamOperatorTestHarness.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/KeyedOneInputStreamOperatorTestHarness.java
@@ -23,33 +23,23 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.ClosureCleaner;
import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.StateAssignmentOperation;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
-import org.apache.flink.runtime.state.CheckpointStreamFactory;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.flink.runtime.state.KeyedStateHandle;
-import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.runtime.state.heap.HeapKeyedStateBackend;
-import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.operators.StreamCheckpointedOperator;
import org.apache.flink.streaming.runtime.tasks.OperatorStateHandles;
import org.apache.flink.util.Migration;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
-import java.util.concurrent.RunnableFuture;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.anyInt;
@@ -142,61 +132,6 @@ public class KeyedOneInputStreamOperatorTestHarness<K, IN, OUT>
}
}
- /**
- *
- */
- @Override
- public StreamStateHandle snapshotLegacy(long checkpointId, long timestamp) throws Exception {
- // simply use an in-memory handle
- MemoryStateBackend backend = new MemoryStateBackend();
-
- CheckpointStreamFactory streamFactory = backend.createStreamFactory(new JobID(), "test_op");
- CheckpointStreamFactory.CheckpointStateOutputStream outStream =
- streamFactory.createCheckpointStateOutputStream(checkpointId, timestamp);
-
- if (operator instanceof StreamCheckpointedOperator) {
- ((StreamCheckpointedOperator) operator).snapshotState(outStream, checkpointId, timestamp);
- }
-
- if (keyedStateBackend != null) {
- RunnableFuture<KeyedStateHandle> keyedSnapshotRunnable = keyedStateBackend.snapshot(
- checkpointId,
- timestamp,
- streamFactory,
- CheckpointOptions.forFullCheckpoint());
- if (!keyedSnapshotRunnable.isDone()) {
- Thread runner = new Thread(keyedSnapshotRunnable);
- runner.start();
- }
- outStream.write(1);
- ObjectOutputStream oos = new ObjectOutputStream(outStream);
- oos.writeObject(keyedSnapshotRunnable.get());
- oos.flush();
- } else {
- outStream.write(0);
- }
- return outStream.closeAndGetHandle();
- }
-
- /**
- *
- */
- @Override
- public void restore(StreamStateHandle snapshot) throws Exception {
- try (FSDataInputStream inStream = snapshot.openInputStream()) {
-
- if (operator instanceof StreamCheckpointedOperator) {
- ((StreamCheckpointedOperator) operator).restoreState(inStream);
- }
-
- byte keyedStatePresent = (byte) inStream.read();
- if (keyedStatePresent == 1) {
- ObjectInputStream ois = new ObjectInputStream(inStream);
- this.restoredKeyedState = Collections.singletonList((KeyedStateHandle) ois.readObject());
- }
- }
- }
-
private static boolean hasMigrationHandles(Collection<KeyedStateHandle> allKeyGroupsHandles) {
for (KeyedStateHandle handle : allKeyGroupsHandles) {
if (handle instanceof Migration) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java
index 7e32723..33f32e9 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/OperatorSnapshotUtil.java
@@ -21,7 +21,6 @@ package org.apache.flink.streaming.util;
import org.apache.flink.runtime.checkpoint.savepoint.SavepointV1Serializer;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
-import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.streaming.runtime.tasks.OperatorStateHandles;
import java.io.DataInputStream;
@@ -53,7 +52,8 @@ public class OperatorSnapshotUtil {
dos.writeInt(state.getOperatorChainIndex());
- SavepointV1Serializer.serializeStreamStateHandle(state.getLegacyOperatorState(), dos);
+ // still required for compatibility
+ SavepointV1Serializer.serializeStreamStateHandle(null, dos);
Collection<OperatorStateHandle> rawOperatorState = state.getRawOperatorState();
if (rawOperatorState != null) {
@@ -108,7 +108,8 @@ public class OperatorSnapshotUtil {
try (DataInputStream dis = new DataInputStream(in)) {
int index = dis.readInt();
- StreamStateHandle legacyState = SavepointV1Serializer.deserializeStreamStateHandle(dis);
+ // still required for compatibility to consume the bytes.
+ SavepointV1Serializer.deserializeStreamStateHandle(dis);
List<OperatorStateHandle> rawOperatorState = null;
int numRawOperatorStates = dis.readInt();
@@ -154,7 +155,12 @@ public class OperatorSnapshotUtil {
}
}
- return new OperatorStateHandles(index, legacyState, managedKeyedState, rawKeyedState, managedOperatorState, rawOperatorState);
+ return new OperatorStateHandles(
+ index,
+ managedKeyedState,
+ rawKeyedState,
+ managedOperatorState,
+ rawOperatorState);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/migration/MigrationTestUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/migration/MigrationTestUtil.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/migration/MigrationTestUtil.java
index f723b34..1c95a04 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/migration/MigrationTestUtil.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/migration/MigrationTestUtil.java
@@ -29,22 +29,16 @@ public class MigrationTestUtil {
/**
* Restore from a snapshot taken with an older Flink version.
*
- * @param testHarness the test harness to restore the snapshot to.
- * @param snapshotPath the absolute path to the snapshot.
+ * @param testHarness the test harness to restore the snapshot to.
+ * @param snapshotPath the absolute path to the snapshot.
* @param snapshotFlinkVersion the Flink version of the snapshot.
- *
* @throws Exception
*/
public static void restoreFromSnapshot(
- AbstractStreamOperatorTestHarness<?> testHarness,
- String snapshotPath,
- MigrationVersion snapshotFlinkVersion) throws Exception {
+ AbstractStreamOperatorTestHarness<?> testHarness,
+ String snapshotPath,
+ MigrationVersion snapshotFlinkVersion) throws Exception {
- if (snapshotFlinkVersion == MigrationVersion.v1_1) {
- // Flink 1.1 snapshots should be read using the legacy restore method
- testHarness.initializeStateFromLegacyCheckpoint(snapshotPath);
- } else {
- testHarness.initializeState(OperatorSnapshotUtil.readStateHandle(snapshotPath));
- }
+ testHarness.initializeState(OperatorSnapshotUtil.readStateHandle(snapshotPath));
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/TimeWindowTranslationTest.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/TimeWindowTranslationTest.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/TimeWindowTranslationTest.scala
index 104400f..35a56d7 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/TimeWindowTranslationTest.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/TimeWindowTranslationTest.scala
@@ -25,15 +25,15 @@ import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.transformations.OneInputTransformation
-import org.apache.flink.streaming.api.windowing.assigners.{SlidingAlignedProcessingTimeWindows, SlidingEventTimeWindows, TumblingAlignedProcessingTimeWindows}
+import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
-import org.apache.flink.streaming.runtime.operators.windowing.{AccumulatingProcessingTimeWindowOperator, AggregatingProcessingTimeWindowOperator, WindowOperator}
+import org.apache.flink.streaming.runtime.operators.windowing.WindowOperator
import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase
import org.apache.flink.util.Collector
import org.junit.Assert._
-import org.junit.{Ignore, Test}
+import org.junit.Test
/**
* These tests verify that the api calls on [[WindowedStream]] that use the "time" shortcut
@@ -85,59 +85,6 @@ class TimeWindowTranslationTest extends StreamingMultipleProgramsTestBase {
assertTrue(operator2.isInstanceOf[WindowOperator[_, _, _, _, _]])
}
- /**
- * These tests ensure that the fast aligned time windows operator is used if the
- * conditions are right.
- */
- @Test
- def testReduceAlignedTimeWindows(): Unit = {
- val env = StreamExecutionEnvironment.getExecutionEnvironment
-
- val source = env.fromElements(("hello", 1), ("hello", 2))
-
- val window1 = source
- .keyBy(0)
- .window(SlidingAlignedProcessingTimeWindows.of(Time.seconds(1), Time.milliseconds(100)))
- .reduce(new DummyReducer())
-
- val transform1 = window1.javaStream.getTransformation
- .asInstanceOf[OneInputTransformation[(String, Int), (String, Int)]]
-
- val operator1 = transform1.getOperator
-
- assertTrue(operator1.isInstanceOf[AggregatingProcessingTimeWindowOperator[_, _]])
- }
-
- /**
- * These tests ensure that the fast aligned time windows operator is used if the
- * conditions are right.
- */
- @Test
- def testApplyAlignedTimeWindows(): Unit = {
- val env = StreamExecutionEnvironment.getExecutionEnvironment
- env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
-
- val source = env.fromElements(("hello", 1), ("hello", 2))
-
- val window1 = source
- .keyBy(0)
- .window(TumblingAlignedProcessingTimeWindows.of(Time.minutes(1)))
- .apply(new WindowFunction[(String, Int), (String, Int), Tuple, TimeWindow]() {
- def apply(
- key: Tuple,
- window: TimeWindow,
- values: Iterable[(String, Int)],
- out: Collector[(String, Int)]) { }
- })
-
- val transform1 = window1.javaStream.getTransformation
- .asInstanceOf[OneInputTransformation[(String, Int), (String, Int)]]
-
- val operator1 = transform1.getOperator
-
- assertTrue(operator1.isInstanceOf[AccumulatingProcessingTimeWindowOperator[_, _, _]])
- }
-
@Test
def testReduceEventTimeWindows(): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
index cad6693..99fb6ef 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/RescalingITCase.java
@@ -44,7 +44,6 @@ import org.apache.flink.runtime.state.filesystem.FsStateBackendFactory;
import org.apache.flink.runtime.testingUtils.TestingCluster;
import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
-import org.apache.flink.streaming.api.checkpoint.CheckpointedRestoring;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -975,7 +974,7 @@ public class RescalingITCase extends TestLogger {
}
}
- private static class PartitionedStateSource extends StateSourceBase implements CheckpointedFunction, CheckpointedRestoring<Integer> {
+ private static class PartitionedStateSource extends StateSourceBase implements CheckpointedFunction {
private static final long serialVersionUID = -359715965103593462L;
private static final int NUM_PARTITIONS = 7;
@@ -1030,10 +1029,5 @@ public class RescalingITCase extends TestLogger {
checkCorrectRestore[getRuntimeContext().getIndexOfThisSubtask()] = counter;
}
}
-
- @Override
- public void restoreState(Integer state) throws Exception {
- counterPartitions.add(state);
- }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
index cc23545..1b7dafa 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
@@ -53,7 +53,7 @@ import org.apache.flink.runtime.messages.JobManagerMessages.CancelJob;
import org.apache.flink.runtime.messages.JobManagerMessages.DisposeSavepoint;
import org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepoint;
import org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepointSuccess;
-import org.apache.flink.runtime.state.StreamStateHandle;
+import org.apache.flink.runtime.state.OperatorStateHandle;
import org.apache.flink.runtime.state.filesystem.FileStateHandle;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackendFactory;
@@ -351,10 +351,6 @@ public class SavepointITCase extends TestLogger {
OperatorSubtaskState subtaskState = operatorState.getState(tdd.getSubtaskIndex());
assertNotNull(subtaskState);
-
- errMsg = "Initial operator state mismatch.";
- assertEquals(errMsg, subtaskState.getLegacyOperatorState(),
- tdd.getTaskStateHandles().getSubtaskStateByOperatorID(operatorState.getOperatorID()).getLegacyOperatorState());
}
}
@@ -377,17 +373,18 @@ public class SavepointITCase extends TestLogger {
assertTrue(errMsg, resp.getClass() == getDisposeSavepointSuccess().getClass());
// - Verification START -------------------------------------------
-
// The checkpoint files
List<File> checkpointFiles = new ArrayList<>();
for (OperatorState stateForTaskGroup : savepoint.getOperatorStates()) {
for (OperatorSubtaskState subtaskState : stateForTaskGroup.getStates()) {
- StreamStateHandle streamTaskState = subtaskState.getLegacyOperatorState();
+ Collection<OperatorStateHandle> streamTaskState = subtaskState.getManagedOperatorState();
- if (streamTaskState != null) {
- FileStateHandle fileStateHandle = (FileStateHandle) streamTaskState;
- checkpointFiles.add(new File(fileStateHandle.getFilePath().toUri()));
+ if (streamTaskState != null && !streamTaskState.isEmpty()) {
+ for (OperatorStateHandle osh : streamTaskState) {
+ FileStateHandle fileStateHandle = (FileStateHandle) osh.getDelegateStateHandle();
+ checkpointFiles.add(new File(fileStateHandle.getFilePath().toUri()));
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
index 21be7ba..eccc7e9 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
@@ -25,6 +25,7 @@ import org.apache.flink.client.program.StandaloneClusterClient;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointSerializers;
import org.apache.flink.runtime.client.JobListeningContext;
import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -39,6 +40,7 @@ import org.apache.flink.test.util.TestBaseUtils;
import org.apache.commons.io.FileUtils;
import org.junit.After;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.rules.TemporaryFolder;
import org.slf4j.Logger;
@@ -64,6 +66,11 @@ import static junit.framework.Assert.fail;
*/
public class SavepointMigrationTestBase extends TestBaseUtils {
+ @BeforeClass
+ public static void before() {
+ SavepointSerializers.setFailWhenLegacyStateDetected(false);
+ }
+
@Rule
public TemporaryFolder tempFolder = new TemporaryFolder();