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

Code before:

        final TypeSerializer<V> dupSerializer =
                ((ListSerializer<V>) safeValueSerializer).getElementSerializer();

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(baos);

        // write the same as RocksDB writes lists, with one ',' separator
        for (int i = 0; i < result.size(); i++) {
            dupSerializer.serialize(result.get(i), view);
            if (i < result.size() - 1) {
                view.writeByte(',');
            }
        }
        view.flush();

        return baos.toByteArray();
    }

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

        Preconditions.checkArgument(
                priorSerializer instanceof TtlAwareSerializer.TtlAwareListSerializer);
        Preconditions.checkArgument(
                newSerializer instanceof TtlAwareSerializer.TtlAwareListSerializer);

        TtlAwareSerializer<V, ?> priorTtlAwareElementSerializer =
                ((TtlAwareSerializer.TtlAwareListSerializer<V>) priorSerializer)
                        .getElementSerializer();
        TtlAwareSerializer<V, ?> newTtlAwareElementSerializer =
                ((TtlAwareSerializer.TtlAwareListSerializer<V>) newSerializer)
                        .getElementSerializer();

        if (!priorTtlAwareElementSerializer.isTtlEnabled()
                && newTtlAwareElementSerializer.isTtlEnabled()) {
            state.replaceAll(v -> (V) new TtlValue<>(v, ttlTimeProvider.currentTimestamp()));
        } else if (priorTtlAwareElementSerializer.isTtlEnabled()
                && !newTtlAwareElementSerializer.isTtlEnabled()) {
            state.replaceAll(v -> (V) ((TtlValue) v).getUserValue());
        }
        return state;
    }

    // ------------------------------------------------------------------------
    //  state merging
    // ------------------------------------------------------------------------


Review comment (suggestion):
it looks like there is suplication of code, can we re arrange the code so the following is not repeated for the different cases
```

              priorSerializer instanceof TtlAwareSerializer.TtlAwareListSerializer);
        Preconditions.checkArgument(
                newSerializer instanceof TtlAwareSerializer.TtlAwareListSerializer);

        TtlAwareSerializer<V, ?> priorTtlAwareElementSerializer =
                ((TtlAwareSerializer.TtlAwareListSerializer<V>) priorSerializer)
                        .getElementSerializer();
        TtlAwareSerializer<V, ?> newTtlAwareElementSerializer =
                ((TtlAwareSerializer.TtlAwareListSerializer<V>) newSerializer)
                        .getElementSerializer();
```

Code after:

        final TypeSerializer<V> dupSerializer =
                ((ListSerializer<V>) safeValueSerializer).getElementSerializer();

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputViewStreamWrapper view = new DataOutputViewStreamWrapper(baos);

        // write the same as RocksDB writes lists, with one ',' separator
        for (int i = 0; i < result.size(); i++) {
            dupSerializer.serialize(result.get(i), view);
            if (i < result.size() - 1) {
                view.writeByte(',');
            }
        }
        view.flush();

        return baos.toByteArray();
    }

    @SuppressWarnings("unchecked")
    public List<V> migrateTtlValue(
            List<V> stateValue,
            TtlAwareSerializer<List<V>, ?> currentTtlAwareSerializer,
            TtlTimeProvider ttlTimeProvider) {
        if (currentTtlAwareSerializer.isTtlEnabled()) {
            stateValue.replaceAll(v -> (V) new TtlValue<>(v, ttlTimeProvider.currentTimestamp()));
        } else {
            stateValue.replaceAll(v -> (V) ((TtlValue<?>) v).getUserValue());
        }

        return stateValue;
    }

    // ------------------------------------------------------------------------
    //  state merging
    // ------------------------------------------------------------------------

    @Override
    protected List<V> mergeState(List<V> a, List<V> b) {
        a.addAll(b);
        return a;
    }

    @Override
    public void update(List<V> values) throws Exception {
        Preconditions.checkNotNull(values, "List of values to add cannot be null.");

        if (values.isEmpty()) {
            clear();
            return;
        }