Versions Compared

Old Version 22

changes.mady.by.user Andrey Zagrebin

Saved on

compared with

New Version 23

changes.mady.by.user Andrey Zagrebin

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

The original state factory provided in backends is wrapped with TtlStateFactory if TTL is enabled:

state = stateDesc.getTtlConfig().isEnabled() ?


  

 new TtlStateFactory(originalStateFactory,..).createState(...) :

 

 

 

originalStateFactory.createState(...);

TtlStateFactory decorates the states produced by the original factory with TTL logic wrappers and adds TtlValue serialisation logic:

TtlStateFactory {
     KeyedStateFactory originalStateFactory;

    TtlTimeProvider timeProvider; // e.g. System.currentTimeMillis()

    <V> TtlValueState<V> createValueState(valueDesc) {

        serializer = new TtlValueSerializer(valueDesc.getSerializer);
        ttlValueDesc = new ValueDesc(valueDesc.name, serializer);
        originalStateWithTtl = originalStateFactory.createValueState(valueDesc);

        returnnew TtlValueState(originalStateWithTtl, timeProvider);

    }

    // List, Map, ...
}

...

TtlValueState<V> implements ValueState<V> {

    // List, Map, ....
     ValueState<TtlValue<V>> underlyingState;

    TtlTimeProvider timeProvider;


     V value() {
         TtlValue<V> valueWithTtl = underlyingState.get();
         // ttl logic here (e.g. update timestamp)
         return valueWithTtl.getValue();
     }

    void update() {

        ...

        underlyingState.update(valueWithTtl);

        ...

    }
}

Cleanup: issue delete/rewrite upon realising expiration during access/modification.

...

Filter out expired entries in case of full state scan to take a full checkpoint. This approach does not reduce the size of local state used in running job but reduces the size of taken snapshot. In case of restore from such a checkpoint, the local state will not contain expired entries as well.The implementation is based on extending backends to support custom state transformers. The backends call the transformer for each state entry during the full snapshot scan and the transformer decide whether to keep, modify or drop the state entry. TTL has its own relevant implementation of state transformers to check timestamp and filter out expired entries.

public interface StateSnapshotTransformer<T> {

    @Nullable
    T filterOrTransform(T value);
}

To avoid concurrent usage of transformer objects (can be relevant for performance and reuse of serialization buffers), each snapshotting thread uses a factory to produce a thread-confined transformer.

...