[ https://issues.apache.org/jira/browse/KAFKA-7663?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16999963#comment-16999963 ]
Levani Kokhreidze edited comment on KAFKA-7663 at 12/19/19 11:03 AM: --------------------------------------------------------------------- Hello, I've stumbled upon this issue. I have a use-case where I don't want to restore all the records from the topic for global store and would like to define some "filtering" logic for the restoration process. If I am not mistaken, (and it was already pointed out) `GlobalStateManagerImpl#restoreState` is called way before processor `init` method is called, therefore during restoration calling defined processor may not be feasible nor intuitive for the user since we would always expect `init` to be called before the `process`. I was thinking, what if we add one more additional parameter to the `addGlobalStore` method where users can define record restoration filtering logic, something like: {code:java} public synchronized void addGlobalStore( final StoreBuilder<KeyValueStore> storeBuilder, final String sourceName, final String topic, final ConsumedInternal consumed, final String processorName, final ProcessorSupplier stateUpdateSupplier, // new method parameter final GlobalStateRestoreFilter globalStateRestoreFilter ) { ... }{code} Signature of the interface may look something like this {code:java} public interface GlobalStateRestoreFilter<K, V> { boolean shouldRestore(String topic, K key, V value); } {code} This change will require KIP, happy to work on it, but would like to hear initial ideas around this. With this end-users will have clear contract if they would like to filter out some records during restoration process. cc [~vvcephei] [~ableegoldman] [~pkleindl] [~twbecker] was (Author: lkokhreidze): Hello, I've stumbled upon this issue. I have a use-case where I don't want to restore all the records from the topic for global store and would like to define some "filtering" logic for the restoration process. If I am not mistaken, (and it was already pointed out) `GlobalStateManagerImpl#restoreState` is called way before processor `init` method is called, therefore during restoration calling defined processor may not be feasible nor intuitive for the user since we would always expect `init` to be called before the `process`. I was thinking, what if we add one more additional parameter to the `addGlobalStore` method where users can define record restoration filtering logic, something like: {code:java} public synchronized void addGlobalStore(final StoreBuilder<KeyValueStore> storeBuilder, final String sourceName, final String topic, final ConsumedInternal consumed, final String processorName, final ProcessorSupplier stateUpdateSupplier, // new method parameter final GlobalStateRestoreFilter globalStateRestoreFilter) { ... }{code} Signature of the interface may look something like this {code:java} public interface GlobalStateRestoreFilter<K, V> { boolean shouldRestore(String topic, K key, V value); } {code} This change will require KIP, happy to work on it, but would like to hear initial ideas around this. With this end-users will have clear contract if they would like to filter out some records during restoration process. cc [~vvcephei] [~ableegoldman] [~pkleindl] [~twbecker] > Custom Processor supplied on addGlobalStore is not used when restoring state > from topic > --------------------------------------------------------------------------------------- > > Key: KAFKA-7663 > URL: https://issues.apache.org/jira/browse/KAFKA-7663 > Project: Kafka > Issue Type: Improvement > Components: streams > Affects Versions: 1.0.0 > Reporter: Frederic Tardif > Priority: Major > Attachments: image-2018-11-20-11-42-14-697.png > > > I have implemented a StreamBuilder#{{addGlobalStore}} supplying a custom > processor responsible to transform a K,V record from the input stream into a > V,K records. It works fine and my {{store.all()}} does print the correct > persisted V,K records. However, if I clean the local store and restart the > stream app, the global table is reloaded but without going through the > processor supplied; instead, it calls {{GlobalStateManagerImp#restoreState}} > which simply stores the input topic K,V records into rocksDB (hence bypassing > the mapping function of my custom processor). I believe this must not be the > expected result? > This is a follow up on stackoverflow discussion around storing a K,V topic > as a global table with some stateless transformations based on a "custom" > processor added on the global store: > [https://stackoverflow.com/questions/50993292/kafka-streams-shared-changelog-topic#comment93591818_50993729] > If we address this issue, we should also apply > `default.deserialization.exception.handler` during restore (cf. KAFKA-8037) > -- This message was sent by Atlassian Jira (v8.3.4#803005)