markap14 commented on code in PR #10986:
URL: https://github.com/apache/nifi/pull/10986#discussion_r3154687830
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java:
##########
@@ -2956,7 +2958,7 @@ public Set<String> getClusterMembers() {
@Override
public Optional<String> getCurrentNode() {
if (isClustered() && getNodeId() != null) {
- return Optional.of(getNodeId().getApiAddress());
+ return Optional.of("%s:%d".formatted(getNodeId().getApiAddress(),
getNodeId().getApiPort()));
Review Comment:
This is a risky change, as they break backward compatibility because it
changes what is returned by the `NodeTypeProvider`. In particular, this affects
`EmbeddedHazelcastCacheManager`.
After this change:
- `m + PORT_SEPARATOR + port` (line 141) becomes
`"address:nifiApiPort:hazelcastPort"` — Hazelcast will fail to bind / connect.
- `hazelcastMembers.contains(getCurrentNode().get())` (line 152) compares
user-configured hostnames against `"address:port"` and will always return
`false`, sending the local node down the client path instead of becoming a
member.
- The `customValidate` check at lines 224–236 compares user-configured
Hazelcast hosts against getClusterMembers() and will always report them as "not
part of the NiFi cluster".
GetAzureEventHub.getClientIdentifier() also folds the value into a client
identifier, which is now silently changed. That is less likely to break things
but is still a behavioral change.
The export/import feature only needs a stable, unique key for ordinal
calculation. Mixing port into a public `NodeTypeProvider` method that other
extensions already depend on is the wrong layer to fix it. One option:
Introduce a new framework-internal helper that returns
`Set<NodeIdentifier>`/sorted IDs and use it only from `StandardProcessGroup` /
`StandardNiFiServiceFacade`. Leave `getClusterMembers()` / `getCurrentNode()`
alone
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusterFlowDefinitionExportImportStateIT.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.tests.system.pg;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.flow.VersionedComponentState;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.registry.flow.RegisteredFlowSnapshot;
+import org.apache.nifi.tests.system.NiFiInstanceFactory;
+import org.apache.nifi.tests.system.NiFiSystemIT;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+import org.apache.nifi.web.api.dto.StateEntryDTO;
+import org.apache.nifi.web.api.entity.ComponentStateEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupImportEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupReplaceRequestEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ClusterFlowDefinitionExportImportStateIT extends NiFiSystemIT {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper()
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+
+ @Override
+ public NiFiInstanceFactory getInstanceFactory() {
+ return createTwoNodeInstanceFactory();
+ }
+
+ @Test
+ public void testClusterExportCapturesClusterState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("GenerateFlowFile", pg.getId());
+ getClientUtil().updateProcessorProperties(stateful,
Collections.singletonMap("State Scope", "CLUSTER"));
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.CLUSTER);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st13-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor proc =
findProcessorByType(snapshot.getFlowContents(), "GenerateFlowFile");
+ assertNotNull(proc);
+ assertNotNull(proc.getComponentState());
+ assertNotNull(proc.getComponentState().getClusterState());
+ assertNotNull(proc.getComponentState().getClusterState().get("count"));
+ }
+
+ @Test
+ public void testClusterExportCapturesLocalStateFromBothNodes() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("GenerateFlowFile", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st14-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor proc =
findProcessorByType(snapshot.getFlowContents(), "GenerateFlowFile");
+ assertNotNull(proc);
+ assertNotNull(proc.getComponentState());
+ assertNotNull(proc.getComponentState().getLocalNodeStates());
+ assertEquals(2, proc.getComponentState().getLocalNodeStates().size(),
+ "Should have local state from both nodes");
+ assertNotNull(proc.getComponentState().getLocalNodeStates().get(0));
+ assertNotNull(proc.getComponentState().getLocalNodeStates().get(1));
+ }
+
+ @Test
+ public void testClusterExportCapturesBothScopes() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.CLUSTER);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st15-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor proc =
findProcessorByType(snapshot.getFlowContents(), "StatefulCountProcessor");
+ assertNotNull(proc);
+
+ final VersionedComponentState state = proc.getComponentState();
+ assertNotNull(state);
+ assertNotNull(state.getClusterState(), "Cluster state should be
present");
+ assertNotNull(state.getClusterState().get("count"), "Cluster state
should contain count");
+ assertNotNull(state.getLocalNodeStates(), "Local node states should be
present");
+ assertEquals(2, state.getLocalNodeStates().size());
+
assertNotNull(state.getLocalNodeStates().get(0).getState().get("count"));
+
assertNotNull(state.getLocalNodeStates().get(1).getState().get("count"));
+ }
+
+ @Test
+ public void testClusterRoundTripSameTopologyBothScopes() throws
NiFiClientException, IOException, InterruptedException {
Review Comment:
This test only re-reads cluster state after import. It does not query each
node's local state to confirm Node 0's exported state ends up on Node 0 (and
not on Node 1). Add an assertion that uses getProcessorState(... Scope.LOCAL)
per node.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -4080,6 +4096,104 @@ private Map<String, String> getPropertyValues(final
ComponentNode componentNode)
return propertyValues;
}
+ private void validateLocalStateTopology(final VersionedProcessGroup
proposed) {
+ final int connectedNodeCount = context.getConnectedNodeCount();
Review Comment:
`connectedNodeCount` is a misleading name
This made says it compares "S source nodes" against "D connected nodes" (per
the PR description), but the `connectedNodeCount` it reads is actually
`Math.max(sortedMembers.size(), 1)` — the size of `getClusterMembers()`, which
is all members regardless of connection state. So a 5-node cluster with only 2
connected nodes will accept a 4-node export and silently leave two of the four
state slices unrestored. We should either rename the field/builder method to
`clusterMemberCount` and document, or compute it from connected nodes only.
--
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]
