flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tzuli...@apache.org
Subject [2/3] flink git commit: [FLINK-6566] [core] More restricted interface for VersionedIOReadableWritable hooks
Date Sat, 13 May 2017 09:03:42 GMT
[FLINK-6566] [core] More restricted interface for VersionedIOReadableWritable hooks

This commit makes the method hooks for defining compatibile
serialization versions of VersionedIOReadableWritables more restricted.

Functionally everything remains the same, but with lesser space for
error-prone user implementations. It also allows for a better error
message to indicate version mismatch.

This closes #3883.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/347100de
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/347100de
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/347100de

Branch: refs/heads/master
Commit: 347100de7527dc4ba3664b8e8306a081834f84a4
Parents: c594af0
Author: Tzu-Li (Gordon) Tai <tzulitai@apache.org>
Authored: Fri May 12 20:06:01 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tzulitai@apache.org>
Committed: Sat May 13 14:45:25 2017 +0800

----------------------------------------------------------------------
 .../typeutils/TypeSerializerConfigSnapshot.java | 23 ---------
 .../core/io/VersionedIOReadableWritable.java    | 53 ++++++++++++--------
 .../flink/core/io/VersionedIOWriteableTest.java | 16 ++----
 .../state/KeyedBackendSerializationProxy.java   | 20 ++------
 .../OperatorBackendSerializationProxy.java      | 17 ++-----
 .../state/heap/HeapKeyedStateBackend.java       |  2 +-
 6 files changed, 42 insertions(+), 89 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/347100de/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java
b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java
index 27369b9..389d141 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerConfigSnapshot.java
@@ -20,7 +20,6 @@ package org.apache.flink.api.common.typeutils;
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.core.io.VersionMismatchException;
 import org.apache.flink.core.io.VersionedIOReadableWritable;
 import org.apache.flink.util.Preconditions;
 
@@ -51,22 +50,6 @@ public abstract class TypeSerializerConfigSnapshot extends VersionedIOReadableWr
 	/** The user code class loader; only relevant if this configuration instance was deserialized
from binary form. */
 	private ClassLoader userCodeClassLoader;
 
-	/** The snapshot version of this configuration. */
-	private Integer snapshotVersion;
-
-	/**
-	 * Returns the version of the configuration at the time its snapshot was taken.
-	 *
-	 * @return the snapshot configuration's version.
-	 */
-	public int getSnapshotVersion() {
-		if (snapshotVersion == null) {
-			return getVersion();
-		} else {
-			return snapshotVersion;
-		}
-	}
-
 	/**
 	 * Set the user code class loader.
 	 * Only relevant if this configuration instance was deserialized from binary form.
@@ -91,12 +74,6 @@ public abstract class TypeSerializerConfigSnapshot extends VersionedIOReadableWr
 		return userCodeClassLoader;
 	}
 
-	@Override
-	protected void resolveVersionRead(int foundVersion) throws VersionMismatchException {
-		super.resolveVersionRead(foundVersion);
-		this.snapshotVersion = foundVersion;
-	}
-
 	public abstract boolean equals(Object obj);
 
 	public abstract int hashCode();

http://git-wip-us.apache.org/repos/asf/flink/blob/347100de/flink-core/src/main/java/org/apache/flink/core/io/VersionedIOReadableWritable.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/io/VersionedIOReadableWritable.java
b/flink-core/src/main/java/org/apache/flink/core/io/VersionedIOReadableWritable.java
index bad9cef..b4a0b2f 100644
--- a/flink-core/src/main/java/org/apache/flink/core/io/VersionedIOReadableWritable.java
+++ b/flink-core/src/main/java/org/apache/flink/core/io/VersionedIOReadableWritable.java
@@ -18,20 +18,23 @@
 
 package org.apache.flink.core.io;
 
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 
 import java.io.IOException;
+import java.util.Arrays;
 
 /**
  * This is the abstract base class for {@link IOReadableWritable} which allows to differentiate
between serialization
  * versions. Concrete subclasses should typically override the {@link #write(DataOutputView)}
and
  * {@link #read(DataInputView)}, thereby calling super to ensure version checking.
  */
