You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2020/06/03 13:06:29 UTC
[flink] 02/04: [FLINK-17376] Don't restore from Flink <= 1.2 state
in Kafka connector
This is an automated email from the ASF dual-hosted git repository.
aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4ff270e1d977cfcd6db1804ae16fc9002184fab3
Author: Aljoscha Krettek <al...@apache.org>
AuthorDate: Thu May 28 15:11:27 2020 +0200
[FLINK-17376] Don't restore from Flink <= 1.2 state in Kafka connector
This code was using the deprecated getSerializedListState(), which we
want to remove.
---
.../connectors/kafka/FlinkKafkaConsumerBase.java | 40 ++++-----------------
.../kafka/FlinkKafkaConsumerBaseMigrationTest.java | 38 --------------------
...er-migration-test-flink1.3-empty-state-snapshot | Bin 473 -> 0 bytes
...kafka-consumer-migration-test-flink1.3-snapshot | Bin 1255 -> 0 bytes
4 files changed, 6 insertions(+), 72 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
index e8c0ae4..0e8c261 100644
--- a/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
+++ b/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java
@@ -36,7 +36,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.CheckpointListener;
-import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
@@ -193,12 +192,6 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
/** Accessor for state in the operator state backend. */
private transient ListState<Tuple2<KafkaTopicPartition, Long>> unionOffsetStates;
- /**
- * Flag indicating whether the consumer is restored from older state written with Flink 1.1 or 1.2.
- * When the current run is restored from older state, partition discovery is disabled.
- */
- private boolean restoredFromOldState;
-
/** Discovery loop, executed in a separate thread. */
private transient volatile Thread discoveryLoopThread;
@@ -566,17 +559,11 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
}
for (Map.Entry<KafkaTopicPartition, Long> restoredStateEntry : restoredState.entrySet()) {
- if (!restoredFromOldState) {
- // seed the partition discoverer with the union state while filtering out
- // restored partitions that should not be subscribed by this subtask
- if (KafkaTopicPartitionAssigner.assign(
- restoredStateEntry.getKey(), getRuntimeContext().getNumberOfParallelSubtasks())
- == getRuntimeContext().getIndexOfThisSubtask()){
- subscribedPartitionsToStartOffsets.put(restoredStateEntry.getKey(), restoredStateEntry.getValue());
- }
- } else {
- // when restoring from older 1.1 / 1.2 state, the restored state would not be the union state;
- // in this case, just use the restored state as the subscribed partitions
+ // seed the partition discoverer with the union state while filtering out
+ // restored partitions that should not be subscribed by this subtask
+ if (KafkaTopicPartitionAssigner.assign(
+ restoredStateEntry.getKey(), getRuntimeContext().getNumberOfParallelSubtasks())
+ == getRuntimeContext().getIndexOfThisSubtask()){
subscribedPartitionsToStartOffsets.put(restoredStateEntry.getKey(), restoredStateEntry.getValue());
}
}
@@ -907,27 +894,12 @@ public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFuncti
OperatorStateStore stateStore = context.getOperatorStateStore();
- ListState<Tuple2<KafkaTopicPartition, Long>> oldRoundRobinListState =
- stateStore.getSerializableListState(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME);
-
this.unionOffsetStates = stateStore.getUnionListState(new ListStateDescriptor<>(OFFSETS_STATE_NAME,
createStateSerializer(getRuntimeContext().getExecutionConfig())));
- if (context.isRestored() && !restoredFromOldState) {
+ if (context.isRestored()) {
restoredState = new TreeMap<>(new KafkaTopicPartition.Comparator());
- // migrate from 1.2 state, if there is any
- for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : oldRoundRobinListState.get()) {
- restoredFromOldState = true;
- unionOffsetStates.add(kafkaOffset);
- }
- oldRoundRobinListState.clear();
-
- if (restoredFromOldState && discoveryIntervalMillis != PARTITION_DISCOVERY_DISABLED) {
- throw new IllegalArgumentException(
- "Topic / partition discovery cannot be enabled if the job is restored from a savepoint from Flink 1.2.x.");
- }
-
// populate actual holder for restored state
for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : unionOffsetStates.get()) {
restoredState.put(kafkaOffset.f0, kafkaOffset.f1);
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
index 2dfe1af..79fa356 100644
--- a/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
+++ b/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseMigrationTest.java
@@ -38,7 +38,6 @@ import org.apache.flink.streaming.util.OperatorSnapshotUtil;
import org.apache.flink.testutils.migration.MigrationVersion;
import org.apache.flink.util.SerializedValue;
-import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -57,8 +56,6 @@ import java.util.stream.Collectors;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-import static org.junit.Assume.assumeTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.powermock.api.mockito.PowerMockito.doAnswer;
@@ -99,7 +96,6 @@ public class FlinkKafkaConsumerBaseMigrationTest {
@Parameterized.Parameters(name = "Migration Savepoint: {0}")
public static Collection<MigrationVersion> parameters () {
return Arrays.asList(
- MigrationVersion.v1_3,
MigrationVersion.v1_4,
MigrationVersion.v1_5,
MigrationVersion.v1_6,
@@ -325,40 +321,6 @@ public class FlinkKafkaConsumerBaseMigrationTest {
consumerOperator.cancel();
}
- /**
- * Test restoring from savepoints before version Flink 1.3 should fail if discovery is enabled.
- */
- @Test
- public void testRestoreFailsWithNonEmptyPreFlink13StatesIfDiscoveryEnabled() throws Exception {
- assumeTrue(testMigrateVersion == MigrationVersion.v1_3);
-
- final List<KafkaTopicPartition> partitions = new ArrayList<>(PARTITION_STATE.keySet());
-
- final DummyFlinkKafkaConsumer<String> consumerFunction =
- new DummyFlinkKafkaConsumer<>(TOPICS, partitions, 1000L); // discovery enabled
-
- StreamSource<String, DummyFlinkKafkaConsumer<String>> consumerOperator =
- new StreamSource<>(consumerFunction);
-
- final AbstractStreamOperatorTestHarness<String> testHarness =
- new AbstractStreamOperatorTestHarness<>(consumerOperator, 1, 1, 0);
-
- testHarness.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);
-
- testHarness.setup();
-
- // restore state from binary snapshot file; should fail since discovery is enabled
- try {
- testHarness.initializeState(
- OperatorSnapshotUtil.getResourceFilename(
- "kafka-consumer-migration-test-flink" + testMigrateVersion + "-snapshot"));
-
- fail("Restore from savepoints from version before Flink 1.3.x should have failed if discovery is enabled.");
- } catch (Exception e) {
- Assert.assertTrue(e instanceof IllegalArgumentException);
- }
- }
-
// ------------------------------------------------------------------------
private static class DummyFlinkKafkaConsumer<T> extends FlinkKafkaConsumerBase<T> {
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/resources/kafka-consumer-migration-test-flink1.3-empty-state-snapshot b/flink-connectors/flink-connector-kafka-base/src/test/resources/kafka-consumer-migration-test-flink1.3-empty-state-snapshot
deleted file mode 100644
index 1a5aad1..0000000
Binary files a/flink-connectors/flink-connector-kafka-base/src/test/resources/kafka-consumer-migration-test-flink1.3-empty-state-snapshot and /dev/null differ
diff --git a/flink-connectors/flink-connector-kafka-base/src/test/resources/kafka-consumer-migration-test-flink1.3-snapshot b/flink-connectors/flink-connector-kafka-base/src/test/resources/kafka-consumer-migration-test-flink1.3-snapshot
deleted file mode 100644
index dc820ef..0000000
Binary files a/flink-connectors/flink-connector-kafka-base/src/test/resources/kafka-consumer-migration-test-flink1.3-snapshot and /dev/null differ