(nifi) branch main updated: NIFI-16490 Return Cluster Members from the Cluster Coordinator during flow synchronization (#11834)

[exceptionfactory]

This is an automated email from the ASF dual-hosted git repository.

exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git


The following commit(s) were added to refs/heads/main by this push:
     new 8d4d520ad2a NIFI-16490 Return Cluster Members from the Cluster 
Coordinator during flow synchronization (#11834)
8d4d520ad2a is described below

commit 8d4d520ad2a6e9a70601319c3bb04977b0cccd59
Author: Pierre Villard <pierre.villard...@gmail.com>
AuthorDate: Thu Oct 22 23:40:54 2026 +0200

    NIFI-16490 Return Cluster Members from the Cluster Coordinator during flow 
synchronization (#11834)
    
    Signed-off-by: David Handermann <exceptionfact...@apache.org>
---
 .../java/org/apache/nifi/controller/FlowController.java   |  2 +-
 .../org/apache/nifi/controller/FlowControllerTest.java    | 15 +++++++++++++--
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 347729b6710..6f1c076bfd4 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -3054,7 +3054,7 @@ public class FlowController implements 
ReportingTaskProvider, FlowAnalysisRulePr
 
     @Override
     public Set<String> getClusterMembers() {
-        if (isClustered()) {
+        if (configuredForClustering) {
             return clusterCoordinator.getConnectionStatuses().stream().map(s 
-> s.getNodeIdentifier().getApiAddress()).collect(Collectors.toSet());
         } else {
             return Collections.emptySet();
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/FlowControllerTest.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/FlowControllerTest.java
index e0d9afe7abb..4627d6f5d0c 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/FlowControllerTest.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/FlowControllerTest.java
@@ -74,6 +74,7 @@ import java.util.stream.Stream;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.params.provider.Arguments.arguments;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
 class FlowControllerTest {
@@ -81,6 +82,8 @@ class FlowControllerTest {
     private static final String CLUSTERED_NODE = "clustered-node";
     private static final String STANDALONE_NODE = "standalone-node";
     private static final String LOCALHOST = "localhost";
+    private static final String JOINING_NODE = "joining-node";
+    private static final String JOINING_HOST = "joining-host";
 
     private final List<FlowController> flowControllers = new ArrayList<>();
     private final AtomicInteger controllerCounter = new AtomicInteger();
@@ -121,6 +124,7 @@ class FlowControllerTest {
         flowController.setConnectionStatus(new 
NodeConnectionStatus(createNodeIdentifier(STANDALONE_NODE), 
org.apache.nifi.cluster.coordination.node.NodeConnectionState.CONNECTED));
 
         assertEquals(NodeConnectionState.STANDALONE, 
flowController.getNodeConnectionState());
+        assertEquals(Set.of(), flowController.getClusterMembers());
     }
 
     @Test
@@ -148,10 +152,16 @@ class FlowControllerTest {
     @Test
     void testProvideNodeConnectionStateDuringFlowSynchronization() throws 
Exception {
         final FlowController flowController = createClusteredFlowController();
-        flowController.setConnectionStatus(new NodeConnectionStatus(
+        final NodeConnectionStatus localNodeStatus = new NodeConnectionStatus(
                 createNodeIdentifier(CLUSTERED_NODE),
                 
org.apache.nifi.cluster.coordination.node.NodeConnectionState.CONNECTING
-        ));
+        );
+        final NodeConnectionStatus joiningNodeStatus = new 
NodeConnectionStatus(
+                new NodeIdentifier(JOINING_NODE, JOINING_HOST, 8443, 
JOINING_HOST, 9090, JOINING_HOST, 10000, 10001, true),
+                
org.apache.nifi.cluster.coordination.node.NodeConnectionState.CONNECTING
+        );
+        
when(clusterCoordinator.getConnectionStatuses()).thenReturn(List.of(localNodeStatus,
 joiningNodeStatus));
+        flowController.setConnectionStatus(localNodeStatus);
 
         final ExecutorService lifecycleExecutor = 
Executors.newSingleThreadExecutor();
         final AtomicReference<Future<NodeConnectionState>> 
nodeStateFutureReference = new AtomicReference<>();
@@ -163,6 +173,7 @@ class FlowControllerTest {
 
                 try {
                     assertEquals(NodeConnectionState.CONNECTING, 
nodeStateFuture.get(5, TimeUnit.SECONDS));
+                    assertEquals(Set.of(LOCALHOST, JOINING_HOST), 
controller.getClusterMembers());
                 } catch (final InterruptedException e) {
                     Thread.currentThread().interrupt();
                     throw new AssertionFailedError("Interrupted while waiting 
for node connection state", e);