-@PublicEvolving
+@Internal
 public abstract class VersionedIOReadableWritable implements IOReadableWritable, Versioned
{
 
+	private int readVersion = Integer.MIN_VALUE;
+
 	@Override
 	public void write(DataOutputView out) throws IOException {
 		out.writeInt(getVersion());
@@ -39,34 +42,42 @@ public abstract class VersionedIOReadableWritable implements IOReadableWritable,
 
 	@Override
 	public void read(DataInputView in) throws IOException {
-		int foundVersion = in.readInt();
-		resolveVersionRead(foundVersion);
+		this.readVersion = in.readInt();
+		resolveVersionRead(readVersion);
 	}
 
 	/**
-	 * This method is a hook to react on the version tag that we find during read. This can
also be used to initialize
-	 * further read logic w.r.t. the version at hand.
-	 * Default implementation of this method just checks the compatibility of a version number
against the own version.
+	 * Returns the found serialization version. If this instance was not read from serialized
bytes
+	 * but simply instantiated, then the current version is returned.
 	 *
-	 * @param foundVersion the version found from reading the input stream
-	 * @throws VersionMismatchException thrown when serialization versions mismatch
+	 * @return the read serialization version, or the current version if the instance was not
read from bytes.
 	 */
-	protected void resolveVersionRead(int foundVersion) throws VersionMismatchException {
-		if (!isCompatibleVersion(foundVersion)) {
-			int expectedVersion = getVersion();
-			throw new VersionMismatchException(
-					"Incompatible version: found " + foundVersion + ", required " + expectedVersion);
-		}
+	public int getReadVersion() {
+		return (readVersion == Integer.MIN_VALUE) ? getVersion() : readVersion;
 	}
 
 	/**
-	 * Checks for compatibility between this and the found version. Subclasses can override
this methods in case of
-	 * intended backwards backwards compatibility.
+	 * Returns the compatible version values.
 	 *
-	 * @param version version number to compare against.
-	 * @return true, iff this is compatible to the passed version.
+	 * <p>By default, the base implementation recognizes only the current version (identified
by {@link #getVersion()})
+	 * as compatible. This method can be used as a hook and may be overridden to identify more
compatible versions.
+	 *
+	 * @return an array of integers representing the compatible version values.
 	 */
-	public boolean isCompatibleVersion(int version) {
-		return getVersion() == version;
+	public int[] getCompatibleVersions() {
+		return new int[] {getVersion()};
+	}
+
+	private void resolveVersionRead(int readVersion) throws VersionMismatchException {
+
+		int[] compatibleVersions = getCompatibleVersions();
+		for (int compatibleVersion : compatibleVersions) {
+			if (compatibleVersion == readVersion) {
+				return;
+			}
+		}
+
+		throw new VersionMismatchException(
+			"Incompatible version: found " + readVersion + ", compatible versions are " + Arrays.toString(compatibleVersions));
 	}
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/347100de/flink-core/src/test/java/org/apache/flink/core/io/VersionedIOWriteableTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/io/VersionedIOWriteableTest.java
b/flink-core/src/test/java/org/apache/flink/core/io/VersionedIOWriteableTest.java
index b7b6d6f..ec5f792 100644
--- a/flink-core/src/test/java/org/apache/flink/core/io/VersionedIOWriteableTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/io/VersionedIOWriteableTest.java
@@ -65,8 +65,8 @@ public class VersionedIOWriteableTest {
 
 		testWriteable = new TestWriteable(2) {
 			@Override
-			public boolean isCompatibleVersion(int version) {
-				return getVersion() >= version;
+			public int[] getCompatibleVersions() {
+				return new int[] {1, 2};
 			}
 		};
 		try (ByteArrayInputStreamWithPos in = new ByteArrayInputStreamWithPos(serialized)) {
@@ -130,19 +130,9 @@ public class VersionedIOWriteableTest {
 			this.data = in.readUTF();
 		}
 
-		@Override
-		protected void resolveVersionRead(int foundVersion) throws VersionMismatchException {
-			super.resolveVersionRead(foundVersion);
-		}
-
-		@Override
-		public boolean isCompatibleVersion(int version) {
-			return super.isCompatibleVersion(version);
-		}
-
 		public String getData() {
 			return data;
 		}
 	}
 
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/347100de/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
index a389c4f..a20628c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedBackendSerializationProxy.java
@@ -20,7 +20,6 @@ package org.apache.flink.runtime.state;
 
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializerSerializationProxy;
-import org.apache.flink.core.io.VersionMismatchException;
 import org.apache.flink.core.io.VersionedIOReadableWritable;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
@@ -41,7 +40,6 @@ public class KeyedBackendSerializationProxy extends VersionedIOReadableWritable
 	private TypeSerializer<?> keySerializer;
 	private List<RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?>> stateMetaInfoSnapshots;
 
-	private int restoredVersion;
 	private ClassLoader userCodeClassLoader;
 
 	public KeyedBackendSerializationProxy(ClassLoader userCodeClassLoader) {
@@ -57,8 +55,6 @@ public class KeyedBackendSerializationProxy extends VersionedIOReadableWritable
 		Preconditions.checkNotNull(stateMetaInfoSnapshots);
 		Preconditions.checkArgument(stateMetaInfoSnapshots.size() <= Short.MAX_VALUE);
 		this.stateMetaInfoSnapshots = stateMetaInfoSnapshots;
-
-		this.restoredVersion = VERSION;
 	}
 
 	public List<RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?>> getStateMetaInfoSnapshots()
{
@@ -74,20 +70,10 @@ public class KeyedBackendSerializationProxy extends VersionedIOReadableWritable
 		return VERSION;
 	}
 
-	public int getRestoredVersion() {
-		return restoredVersion;
-	}
-
-	@Override
-	protected void resolveVersionRead(int foundVersion) throws VersionMismatchException {
-		super.resolveVersionRead(foundVersion);
-		this.restoredVersion = foundVersion;
-	}
-
 	@Override
-	public boolean isCompatibleVersion(int version) {
+	public int[] getCompatibleVersions() {
 		// we are compatible with version 3 (Flink 1.3.x) and version 1 & 2 (Flink 1.2.x)
-		return super.isCompatibleVersion(version) || version == 2 || version == 1;
+		return new int[] {VERSION, 2, 1};
 	}
 
 	@Override
@@ -119,7 +105,7 @@ public class KeyedBackendSerializationProxy extends VersionedIOReadableWritable
 		for (int i = 0; i < numKvStates; i++) {
 			stateMetaInfoSnapshots.add(
 				KeyedBackendStateMetaInfoSnapshotReaderWriters
-					.getReaderForVersion(restoredVersion, userCodeClassLoader)
+					.getReaderForVersion(getReadVersion(), userCodeClassLoader)
 					.readStateMetaInfo(in));
 		}
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/347100de/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java
index 91d7aab..074d84e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.state;
 
-import org.apache.flink.core.io.VersionMismatchException;
 import org.apache.flink.core.io.VersionedIOReadableWritable;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
@@ -39,8 +38,6 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
 	private List<RegisteredOperatorBackendStateMetaInfo.Snapshot<?>> stateMetaInfoSnapshots;
 	private ClassLoader userCodeClassLoader;
 
-	private int restoredVersion;
-
 	public OperatorBackendSerializationProxy(ClassLoader userCodeClassLoader) {
 		this.userCodeClassLoader = Preconditions.checkNotNull(userCodeClassLoader);
 	}
@@ -50,8 +47,6 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
 
 		this.stateMetaInfoSnapshots = Preconditions.checkNotNull(stateMetaInfoSnapshots);
 		Preconditions.checkArgument(stateMetaInfoSnapshots.size() <= Short.MAX_VALUE);
-
-		this.restoredVersion = VERSION;
 	}
 
 	@Override
@@ -60,15 +55,9 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
 	}
 
 	@Override
-	protected void resolveVersionRead(int foundVersion) throws VersionMismatchException {
-		super.resolveVersionRead(foundVersion);
-		this.restoredVersion = foundVersion;
-	}
-
-	@Override
-	public boolean isCompatibleVersion(int version) {
+	public int[] getCompatibleVersions() {
 		// we are compatible with version 2 (Flink 1.3.x) and version 1 (Flink 1.2.x)
-		return super.isCompatibleVersion(version) || version == 1;
+		return new int[] {VERSION, 1};
 	}
 
 	@Override
@@ -92,7 +81,7 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
 		for (int i = 0; i < numKvStates; i++) {
 			stateMetaInfoSnapshots.add(
 				OperatorBackendStateMetaInfoSnapshotReaderWriters
-					.getReaderForVersion(restoredVersion, userCodeClassLoader)
+					.getReaderForVersion(getReadVersion(), userCodeClassLoader)
 					.readStateMetaInfo(in));
 		}
 	}

http://git-wip-us.apache.org/repos/asf/flink/blob/347100de/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
index bc314df..11e7760 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
@@ -448,7 +448,7 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K>
{
 						StateTableByKeyGroupReader keyGroupReader =
 								StateTableByKeyGroupReaders.readerForVersion(
 										stateTable,
-										serializationProxy.getRestoredVersion());
+										serializationProxy.getReadVersion());
 
 						keyGroupReader.readMappingsInKeyGroup(inView, keyGroupIndex);
 					}


Mime
View raw message