You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2018/12/11 09:49:39 UTC
[GitHub] asfgit closed pull request #7256: [FLINK-11087] [state] Incorrect
K/V serializer association when reading broadcast state from 1.5.x snapshots
asfgit closed pull request #7256: [FLINK-11087] [state] Incorrect K/V serializer association when reading broadcast state from 1.5.x snapshots
URL: https://github.com/apache/flink/pull/7256
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/docs/ops/upgrading.md b/docs/ops/upgrading.md
index 0451940723e..22cea4c53f0 100644
--- a/docs/ops/upgrading.md
+++ b/docs/ops/upgrading.md
@@ -274,7 +274,10 @@ Savepoints are compatible across Flink versions as indicated by the table below:
<td class="text-center">O</td>
<td class="text-center">O</td>
<td class="text-center">O</td>
- <td class="text-left"></td>
+ <td class="text-left">There is a known issue with resuming broadcast state created with 1.5.x in versions
+ 1.6.x up to 1.6.2, and 1.7.0: <a href="https://issues.apache.org/jira/browse/FLINK-11087">FLINK-11087</a>. Users
+ upgrading to 1.6.x or 1.7.x series need to directly migrate to minor versions higher than 1.6.2 and 1.7.0,
+ respectively.</td>
</tr>
<tr>
<td class="text-center"><strong>1.6.x</strong></td>
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java
index 77c267adff1..836edef0aac 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/LegacyStateMetaInfoReaders.java
@@ -132,11 +132,6 @@ public StateMetaInfoSnapshot readStateMetaInfoSnapshot(
static final OperatorBackendStateMetaInfoReaderV2V3 INSTANCE = new OperatorBackendStateMetaInfoReaderV2V3();
- private static final String[] ORDERED_KEY_STRINGS =
- new String[]{
- StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER.toString(),
- StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString()};
-
@Nonnull
@Override
public StateMetaInfoSnapshot readStateMetaInfoSnapshot(
@@ -156,17 +151,25 @@ public StateMetaInfoSnapshot readStateMetaInfoSnapshot(
final int listSize = stateSerializerAndConfigList.size();
StateMetaInfoSnapshot.BackendStateType stateType = listSize == 1 ?
StateMetaInfoSnapshot.BackendStateType.OPERATOR : StateMetaInfoSnapshot.BackendStateType.BROADCAST;
- Map<String, TypeSerializerSnapshot<?>> serializerConfigsMap = new HashMap<>(listSize);
- for (int i = 0; i < listSize; ++i) {
- Tuple2<TypeSerializer<?>, TypeSerializerSnapshot<?>> serializerAndConf =
- stateSerializerAndConfigList.get(i);
-
- // this particular mapping happens to support both, V2 and V3
- String serializerKey = ORDERED_KEY_STRINGS[ORDERED_KEY_STRINGS.length - 1 - i];
- serializerConfigsMap.put(
- serializerKey,
- serializerAndConf.f1);
+ Map<String, TypeSerializerSnapshot<?>> serializerConfigsMap = new HashMap<>(listSize);
+ switch (stateType) {
+ case OPERATOR:
+ serializerConfigsMap.put(
+ StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString(),
+ stateSerializerAndConfigList.get(0).f1);
+ break;
+ case BROADCAST:
+ serializerConfigsMap.put(
+ StateMetaInfoSnapshot.CommonSerializerKeys.KEY_SERIALIZER.toString(),
+ stateSerializerAndConfigList.get(0).f1);
+
+ serializerConfigsMap.put(
+ StateMetaInfoSnapshot.CommonSerializerKeys.VALUE_SERIALIZER.toString(),
+ stateSerializerAndConfigList.get(1).f1);
+ break;
+ default:
+ throw new IllegalStateException("Unknown operator state type " + stateType);
}
return new StateMetaInfoSnapshot(
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobWBroadcastStateMigrationITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobWBroadcastStateMigrationITCase.java
index f7923620850..3f49d84d595 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobWBroadcastStateMigrationITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/StatefulJobWBroadcastStateMigrationITCase.java
@@ -113,17 +113,17 @@ public void testSavepoint() throws Exception {
expectedFirstState.put(2L, 2L);
expectedFirstState.put(3L, 3L);
- final Map<String, String> expectedSecondState = new HashMap<>();
- expectedSecondState.put("0", "0");
- expectedSecondState.put("1", "1");
- expectedSecondState.put("2", "2");
- expectedSecondState.put("3", "3");
-
- final Map<String, String> expectedThirdState = new HashMap<>();
- expectedThirdState.put("0", "0");
- expectedThirdState.put("1", "1");
- expectedThirdState.put("2", "2");
- expectedThirdState.put("3", "3");
+ final Map<String, Long> expectedSecondState = new HashMap<>();
+ expectedSecondState.put("0", 0L);
+ expectedSecondState.put("1", 1L);
+ expectedSecondState.put("2", 2L);
+ expectedSecondState.put("3", 3L);
+
+ final Map<Long, String> expectedThirdState = new HashMap<>();
+ expectedThirdState.put(0L, "0");
+ expectedThirdState.put(1L, "1");
+ expectedThirdState.put(2L, "2");
+ expectedThirdState.put(3L, "3");
if (executionMode == StatefulJobSavepointMigrationITCase.ExecutionMode.PERFORM_SAVEPOINT) {
nonParallelSource = new MigrationTestUtils.CheckpointingNonParallelSourceWithListState(NUM_SOURCE_ELEMENTS);
@@ -171,12 +171,12 @@ public Long getKey(Tuple2<Long, Long> value) throws Exception {
"broadcast-state-1", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
);
- final MapStateDescriptor<String, String> secondBroadcastStateDesc = new MapStateDescriptor<>(
- "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ final MapStateDescriptor<String, Long> secondBroadcastStateDesc = new MapStateDescriptor<>(
+ "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
);
- final MapStateDescriptor<String, String> thirdBroadcastStateDesc = new MapStateDescriptor<>(
- "broadcast-state-3", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ final MapStateDescriptor<Long, String> thirdBroadcastStateDesc = new MapStateDescriptor<>(
+ "broadcast-state-3", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
);
BroadcastStream<Tuple2<Long, Long>> npBroadcastStream = env
@@ -234,7 +234,7 @@ private String getBroadcastSavepointPath(MigrationVersion savepointVersion, Stri
private MapStateDescriptor<Long, Long> firstStateDesc;
- private MapStateDescriptor<String, String> secondStateDesc;
+ private MapStateDescriptor<String, Long> secondStateDesc;
@Override
public void open(Configuration parameters) throws Exception {
@@ -245,7 +245,7 @@ public void open(Configuration parameters) throws Exception {
);
secondStateDesc = new MapStateDescriptor<>(
- "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
);
}
@@ -257,7 +257,7 @@ public void processElement(Tuple2<Long, Long> value, ReadOnlyContext ctx, Collec
@Override
public void processBroadcastElement(Tuple2<Long, Long> value, Context ctx, Collector<Tuple2<Long, Long>> out) throws Exception {
ctx.getBroadcastState(firstStateDesc).put(value.f0, value.f1);
- ctx.getBroadcastState(secondStateDesc).put(Long.toString(value.f0), Long.toString(value.f1));
+ ctx.getBroadcastState(secondStateDesc).put(Long.toString(value.f0), value.f1);
}
}
@@ -269,14 +269,14 @@ public void processBroadcastElement(Tuple2<Long, Long> value, Context ctx, Colle
private static final long serialVersionUID = 1333992081671604521L;
- private MapStateDescriptor<String, String> stateDesc;
+ private MapStateDescriptor<Long, String> stateDesc;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
stateDesc = new MapStateDescriptor<>(
- "broadcast-state-3", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ "broadcast-state-3", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
);
}
@@ -287,7 +287,7 @@ public void processElement(Tuple2<Long, Long> value, ReadOnlyContext ctx, Collec
@Override
public void processBroadcastElement(Tuple2<Long, Long> value, Context ctx, Collector<Tuple2<Long, Long>> out) throws Exception {
- ctx.getBroadcastState(stateDesc).put(Long.toString(value.f0), Long.toString(value.f1));
+ ctx.getBroadcastState(stateDesc).put(value.f0, Long.toString(value.f1));
}
}
@@ -301,13 +301,13 @@ public void processBroadcastElement(Tuple2<Long, Long> value, Context ctx, Colle
private final Map<Long, Long> expectedFirstState;
- private final Map<String, String> expectedSecondState;
+ private final Map<String, Long> expectedSecondState;
private MapStateDescriptor<Long, Long> firstStateDesc;
- private MapStateDescriptor<String, String> secondStateDesc;
+ private MapStateDescriptor<String, Long> secondStateDesc;
- CheckingKeyedBroadcastFunction(Map<Long, Long> firstState, Map<String, String> secondState) {
+ CheckingKeyedBroadcastFunction(Map<Long, Long> firstState, Map<String, Long> secondState) {
this.expectedFirstState = firstState;
this.expectedSecondState = secondState;
}
@@ -321,7 +321,7 @@ public void open(Configuration parameters) throws Exception {
);
secondStateDesc = new MapStateDescriptor<>(
- "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ "broadcast-state-2", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
);
}
@@ -334,8 +334,8 @@ public void processElement(Tuple2<Long, Long> value, ReadOnlyContext ctx, Collec
}
Assert.assertEquals(expectedFirstState, actualFirstState);
- final Map<String, String> actualSecondState = new HashMap<>();
- for (Map.Entry<String, String> entry: ctx.getBroadcastState(secondStateDesc).immutableEntries()) {
+ final Map<String, Long> actualSecondState = new HashMap<>();
+ for (Map.Entry<String, Long> entry: ctx.getBroadcastState(secondStateDesc).immutableEntries()) {
actualSecondState.put(entry.getKey(), entry.getValue());
}
Assert.assertEquals(expectedSecondState, actualSecondState);
@@ -357,11 +357,11 @@ public void processBroadcastElement(Tuple2<Long, Long> value, Context ctx, Colle
private static final long serialVersionUID = 1333992081671604521L;
- private final Map<String, String> expectedState;
+ private final Map<Long, String> expectedState;
- private MapStateDescriptor<String, String> stateDesc;
+ private MapStateDescriptor<Long, String> stateDesc;
- CheckingKeyedSingleBroadcastFunction(Map<String, String> state) {
+ CheckingKeyedSingleBroadcastFunction(Map<Long, String> state) {
this.expectedState = state;
}
@@ -370,14 +370,14 @@ public void open(Configuration parameters) throws Exception {
super.open(parameters);
stateDesc = new MapStateDescriptor<>(
- "broadcast-state-3", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
+ "broadcast-state-3", BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO
);
}
@Override
public void processElement(Tuple2<Long, Long> value, ReadOnlyContext ctx, Collector<Tuple2<Long, Long>> out) throws Exception {
- final Map<String, String> actualState = new HashMap<>();
- for (Map.Entry<String, String> entry: ctx.getBroadcastState(stateDesc).immutableEntries()) {
+ final Map<Long, String> actualState = new HashMap<>();
+ for (Map.Entry<Long, String> entry: ctx.getBroadcastState(stateDesc).immutableEntries()) {
actualState.put(entry.getKey(), entry.getValue());
}
Assert.assertEquals(expectedState, actualState);
diff --git a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-rocksdb-savepoint/_metadata b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-rocksdb-savepoint/_metadata
index aed61836bd9..2e02f1bf87d 100644
Binary files a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-rocksdb-savepoint/_metadata and b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-rocksdb-savepoint/_metadata differ
diff --git a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-savepoint/_metadata b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-savepoint/_metadata
index 13e8d321036..4ca91d33ccb 100644
Binary files a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-savepoint/_metadata and b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.5-savepoint/_metadata differ
diff --git a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-rocksdb-savepoint/_metadata b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-rocksdb-savepoint/_metadata
index 9e2b08654b2..8b2a63ab41b 100644
Binary files a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-rocksdb-savepoint/_metadata and b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-rocksdb-savepoint/_metadata differ
diff --git a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-savepoint/_metadata b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-savepoint/_metadata
index cceb5bf7f89..cde6d5c44bf 100644
Binary files a/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-savepoint/_metadata and b/flink-tests/src/test/resources/new-stateful-broadcast-udf-migration-itcase-flink1.6-savepoint/_metadata differ
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services