Versions Compared

Old Version 21

changes.mady.by.user Andrey Zagrebin

Saved on

compared with

New Version 22

changes.mady.by.user Andrey Zagrebin

Saved on

Key

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

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Table of Contents

Motivation

Reasons to introduce Flink native state expiration:

...

TTL can be configured and enabled in the abstract StateDescriptor and become available in all subclasses:

enum UpdateType { Disabled, OnCreateAndWrite, OnReadAndWrite }

enum StateVisibility { ReturnExpiredIfNotCleanedUp, NeverReturnExpired }

enum TtlTimeCharacteristic { ProcessingTime, EventTime }

class StateTtlConfig {
    UpdateType updateType;    StateVisibility stateVisibility;
     TtlTimeCharacteristic timeCharacteristic;
     Time ttl;
}

abstract class StateDescriptor {   

    void enableTimeToLive(StateTtlConfig ttlConfig) { … }

}

StateTtlConfig ttlConfig = StateTtlConfig

    .newBuilder(Time.seconds(1))

    .setUpdateType(..)   

    …   

    .build();

XxxStateDescriptor stateDesc =

new

new XxxStateDescriptor(...);

stateDesc.enableTimeToLive(ttlConfig);

State value with timestamp

The main idea is to wrap user state value with a class holding the value and the last access timestamp (maybe meta data in future) and use the new object as a value in the existing implementations:

class TtlValue<V> {
     V value;
     long lastAccessTimestamp;
}

Wrapping state factory

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, ...
}

TTL serializer should add expiration timestamp.

...

TTL state decorators use original state with packed TTL and add TTL logic using time provider:

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.

...