[
https://issues.apache.org/jira/browse/KAFKA-7663?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17000278#comment-17000278
]
Levani Kokhreidze commented on KAFKA-7663:
------------------------------------------
Hello Matthias,
When `GlobalStateManagerImpl#restoreState` is triggered `Processor#process`
won't have access to state store since init is not called yet. If I am not
mistaken, in that case we would have to manually init processor with some
ProcessorContext implementation. Do you think that's better approach compared
to giving user one more parameter where he/she can transform records during
restore? Actually, maybe additional parameter should be mapper rather than
predicate, since we could different types of records between input topic and
the actual global store, where null values are filtered out during restore
process?
> 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)
