Tzu-Li (Gordon) Tai created FLINK-11073:
-------------------------------------------
Summary: Make serializers immutable / provide option
TypeSerializerSchemaCompatibility.compatibleWithReconfiguredSerializer
Key: FLINK-11073
URL: https://issues.apache.org/jira/browse/FLINK-11073
Project: Flink
Issue Type: Improvement
Components: Type Serialization System
Reporter: Tzu-Li (Gordon) Tai
Assignee: Tzu-Li (Gordon) Tai
Fix For: 1.8.0
h2. Motivation
Right now, when a new serializer is provided to the old serializer (or, to be
more specific, the old serializer's snapshot) for state schema compatibility
checks, if the new serializer is reconfigurable so that it may be compatible,
the only possible way to do this is reconfigure the new serializer in-place and
return {{TypeSerializerSchemaCompatibility.compatibleAsIs()}} as the result of
the compatibility check.
One solid example is the {{KryoSerializer}}. The {{KryoSerializer}} contains as
configuration a map of serialized classes to their registered ids. This mapping
may change on restore executions, and the new {{KryoSerializer}} must
reconfigure this mapping to match with the previous execution before the new
{{KryoSerializer}} can be used for state access.
Right now, this is performed by directly mutating the map in the new serializer
instance.
This mutative behaviour is fragile, especially when taking into account scale
down / up scenarios which could easily result in mismatching state serializer
configurations across TMs.
h2. Proposed Approach
# The {{TypeSerializerSchemaCompatibility}} result class should be extended to
contain an option {{compatibleWithReconfiguredSerializer(TypeSerializer)}},
which would wrap a new instance of a reconfigured version of the new serializer.
# Callers of the compatibility check needs to be aware of this case and
respect it, using the provided reconfigured serializer instance when one is
provided. In Flink, there are two places which performs compatibility checks on
serializers: 1) composite serializers which contain nested serializers, and
therefore needs to check compatibility of its nested serializers, and 2) in
state backends, checking the compatibility of the new serializer with the old
serializer.
# Introduce {{CompositeTypeSerializerSnapshot}} to encapsulate logic of
handling reconfiguration of nested serializers: if a composite serializer has a
nested serializer that returns a new reconfigured instance of itself, than the
result of the compatibility check on the composite serializer should also wrap
a reconfigured version of the composite serializer that holds the reconfigured
nested serializer. This logic should be captured in a base abstract class, say
{{CompositeTypeSerializerSnapshot}} so that it can be commonly shared by many
of Flink's composite serializers.
# For composite serializers that is still using the legacy, less-powerful
{{TypeSerializerConfigSnapshot}} and {{CompatibilityResult}} abstractions,
while its nested serializer is signaling that it has reconfigured itself, this
should be detected and an error is thrown complaining that the outer composite
serializer needs to be upgraded to use the new serializer snapshot and
compatibility abstractions. This approach follows the same way we handled
bridging the new {{TypeSerializerSchemaCompatibility}} and old
{{CompatibilityResult}} class in Flink 1.7.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
