pvillard31 commented on code in PR #11837:
URL: https://github.com/apache/nifi/pull/11837#discussion_r4392555602
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/StateMapSerDe.java:
##########
@@ -28,14 +28,33 @@
import java.util.Map;
import java.util.Optional;
+/**
+ * Serializes component state for the write-ahead log. Full records, which are
used for snapshots, always hold the whole state map.
+ * Starting with version 1, an edit in the journal holds either the whole
state map or only the patches. An edit that holds
+ * only the patches is applied on top of the component's previous record when
the journal is recovered. Version 0 edits always
+ * hold the whole state map, in the same layout as full records.
+ */
public class StateMapSerDe implements SerDe<StateMapUpdate> {
private static final long EMPTY_VERSION = -1;
- private static final int VERSION = 0;
+ private static final int PATCHES_VERSION = 1;
Review Comment:
Can we add a note to the [Migration
Guidance](https://cwiki.apache.org/confluence/spaces/NIFI/pages/57905503/Migration+Guidance)
that older NiFi versions cannot read state storage written with serializer
version 1?
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java:
##########
@@ -2597,12 +2682,14 @@ public void testRollbackAfterCheckpointStoresState()
throws IOException {
session.checkpoint();
session.setState(Collections.singletonMap("abc", "2"), Scope.LOCAL);
+ session.patchState(Map.of("def", "2"), Scope.LOCAL);
Review Comment:
Can we extend this test to checkpoint a patch, apply another patch, then
roll back and commit, verifying that only the checkpointed patch is stored?
##########
nifi-mock/src/main/java/org/apache/nifi/util/MockProcessSession.java:
##########
@@ -1405,21 +1418,64 @@ public ProvenanceReporter getProvenanceReporter() {
@Override
public void setState(final Map<String, String> state, final Scope scope)
throws IOException {
stateManager.setState(state, scope);
+ pendingStatePatches.remove(scope);
}
@Override
public StateMap getState(final Scope scope) throws IOException {
- return stateManager.getState(scope);
+ final StateMap storedState = stateManager.getState(scope);
+ final Map<String, String> patches = pendingStatePatches.get(scope);
+ if (patches == null) {
+ return storedState;
+ }
+
+ final long version =
storedState.getStateVersion().map(Long::parseLong).orElse(-1L);
Review Comment:
Should pending patches advance the mock session's state version as they do
in `StandardProcessSession`, so a stale state handle after a same-value patch
is rejected consistently?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
