markap14 commented on code in PR #11837:
URL: https://github.com/apache/nifi/pull/11837#discussion_r4401948462
##########
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:
[Claude Opus 5.5] Good catch. `MockProcessSession` now counts the patches it
is holding for each scope, and `getState` returns a version equal to the stored
version plus that count, the same way `StandardProcessSession` gives the
session's state a new version on each patch. A `StateMap` obtained before a
patch therefore no longer matches in `replaceState`, even if the patch wrote
the same values.
`TestMockProcessSession.RegardingState.canSetLocalStateWhenDeclaredStateful`
now checks this case.
##########
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:
[Claude Opus 5.5] Done. `testRollbackAfterCheckpointStoresState` now patches
`abc`, checkpoints, patches `def`, rolls back, and then commits. It verifies
that `patchState` is called on the state manager with only `{abc=1}`, and that
the stored state is `{abc=1}`, so the patch made after the checkpoint is
discarded and the checkpointed patch is kept.
##########
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:
[Claude Opus 5.5] Agreed that this needs a migration note. I don't have edit
access to the wiki, so here is suggested text for the migration guide. Could
you or someone with access add it?
> Local component state written by this version uses a newer write-ahead log
format (serializer version 1). Older versions of NiFi cannot read it and fail
to start with an error saying the journal was written by a newer serializer
version. To downgrade, restore a backup of the `state/local` directory taken
before the upgrade, or remove that directory, which discards all local
component state.
##########
nifi-mock/src/main/java/org/apache/nifi/util/MockProcessSession.java:
##########
@@ -1405,21 +1418,70 @@ public ProvenanceReporter getProvenanceReporter() {
@Override
public void setState(final Map<String, String> state, final Scope scope)
throws IOException {
stateManager.setState(state, scope);
+ pendingStatePatches.remove(scope);
}
+ /**
+ * Returns the stored state with any patches that have not yet been
committed applied. Each patch gives the returned state a new version,
+ * as the framework does, so that a state obtained before a patch no
longer matches in {@link #replaceState(StateMap, Map, Scope)}.
+ */
@Override
public StateMap getState(final Scope scope) throws IOException {
- return stateManager.getState(scope);
+ final StateMap storedState = stateManager.getState(scope);
+ final PendingStatePatches patches = pendingStatePatches.get(scope);
+ if (patches == null) {
+ return storedState;
+ }
+
+ final long storedVersion =
storedState.getStateVersion().map(Long::parseLong).orElse(-1L);
+ return new MockStateMap(storedState.toMap(patches.values),
storedVersion + patches.count);
}
+ /**
+ * Replaces the state if the given old value matches the session's view of
the state. When the session holds patches that have not yet
+ * been committed, the old value must match the stored state with those
patches applied, and a successful replacement discards the patches.
+ */
@Override
public boolean replaceState(final StateMap oldValue, final Map<String,
String> newValue, final Scope scope) throws IOException {
- return stateManager.replace(oldValue, newValue, scope);
+ if (!pendingStatePatches.containsKey(scope)) {
+ return stateManager.replace(oldValue, newValue, scope);
+ }
+
+ final StateMap currentState = getState(scope);
+ if (oldValue == null ||
!oldValue.getStateVersion().equals(currentState.getStateVersion()) ||
!oldValue.toMap().equals(currentState.toMap())) {
+ return false;
+ }
+
+ final boolean replaced =
stateManager.replace(stateManager.getState(scope), newValue, scope);
Review Comment:
[Claude Opus 5.5] You're right. `replaceState` compared against one read of
the stored state and then passed a second read to `stateManager.replace`, so an
update stored between the two reads would be overwritten. It now reads the
stored state once, applies the pending patches to that copy for the comparison,
and passes the same copy to `stateManager.replace`. If anything else updates
the state in between, the replacement fails. I extended
`canSetLocalStateWhenDeclaredStateful` with a state manager that stores an
update right after the session reads the state. That check failed before this
change and passes now.
--
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]
