[jira] [Commented] (FLINK-26372) Allow to configure Changelog Storage per program

[Roman Khachatryan (Jira)]

Title: Message Title

Roman Khachatryan commented on FLINK-26372   Re: Allow to configure Changelog Storage per program Thanks a lot for the proposal Feifan Wang! > 1. add a class 'StateChangelogConfig' contains setPeriodicMaterializationInterval(Duration), setMaterializationMaxFailuresAllowed(Integer), setStateChangelogStorage(StateChangelogStorage) That makes sense to me, except > setStateChangelogStorage(StateChangelogStorage) > 2. make StateChangelogStorage extends java.io.Serializable. The intention is that StateChangelogStorage could be created on client or JM and then passed to TM, right? This doesn't seem a good idea to me because StateChangelogStorage is a factory of readers and writers; and what we need is only a configuration. Can we set the class name of StateChangelogStorage instead? > 3. add field 'stateChangelogConfig' and method getStateChangelogConfig() to StreamExecutionEnvironment I think this is another aspect - having more changelog configuration via java API (rather than enabled/disabled). While it would be valuable, it probably should be a separate ticket becase it would require updating Scala and Python APIs, as well as documentation it makes sense to consider how further details of a specific implementation should be configured, e.g. FS path > 4. save StateChangelogConfig to taskConfigration I assume you mean saving it in TaskInformation.taskConfiguration Configuration object. If we don't add any new config parameters (and only keep "enabled"), as I proposed in (3), this is already done. Note, that reading this config from within the TaskExecutor (where the storage is created) overlaps with https://issues.apache.org/jira/browse/FLINK-28286 . > 5. StateChangelogStorageLoader.load(Configuration,TaskManagerJobMetricGroup) add parameter '@Nullable StateChangelogStorage fromApplication' > 6. extract stateChangelogStorageFromApplication from taskConfiguration and pass it to StateChangelogStorageLoader.load() If we don't pass StateChangelogStorage around (2), this is not needed. To summarize, I'd limit the scope of this task to just reading "changelog.enabled" from job configuration instead of cluster; without changing public API or reading more job configuration. The last two things can be implemented incrementally. WDYT? Add Comment   This message was sent by Atlassian Jira (v8.20.10#820010-sha1:ace47f9)