[ https://issues.apache.org/jira/browse/FLINK-9513?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Andrey Zagrebin updated FLINK-9513: ----------------------------------- Description: The main idea is to wrap user state value with a class holding the value and the expiration timestamp (maybe meta data in future) and use the new object as a value in the existing implementations: {code:java} class TtlValue<V> { V value; long expirationTimestamp; } {code} The original state binder factory is wrapped with TtlStateBinder if TTL is enabled: {code:java} state = ttlConfig.updateType == DISABLED ? bind(binder) : bind(new TtlStateBinder(binder, timerService)); {code} TtlStateBinder decorates the states produced by the original binder with TTL logic wrappers and adds TtlValue serialisation logic: {code:java} TtlStateBinder { StateBinder binder; ProcessingTimeProvier timeProvider; // System.currentTimeMillis() <V> TtlValueState<V> createValueState(valueDesc) { serializer = new TtlValueSerializer(valueDesc.getSerializer); ttlValueDesc = new ValueDesc(serializer, ...); // or implement custom TypeInfo originalStateWithTtl = binder.createValueState(valueDesc); return new TtlValueState(originalStateWithTtl, timeProvider); } // List, Map, ... } {code} TTL serializer should add expiration timestamp was: The main idea is to wrap user state value with a class holding the value and the expiration timestamp (maybe meta data in future) and use the new object as a value in the existing implementations: {code:java} class TtlValue<V> { V value; long expirationTimestamp; } {code} The original state binder factory is wrapped with TtlStateBinder if TTL is enabled: {code:java} state = ttlConfig.updateType == DISABLED ? bind(binder) : bind(new TtlStateBinder(binder, timerService)); {code} TtlStateBinder decorates the states produced by the original binder with TTL logic wrappers and adds TtlValue serialisation logic: {code:java} TtlStateBinder { StateBinder binder; ProcessingTimeProvier timeProvider; // System.currentTimeMillis() <V> TtlValueState<V> createValueState(valueDesc) { serializer = new TtlValueSerializer(valueDesc.getSerializer); ttlValueDesc = new ValueDesc(serializer, ...); // or implement custom TypeInfo originalStateWithTtl = binder.createValueState(valueDesc); return new TtlValueState(originalStateWithTtl, timeProvider); } // List, Map, ... } {code} | | | | TTL serializer should add expiration timestamp > Wrap state binder with TTL logic > -------------------------------- > > Key: FLINK-9513 > URL: https://issues.apache.org/jira/browse/FLINK-9513 > Project: Flink > Issue Type: Sub-task > Components: State Backends, Checkpointing > Affects Versions: 1.6.0 > Reporter: Andrey Zagrebin > Priority: Major > Fix For: 1.6.0 > > > The main idea is to wrap user state value with a class holding the value and > the expiration timestamp (maybe meta data in future) and use the new object > as a value in the existing implementations: > {code:java} > class TtlValue<V> { > V value; > long expirationTimestamp; > } > {code} > The original state binder factory is wrapped with TtlStateBinder if TTL is > enabled: > {code:java} > state = ttlConfig.updateType == DISABLED ? > bind(binder) : bind(new TtlStateBinder(binder, timerService)); > {code} > TtlStateBinder decorates the states produced by the original binder with TTL > logic wrappers and adds TtlValue serialisation logic: > {code:java} > TtlStateBinder { > StateBinder binder; > ProcessingTimeProvier timeProvider; // System.currentTimeMillis() > <V> TtlValueState<V> createValueState(valueDesc) { > serializer = new TtlValueSerializer(valueDesc.getSerializer); > ttlValueDesc = new ValueDesc(serializer, ...); > // or implement custom TypeInfo > originalStateWithTtl = binder.createValueState(valueDesc); > return new TtlValueState(originalStateWithTtl, timeProvider); > } > // List, Map, ... > } > {code} > TTL serializer should add expiration timestamp -- This message was sent by Atlassian JIRA (v7.6.3#76005)