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:34 UTC
[03/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-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom11MigrationITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom11MigrationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom11MigrationITCase.java
deleted file mode 100644
index da6e035..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom11MigrationITCase.java
+++ /dev/null
@@ -1,562 +0,0 @@
-/*
- * 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.checkpointing.utils;
-
-import org.apache.flink.api.common.accumulators.IntCounter;
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.api.common.functions.RichFlatMapFunction;
-import org.apache.flink.api.common.state.ValueState;
-import org.apache.flink.api.common.state.ValueStateDescriptor;
-import org.apache.flink.api.common.typeinfo.TypeHint;
-import org.apache.flink.api.common.typeutils.base.LongSerializer;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
-import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.core.memory.DataInputViewStreamWrapper;
-import org.apache.flink.runtime.state.memory.MemoryStateBackend;
-import org.apache.flink.streaming.api.TimeCharacteristic;
-import org.apache.flink.streaming.api.checkpoint.Checkpointed;
-import org.apache.flink.streaming.api.checkpoint.CheckpointedRestoring;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
-import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.operators.TimestampedCollector;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import org.apache.flink.util.Collector;
-
-import org.junit.Ignore;
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * This verifies that we can restore a complete job from a Flink 1.1 savepoint.
- *
- * <p>The test pipeline contains both "Checkpointed" state and keyed user state.
- */
-public class StatefulJobSavepointFrom11MigrationITCase extends SavepointMigrationTestBase {
- private static final int NUM_SOURCE_ELEMENTS = 4;
- private static final String EXPECTED_ELEMENTS_ACCUMULATOR = "NUM_EXPECTED_ELEMENTS";
- private static final String SUCCESSFUL_CHECK_ACCUMULATOR = "SUCCESSFUL_CHECKS";
-
- /**
- * This has to be manually executed to create the savepoint on Flink 1.1.
- */
- @Test
- @Ignore
- public void testCreateSavepointOnFlink11() throws Exception {
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
- // we only test memory state backend yet
- env.setStateBackend(new MemoryStateBackend());
- env.enableCheckpointing(500);
- env.setParallelism(4);
- env.setMaxParallelism(4);
-
- // create source
- env
- .addSource(new LegacyCheckpointedSource(NUM_SOURCE_ELEMENTS)).setMaxParallelism(1).uid("LegacyCheckpointedSource")
- .flatMap(new LegacyCheckpointedFlatMap()).startNewChain().uid("LegacyCheckpointedFlatMap")
- .keyBy(0)
- .flatMap(new LegacyCheckpointedFlatMapWithKeyedState()).startNewChain().uid("LegacyCheckpointedFlatMapWithKeyedState")
- .keyBy(0)
- .flatMap(new KeyedStateSettingFlatMap()).startNewChain().uid("KeyedStateSettingFlatMap")
- .keyBy(0)
- .transform(
- "custom_operator",
- new TypeHint<Tuple2<Long, Long>>() {}.getTypeInfo(),
- new CheckpointedUdfOperator(new LegacyCheckpointedFlatMapWithKeyedState())).uid("LegacyCheckpointedOperator")
- .addSink(new AccumulatorCountingSink<Tuple2<Long, Long>>(EXPECTED_ELEMENTS_ACCUMULATOR));
-
- executeAndSavepoint(
- env,
- "src/test/resources/stateful-udf-migration-itcase-flink1.1-savepoint",
- new Tuple2<>(EXPECTED_ELEMENTS_ACCUMULATOR, NUM_SOURCE_ELEMENTS));
- }
-
- /**
- * This has to be manually executed to create the savepoint on Flink 1.1.
- */
- @Test
- @Ignore
- public void testCreateSavepointOnFlink11WithRocksDB() throws Exception {
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
- RocksDBStateBackend rocksBackend =
- new RocksDBStateBackend(new MemoryStateBackend());
-// rocksBackend.enableFullyAsyncSnapshots();
- env.setStateBackend(rocksBackend);
- env.enableCheckpointing(500);
- env.setParallelism(4);
- env.setMaxParallelism(4);
-
- // create source
- env
- .addSource(new LegacyCheckpointedSource(NUM_SOURCE_ELEMENTS)).setMaxParallelism(1).uid("LegacyCheckpointedSource")
- .flatMap(new LegacyCheckpointedFlatMap()).startNewChain().uid("LegacyCheckpointedFlatMap")
- .keyBy(0)
- .flatMap(new LegacyCheckpointedFlatMapWithKeyedState()).startNewChain().uid("LegacyCheckpointedFlatMapWithKeyedState")
- .keyBy(0)
- .flatMap(new KeyedStateSettingFlatMap()).startNewChain().uid("KeyedStateSettingFlatMap")
- .keyBy(0)
- .transform(
- "custom_operator",
- new TypeHint<Tuple2<Long, Long>>() {}.getTypeInfo(),
- new CheckpointedUdfOperator(new LegacyCheckpointedFlatMapWithKeyedState())).uid("LegacyCheckpointedOperator")
- .addSink(new AccumulatorCountingSink<Tuple2<Long, Long>>(EXPECTED_ELEMENTS_ACCUMULATOR));
-
- executeAndSavepoint(
- env,
- "src/test/resources/stateful-udf-migration-itcase-flink1.1-rocksdb-savepoint",
- new Tuple2<>(EXPECTED_ELEMENTS_ACCUMULATOR, NUM_SOURCE_ELEMENTS));
- }
-
- @Test
- public void testSavepointRestoreFromFlink11() throws Exception {
-
- final int expectedSuccessfulChecks = 21;
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
- // we only test memory state backend yet
- env.setStateBackend(new MemoryStateBackend());
- env.enableCheckpointing(500);
- env.setParallelism(4);
- env.setMaxParallelism(4);
-
- // create source
- env
- .addSource(new RestoringCheckingSource(NUM_SOURCE_ELEMENTS)).setMaxParallelism(1).uid("LegacyCheckpointedSource")
- .flatMap(new RestoringCheckingFlatMap()).startNewChain().uid("LegacyCheckpointedFlatMap")
- .keyBy(0)
- .flatMap(new RestoringCheckingFlatMapWithKeyedState()).startNewChain().uid("LegacyCheckpointedFlatMapWithKeyedState")
- .keyBy(0)
- .flatMap(new KeyedStateCheckingFlatMap()).startNewChain().uid("KeyedStateSettingFlatMap")
- .keyBy(0)
- .transform(
- "custom_operator",
- new TypeHint<Tuple2<Long, Long>>() {}.getTypeInfo(),
- new RestoringCheckingUdfOperator(new RestoringCheckingFlatMapWithKeyedState())).uid("LegacyCheckpointedOperator")
- .addSink(new AccumulatorCountingSink<Tuple2<Long, Long>>(EXPECTED_ELEMENTS_ACCUMULATOR));
-
- restoreAndExecute(
- env,
- getResourceFilename("stateful-udf-migration-itcase-flink1.1-savepoint"),
- new Tuple2<>(SUCCESSFUL_CHECK_ACCUMULATOR, expectedSuccessfulChecks));
- }
-
- @Test
- public void testSavepointRestoreFromFlink11FromRocksDB() throws Exception {
-
- final int expectedSuccessfulChecks = 21;
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
- // we only test memory state backend yet
- env.setStateBackend(new RocksDBStateBackend(new MemoryStateBackend()));
- env.enableCheckpointing(500);
- env.setParallelism(4);
- env.setMaxParallelism(4);
-
- // create source
- env
- .addSource(new RestoringCheckingSource(NUM_SOURCE_ELEMENTS)).setMaxParallelism(1).uid("LegacyCheckpointedSource")
- .flatMap(new RestoringCheckingFlatMap()).startNewChain().uid("LegacyCheckpointedFlatMap")
- .keyBy(0)
- .flatMap(new RestoringCheckingFlatMapWithKeyedState()).startNewChain().uid("LegacyCheckpointedFlatMapWithKeyedState")
- .keyBy(0)
- .flatMap(new KeyedStateCheckingFlatMap()).startNewChain().uid("KeyedStateSettingFlatMap")
- .keyBy(0)
- .transform(
- "custom_operator",
- new TypeHint<Tuple2<Long, Long>>() {}.getTypeInfo(),
- new RestoringCheckingUdfOperator(new RestoringCheckingFlatMapWithKeyedState())).uid("LegacyCheckpointedOperator")
- .addSink(new AccumulatorCountingSink<Tuple2<Long, Long>>(EXPECTED_ELEMENTS_ACCUMULATOR));
-
- restoreAndExecute(
- env,
- getResourceFilename("stateful-udf-migration-itcase-flink1.1-rocksdb-savepoint"),
- new Tuple2<>(SUCCESSFUL_CHECK_ACCUMULATOR, expectedSuccessfulChecks));
- }
-
- private static class LegacyCheckpointedSource
- implements SourceFunction<Tuple2<Long, Long>>, Checkpointed<String> {
-
- public static String checkpointedString = "Here be dragons!";
-
- private static final long serialVersionUID = 1L;
-
- private volatile boolean isRunning = true;
-
- private final int numElements;
-
- public LegacyCheckpointedSource(int numElements) {
- this.numElements = numElements;
- }
-
- @Override
- public void run(SourceContext<Tuple2<Long, Long>> ctx) throws Exception {
-
- synchronized (ctx.getCheckpointLock()) {
- for (long i = 0; i < numElements; i++) {
- ctx.collect(new Tuple2<>(i, i));
- }
- }
- while (isRunning) {
- Thread.sleep(20);
- }
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
-
- @Override
- public void restoreState(String state) throws Exception {
- assertEquals(checkpointedString, state);
- }
-
- @Override
- public String snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- return checkpointedString;
- }
- }
-
- private static class RestoringCheckingSource
- extends RichSourceFunction<Tuple2<Long, Long>>
- implements CheckpointedRestoring<String> {
-
- private static final long serialVersionUID = 1L;
-
- private volatile boolean isRunning = true;
-
- private final int numElements;
-
- private String restoredState;
-
- public RestoringCheckingSource(int numElements) {
- this.numElements = numElements;
- }
-
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
-
- getRuntimeContext().addAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR, new IntCounter());
- }
-
- @Override
- public void run(SourceContext<Tuple2<Long, Long>> ctx) throws Exception {
- assertEquals(LegacyCheckpointedSource.checkpointedString, restoredState);
- getRuntimeContext().getAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR).add(1);
-
- synchronized (ctx.getCheckpointLock()) {
- for (long i = 0; i < numElements; i++) {
- ctx.collect(new Tuple2<>(i, i));
- }
- }
-
- while (isRunning) {
- Thread.sleep(20);
- }
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
-
- @Override
- public void restoreState(String state) throws Exception {
- restoredState = state;
- }
- }
-
- private static class LegacyCheckpointedFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements Checkpointed<Tuple2<String, Long>> {
-
- private static final long serialVersionUID = 1L;
-
- public static Tuple2<String, Long> checkpointedTuple =
- new Tuple2<>("hello", 42L);
-
- @Override
- public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
- out.collect(value);
- }
-
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- }
-
- @Override
- public Tuple2<String, Long> snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- return checkpointedTuple;
- }
- }
-
- private static class RestoringCheckingFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements CheckpointedRestoring<Tuple2<String, Long>> {
-
- private static final long serialVersionUID = 1L;
-
- private transient Tuple2<String, Long> restoredState;
-
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
-
- getRuntimeContext().addAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR, new IntCounter());
- }
-
- @Override
- public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
- out.collect(value);
-
- assertEquals(LegacyCheckpointedFlatMap.checkpointedTuple, restoredState);
- getRuntimeContext().getAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR).add(1);
-
- }
-
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- restoredState = state;
- }
- }
-
- private static class LegacyCheckpointedFlatMapWithKeyedState
- extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements Checkpointed<Tuple2<String, Long>> {
-
- private static final long serialVersionUID = 1L;
-
- public static Tuple2<String, Long> checkpointedTuple =
- new Tuple2<>("hello", 42L);
-
- private final ValueStateDescriptor<Long> stateDescriptor =
- new ValueStateDescriptor<Long>("state-name", LongSerializer.INSTANCE);
-
- @Override
- public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
- out.collect(value);
-
- getRuntimeContext().getState(stateDescriptor).update(value.f1);
- }
-
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- }
-
- @Override
- public Tuple2<String, Long> snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- return checkpointedTuple;
- }
- }
-
- private static class RestoringCheckingFlatMapWithKeyedState extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements CheckpointedRestoring<Tuple2<String, Long>> {
-
- private static final long serialVersionUID = 1L;
-
- private transient Tuple2<String, Long> restoredState;
-
- private final ValueStateDescriptor<Long> stateDescriptor =
- new ValueStateDescriptor<Long>("state-name", LongSerializer.INSTANCE);
-
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
-
- getRuntimeContext().addAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR, new IntCounter());
- }
-
- @Override
- public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
- out.collect(value);
-
- ValueState<Long> state = getRuntimeContext().getState(stateDescriptor);
- if (state == null) {
- throw new RuntimeException("Missing key value state for " + value);
- }
-
- assertEquals(value.f1, state.value());
- assertEquals(LegacyCheckpointedFlatMap.checkpointedTuple, restoredState);
- getRuntimeContext().getAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR).add(1);
- }
-
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- restoredState = state;
- }
- }
-
- private static class KeyedStateSettingFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
-
- private static final long serialVersionUID = 1L;
-
- private final ValueStateDescriptor<Long> stateDescriptor =
- new ValueStateDescriptor<Long>("state-name", LongSerializer.INSTANCE);
-
- @Override
- public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
- out.collect(value);
-
- getRuntimeContext().getState(stateDescriptor).update(value.f1);
- }
- }
-
- private static class KeyedStateCheckingFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
-
- private static final long serialVersionUID = 1L;
-
- private final ValueStateDescriptor<Long> stateDescriptor =
- new ValueStateDescriptor<Long>("state-name", LongSerializer.INSTANCE);
-
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
-
- getRuntimeContext().addAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR, new IntCounter());
- }
-
- @Override
- public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
- out.collect(value);
-
- ValueState<Long> state = getRuntimeContext().getState(stateDescriptor);
- if (state == null) {
- throw new RuntimeException("Missing key value state for " + value);
- }
-
- assertEquals(value.f1, state.value());
- getRuntimeContext().getAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR).add(1);
- }
- }
-
- private static class CheckpointedUdfOperator
- extends AbstractUdfStreamOperator<Tuple2<Long, Long>, FlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>>
- implements OneInputStreamOperator<Tuple2<Long, Long>, Tuple2<Long, Long>> {
- private static final long serialVersionUID = 1L;
-
- private static final String CHECKPOINTED_STRING = "Oh my, that's nice!";
-
- public CheckpointedUdfOperator(FlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> userFunction) {
- super(userFunction);
- }
-
- @Override
- public void processElement(StreamRecord<Tuple2<Long, Long>> element) throws Exception {
- output.collect(element);
- }
-
- @Override
- public void processWatermark(Watermark mark) throws Exception {
- output.emitWatermark(mark);
- }
-
- // Flink 1.1
-// @Override
-// public StreamTaskState snapshotOperatorState(
-// long checkpointId, long timestamp) throws Exception {
-// StreamTaskState result = super.snapshotOperatorState(checkpointId, timestamp);
-//
-// AbstractStateBackend.CheckpointStateOutputView out = getStateBackend().createCheckpointStateOutputView(
-// checkpointId,
-// timestamp);
-//
-// out.writeUTF(checkpointedString);
-//
-// result.setOperatorState(out.closeAndGetHandle());
-//
-// return result;
-// }
- }
-
- private static class RestoringCheckingUdfOperator
- extends AbstractUdfStreamOperator<Tuple2<Long, Long>, FlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>>
- implements OneInputStreamOperator<Tuple2<Long, Long>, Tuple2<Long, Long>> {
- private static final long serialVersionUID = 1L;
-
- private String restoredState;
-
- public RestoringCheckingUdfOperator(FlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> userFunction) {
- super(userFunction);
- }
-
- @Override
- public void open() throws Exception {
- super.open();
- }
-
- @Override
- public void processElement(StreamRecord<Tuple2<Long, Long>> element) throws Exception {
- userFunction.flatMap(element.getValue(), new TimestampedCollector<>(output));
-
- assertEquals(CheckpointedUdfOperator.CHECKPOINTED_STRING, restoredState);
- getRuntimeContext().getAccumulator(SUCCESSFUL_CHECK_ACCUMULATOR).add(1);
- }
-
- @Override
- public void processWatermark(Watermark mark) throws Exception {
- output.emitWatermark(mark);
- }
-
- @Override
- public void restoreState(FSDataInputStream in) throws Exception {
- super.restoreState(in);
-
- DataInputViewStreamWrapper streamWrapper = new DataInputViewStreamWrapper(in);
-
- restoredState = streamWrapper.readUTF();
- }
- }
-
- private static class AccumulatorCountingSink<T> extends RichSinkFunction<T> {
- private static final long serialVersionUID = 1L;
-
- private final String accumulatorName;
-
- int count = 0;
-
- public AccumulatorCountingSink(String accumulatorName) {
- this.accumulatorName = accumulatorName;
- }
-
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
-
- getRuntimeContext().addAccumulator(accumulatorName, new IntCounter());
- }
-
- @Override
- public void invoke(T value) throws Exception {
- count++;
- getRuntimeContext().getAccumulator(accumulatorName).add(1);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom12MigrationITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom12MigrationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom12MigrationITCase.java
index 7dd1144..6859c2d 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom12MigrationITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobSavepointFrom12MigrationITCase.java
@@ -29,14 +29,8 @@ import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
-import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.core.fs.FSDataOutputStream;
-import org.apache.flink.core.memory.DataInputViewStreamWrapper;
-import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.TimeCharacteristic;
-import org.apache.flink.streaming.api.checkpoint.Checkpointed;
-import org.apache.flink.streaming.api.checkpoint.CheckpointedRestoring;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
@@ -60,10 +54,13 @@ import static org.junit.Assert.assertEquals;
/**
* This verifies that we can restore a complete job from a Flink 1.2 savepoint.
*
- * <p>The test pipeline contains both "Checkpointed" state and keyed user state.
+ * <p>The test for checkpointed (legacy state) was removed from this test for Flink 1.4 because compatibility with
+ * Flink 1.1 is removed. The legacy state in the binary savepoints is ignored by the tests now.
*
* <p>The tests will time out if they don't see the required number of successful checks within
* a time limit.
+ *
+ *
*/
public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigrationTestBase {
private static final int NUM_SOURCE_ELEMENTS = 4;
@@ -247,7 +244,7 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
}
private static class LegacyCheckpointedSource
- implements SourceFunction<Tuple2<Long, Long>>, Checkpointed<String> {
+ implements SourceFunction<Tuple2<Long, Long>> {
public static String checkpointedString = "Here be dragons!";
@@ -283,21 +280,10 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
public void cancel() {
isRunning = false;
}
-
- @Override
- public void restoreState(String state) throws Exception {
- assertEquals(checkpointedString, state);
- }
-
- @Override
- public String snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- return checkpointedString;
- }
}
private static class CheckingRestoringSource
- extends RichSourceFunction<Tuple2<Long, Long>>
- implements CheckpointedRestoring<String> {
+ extends RichSourceFunction<Tuple2<Long, Long>> {
private static final long serialVersionUID = 1L;
@@ -322,7 +308,6 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
@Override
public void run(SourceContext<Tuple2<Long, Long>> ctx) throws Exception {
- assertEquals(LegacyCheckpointedSource.checkpointedString, restoredState);
getRuntimeContext().getAccumulator(SUCCESSFUL_RESTORE_CHECK_ACCUMULATOR).add(1);
// immediately trigger any set timers
@@ -343,15 +328,9 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
public void cancel() {
isRunning = false;
}
-
- @Override
- public void restoreState(String state) throws Exception {
- restoredState = state;
- }
}
- private static class LegacyCheckpointedFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements Checkpointed<Tuple2<String, Long>> {
+ private static class LegacyCheckpointedFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
private static final long serialVersionUID = 1L;
@@ -362,19 +341,9 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
out.collect(value);
}
-
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- }
-
- @Override
- public Tuple2<String, Long> snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- return checkpointedTuple;
- }
}
- private static class CheckingRestoringFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements CheckpointedRestoring<Tuple2<String, Long>> {
+ private static class CheckingRestoringFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
private static final long serialVersionUID = 1L;
@@ -393,20 +362,14 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Long>> out) throws Exception {
out.collect(value);
- assertEquals(LegacyCheckpointedFlatMap.checkpointedTuple, restoredState);
getRuntimeContext().getAccumulator(SUCCESSFUL_RESTORE_CHECK_ACCUMULATOR).add(1);
}
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- restoredState = state;
- }
}
private static class LegacyCheckpointedFlatMapWithKeyedState
- extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements Checkpointed<Tuple2<String, Long>> {
+ extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
private static final long serialVersionUID = 1L;
@@ -424,19 +387,10 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
assertEquals(value.f1, getRuntimeContext().getState(stateDescriptor).value());
}
-
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- }
-
- @Override
- public Tuple2<String, Long> snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- return checkpointedTuple;
- }
}
- private static class CheckingRestoringFlatMapWithKeyedState extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements CheckpointedRestoring<Tuple2<String, Long>> {
+ private static class CheckingRestoringFlatMapWithKeyedState
+ extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
private static final long serialVersionUID = 1L;
@@ -464,18 +418,12 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
}
assertEquals(value.f1, state.value());
- assertEquals(LegacyCheckpointedFlatMap.checkpointedTuple, restoredState);
getRuntimeContext().getAccumulator(SUCCESSFUL_RESTORE_CHECK_ACCUMULATOR).add(1);
}
-
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- restoredState = state;
- }
}
- private static class CheckingRestoringFlatMapWithKeyedStateInOperator extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>
- implements CheckpointedRestoring<Tuple2<String, Long>> {
+ private static class CheckingRestoringFlatMapWithKeyedStateInOperator
+ extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
private static final long serialVersionUID = 1L;
@@ -503,14 +451,8 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
}
assertEquals(value.f1, state.value());
- assertEquals(LegacyCheckpointedFlatMap.checkpointedTuple, restoredState);
getRuntimeContext().getAccumulator(SUCCESSFUL_RESTORE_CHECK_ACCUMULATOR).add(1);
}
-
- @Override
- public void restoreState(Tuple2<String, Long> state) throws Exception {
- restoredState = state;
- }
}
private static class KeyedStateSettingFlatMap extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
@@ -578,17 +520,6 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
public void processWatermark(Watermark mark) throws Exception {
output.emitWatermark(mark);
}
-
- @Override
- public void snapshotState(
- FSDataOutputStream out, long checkpointId, long timestamp) throws Exception {
- super.snapshotState(out, checkpointId, timestamp);
-
- DataOutputViewStreamWrapper streamWrapper = new DataOutputViewStreamWrapper(out);
-
- streamWrapper.writeUTF(CHECKPOINTED_STRING);
- streamWrapper.flush();
- }
}
private static class CheckingRestoringUdfOperator
@@ -615,8 +546,6 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
@Override
public void processElement(StreamRecord<Tuple2<Long, Long>> element) throws Exception {
userFunction.flatMap(element.getValue(), new TimestampedCollector<>(output));
-
- assertEquals(CheckpointedUdfOperator.CHECKPOINTED_STRING, restoredState);
getRuntimeContext().getAccumulator(SUCCESSFUL_RESTORE_CHECK_ACCUMULATOR).add(1);
}
@@ -624,15 +553,6 @@ public class StatefulJobSavepointFrom12MigrationITCase extends SavepointMigratio
public void processWatermark(Watermark mark) throws Exception {
output.emitWatermark(mark);
}
-
- @Override
- public void restoreState(FSDataInputStream in) throws Exception {
- super.restoreState(in);
-
- DataInputViewStreamWrapper streamWrapper = new DataInputViewStreamWrapper(in);
-
- restoredState = streamWrapper.readUTF();
- }
}
private static class TimelyStatefulOperator
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/LegacyCheckpointedStreamingProgram.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/LegacyCheckpointedStreamingProgram.java b/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/LegacyCheckpointedStreamingProgram.java
deleted file mode 100644
index 1431d96..0000000
--- a/flink-tests/src/test/java/org/apache/flink/test/classloading/jar/LegacyCheckpointedStreamingProgram.java
+++ /dev/null
@@ -1,143 +0,0 @@
-/*
- * 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.classloading.jar;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.restartstrategy.RestartStrategies;
-import org.apache.flink.runtime.state.CheckpointListener;
-import org.apache.flink.streaming.api.checkpoint.Checkpointed;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.sink.SinkFunction;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-
-/**
- * This test is the same as the {@link CheckpointedStreamingProgram} but using the
- * old and deprecated {@link Checkpointed} interface. It stays here in order to
- * guarantee that although deprecated, the old Checkpointed interface is still supported.
- * This is necessary to not break user code.
- * */
-public class LegacyCheckpointedStreamingProgram {
-
- private static final int CHECKPOINT_INTERVALL = 100;
-
- public static void main(String[] args) throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.getConfig().disableSysoutLogging();
- env.enableCheckpointing(CHECKPOINT_INTERVALL);
- env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, 10000));
- env.disableOperatorChaining();
-
- DataStream<String> text = env.addSource(new SimpleStringGenerator());
- text.map(new StatefulMapper()).addSink(new NoOpSink());
- env.setParallelism(1);
- env.execute("Checkpointed Streaming Program");
- }
-
- // with Checkpointing
- private static class SimpleStringGenerator implements SourceFunction<String>, Checkpointed<Integer> {
-
- private static final long serialVersionUID = 3700033137820808611L;
-
- public boolean running = true;
-
- @Override
- public void run(SourceContext<String> ctx) throws Exception {
- while (running) {
- Thread.sleep(1);
- ctx.collect("someString");
- }
- }
-
- @Override
- public void cancel() {
- running = false;
- }
-
- @Override
- public Integer snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- return null;
- }
-
- @Override
- public void restoreState(Integer state) {
-
- }
- }
-
- private static class StatefulMapper implements MapFunction<String, String>, Checkpointed<StatefulMapper>, CheckpointListener {
-
- private static final long serialVersionUID = 2703630582894634440L;
-
- private String someState;
- private boolean atLeastOneSnapshotComplete = false;
- private boolean restored = false;
-
- @Override
- public StatefulMapper snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- return this;
- }
-
- @Override
- public void restoreState(StatefulMapper state) {
- restored = true;
- this.someState = state.someState;
- this.atLeastOneSnapshotComplete = state.atLeastOneSnapshotComplete;
- }
-
- @Override
- public String map(String value) throws Exception {
- if (!atLeastOneSnapshotComplete) {
- // throttle consumption by the checkpoint interval until we have one snapshot.
- Thread.sleep(CHECKPOINT_INTERVALL);
- }
- if (atLeastOneSnapshotComplete && !restored) {
- throw new RuntimeException("Intended failure, to trigger restore");
- }
- if (restored) {
- throw new SuccessException();
- //throw new RuntimeException("All good");
- }
- someState = value; // update our state
- return value;
- }
-
- @Override
- public void notifyCheckpointComplete(long checkpointId) throws Exception {
- atLeastOneSnapshotComplete = true;
- }
- }
- // --------------------------------------------------------------------------------------------
-
- /**
- * We intentionally use a user specified failure exception.
- */
- private static class SuccessException extends Exception {
-
- private static final long serialVersionUID = 7073311460437532086L;
- }
-
- private static class NoOpSink implements SinkFunction<String> {
- private static final long serialVersionUID = 2381410324190818620L;
-
- @Override
- public void invoke(String value) throws Exception {
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6642768a/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java
index 3d78242..00d0b2c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.akka.ListeningBehaviour;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointSerializers;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
@@ -89,6 +90,11 @@ public abstract class AbstractOperatorRestoreTestBase extends TestLogger {
private static final FiniteDuration timeout = new FiniteDuration(30L, TimeUnit.SECONDS);
@BeforeClass
+ public static void beforeClass() {
+ SavepointSerializers.setFailWhenLegacyStateDetected(false);
+ }
+
+ @BeforeClass
public static void setupCluster() throws Exception {
final Configuration configuration = new Configuration();