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/11/05 09:12:16 UTC
[GitHub] asfgit closed pull request #6910: [FLINK-10490][tests] OperatorSnapshotUtil should use SavepointV2Seria…
asfgit closed pull request #6910: [FLINK-10490][tests] OperatorSnapshotUtil should use SavepointV2Seria…
URL: https://github.com/apache/flink/pull/6910
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/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java
index faee588c48d..fa8407729d9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2Serializer.java
@@ -18,6 +18,8 @@
package org.apache.flink.runtime.checkpoint.savepoint;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.checkpoint.MasterState;
import org.apache.flink.runtime.checkpoint.OperatorState;
@@ -67,7 +69,9 @@
* +--------------+---------------------+---------+------+---------------+
* </pre>
*/
-class SavepointV2Serializer implements SavepointSerializer<SavepointV2> {
+@Internal
+@VisibleForTesting
+public class SavepointV2Serializer implements SavepointSerializer<SavepointV2> {
/** Random magic number for consistency checks */
private static final int MASTER_STATE_MAGIC_NUMBER = 0xc96b1696;
@@ -320,7 +324,8 @@ private static OperatorSubtaskState deserializeSubtaskState(DataInputStream dis)
keyedStateStream);
}
- private static void serializeKeyedStateHandle(
+ @VisibleForTesting
+ public static void serializeKeyedStateHandle(
KeyedStateHandle stateHandle, DataOutputStream dos) throws IOException {
if (stateHandle == null) {
@@ -380,7 +385,8 @@ private static void serializeStreamStateHandleMap(
return result;
}
- private static KeyedStateHandle deserializeKeyedStateHandle(DataInputStream dis) throws IOException {
+ @VisibleForTesting
+ public static KeyedStateHandle deserializeKeyedStateHandle(DataInputStream dis) throws IOException {
final int type = dis.readByte();
if (NULL_HANDLE == type) {
@@ -433,7 +439,8 @@ private static KeyedStateHandle deserializeKeyedStateHandle(DataInputStream dis)
}
}
- private static void serializeOperatorStateHandle(
+ @VisibleForTesting
+ public static void serializeOperatorStateHandle(
OperatorStateHandle stateHandle, DataOutputStream dos) throws IOException {
if (stateHandle != null) {
@@ -461,7 +468,8 @@ private static void serializeOperatorStateHandle(
}
}
- private static OperatorStateHandle deserializeOperatorStateHandle(
+ @VisibleForTesting
+ public static OperatorStateHandle deserializeOperatorStateHandle(
DataInputStream dis) throws IOException {
final int type = dis.readByte();
@@ -492,7 +500,8 @@ private static OperatorStateHandle deserializeOperatorStateHandle(
}
}
- private static void serializeStreamStateHandle(
+ @VisibleForTesting
+ public static void serializeStreamStateHandle(
StreamStateHandle stateHandle, DataOutputStream dos) throws IOException {
if (stateHandle == null) {
@@ -518,7 +527,7 @@ private static void serializeStreamStateHandle(
dos.flush();
}
- private static StreamStateHandle deserializeStreamStateHandle(DataInputStream dis) throws IOException {
+ public static StreamStateHandle deserializeStreamStateHandle(DataInputStream dis) throws IOException {
final int type = dis.read();
if (NULL_HANDLE == type) {
return null;
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 1b5113da2cb..53627d5ad49 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
@@ -20,7 +20,7 @@
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.runtime.checkpoint.StateObjectCollection;
-import org.apache.flink.runtime.checkpoint.savepoint.SavepointV1Serializer;
+import org.apache.flink.runtime.checkpoint.savepoint.SavepointV2Serializer;
import org.apache.flink.runtime.state.KeyedStateHandle;
import org.apache.flink.runtime.state.OperatorStateHandle;
@@ -55,13 +55,13 @@ public static void writeStateHandle(OperatorSubtaskState state, String path) thr
dos.writeInt(0);
// still required for compatibility
- SavepointV1Serializer.serializeStreamStateHandle(null, dos);
+ SavepointV2Serializer.serializeStreamStateHandle(null, dos);
Collection<OperatorStateHandle> rawOperatorState = state.getRawOperatorState();
if (rawOperatorState != null) {
dos.writeInt(rawOperatorState.size());
for (OperatorStateHandle operatorStateHandle : rawOperatorState) {
- SavepointV1Serializer.serializeOperatorStateHandle(operatorStateHandle, dos);
+ SavepointV2Serializer.serializeOperatorStateHandle(operatorStateHandle, dos);
}
} else {
// this means no states, not even an empty list
@@ -72,7 +72,7 @@ public static void writeStateHandle(OperatorSubtaskState state, String path) thr
if (managedOperatorState != null) {
dos.writeInt(managedOperatorState.size());
for (OperatorStateHandle operatorStateHandle : managedOperatorState) {
- SavepointV1Serializer.serializeOperatorStateHandle(operatorStateHandle, dos);
+ SavepointV2Serializer.serializeOperatorStateHandle(operatorStateHandle, dos);
}
} else {
// this means no states, not even an empty list
@@ -83,7 +83,7 @@ public static void writeStateHandle(OperatorSubtaskState state, String path) thr
if (rawKeyedState != null) {
dos.writeInt(rawKeyedState.size());
for (KeyedStateHandle keyedStateHandle : rawKeyedState) {
- SavepointV1Serializer.serializeKeyedStateHandle(keyedStateHandle, dos);
+ SavepointV2Serializer.serializeKeyedStateHandle(keyedStateHandle, dos);
}
} else {
// this means no operator states, not even an empty list
@@ -94,7 +94,7 @@ public static void writeStateHandle(OperatorSubtaskState state, String path) thr
if (managedKeyedState != null) {
dos.writeInt(managedKeyedState.size());
for (KeyedStateHandle keyedStateHandle : managedKeyedState) {
- SavepointV1Serializer.serializeKeyedStateHandle(keyedStateHandle, dos);
+ SavepointV2Serializer.serializeKeyedStateHandle(keyedStateHandle, dos);
}
} else {
// this means no operator states, not even an empty list
@@ -113,14 +113,14 @@ public static OperatorSubtaskState readStateHandle(String path) throws IOExcepti
dis.readInt();
// still required for compatibility to consume the bytes.
- SavepointV1Serializer.deserializeStreamStateHandle(dis);
+ SavepointV2Serializer.deserializeStreamStateHandle(dis);
List<OperatorStateHandle> rawOperatorState = null;
int numRawOperatorStates = dis.readInt();
if (numRawOperatorStates >= 0) {
rawOperatorState = new ArrayList<>();
for (int i = 0; i < numRawOperatorStates; i++) {
- OperatorStateHandle operatorState = SavepointV1Serializer.deserializeOperatorStateHandle(
+ OperatorStateHandle operatorState = SavepointV2Serializer.deserializeOperatorStateHandle(
dis);
rawOperatorState.add(operatorState);
}
@@ -131,7 +131,7 @@ public static OperatorSubtaskState readStateHandle(String path) throws IOExcepti
if (numManagedOperatorStates >= 0) {
managedOperatorState = new ArrayList<>();
for (int i = 0; i < numManagedOperatorStates; i++) {
- OperatorStateHandle operatorState = SavepointV1Serializer.deserializeOperatorStateHandle(
+ OperatorStateHandle operatorState = SavepointV2Serializer.deserializeOperatorStateHandle(
dis);
managedOperatorState.add(operatorState);
}
@@ -142,7 +142,7 @@ public static OperatorSubtaskState readStateHandle(String path) throws IOExcepti
if (numRawKeyedStates >= 0) {
rawKeyedState = new ArrayList<>();
for (int i = 0; i < numRawKeyedStates; i++) {
- KeyedStateHandle keyedState = SavepointV1Serializer.deserializeKeyedStateHandle(
+ KeyedStateHandle keyedState = SavepointV2Serializer.deserializeKeyedStateHandle(
dis);
rawKeyedState.add(keyedState);
}
@@ -153,7 +153,7 @@ public static OperatorSubtaskState readStateHandle(String path) throws IOExcepti
if (numManagedKeyedStates >= 0) {
managedKeyedState = new ArrayList<>();
for (int i = 0; i < numManagedKeyedStates; i++) {
- KeyedStateHandle keyedState = SavepointV1Serializer.deserializeKeyedStateHandle(
+ KeyedStateHandle keyedState = SavepointV2Serializer.deserializeKeyedStateHandle(
dis);
managedKeyedState.add(keyedState);
}
----------------------------------------------------------------
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