dawidwys commented on code in PR #21635:
URL: https://github.com/apache/flink/pull/21635#discussion_r1090379604
##########
flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V1TestTypeSerializerSnapshot.java:
##########
@@ -36,16 +36,14 @@ public int getCurrentVersion() {
@Override
public TypeSerializerSchemaCompatibility<TestType>
resolveSchemaCompatibility(
- TypeSerializer<TestType> newSerializer) {
- if (newSerializer instanceof TestType.V1TestTypeSerializer) {
+ TypeSerializerSnapshot<TestType> oldSerializerSnapshot) {
+ if (oldSerializerSnapshot instanceof V1TestTypeSerializerSnapshot) {
return TypeSerializerSchemaCompatibility.compatibleAsIs();
- } else if (newSerializer instanceof TestType.V2TestTypeSerializer) {
- return
TypeSerializerSchemaCompatibility.compatibleAfterMigration();
- } else if (newSerializer instanceof
TestType.ReconfigurationRequiringTestTypeSerializer) {
- // we mimic the reconfiguration by just re-instantiating the
correct serializer
- return
TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
- new TestType.V1TestTypeSerializer());
- } else if (newSerializer instanceof
TestType.IncompatibleTestTypeSerializer) {
+ } else if (oldSerializerSnapshot
+ instanceof
TestType.ReconfigurationRequiringTestTypeSerializer
Review Comment:
This is always false, you compare serializer with snapshot.
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/SimpleTypeSerializerSnapshot.java:
##########
@@ -74,9 +74,14 @@ public TypeSerializer<T> restoreSerializer() {
@Override
public TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer) {
-
- return newSerializer.getClass() == serializerSupplier.get().getClass()
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ if (!(oldSerializerSnapshot instanceof SimpleTypeSerializerSnapshot)) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ SimpleTypeSerializerSnapshot<T> oldSimpleTypeSerializerSnapshot =
+ (SimpleTypeSerializerSnapshot<T>) oldSerializerSnapshot;
+ return
oldSimpleTypeSerializerSnapshot.serializerSupplier.get().getClass()
Review Comment:
Let's use `restoreSerializer` here.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/V2TestTypeSerializerSnapshot.java:
##########
@@ -36,18 +36,14 @@ public int getCurrentVersion() {
@Override
public TypeSerializerSchemaCompatibility<TestType>
resolveSchemaCompatibility(
- TypeSerializer<TestType> newSerializer) {
- if (newSerializer instanceof TestType.V2TestTypeSerializer) {
+ TypeSerializerSnapshot<TestType> oldSerializerSnapshot) {
+ if (oldSerializerSnapshot instanceof V2TestTypeSerializerSnapshot) {
return TypeSerializerSchemaCompatibility.compatibleAsIs();
- } else if (newSerializer instanceof
TestType.ReconfigurationRequiringTestTypeSerializer) {
- // we mimic the reconfiguration by just re-instantiating the
correct serializer
- return
TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer(
- new TestType.V2TestTypeSerializer());
- } else if (
- // migrating from V2 -> V1 is not supported
- newSerializer instanceof TestType.V1TestTypeSerializer
- || newSerializer instanceof
TestType.IncompatibleTestTypeSerializer) {
-
+ } else if (oldSerializerSnapshot instanceof
V1TestTypeSerializerSnapshot) {
+ return
TypeSerializerSchemaCompatibility.compatibleAfterMigration();
+ } else if (oldSerializerSnapshot
+ instanceof
TestType.ReconfigurationRequiringTestTypeSerializer
+ || oldSerializerSnapshot instanceof
TestType.IncompatibleTestTypeSerializer) {
Review Comment:
ditto, also always false, you compare snapshot with a serializer
##########
flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializerSnapshot.java:
##########
@@ -124,11 +124,40 @@ void readSnapshot(int readVersion, DataInputView in,
ClassLoader userCodeClassLo
* program's serializer re-serializes the data, thus converting the format
during the restore
* operation.
*
+ * @deprecated This method has been replaced by {@link
TypeSerializerSnapshot
+ * #resolveSchemaCompatibility(TypeSerializerSnapshot)}.
* @param newSerializer the new serializer to check.
* @return the serializer compatibility result.
*/
- TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
- TypeSerializer<T> newSerializer);
+ @Deprecated
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializer<T> newSerializer) {
+ return
newSerializer.snapshotConfiguration().resolveSchemaCompatibility(this);
+ }
+
+ /**
+ * Checks current serializer's compatibility to read data written by the
prior serializer.
+ *
+ * <p>When a checkpoint/savepoint is restored, this method checks whether
the serialization
+ * format of the data in the checkpoint/savepoint is compatible for the
format of the serializer
+ * used by the program that restores the checkpoint/savepoint. The outcome
can be that the
+ * serialization format is compatible, that the program's serializer needs
to reconfigure itself
+ * (meaning to incorporate some information from the
TypeSerializerSnapshot to be compatible),
+ * that the format is outright incompatible, or that a migration needed.
In the latter case, the
+ * TypeSerializerSnapshot produces a serializer to deserialize the data,
and the restoring
+ * program's serializer re-serializes the data, thus converting the format
during the restore
+ * operation.
+ *
+ * <p>This method must be implemented to clarify the compatibility. See
FLIP-263 for more
+ * details.
+ *
+ * @param oldSerializerSnapshot the old serializer snapshot to check.
+ * @return the serializer compatibility result.
+ */
+ default TypeSerializerSchemaCompatibility<T> resolveSchemaCompatibility(
+ TypeSerializerSnapshot<T> oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
Review Comment:
What happens if a user has a custom serializer that embeds e.g. a
`PojoSerializer` and want to migrate to the new method. A situation like:
```
class CustomCollectionSerializerSnapshot<T> extends
TypeSerializerSnapshot<Collection<T>>{
private final TypeSerializerSnapshot innerSnapshot;
public ... resolveSchemaCompatibility(TypeSerializerSnapshot oldSnapshot) {
...
return
innerSnapshot.resolveSchemaCompatibility(oldSnapshot.innerSnapshot);
}
}
CustomCollectionSerializer<? extends POJO> serializer
```
This would fail, wouldn't it?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/testutils/statemigration/TestType.java:
##########
@@ -200,8 +237,34 @@ public TestType deserialize(DataInputView source) throws
IOException {
@Override
public TypeSerializerSnapshot<TestType> snapshotConfiguration() {
- throw new UnsupportedOperationException(
- "This is an incompatible serializer; shouldn't be used.");
+ return new TypeSerializerSnapshot<TestType>() {
+ @Override
+ public int getCurrentVersion() {
+ return 0;
+ }
+
+ @Override
+ public void writeSnapshot(DataOutputView out) {
+ // do nothing
+ }
+
+ @Override
+ public void readSnapshot(
+ int readVersion, DataInputView in, ClassLoader
userCodeClassLoader) {
+ // do nothing
+ }
+
+ @Override
+ public TypeSerializer<TestType> restoreSerializer() {
+ return new IncompatibleTestTypeSerializer();
+ }
+
+ @Override
+ public TypeSerializerSchemaCompatibility<TestType>
resolveSchemaCompatibility(
+ TypeSerializerSnapshot<TestType>
oldSerializerSnapshot) {
+ return TypeSerializerSchemaCompatibility.incompatible();
+ }
+ };
Review Comment:
We shouldn't do that. Let's implement the original intent in a proper way.
If I understand the issue (why the code needs to be changed in the first
place) is that you need to perform
`newSerializer.snapshotConfiguration().resolveSchemaCompatibility`. Let's
implement that properly.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/state/InternalPriorityQueueTestBase.java:
##########
@@ -546,13 +546,13 @@ public TypeSerializer<TestElement> restoreSerializer() {
@Override
public TypeSerializerSchemaCompatibility<TestElement>
resolveSchemaCompatibility(
- TypeSerializer<TestElement> newSerializer) {
- if (!(newSerializer instanceof TestElementSerializer)) {
+ TypeSerializerSnapshot<TestElement> oldSerializerSnapshot)
{
+ if (!(oldSerializerSnapshot instanceof Snapshot)) {
return TypeSerializerSchemaCompatibility.incompatible();
}
- TestElementSerializer testElementSerializer =
(TestElementSerializer) newSerializer;
- return (revision <= testElementSerializer.getRevision())
+ Snapshot snapshot = (Snapshot) oldSerializerSnapshot;
+ return (revision <= snapshot.getRevision())
Review Comment:
That should be the other way round, shouldn't it? Old revision should be
lower than the new one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
