rkhachatryan commented on a change in pull request #15200:
URL: https://github.com/apache/flink/pull/15200#discussion_r640964347
##########
File path:
flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogListState.java
##########
@@ -39,53 +42,79 @@
extends AbstractChangelogState<K, N, List<V>, InternalListState<K, N,
V>>
implements InternalListState<K, N, V> {
- ChangelogListState(InternalListState<K, N, V> delegatedState) {
- super(delegatedState);
+ ChangelogListState(
+ InternalListState<K, N, V> delegatedState,
+ KvStateChangeLogger<List<V>, N> changeLogger) {
+ super(delegatedState, changeLogger);
}
@Override
public void update(List<V> values) throws Exception {
+ changeLogger.stateUpdated(values, getCurrentNamespace());
delegatedState.update(values);
}
@Override
public void addAll(List<V> values) throws Exception {
+ changeLogger.stateAdded(values, getCurrentNamespace());
delegatedState.addAll(values);
}
@Override
public void updateInternal(List<V> valueToStore) throws Exception {
+ changeLogger.stateUpdated(valueToStore, getCurrentNamespace());
delegatedState.updateInternal(valueToStore);
}
@Override
public void add(V value) throws Exception {
+ if (getValueSerializer() instanceof ListSerializer) {
Review comment:
Yes, it should be handled during the deserialization.
According to the contract, `null` shouldn't change the state, but in the
code I see that:
1. list states actually don't allow nulls
2. reducing states use null to clear state.
So I think it's safer to pass `nulls` further to delegated state here and
log the change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
