Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/AbstractHeapState.java (Java)

Code before:
            throws Exception {

        Preconditions.checkNotNull(serializedKeyAndNamespace);
        Preconditions.checkNotNull(safeKeySerializer);
        Preconditions.checkNotNull(safeNamespaceSerializer);
        Preconditions.checkNotNull(safeValueSerializer);

        Tuple2<K, N> keyAndNamespace =
                KvStateSerializer.deserializeKeyAndNamespace(
                        serializedKeyAndNamespace, safeKeySerializer, safeNamespaceSerializer);

        SV result = stateTable.get(keyAndNamespace.f0, keyAndNamespace.f1);

        if (result == null) {
            return null;
        }
        return KvStateSerializer.serializeValue(result, safeValueSerializer);
    }

    public SV migrateValueForTtlStateMigration(
            SV state,
            TypeSerializer<SV> priorSerializer,
            TypeSerializer<SV> newSerializer,
            TtlTimeProvider ttlTimeProvider) {

        Preconditions.checkArgument(priorSerializer instanceof TtlAwareSerializer);
        Preconditions.checkArgument(newSerializer instanceof TtlAwareSerializer);
        TtlAwareSerializer<SV, ?> previousTtlAwareSerializer =
                (TtlAwareSerializer<SV, ?>) priorSerializer;
        TtlAwareSerializer<SV, ?> currentTtlAwareSerializer =
                (TtlAwareSerializer<SV, ?>) newSerializer;

        if (!previousTtlAwareSerializer.isTtlEnabled()
                && currentTtlAwareSerializer.isTtlEnabled()) {
            return (SV) new TtlValue<>(state, ttlTimeProvider.currentTimestamp());
        } else if (previousTtlAwareSerializer.isTtlEnabled()
                && !currentTtlAwareSerializer.isTtlEnabled()) {
            return (SV) ((TtlValue) state).getUserValue();
        }
        return state;
    }

    /** This should only be used for testing. */
    @VisibleForTesting
    public StateTable<K, N, SV> getStateTable() {
        return stateTable;
    }

    protected SV getDefaultValue() {
        if (defaultValue != null) {
            return valueSerializer.copy(defaultValue);

Review comment (suggestion):
nit: in the variable names we use  prior and new and previous and current. I suggest being consistent; 
previousSerializer previousAwareSerializer
currentSerializer currentTtlAwareSerializer

Code after:
            throws Exception {

        Preconditions.checkNotNull(serializedKeyAndNamespace);
        Preconditions.checkNotNull(safeKeySerializer);
        Preconditions.checkNotNull(safeNamespaceSerializer);
        Preconditions.checkNotNull(safeValueSerializer);

        Tuple2<K, N> keyAndNamespace =
                KvStateSerializer.deserializeKeyAndNamespace(
                        serializedKeyAndNamespace, safeKeySerializer, safeNamespaceSerializer);

        SV result = stateTable.get(keyAndNamespace.f0, keyAndNamespace.f1);

        if (result == null) {
            return null;
        }
        return KvStateSerializer.serializeValue(result, safeValueSerializer);
    }

    @SuppressWarnings("unchecked")
    public SV migrateTtlValue(
            SV stateValue,
            TtlAwareSerializer<SV, ?> currentTtlAwareSerializer,
            TtlTimeProvider ttlTimeProvider) {
        if (currentTtlAwareSerializer.isTtlEnabled()) {
            return (SV) new TtlValue<>(stateValue, ttlTimeProvider.currentTimestamp());
        }

        return (SV) ((TtlValue<?>) stateValue).getUserValue();
    }

    /** This should only be used for testing. */
    @VisibleForTesting
    public StateTable<K, N, SV> getStateTable() {
        return stateTable;
    }

    protected SV getDefaultValue() {
        if (defaultValue != null) {
            return valueSerializer.copy(defaultValue);
        } else {
            return null;
        }
    }

    protected AbstractHeapState<K, N, SV> setNamespaceSerializer(
            TypeSerializer<N> namespaceSerializer) {
        this.namespaceSerializer = namespaceSerializer;
        return this;
    }